scripts.claude_turn_capture

Reference › scripts › claude_turn_capture

The Claude Code conversation producer for SPEC-021 (multi-platform producers). Claude Code runs it as a hook, and it turns the hook events into conversation messages placed on a Redis queue. A consumer elsewhere stores them; see run_conversation_consumer. tests/scripts/test_claude_turn_capture.py imports it, and it uses the @traces decorator from traces.

Invocation: main takes no options. It reads one hook event as JSON from standard input and the environment, and always exits 0 quickly, so a capture problem never blocks the Claude session. Failures are logged to a capture log (_capture_log_path, _log_drop) rather than raised. main is a thin wrapper around _run, the only function tests drive directly.

Two events are handled. On a user prompt submission, _handle_user_prompt_submit builds a user message. On a stop event, _handle_stop reads the transcript file (_read_assistant_entries), takes only the assistant entries since the last captured one (_delta_since), extracts their text and tool calls (_extract_assistant_text, _extract_tool_calls) and enqueues them. _resolve_project works out which project the session belongs to from the environment, the working directory or the transcript path. _get_or_create_session_seq keeps a per-session sequence number in Redis, _build_redis_client and _conversation_redis_config read the connection from the environment, and _enqueue pushes the message onto the queue named by _queue_key.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

API

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/scripts/claude_turn_capture.py

#!/usr/bin/env python3
"""Claude Code conversation producer (SPEC-021 multi-platform producers).

Design authority: ``plans/spec-021-multiplatform-producers.md`` § "1. Claude
Code producer", generalizing the settled ontology contract
(``plans/contracts/spec-021-ontology.contract.json``) to a second platform.
The landed consumer this producer's messages must satisfy is
``orchestrata_ultimate.knowledge_graph.conversation_consumer`` — message v2
exactly (``schema:2``, ``project``, ``conversation_id``, ``session_seq``,
``turn_id``, ``turn_seq``, ``role``, ``content``, ``timestamp``, ``platform``,
``tool_call_ids?``), pushed onto the Redis LIST ``orchestrata:conversation:
queue`` (bounded, LPUSH + LTRIM). The consumer never whitelists ``platform``,
so a new producer needs no consumer-side change.

Invoked by two ADDITIVE Claude Code hook registrations (``.claude/settings.
json``):

- ``UserPromptSubmit`` — captures the user's prompt (stdin field ``prompt``)
  as one ``role=user`` turn.
- ``Stop`` — parses the session transcript JSONL (stdin field
  ``transcript_path``) and captures every assistant text turn appended since
  the last-captured message uuid (tracked in Redis, per conversation).

Hard contract for a Claude Code hook script (never negotiable):
    * FAIL-OPEN: every error path (Redis unreachable, malformed transcript,
      malformed stdin, missing fields) logs to stderr and returns — the
      process ALWAYS exits 0, fast (well under 1s in the normal path; the
      Redis client below uses a short socket-connect/read timeout so an
      unreachable Redis fails fast rather than hanging the host session).
    * NEVER writes to stdout — some hook events feed stdout into the model's
      context; all diagnostics go to stderr via ``logging``.
    * Unknown/missing stdin fields are a silent no-op, never an error.

Field-robustness choices (stdin JSON has no formal schema across Claude Code
versions, so every field is read defensively via ``.get`` and type-checked
before use):
    * ``project`` resolution prefers the hook's ``cwd`` field (given
      directly, no assumption about Claude's internal transcript storage
      layout) and falls back to the transcript path's parent directory name
      only when ``cwd`` is absent — see :func:`_resolve_project`.
    * ``turn_id`` is the transcript message ``uuid`` for assistant turns
      (deterministic -> a re-fired ``Stop`` hook re-enqueues the same id,
      which the consumer MERGEs idempotently) and a fresh UUIDv4 for user
      prompts (``UserPromptSubmit`` carries no uuid).

Test-only seam: ``ORCHESTRATA_CONVERSATION_QUEUE_KEY`` overrides the queue
key (default: the contract's ``orchestrata:conversation:queue``, unchanged
in production). This is a deliberate, minor deviation from the design doc
(which fixes the queue key name) purely for test isolation, mirroring how
``RedisQueueSource`` itself already accepts a ``queue_key`` override for the
same reason on the consumer side — the shared production queue can't be
scoped by ``test_id`` the way project-qualified keys can.
"""

