orchestrata_ultimate.knowledge_graph.conversation_consumer

Reference › orchestrata_ultimate › knowledge_graph › conversation_consumer

The consumer for conversation capture (SPEC-021, US-021-007): it drains a Redis queue of conversation turns into a dedicated conversation graph. That graph has its own endpoint and is neither the operational orchkg database nor the shared memgraph semantic graph; resolve_conversation_endpoint returns its URI and database from CONVERSATION_KG_URI and CONVERSATION_KG_DATABASE, and resolve_conversation_redis_endpoint returns the queue's Redis from CONVERSATION_REDIS_HOST, CONVERSATION_REDIS_PORT and CONVERSATION_REDIS_DB. Both raise RuntimeError when a key is unset or blank, and there is no fallback to the ambient REDIS_* or profile-driven KG keys (REQ-021-069). create_conversation_redis and create_conversation_graph_repository build the clients from those endpoints.

The design follows an ontology-before-import ruling and avoids three defects of the original plugin. ConversationTurnMessage.parse validates schema version 2, which requires a project; anything without it, including every older message, raises ConversationMessageError and is quarantined, never guessed. RedisQueueSource implements the QueueSource protocol with an atomic LMOVE claim from the queue to a processing list, ack after a successful write, requeue_processing for crash recovery, quarantine and depth. TurnGraphMapper.to_statements is a pure mapper from a message to Cypher statements that MERGE Conversation and ConversationTurn nodes on project-qualified, producer-deterministic ids from canonical_schema; turn order is the turn_seq property, so redelivery is a safe no-op. ConversationQueueConsumer.flush requeues orphaned in-flight messages, drains the queue and returns a FlushReport (depth before, synced, quarantined, failed). It never raises (REQ-021-070): a failed write leaves the message in the processing list, counted as failed, and the next flush redelivers it. The kernel CLI and the conversation tools import it.

At a glance

Class diagram

Dependency diagram

API

Classes

ConversationMessageError

Bases: ValueError · Decorators: traces(spec='021', story='US-021-007', reqs=['REQ-021-063']) · Line 260

A queued message failed :meth:ConversationTurnMessage.parse validation.

ConversationTurnMessage (dataclass)

Decorators: dataclass(frozen=True) · Line 275

One parsed, validated queue message (message schema v2).

QueueSource (protocol)

Bases: Protocol · Decorators: runtime_checkable · Line 381

Reliable-queue seam the consumer drives (LMOVE claim/ack/requeue).

RedisQueueSource

Line 414

Real Redis reliable-queue implementation of :class:QueueSource.

TurnGraphMapper

Line 494

Pure ConversationTurnMessage -> [(cypher, params), ...] mapper.

FlushReport (dataclass)

Decorators: traces(spec='021', story='US-021-007', reqs=['REQ-021-217']), dataclass(frozen=True) · Line 609

Result of one :meth:ConversationQueueConsumer.flush call.

ConversationQueueConsumer

Line 626

Drains the conversation queue into the graph — the SPEC-021 consumer.

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/knowledge_graph/conversation_consumer.py

