scripts.run_conversation_consumer

Reference › scripts › run_conversation_consumer

scripts/run_conversation_consumer.py is the standing SPEC-021 daemon that drains the conversation-turn queue into the knowledge graph. It is started as a long-running process; tests/scripts/test_run_conversation_consumer.py covers it. The consumer logic is ConversationQueueConsumer in orchestrata_ultimate.knowledge_graph.conversation_consumer; the kimi and IRC producer helper is _irc_conversation.

_load_dotenv first loads the RUL-019 two-phase environment contract through orchestrata_ultimate.core.env_bootstrap. _build_consumer constructs the consumer, and run drains the queue continuously until signalled to stop. Options: --interval (seconds to sleep when the queue is empty, default 5, from CONVERSATION_CONSUMER_INTERVAL), --queue-key (the Redis list key, from CONVERSATION_QUEUE_KEY) and --once (run a single flush and exit, for verification). It reads the Redis queue and writes to the graph.

At a glance

Class diagram

Dependency diagram

API

Classes

_StopFlag

Line 85

SIGTERM/SIGINT-driven cooperative stop for the drain loop.

Functions

Source

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

#!/usr/bin/env python
"""Standing SPEC-021 conversation-turn consumer daemon.

The capture *producer* (``scripts/claude_turn_capture.py``, a user-level Claude
Code hook) writes one message per conversation turn into the Redis reliable
queue ``orchestrata:conversation:queue`` on every turn — automatically, whether
or not any kernel is running. Until now nothing drained that queue on a standing
cadence: the kernel only flushes once at boot and once at shutdown
(``kernel/cli_main.py::_boot_conversation_consumer`` /
``_ConversationConsumerComponent.stop``), so between kernel bounces turns pile up
unsynced. This daemon closes that gap — it is the continuous drainer that makes
conversation-turn capture *automatic and synced*.

Design
------
* Decoupled from the kernel on purpose: turns are captured on every Claude turn
  independent of kernel uptime, so the drainer must run independently and
  survive kernel restarts (same rationale as ``scripts/ast_live_sync.py``).
* :meth:`ConversationQueueConsumer.flush` drains everything currently available
  (claim -> parse -> map -> write -> ack) and returns a :class:`FlushReport`
  without ever raising (REQ-021-070). ``RedisQueueSource.claim`` is a
  non-blocking ``LMOVE``, so we poll: fast-loop while a flush did work (drains a
  backlog quickly), then ``sleep(--interval)`` once a flush comes back empty.
* The Redis reliable queue makes concurrent consumers safe, so this daemon may
  run alongside the kernel's own boot/stop flushes without double-processing.

Infra (Redis + graph endpoints) resolves from the process environment at call
time (REQ-021-069) via ``create_sync_redis`` / ``create_graph_repository``,
exactly as the kernel boot path does. Under the systemd unit that environment
is the repo ``.env`` (prod: orchkg on 10.53.1.160:7691, Redis 10.53.1.10).
"""

from __future__ import annotations

import argparse
import logging
import os
import signal
import sys
import time
from collections.abc import Mapping
from pathlib import Path

from orchestrata_ultimate.core.env_bootstrap import load_env_contract
from orchestrata_ultimate.knowledge_graph.conversation_consumer import (
    ConversationQueueConsumer,
    RedisQueueSource,
    create_conversation_graph_repository,
    create_conversation_redis,
)

_logger = logging.getLogger("conversation_consumer_daemon")

_DEFAULT_QUEUE_KEY = "orchestrata:conversation:queue"
_DEFAULT_IDLE_INTERVAL = 5.0

#: Seconds between cumulative "totals" log lines (the daemon is otherwise
#: silent while idle, which made a healthy-but-starved drain look dead).
_TOTALS_LOG_INTERVAL = 300.0

_PROJECT_ROOT = Path(__file__).resolve().parent.parent


def _load_dotenv() -> None:
    """Load the RUL-019 two-phase environment contract.

    Delegates to ``core.env_bootstrap.load_env_contract`` — THE canonical
    loader. What stood here loaded ``.env`` and stopped, and ``.env`` is the
    prod-BASE layer: it carries ``CXP_GRPC_HOST=localhost`` for the host
    user-units, so a script that stops after phase 1 resolves PRODUCTION
    endpoints while believing it is on the test fleet.

    ``override=False`` was the right instinct aimed at the wrong half. It was
    added on 2026-07-26 after a bootstrap intended for the test node wrote 592
    chunks into the production cluster, because ``.env``'s value had replaced
    an explicitly-exported one. The canonical loader keeps that property for
    phase 1 (``.env`` only fills keys the environment has not already set) and
    then applies the ``ORCHESTRATA_ENV_PROFILE`` overlay, which is the half
    that actually points the process at the right infrastructure — and honours
    ``declare_owned`` so it cannot reset a test worker's per-slot isolation.
    """
    load_env_contract()