from __future__ import annotations

import json
import logging
import os
import sys
import uuid
from datetime import UTC, datetime
from pathlib import Path
from typing import TYPE_CHECKING, Any

from orchestrata_ultimate.core.base.traces import traces

if TYPE_CHECKING:
    import redis

_logger = logging.getLogger(__name__)

#: Production queue key (REQ-021-060 / FR-011 — name unchanged). Overridable
#: only for test isolation, see module docstring.
_DEFAULT_QUEUE_KEY = "orchestrata:conversation:queue"

#: Message schema version this producer emits — the consumer's only
#: accepted version (``conversation_consumer._SCHEMA_VERSION``).
_SCHEMA_VERSION = 2

#: content re-truncation on the producer side (the consumer re-enforces
#: this too, but the contract requires the producer to truncate as well).
_MAX_CONTENT_LEN = 2000

#: Platform label carried on every message this producer emits.
_PLATFORM = "claude-code"

#: conversation_id / key slug for this platform (contract's generalized
#: ``{platform_slug}-{session_seq}`` scheme -> ``claude-{seq}``).
_PLATFORM_SLUG = "claude"

#: Short timeouts so an unreachable Redis fails fast (hook budget: <1s).
_SOCKET_CONNECT_TIMEOUT = 0.3
_SOCKET_TIMEOUT = 0.5


@traces(spec="021", story="US-021-007", reqs=["REQ-021-060"])
def _queue_key(env: dict[str, str]) -> str:
    return env.get("ORCHESTRATA_CONVERSATION_QUEUE_KEY", _DEFAULT_QUEUE_KEY)


@traces(spec="021", story="US-021-007", reqs=["REQ-021-066"])
def _now_iso() -> str:
    return datetime.now(UTC).isoformat()


#: The conversation queue's OWN Redis endpoint keys (same names as
#: ``knowledge_graph.conversation_consumer.CONVERSATION_REDIS_*_KEY``). NO
#: fallback to the ambient ``REDIS_*`` keys: a hook launched where ``REDIS_*``
#: names a different Redis than the daemon drains writes into a void.
_REDIS_KEYS = (
    "CONVERSATION_REDIS_HOST",
    "CONVERSATION_REDIS_PORT",
    "CONVERSATION_REDIS_DB",
)


@traces(spec="021", story="US-021-007", reqs=["REQ-021-070"])
def _conversation_redis_config(env: dict[str, str]) -> tuple[str, int, int]:
    """``(host, port, db)`` from ``CONVERSATION_REDIS_*``; ``ValueError`` when
    any is unset/blank or port/db is not an integer."""
    values = {k: (env.get(k) or "").strip() for k in _REDIS_KEYS}
    missing = [k for k, v in values.items() if not v]
    if missing:
        raise ValueError(f"{' and '.join(missing)} not set")
    return (
        values[_REDIS_KEYS[0]],
        int(values[_REDIS_KEYS[1]]),
        int(values[_REDIS_KEYS[2]]),
    )


@traces(spec="021", story="US-021-007", reqs=["REQ-021-070"])
def _capture_log_path(env: dict[str, str]) -> Path:
    """``$XDG_STATE_HOME/orchestrata/conversation_capture.log`` (default
    ``~/.local/state/...``)."""
    state = env.get("XDG_STATE_HOME", "").strip()
    base = (
        Path(state) if state else Path(env.get("HOME") or Path.home()) / ".local/state"
    )
    return base / "orchestrata" / "conversation_capture.log"