"""Conversation-capture queue consumer (SPEC-021, US-021-007).

Ontology-before-import (user ruling: never dump raw nodes — see
``plans/contracts/spec-021-ontology.contract.json``, the settled design
authority for this module). The OpenCode plugin's original implementation
(``orchestrata/open-code/plugins/orchestrata.ts``) had three defects this
module deliberately does NOT reproduce (deviations D1/D2/D3 in the contract):

1. **No project field** — the v1 push shape carried no project identity,
   which is unmappable under the multi-project KG rule (100+ projects share
   one graph; a bare ``session_id`` like ``"opencode-227"`` collides across
   every one of them). Message v2 (:class:`ConversationTurnMessage`) REQUIRES
   ``project``; anything without it (including every v1 message) is
   quarantined, never guessed.
2. **Bare LPOP + CREATE** — a crash between pop and graph write silently lost
   a turn, and a re-consumed message minted a *new* node (flush-time
   enumerate index, not a stable id). This module claims via LMOVE
   (queue -> processing, atomic) and only ever MERGEs on a producer-
   deterministic id, so redelivery after a crash is a safe no-op.
3. **Pre-grouping by session_seq** — unnecessary once every message carries
   its own full conversation identity; each message maps independently and
   order-independently (grouping falls out of MERGE).

Identity: :func:`~orchestrata_ultimate.knowledge_graph.canonical_schema.conversation_entity_id`
and ``conversation_turn_entity_id`` project-qualify both new labels
(``Conversation``, ``ConversationTurn``) per the multi-project KG ruling —
see ``canonical_schema.py``'s Tier I section. Turn ORDER is the ``turn_seq``
property (a producer-side Redis INCR), not a linked list: list-pointer edges
are not MERGE-idempotent under out-of-order/duplicate redelivery, property
ordering is.

Never raises out of :meth:`ConversationQueueConsumer.flush` (REQ-021-070,
ERR-008): a graph-write failure leaves the claimed message in the
``processing`` list, counted as ``failed`` — the NEXT ``flush()`` call
requeues it (:meth:`QueueSource.requeue_processing`) and, being an
idempotent MERGE, safely redelivers it.
"""

from __future__ import annotations

import json
import logging
import os
from collections.abc import Mapping
from dataclasses import dataclass
from typing import TYPE_CHECKING, Any, Protocol, cast, runtime_checkable

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.tracing.tracer import Tracer
from orchestrata_ultimate.knowledge_graph import canonical_schema as cs

if TYPE_CHECKING:
    import redis

    from orchestrata_ultimate.core.graph.interfaces import SyncGraphRepository

_logger = logging.getLogger(__name__)

#: The conversation graph's OWN endpoint keys. Readers and writers must both
#: resolve through these and nothing else — see
#: :func:`create_conversation_graph_repository`.
CONVERSATION_KG_URI_KEY = "CONVERSATION_KG_URI"
CONVERSATION_KG_DATABASE_KEY = "CONVERSATION_KG_DATABASE"

#: The conversation pipeline's OWN Redis endpoint keys (the queue half of the
#: pipeline, as ``CONVERSATION_KG_*`` is the graph half). Producers and
#: consumers resolve the queue's Redis through these and NEVER through the
#: ambient ``REDIS_*`` keys — see :func:`resolve_conversation_redis_endpoint`.
CONVERSATION_REDIS_HOST_KEY = "CONVERSATION_REDIS_HOST"
CONVERSATION_REDIS_PORT_KEY = "CONVERSATION_REDIS_PORT"
CONVERSATION_REDIS_DB_KEY = "CONVERSATION_REDIS_DB"


@traces(spec="021", story="US-021-007", reqs=["REQ-021-069"])
def resolve_conversation_redis_endpoint(
    env: Mapping[str, str] | None = None,
) -> tuple[str, int, int]:
    """Return ``(host, port, db)`` of the conversation queue's Redis, or raise.

    Producers (Claude hook, kimi/IRC) and consumers (daemon, kernel boot) used
    to build Redis from the ambient ``REDIS_HOST``/``REDIS_PORT``/``REDIS_DB``,
    so a producer launched from a shell with a different ``REDIS_*`` than the
    systemd daemon wrote turns into a queue nobody drained. The dedicated keys
    live in the secrets-only ``.env`` files and are deliberately absent from
    every ``env/*.env`` overlay. There is NO fallback to ``REDIS_*``.

    Raises:
        RuntimeError: When any key is unset/blank, or port/db is not an int.
    """
    environ = os.environ if env is None else env
    values = {
        key: (environ.get(key) or "").strip()
        for key in (
            CONVERSATION_REDIS_HOST_KEY,
            CONVERSATION_REDIS_PORT_KEY,
            CONVERSATION_REDIS_DB_KEY,
        )
    }
    missing = [key for key, value in values.items() if not value]
    if missing:
        raise RuntimeError(
            f"{' and '.join(missing)} not set — the conversation queue has its "
            "own Redis endpoint and must never be resolved through the ambient "
            "REDIS_HOST/REDIS_PORT/REDIS_DB keys. Set them in the secrets-only "
            ".env (they are deliberately absent from every env/*.env profile "
            "overlay)."
        )
    try:
        port = int(values[CONVERSATION_REDIS_PORT_KEY])
        db = int(values[CONVERSATION_REDIS_DB_KEY])
    except ValueError as exc:
        raise RuntimeError(
            f"{CONVERSATION_REDIS_PORT_KEY}/{CONVERSATION_REDIS_DB_KEY} must be "
            f"integers: {exc}"
        ) from exc
    return values[CONVERSATION_REDIS_HOST_KEY], port, db