class _StopFlag:
    """SIGTERM/SIGINT-driven cooperative stop for the drain loop."""

    def __init__(self) -> None:
        self._stop = False

    def request(self, signum: int, *_: object) -> None:
        _logger.info("received signal %s — stopping after current flush", signum)
        self._stop = True

    @property
    def stopping(self) -> bool:
        return self._stop


def _build_consumer(
    queue_key: str, env: Mapping[str, str] | None = None
) -> tuple[ConversationQueueConsumer, object, object]:
    """Wire the real Redis source + graph repo into a consumer.

    Mirrors ``kernel/cli_main.py::_boot_conversation_consumer`` construction so
    the daemon and the kernel drain the identical pipeline. Returns the consumer
    plus the two owned resources (redis client, graph repo) for clean shutdown.
    """
    # The conversation queue's OWN Redis (CONVERSATION_REDIS_*), never the
    # ambient REDIS_* a producer in another shell may disagree with. Raises
    # when unset.
    redis_client = create_conversation_redis(env)
    # The CONVERSATION graph's own endpoint, not the profile-driven one. A bare
    # create_graph_repository() resolves KG_ROUTING_URI/KG_URI, which env/
    # test.env overrides -- so running this daemon from a direnv'd dev shell
    # wrote turns into the TEST graph while every reader stayed on prod.
    repo = create_conversation_graph_repository(env)
    source = RedisQueueSource(redis_client, queue_key=queue_key)
    consumer = ConversationQueueConsumer(source=source, repo=repo)
    return consumer, redis_client, repo


def run(interval: float, queue_key: str, once: bool) -> int:
    """Drain the conversation queue continuously until signalled to stop.

    Returns a process exit code (0 = clean stop).
    """
    stop = _StopFlag()
    signal.signal(signal.SIGTERM, stop.request)
    signal.signal(signal.SIGINT, stop.request)

    consumer, redis_client, repo = _build_consumer(queue_key)
    _logger.info(
        "conversation consumer daemon started (queue=%s, idle_interval=%.1fs, once=%s)",
        queue_key,
        interval,
        once,
    )

    try:
        totals = {"synced": 0, "quarantined": 0, "failed": 0}
        last_totals_log = time.monotonic()
        while not stop.stopping:
            report = consumer.flush()
            totals["synced"] += report.synced
            totals["quarantined"] += report.quarantined
            totals["failed"] += report.failed
            if report.synced or report.quarantined or report.failed:
                _logger.info("flush: %s", report.to_summary())
            if time.monotonic() - last_totals_log >= _TOTALS_LOG_INTERVAL:
                _logger.info(
                    "conversation consumer totals: synced=%d quarantined=%d failed=%d",
                    totals["synced"],
                    totals["quarantined"],
                    totals["failed"],
                )
                last_totals_log = time.monotonic()
            if once:
                break
            # Fast-drain a backlog: only sleep once a flush comes back empty.
            if report.synced == 0:
                # Interruptible idle wait so a signal is honoured promptly.
                slept = 0.0
                while slept < interval and not stop.stopping:
                    time.sleep(min(0.5, interval - slept))
                    slept += 0.5
    finally:
        # Final drain, then release owned resources.
        try:
            final = consumer.flush()
            if final.synced or final.quarantined or final.failed:
                _logger.info("final flush: %s", final.to_summary())
        except Exception as exc:  # flush is REQ-021-070 no-raise; belt-and-suspenders
            _logger.error("error on final flush: %s", exc)
        for name, resource in (("graph repo", repo), ("redis client", redis_client)):
            close = getattr(resource, "close", None)
            if close is not None:
                try:
                    close()
                except Exception as exc:
                    _logger.warning("error closing %s: %s", name, exc)
    _logger.info("conversation consumer daemon stopped cleanly")
    return 0


def main(argv: list[str] | None = None) -> int:
    _load_dotenv()
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--interval",
        type=float,
        default=float(
            os.environ.get("CONVERSATION_CONSUMER_INTERVAL", _DEFAULT_IDLE_INTERVAL)
        ),
        help="seconds to sleep when the queue is empty (default: 5)",
    )
    parser.add_argument(
        "--queue-key",
        default=os.environ.get("CONVERSATION_QUEUE_KEY", _DEFAULT_QUEUE_KEY),
        help="Redis list key for the conversation queue",
    )
    parser.add_argument(
        "--once",
        action="store_true",
        help="run a single flush and exit (for verification, not the daemon)",
    )
    args = parser.parse_args(argv)

    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s %(levelname)s %(name)s %(message)s",
    )
    return run(interval=args.interval, queue_key=args.queue_key, once=args.once)


if __name__ == "__main__":
    sys.exit(main())