@traces(spec="021", story="US-021-007", reqs=["REQ-021-070"])
def _log_drop(reason: str, env: dict[str, str], project: str | None) -> None:
    """Append ONE line (timestamp, project, reason) to the capture log so a
    dropped turn is observable. Never raises (fail-open)."""
    try:
        path = _capture_log_path(env)
        path.parent.mkdir(parents=True, exist_ok=True)
        flat = " ".join(reason.split())
        line = f"{_now_iso()} {project or '-'} {flat}\n"
        with path.open("a", encoding="utf-8") as fh:
            fh.write(line)
    except Exception as exc:
        _logger.warning("capture log write failed: %s", exc)


@traces(spec="021", story="US-021-007", reqs=["REQ-021-070"])
def _build_redis_client(env: dict[str, str]) -> redis.Redis:
    """Construct a sync Redis client with a short fail-fast timeout.

    Deliberately does not go through ``core.redis_factory.create_sync_redis``:
    that factory's ``BlockingConnectionPool`` timeout governs connection
    *checkout*, not the socket connect/read timeouts this single-shot,
    sub-1s hook process needs to fail fast against an unreachable host.
    ``retry=Retry(NoBackoff(), 0)`` disables redis-py's default multi-attempt
    backoff retry on connection errors (measured ~10s against a dead port
    otherwise) — mirrors the established fix in
    ``tests/governance/test_error_paths.py``'s ``broken_redis`` fixture.
    """
    import redis as redis_lib
    from redis.backoff import NoBackoff
    from redis.retry import Retry

    host, port, db = _conversation_redis_config(env)
    return redis_lib.Redis(
        host=host,
        port=port,
        db=db,
        decode_responses=True,
        socket_connect_timeout=_SOCKET_CONNECT_TIMEOUT,
        socket_timeout=_SOCKET_TIMEOUT,
        retry=Retry(NoBackoff(), 0),
    )


@traces(spec="021", story="US-021-007", reqs=["REQ-021-065"])
def _resolve_project(
    env: dict[str, str], cwd: str | None, transcript_path: str | None
) -> str | None:
    """Resolve the KG ``Project.slug`` for this hook invocation.

    Precedence: ``ORCHESTRATA_PROJECT_SLUG`` env override, else the hook's
    ``cwd`` field's basename (the most robust choice — given directly by
    Claude Code, no assumption about internal transcript storage layout),
    else the transcript path's parent directory name (Claude Code nests
    transcripts under a per-project directory) as a last resort. Returns
    ``None`` when nothing resolves — never guess a project id (a wrong one
    poisons the multi-project graph).
    """
    slug = env.get("ORCHESTRATA_PROJECT_SLUG", "").strip()
    if slug:
        return slug
    if cwd:
        # Canonical Project.slug is pyproject [project].name, NOT the
        # checkout directory name (this repo: 'orchestrata-ultimate' in a
        # dir named 'orchestrata-ultimate-architecture'). The dir-basename
        # first cut split the same repo across two Project identities in
        # the live KG — discovered in the W3 smoke against the opencode
        # producer, which already resolves via pyproject.
        pyproject = Path(cwd) / "pyproject.toml"
        if pyproject.is_file():
            try:
                import tomllib

                with pyproject.open("rb") as fh:
                    name = tomllib.load(fh).get("project", {}).get("name", "")
                if isinstance(name, str) and name.strip():
                    return name.strip()
            except Exception:
                pass  # unreadable/invalid pyproject -> fall through
        name = Path(cwd).name
        if name:
            return name
    if transcript_path:
        name = Path(transcript_path).parent.name
        if name:
            return name
    return None