@traces(spec="021", story="US-021-007", reqs=["REQ-021-069"])
def create_conversation_redis(env: Mapping[str, str] | None = None) -> redis.Redis:
    """Build the sync Redis client for the conversation queue, pinned to its
    own endpoint keys (raises when unset — see
    :func:`resolve_conversation_redis_endpoint`)."""
    from orchestrata_ultimate.core.redis_factory import create_sync_redis

    host, port, db = resolve_conversation_redis_endpoint(env)
    return create_sync_redis(host=host, port=port, db=db)


@traces(spec="021", story="US-021-007", reqs=["REQ-021-069"])
def resolve_conversation_endpoint(
    env: Mapping[str, str] | None = None,
) -> tuple[str, str]:
    """Return ``(uri, database)`` for the CONVERSATION graph, or raise.

    Split out from :func:`create_conversation_graph_repository` so the
    endpoint CONTRACT is verifiable without constructing a driver: the
    repository exposes no public accessor for its config (``BoltSyncRepository``
    keeps it in ``_config``), and a test that reaches into private state to
    prove which server a writer talks to is testing the wrong thing. This
    function is pure — mapping in, tuple out — so the rule can be pinned
    directly, and no test needs to open a connection to prod to check it.

    Args:
        env: Environment mapping; defaults to ``os.environ`` read at CALL time
            (REQ-021-069), which is also the injection seam tests use instead
            of monkeypatch.

    Returns:
        The conversation graph's ``(uri, database)``.

    Raises:
        RuntimeError: When either key is unset or blank.
    """
    environ = os.environ if env is None else env
    uri = (environ.get(CONVERSATION_KG_URI_KEY) or "").strip()
    database = (environ.get(CONVERSATION_KG_DATABASE_KEY) or "").strip()

    missing = [
        key
        for key, value in (
            (CONVERSATION_KG_URI_KEY, uri),
            (CONVERSATION_KG_DATABASE_KEY, database),
        )
        if not value
    ]
    if missing:
        raise RuntimeError(
            f"{' and '.join(missing)} not set — the conversation graph has its "
            "own endpoint and must never be resolved through the "
            "profile-driven KG_ROUTING_URI/KG_URI keys, which a dev/test "
            "overlay redirects. Set them in the secrets-only .env (they are "
            "deliberately absent from every env/*.env profile overlay)."
        )

    return uri, database