@traces(spec="021", story="US-021-007", reqs=["REQ-021-065"])
def _get_or_create_session_seq(
    client: redis.Redis, project: str, claude_session_id: str
) -> int:
    """Map Claude's ``session_id`` uuid -> a stable ``session_seq`` int.

    HSETNX onto ``orchestrata:conversation:session_map:{project}:claude``,
    backed by an INCR allocator on ``orchestrata:conversation:session_seq:
    {project}:claude`` (both key names per the design doc's shared rules) —
    every hook invocation within one Claude session lands on the same
    ``session_seq``, including across resume/compaction (same uuid).
    """
    map_key = f"orchestrata:conversation:session_map:{project}:{_PLATFORM_SLUG}"
    existing = client.hget(map_key, claude_session_id)
    if existing is not None:
        return int(existing)  # pyright: ignore[reportArgumentType]

    alloc_key = f"orchestrata:conversation:session_seq:{project}:{_PLATFORM_SLUG}"
    candidate = int(client.incr(alloc_key))  # pyright: ignore[reportArgumentType]
    claimed = client.hsetnx(map_key, claude_session_id, str(candidate))
    if claimed:
        return candidate
    # Lost the race to another concurrent invocation — use its value.
    winner = client.hget(map_key, claude_session_id)
    if winner is not None:
        return int(winner)  # pyright: ignore[reportArgumentType]
    return candidate


@traces(spec="021", story="US-021-007", reqs=["REQ-021-060"])
def _enqueue(client: redis.Redis, env: dict[str, str], message: dict[str, Any]) -> None:
    """LPUSH one message, then LTRIM to the bounded queue max (drop-oldest)."""
    queue_key = _queue_key(env)
    max_len = int(env.get("ORCHESTRATA_CONVERSATION_QUEUE_MAX", "10000"))
    client.lpush(queue_key, json.dumps(message))
    client.ltrim(queue_key, 0, max_len - 1)


@traces(spec="021", story="US-021-007", reqs=["REQ-021-065", "REQ-021-066"])
def _handle_user_prompt_submit(
    event: dict[str, Any], env: dict[str, str], client: redis.Redis
) -> None:
    """Capture the user's submitted prompt as one ``role=user`` turn."""
    prompt = event.get("prompt")
    session_id = event.get("session_id")
    if not isinstance(prompt, str) or not prompt.strip():
        return
    if not isinstance(session_id, str) or not session_id.strip():
        return

    project = _resolve_project(env, event.get("cwd"), event.get("transcript_path"))
    if not project:
        return

    session_seq = _get_or_create_session_seq(client, project, session_id)
    conversation_id = f"{_PLATFORM_SLUG}-{session_seq}"
    turn_seq = int(
        client.incr(  # pyright: ignore[reportArgumentType]
            f"orchestrata:conversation:seq:{project}:{conversation_id}"
        )
    )
    message = {
        "schema": _SCHEMA_VERSION,
        "project": project,
        "conversation_id": conversation_id,
        "session_seq": session_seq,
        "turn_id": str(uuid.uuid4()),
        "turn_seq": turn_seq,
        "role": "user",
        "content": prompt.strip()[:_MAX_CONTENT_LEN],
        "timestamp": _now_iso(),
        "platform": _PLATFORM,
        "tool_call_ids": [],
    }
    _enqueue(client, env, message)


def _extract_tool_calls(entry: dict[str, Any]) -> list[tuple[str, str]]:
    """``(tool_use_id, tool_name)`` for every tool call in one transcript entry.

    WHY THIS EXISTS. The message contract has carried a ``tool_call_ids`` field
    since schema v2, and the consumer has always written
    ``(:ConversationTurn)-[:TRIGGERED]->(:ToolCall)`` from it — but this
    producer sent ``[]`` on every single message, so the edge was never once
    created. Measured 2026-08-08 against the live conversation graph: 14560
    assistant turns, 0 carrying any per-turn tool signal.

    What that cost: "did work actually follow this turn?" is the only answer to
    "should this turn have started a pipeline?" that is an observation rather
    than an opinion, and it was unavailable — forcing the speech-act corpus
    (tests/orchestrata/data/intent_corpus_speech_acts.json) to rely on a single
    labeller's judgement, with the taxonomy ambiguity that produced.
    """
    message = entry.get("message")
    if not isinstance(message, dict):
        return []
    content = message.get("content")
    if not isinstance(content, list):
        return []
    calls: list[tuple[str, str]] = []
    for block in content:
        if not isinstance(block, dict) or block.get("type") != "tool_use":
            continue
        call_id = block.get("id")
        name = block.get("name")
        if isinstance(call_id, str) and call_id:
            calls.append((call_id, name if isinstance(name, str) else ""))
    return calls