@traces(spec="021", story="US-021-007", reqs=["REQ-021-069"])
def create_conversation_graph_repository(
    env: Mapping[str, str] | None = None,
) -> SyncGraphRepository:
    """Build the graph repository for the CONVERSATION graph, pinned to its own
    endpoint keys.

    The conversation graph is cross-project operational data that lives on the
    prod cluster, and it deliberately does NOT follow
    ``ORCHESTRATA_ENV_PROFILE``: a dev/test overlay must never redirect a
    writer away from where every reader looks. ``~/.claude/tools/
    conversation_recall.py`` (the only sanctioned reader) already enforces
    that by resolving ``CONVERSATION_KG_URI`` / ``CONVERSATION_KG_DATABASE``
    and raising when they are unset.

    The WRITERS did not. Both the standing daemon
    (``scripts/run_conversation_consumer.py``) and the kernel boot flush
    (``kernel/cli_main.py::_boot_conversation_consumer``) called a bare
    ``create_graph_repository()``, whose ``GraphRepositoryConfig.from_env()``
    defers to ``core.graph_factory`` — which reads the GENERIC
    ``KG_ROUTING_URI`` / ``KG_URI`` at driver-construction time. Those keys
    ARE profile-driven. So the endpoint was pinned by construction on the read
    side and by ambient-environment convention on the write side.

    The convention holds under systemd (the unit sets
    ``ORCHESTRATA_ENV_PROFILE=prod`` over the prod checkout) and breaks in any
    developer shell: direnv layers ``env/test.env``, which exports
    ``KG_ROUTING_URI=bolt://10.53.2.10:7687``, and the daemon's own
    ``load_dotenv(..., override=False)`` lets that exported value win over
    ``.env``. Running the consumer from such a shell writes conversation turns
    into the TEST graph while every reader keeps looking at prod — the history
    silently splits and nothing fails.

    Resolving the conversation endpoint here makes the wrong endpoint
    unreachable rather than merely discouraged. Unset keys RAISE (RUL-003's
    inverse rule for required infra endpoints: no literal fallback, not even
    localhost) — a conversation writer with no configured endpoint must stop,
    not invent one.

    Args:
        env: Environment mapping; defaults to ``os.environ`` read at CALL time
            (REQ-021-069), which is also the injection seam tests use instead
            of monkeypatch.

    Returns:
        A ``SyncGraphRepository`` bound to the conversation graph.

    Raises:
        RuntimeError: When either key is unset or blank.
    """
    from orchestrata_ultimate.core.graph.config import GraphRepositoryConfig
    from orchestrata_ultimate.core.graph.repository import create_graph_repository

    uri, database = resolve_conversation_endpoint(env)
    return create_graph_repository(GraphRepositoryConfig(uri=uri, database=database))


#: Message schema version this consumer accepts (REQ-021-063). Anything else
#: — including every v1 message the original plugin ever produced — is
#: quarantined, never guessed at.
_SCHEMA_VERSION = 2

#: Max stored turn content length (REQ-021-068). The producer is expected to
#: truncate too; this is the re-enforcement the contract requires.
_MAX_CONTENT_LEN = 2000

_VALID_ROLES = frozenset({"user", "assistant"})

_REQUIRED_STR_FIELDS = (
    "project",
    "conversation_id",
    "turn_id",
    "role",
    "content",
    "timestamp",
    "platform",
)
_REQUIRED_INT_FIELDS = ("session_seq", "turn_seq")


@traces(spec="021", story="US-021-007", reqs=["REQ-021-063"])
class ConversationMessageError(ValueError):
    """A queued message failed :meth:`ConversationTurnMessage.parse` validation.

    Carries both the machine-readable ``reason`` (logged and, verbatim,
    quarantined alongside the raw message) and the original ``raw`` string —
    the message is never lost, only parked (REQ-021-063).
    """

    def __init__(self, reason: str, raw: str) -> None:
        self.reason = reason
        self.raw = raw
        super().__init__(reason)


@dataclass(frozen=True)
class ConversationTurnMessage:
    """One parsed, validated queue message (message schema v2).

    See ``plans/contracts/spec-021-ontology.contract.json``
    ``queue_message_schema.message_v2`` for the wire shape this parses.
    """

    schema: int
    project: str
    conversation_id: str
    session_seq: int
    turn_id: str
    turn_seq: int
    role: str
    content: str
    timestamp: str
    platform: str
    tool_call_ids: tuple[str, ...] = ()
    # Tool NAMES, positionally parallel to ``tool_call_ids``.
    #
    # Ids alone answer "did anything run"; names answer "was it work". A turn
    # that only read files is not the same event as one that edited them, and
    # that distinction is the whole reason per-turn tool capture exists — it
    # is the only non-opinion answer to "should this turn have started a
    # pipeline?" (plans/nlq-research-2026-08.md R1/R2).
    #
    # Optional and positional rather than a dict: a producer that knows ids but
    # not names stays valid, and ``tool_name`` matches the property existing
    # ToolCall nodes already carry.
    tool_names: tuple[str, ...] = ()

    @classmethod
    @traces(spec="021", story="US-021-007", reqs=["REQ-021-063", "REQ-021-068"])
    def parse(cls, raw: str) -> ConversationTurnMessage:
        """Parse + validate one raw queue message.

        Raises:
            ConversationMessageError: On malformed JSON, an unsupported/
                missing ``schema`` (including every v1 message — it carries
                no ``schema`` field at all), a missing/blank required field,
                an invalid ``project`` (blank or containing ``:``), an
                invalid ``role``, or malformed ``tool_call_ids``. ``content``
                is truncated to :data:`_MAX_CONTENT_LEN`, never rejected for
                being long.
        """
        try:
            data = json.loads(raw)
        except (json.JSONDecodeError, TypeError, ValueError) as exc:
            raise ConversationMessageError("invalid-json", raw) from exc
        if not isinstance(data, dict):
            raise ConversationMessageError("invalid-json", raw)

        if data.get("schema") != _SCHEMA_VERSION:
            raise ConversationMessageError("unsupported-schema", raw)

        values: dict[str, Any] = {}
        for name in _REQUIRED_STR_FIELDS:
            value = data.get(name)
            if not isinstance(value, str) or not value.strip():
                raise ConversationMessageError(f"missing-field:{name}", raw)
            values[name] = value
        for name in _REQUIRED_INT_FIELDS:
            value = data.get(name)
            if not isinstance(value, int) or isinstance(value, bool):
                raise ConversationMessageError(f"missing-field:{name}", raw)
            values[name] = value

        if ":" in values["project"]:
            raise ConversationMessageError("invalid-project", raw)
        if values["role"] not in _VALID_ROLES:
            raise ConversationMessageError("invalid-role", raw)

        tool_call_ids_raw = data.get("tool_call_ids", [])
        if not isinstance(tool_call_ids_raw, list) or not all(
            isinstance(item, str) for item in tool_call_ids_raw
        ):
            raise ConversationMessageError("invalid-tool-call-ids", raw)

        tool_names_raw = data.get("tool_names", [])
        if not isinstance(tool_names_raw, list) or not all(
            isinstance(item, str) for item in tool_names_raw
        ):
            raise ConversationMessageError("invalid-tool-names", raw)
        # Names are positional against ids, so a length mismatch would silently
        # mislabel every call after the first gap. Reject rather than zip-
        # truncate: a producer that cannot supply names sends none at all.
        if tool_names_raw and len(tool_names_raw) != len(tool_call_ids_raw):
            raise ConversationMessageError("tool-names-length-mismatch", raw)

        return cls(
            schema=_SCHEMA_VERSION,
            project=values["project"],
            conversation_id=values["conversation_id"],
            session_seq=values["session_seq"],
            turn_id=values["turn_id"],
            turn_seq=values["turn_seq"],
            role=values["role"],
            content=values["content"][:_MAX_CONTENT_LEN],
            timestamp=values["timestamp"],
            platform=values["platform"],
            tool_call_ids=tuple(tool_call_ids_raw),
            tool_names=tuple(tool_names_raw),
        )