@traces(spec="021", story="US-021-007", reqs=["REQ-021-066"])
def _extract_assistant_text(entry: dict[str, Any]) -> str:
    """Assistant text blocks only (``tool_use`` blocks are skipped).

    Tool calls are read separately by :func:`_extract_tool_calls` — a turn's
    prose and the tools it ran are different properties of the same turn.
    """
    message = entry.get("message")
    if not isinstance(message, dict):
        return ""
    content = message.get("content")
    if isinstance(content, str):
        return content.strip()
    if not isinstance(content, list):
        return ""
    parts: list[str] = []
    for block in content:
        if isinstance(block, dict) and block.get("type") == "text":
            text = block.get("text")
            if isinstance(text, str):
                parts.append(text)
    return "\n".join(parts).strip()


@traces(spec="021", story="US-021-007", reqs=["REQ-021-063"])
def _read_assistant_entries(transcript_path: str) -> list[dict[str, Any]]:
    """Parse the transcript JSONL, skipping malformed lines and non-assistant
    entries (mirrors the consumer's "parse, skip invalid" discipline —
    REQ-021-063 — on the producer side of the same pipeline)."""
    entries: list[dict[str, Any]] = []
    try:
        with Path(transcript_path).open("r", encoding="utf-8") as fh:
            for line in fh:
                line = line.strip()
                if not line:
                    continue
                try:
                    obj = json.loads(line)
                except json.JSONDecodeError, ValueError:
                    continue
                if not isinstance(obj, dict) or obj.get("type") != "assistant":
                    continue
                turn_uuid = obj.get("uuid")
                if isinstance(turn_uuid, str) and turn_uuid:
                    entries.append(obj)
    except OSError as exc:
        _logger.warning("cannot read transcript %s: %s", transcript_path, exc)
    return entries


def _delta_since(
    entries: list[dict[str, Any]], last_uuid: str | None
) -> list[dict[str, Any]]:
    """Entries appended after ``last_uuid`` (or all of them if unset/not found —
    a compacted/rewritten transcript can drop the marker; reprocessing is
    safe because the consumer MERGEs on the deterministic ``turn_id``)."""
    if not last_uuid:
        return entries
    for idx, entry in enumerate(entries):
        if entry.get("uuid") == last_uuid:
            return entries[idx + 1 :]
    return entries