@runtime_checkable
class QueueSource(Protocol):
    """Reliable-queue seam the consumer drives (LMOVE claim/ack/requeue).

    A real Redis-backed implementation (:class:`RedisQueueSource`) is the
    only production adapter; the Protocol exists so tests/future backends
    can swap it without inheritance.
    """

    def claim(self) -> str | None:
        """Atomically move one raw message queue -> processing; ``None`` when empty."""
        ...  # pragma: no cover - protocol method body is not executable

    def ack(self, raw: str) -> None:
        """Remove ``raw`` from the processing list after a successful write."""
        ...  # pragma: no cover - protocol method body is not executable

    def requeue_processing(self) -> int:
        """Crash recovery: drain the processing list back onto the queue.

        Returns:
            The number of messages requeued.
        """
        ...  # pragma: no cover - protocol method body is not executable

    def quarantine(self, raw: str, reason: str) -> None:
        """Park an invalid/legacy message on the quarantine list."""
        ...  # pragma: no cover - protocol method body is not executable

    def depth(self) -> int:
        """Pending message count on the main queue."""
        ...  # pragma: no cover - protocol method body is not executable


class RedisQueueSource:
    """Real Redis reliable-queue implementation of :class:`QueueSource`.

    Replaces the original plugin's bare ``LPOP`` loop (REQ-021-062) with an
    ``LMOVE queue -> processing`` claim: a crash between claim and graph
    write no longer loses the turn — it sits in ``processing`` until the
    next :meth:`requeue_processing` call puts it back on the main queue for
    redelivery.
    """

    def __init__(
        self,
        redis_client: redis.Redis,
        *,
        queue_key: str = "orchestrata:conversation:queue",
        tracer: Tracer | None = None,
    ) -> None:
        """
        Args:
            redis_client: An already-constructed sync Redis client (DI — no
                env reads happen in this class; the boot seam builds the
                client via ``core.redis_factory.create_sync_redis``).
            queue_key: The main queue key. ``processing``/``quarantine`` keys
                are derived from it (stripping a trailing ``:queue`` suffix
                when present, else appending directly) so a test-scoped
                ``queue_key`` gets its own isolated processing/quarantine
                lists too.
            tracer: Injected :class:`Tracer`; defaults to the process-wide
                singleton.
        """
        self._redis = redis_client
        self._queue_key = queue_key
        base = (
            queue_key[: -len(":queue")] if queue_key.endswith(":queue") else queue_key
        )
        self._processing_key = f"{base}:processing"
        self._quarantine_key = f"{base}:quarantine"
        self._tracer = tracer if tracer is not None else Tracer.default()

    @traces(spec="021", story="US-021-007", reqs=["REQ-021-062"])
    def claim(self) -> str | None:
        with self._tracer.span("RedisQueueSource.claim"):
            result = self._redis.lmove(
                self._queue_key, self._processing_key, "RIGHT", "LEFT"
            )
            return cast("str | None", result)

    @traces(spec="021", story="US-021-007", reqs=["REQ-021-062"])
    def ack(self, raw: str) -> None:
        with self._tracer.span("RedisQueueSource.ack"):
            self._redis.lrem(self._processing_key, 1, raw)

    @traces(spec="021", story="US-021-007", reqs=["REQ-021-062"])
    def requeue_processing(self) -> int:
        with self._tracer.span("RedisQueueSource.requeue_processing"):
            count = 0
            while True:
                moved = cast(
                    "str | None",
                    self._redis.lmove(
                        self._processing_key, self._queue_key, "RIGHT", "LEFT"
                    ),
                )
                if moved is None:
                    break
                count += 1
            return count

    @traces(spec="021", story="US-021-007", reqs=["REQ-021-063"])
    def quarantine(self, raw: str, reason: str) -> None:
        with self._tracer.span("RedisQueueSource.quarantine"):
            _logger.warning("quarantining conversation message: %s", reason)
            self._redis.rpush(self._quarantine_key, raw)

    @traces(spec="021", story="US-021-007", reqs=["REQ-021-060"])
    def depth(self) -> int:
        with self._tracer.span("RedisQueueSource.depth"):
            return cast(int, self._redis.llen(self._queue_key))


class TurnGraphMapper:
    """Pure ``ConversationTurnMessage -> [(cypher, params), ...]`` mapper.

    Emits the ordered MERGE statements of
    ``consumer_mapping_table`` (contract JSON) — no I/O. Every identity
    write is MERGE, never CREATE; endpoints of ``TRIGGERED`` (one per
    ``tool_call_ids`` entry) are appended last and skipped entirely when the
    message carries none.
    """

    def __init__(self, tracer: Tracer | None = None) -> None:
        self._tracer = tracer if tracer is not None else Tracer.default()

    @traces(
        spec="021",
        story="US-021-007",
        reqs=["REQ-021-065", "REQ-021-066", "REQ-021-067"],
    )
    def to_statements(
        self, msg: ConversationTurnMessage
    ) -> list[tuple[str, dict[str, Any]]]:
        with self._tracer.span("TurnGraphMapper.to_statements"):
            conv_id = cs.conversation_entity_id(msg.project, msg.conversation_id)
            turn_qid = cs.conversation_turn_entity_id(msg.project, msg.turn_id)
            # D4 (contract): session-CLI Session.session_id VALUE shape,
            # settled here via the graph_audit lazy-MERGE precedent.
            session_qid = f"{msg.project}:session:{msg.session_seq}"

            statements: list[tuple[str, dict[str, Any]]] = [
                (
                    "MERGE (p:Project {slug: $project})",
                    {"project": msg.project},
                ),
                (
                    "MERGE (c:Conversation {id: $conv_id}) "
                    "ON CREATE SET c.natural_id=$conv_natural, c.project=$project, "
                    "c.session_seq=$session_seq, c.platform=$platform, "
                    "c.started_at=$ts "
                    "SET c.last_turn_at=$ts",
                    {
                        "conv_id": conv_id,
                        "conv_natural": msg.conversation_id,
                        "project": msg.project,
                        "session_seq": msg.session_seq,
                        "platform": msg.platform,
                        "ts": msg.timestamp,
                    },
                ),
                (
                    f"MATCH (p:Project {{slug: $project}}) "
                    f"MATCH (c:Conversation {{id: $conv_id}}) "
                    f"MERGE (p)-[:{cs.REL_HAS_CONVERSATION}]->(c)",
                    {"project": msg.project, "conv_id": conv_id},
                ),
                (
                    "MERGE (s:Session {session_id: $session_qid}) "
                    "ON CREATE SET s.seq=$session_seq, s.project=$project "
                    "WITH s MATCH (c:Conversation {id: $conv_id}) "
                    "MERGE (c)-[:IN_SESSION]->(s)",
                    {
                        "session_qid": session_qid,
                        "session_seq": msg.session_seq,
                        "project": msg.project,
                        "conv_id": conv_id,
                    },
                ),
                (
                    "MERGE (t:ConversationTurn {id: $turn_qid}) "
                    "ON CREATE SET t.turn_id=$turn_id, t.turn_seq=$turn_seq, "
                    "t.role=$role, t.content=$content, t.timestamp=$ts, "
                    "t.platform=$platform, t.project=$project "
                    f"WITH t MATCH (c:Conversation {{id: $conv_id}}) "
                    f"MERGE (c)-[:{cs.REL_HAS_TURN}]->(t)",
                    {
                        "turn_qid": turn_qid,
                        "turn_id": msg.turn_id,
                        "turn_seq": msg.turn_seq,
                        "role": msg.role,
                        "content": msg.content,
                        "ts": msg.timestamp,
                        "platform": msg.platform,
                        "project": msg.project,
                        "conv_id": conv_id,
                    },
                ),
            ]
            for index, tc_id in enumerate(msg.tool_call_ids):
                # `tool_name` (not `name`) is the property ToolCall nodes
                # written by the kernel already carry, so conversation-sourced
                # calls stay queryable by the same shape. SET, not ON CREATE
                # SET: a ToolCall may be MERGEd first by the kernel path with
                # no name, and arriving late with the name should fill it in.
                statements.append(
                    (
                        "MERGE (tc:ToolCall {id: $tc_id}) "
                        "SET tc.tool_call_id=$tc_id"
                        + (", tc.tool_name=$tool_name " if msg.tool_names else " ")
                        + "WITH tc MATCH (t:ConversationTurn {id: $turn_qid}) "
                        "MERGE (t)-[:TRIGGERED]->(tc)",
                        {
                            "tc_id": tc_id,
                            "turn_qid": turn_qid,
                            **(
                                {"tool_name": msg.tool_names[index]}
                                if msg.tool_names
                                else {}
                            ),
                        },
                    )
                )
            return statements