@traces(spec="021", story="US-021-007", reqs=["REQ-021-065", "REQ-021-066"])
def _handle_stop(
    event: dict[str, Any], env: dict[str, str], client: redis.Redis
) -> None:
    """Capture assistant turns appended to the transcript since the last Stop."""
    session_id = event.get("session_id")
    transcript_path = event.get("transcript_path")
    if not isinstance(session_id, str) or not session_id.strip():
        return
    if not isinstance(transcript_path, str) or not transcript_path.strip():
        return
    if not Path(transcript_path).is_file():
        return

    project = _resolve_project(env, event.get("cwd"), transcript_path)
    if not project:
        return

    session_seq = _get_or_create_session_seq(client, project, session_id)
    conversation_id = f"{_PLATFORM_SLUG}-{session_seq}"
    last_uuid_key = f"orchestrata:conversation:last_uuid:{project}:{conversation_id}"
    last_uuid = client.get(last_uuid_key)

    entries = _read_assistant_entries(transcript_path)
    delta = _delta_since(entries, last_uuid)  # pyright: ignore[reportArgumentType]

    # Tool calls from entries that carry NO prose (a pure tool_use message)
    # cannot be emitted on their own — the consumer requires non-blank
    # `content` — so they are held and attached to the next entry that does
    # have prose. The carry is kept in Redis, not just in this loop, because a
    # Stop cycle can end on a tool-only entry and the tools it ran still
    # belong to the assistant span the next cycle continues.
    pending_key = f"orchestrata:conversation:pending_tools:{project}:{conversation_id}"
    pending: list[tuple[str, str]] = []
    raw_pending = client.get(pending_key)
    if isinstance(raw_pending, (str, bytes)):
        try:
            pending = [
                (item[0], item[1]) for item in json.loads(raw_pending) if len(item) == 2
            ]
        except ValueError, TypeError, IndexError:
            pending = []

    for entry in delta:
        text = _extract_assistant_text(entry)
        turn_uuid = entry["uuid"]
        pending.extend(_extract_tool_calls(entry))
        if not text:
            # No prose to hang a turn on, but its tool calls are now held in
            # `pending` rather than dropped. Advance so it is not re-scanned.
            client.set(pending_key, json.dumps(pending))
            client.set(last_uuid_key, turn_uuid)
            continue
        turn_seq = int(
            client.incr(  # pyright: ignore[reportArgumentType]
                f"orchestrata:conversation:seq:{project}:{conversation_id}"
            )
        )
        message = {
            "schema": _SCHEMA_VERSION,
            "project": project,
            "conversation_id": conversation_id,
            "session_seq": session_seq,
            "turn_id": turn_uuid,
            "turn_seq": turn_seq,
            "role": "assistant",
            "content": text[:_MAX_CONTENT_LEN],
            "timestamp": entry.get("timestamp") or _now_iso(),
            "platform": _PLATFORM,
            # Positionally parallel: the consumer rejects a length mismatch
            # rather than mislabelling calls after a gap.
            "tool_call_ids": [call_id for call_id, _ in pending],
            "tool_names": [name for _, name in pending],
        }
        _enqueue(client, env, message)
        # Enqueue first, THEN clear — a crash between the two re-sends tools
        # on the next cycle, which the consumer MERGEs idempotently. Clearing
        # first would lose them outright.
        pending = []
        client.delete(pending_key)
        client.set(last_uuid_key, turn_uuid)


@traces(spec="021", story="US-021-007", reqs=["REQ-021-070"])
def _run(raw: str, env: dict[str, str]) -> None:
    """Core dispatch logic: parse one hook event and capture it.

    Pure(-ish) w.r.t. its inputs — ``raw`` (stdin text) and ``env`` are both
    plain parameters, not read from ``sys.stdin``/``os.environ`` directly —
    so this is the injectable seam tests drive in-process. Never raises
    (fail-open, hard): every error path returns/logs to stderr silently.
    """
    event: Any = None
    try:
        if env.get("ORCHESTRATA_CONVERSATION_CAPTURE", "1").strip() == "0":
            return
        if not raw.strip():
            return
        try:
            event = json.loads(raw)
        except json.JSONDecodeError, ValueError:
            return
        if not isinstance(event, dict):
            return

        hook_event_name = event.get("hook_event_name")
        if hook_event_name not in ("UserPromptSubmit", "Stop"):
            return

        client = _build_redis_client(env)
        if hook_event_name == "UserPromptSubmit":
            _handle_user_prompt_submit(event, env, client)
        else:
            _handle_stop(event, env, client)
    except Exception as exc:  # fail-open, hard: never blocks the host session
        _logger.warning("claude_turn_capture failed (fail-open): %s", exc)
        # A drop must be observable: one line to the per-user capture log.
        project = None
        if isinstance(event, dict):
            project = _resolve_project(
                env, event.get("cwd"), event.get("transcript_path")
            )
        _log_drop(f"{type(exc).__name__}: {exc}", env, project)


def main() -> None:
    """Read one hook event from real stdin/env and always exit 0 fast.

    Thin wrapper around :func:`_run` — the only place this module reads
    ``sys.stdin``/``os.environ`` directly (the injection seam tests use
    instead is ``_run`` itself).
    """
    logging.basicConfig(level=logging.WARNING, stream=sys.stderr)
    _run(sys.stdin.read(), dict(os.environ))


if __name__ == "__main__":
    try:
        main()
    except BaseException as exc:  # noqa: BLE001 - last-resort exit-0 guard
        _logger.warning("claude_turn_capture crashed (fail-open): %s", exc)
    sys.exit(0)