@traces(spec="021", story="US-021-007", reqs=["REQ-021-217"])
@dataclass(frozen=True)
class FlushReport:
    """Result of one :meth:`ConversationQueueConsumer.flush` call."""

    depth_before: int
    synced: int = 0
    quarantined: int = 0
    failed: int = 0

    @traces(spec="021", story="US-021-007", reqs=["REQ-021-217"])
    def to_summary(self) -> str:
        """The ``conv_sync`` tool's output line."""
        return (
            f"Flushed {self.synced} conversation turns to graph "
            f"({self.quarantined} quarantined, {self.failed} failed)"
        )


class ConversationQueueConsumer:
    """Drains the conversation queue into the graph — the SPEC-021 consumer.

    Composes a :class:`QueueSource`, a ``SyncGraphRepository``, and a
    :class:`TurnGraphMapper` (pure DI — no env/infra construction inside,
    REQ-021-069). :meth:`flush` never raises (REQ-021-070): a graph-write
    failure leaves the claimed message in ``processing`` for the next
    ``flush()`` to requeue and safely re-MERGE.
    """

    def __init__(
        self,
        source: QueueSource,
        repo: SyncGraphRepository,
        mapper: TurnGraphMapper | None = None,
        tracer: Tracer | None = None,
    ) -> None:
        self._source = source
        self._repo = repo
        self._tracer = tracer if tracer is not None else Tracer.default()
        self._mapper = mapper if mapper is not None else TurnGraphMapper(self._tracer)

    @traces(
        spec="021",
        story="US-021-007",
        reqs=["REQ-021-061", "REQ-021-062", "REQ-021-063", "REQ-021-070"],
    )
    def flush(self) -> FlushReport:
        """Requeue any orphaned in-flight messages, then drain the queue.

        Loop: claim -> parse -> map -> write each statement -> ack.
        ``ConversationMessageError`` -> quarantine + ack (never retried,
        never lost). A graph-write exception leaves the message claimed in
        ``processing`` (retried by the next call's requeue) and is counted
        as ``failed`` — this method itself never raises.
        """
        with self._tracer.span("ConversationQueueConsumer.flush"):
            self._source.requeue_processing()
            depth_before = self._source.depth()
            synced = 0
            quarantined = 0
            failed = 0
            while True:
                raw = self._source.claim()
                if raw is None:
                    break
                try:
                    msg = ConversationTurnMessage.parse(raw)
                except ConversationMessageError as exc:
                    self._source.quarantine(raw, exc.reason)
                    self._source.ack(raw)
                    quarantined += 1
                    continue
                try:
                    for cypher, params in self._mapper.to_statements(msg):
                        self._repo.execute_write(cypher, params)
                except Exception as exc:
                    _logger.error(
                        "conversation graph write failed (left in processing "
                        "for next flush): %s",
                        exc,
                    )
                    failed += 1
                    continue
                self._source.ack(raw)
                synced += 1
            report = FlushReport(
                depth_before=depth_before,
                synced=synced,
                quarantined=quarantined,
                failed=failed,
            )
            if depth_before or synced or quarantined or failed:
                _logger.info(
                    "conversation flush: depth_before=%d synced=%d "
                    "quarantined=%d failed=%d",
                    depth_before,
                    synced,
                    quarantined,
                    failed,
                )
            return report

    @traces(spec="021", story="US-021-007", reqs=["REQ-021-060", "REQ-021-216"])
    def depth(self) -> int:
        """Pending message count (delegates to the source)."""
        return self._source.depth()