scripts._irc_conversation

Reference › scripts › _irc_conversation

The shared conversation-capture helper for the kimi and IRC producer of SPEC-021. It is a library, not a command; _irc_listen and _irc_send use it to turn IRC messages into conversation records. It is the IRC counterpart of claude_turn_capture, and both feed the queue read by run_conversation_consumer.

Every function takes an optional environment mapping. capture_enabled says whether capture is on, project_slug names the project, and conversation_redis_config and build_redis_client read the Redis connection from the environment; the client builder returns nothing when no connection can be made. start_conversation_session allocates a session sequence number in Redis and read_current_session_seq reads the current one. conversation_id_for turns a sequence into a conversation id, and per-project and per-conversation counters live under keys built by _session_seq_key, _current_key and _turn_seq_key. enqueue_message pushes one message (role, platform, content, session and conversation ids) onto the queue, capped at a maximum length (_queue_max). platform_for_nick maps an IRC nick to a platform name. Redis operations are retried once (_with_one_retry), and a message that cannot be captured is recorded in a capture log (log_drop, capture_log_path) instead of raising.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

No internal imports and no importers.

API

Constants

Functions

Source

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

"""Shared conversation-capture helper for the kimi/IRC SPEC-021 producer.

Used by both ``_irc_listen.py`` (assistant-side capture: PRIVMSG lines from
the fleet nicks ``kimi``/``codewhale`` in ``#agents``) and ``_irc_send.py``
(user-side capture: the lines we send) so that one persistent listener
connection and its many short-lived sender invocations agree on exactly one
conversation, per ``plans/spec-021-multiplatform-producers.md`` §
"kimi/IRC producer" (the settled design) and the message-v2 wire shape in
``plans/contracts/spec-021-ontology.contract.json``
(``queue_message_schema.message_v2``), which
``knowledge_graph/conversation_consumer.py`` (``ConversationTurnMessage.parse``)
validates on the other end. Relevant requirement ids: REQ-021-065 (session
identity), REQ-021-066 (turn identity/ordering), REQ-021-068 (content
truncation), REQ-021-070 (non-blocking).

Fail-open EVERYWHERE (REQ-021-070): every public function here catches its
own exceptions, logs to stderr, and returns ``None``/``False`` instead of
raising. The IRC bus (and the watchdog's liveness probe, which only
``pgrep -f _irc_listen.py`` — see ``_irc_watchdog.py``) must never be
affected by a Redis hiccup or a malformed message. Redis clients are built
lazily with short connect/socket timeouts so a dead Redis cannot delay
message handling perceptibly.

This module is a plain top-level script-sibling (not part of the
``orchestrata_ultimate`` package) so ``_irc_listen.py``/``_irc_send.py``
stay standalone-runnable via ``python scripts/_irc_listen.py``.
"""

from __future__ import annotations

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

if TYPE_CHECKING:
    from collections.abc import Callable

    import redis

#: Message schema version this producer emits (matches the consumer's
#: ``_SCHEMA_VERSION`` in knowledge_graph/conversation_consumer.py).
_SCHEMA_VERSION = 2

#: Max stored turn content length (REQ-021-068) — producer-side truncation;
#: the consumer re-enforces the same limit.
_MAX_CONTENT_LEN = 2000

#: Default bounded-queue size (design doc: "LPUSH + LTRIM to
#: ORCHESTRATA_CONVERSATION_QUEUE_MAX (default 10000)").
_DEFAULT_QUEUE_MAX = 10000

_DEFAULT_QUEUE_KEY = "orchestrata:conversation:queue"

#: Short timeouts so a dead Redis never delays IRC message handling
#: perceptibly (design doc: "Redis client created lazily with short
#: timeouts").
_CONNECT_TIMEOUT = 0.3
_SOCKET_TIMEOUT = 0.3

#: This producer's platform slug (design doc: "conversation_id = 'kimi-
#: {session_seq}'"). Fixed — this module is specifically the kimi/IRC
#: producer, not a generic multi-platform one.
PLATFORM_SLUG = "kimi"

#: Fleet nicks whose PRIVMSG lines in #agents are captured as assistant
#: turns, mapped to the `platform` value recorded on the turn (design doc:
#: "platform records the speaker's client" — all turns still land under the
#: same kimi-{seq} conversation). Any other nick's chatter is delegation-
#: adjacent noise and is deliberately not captured.
PLATFORM_FOR_NICK: dict[str, str] = {
    "kimi": "irc-kimi",
    "kimi_agent": "irc-kimi",
    "codewhale": "irc-codewhale",
}


def capture_enabled(env: dict[str, str] | None = None) -> bool:
    """Kill switch: ``ORCHESTRATA_CONVERSATION_CAPTURE=0`` disables capture."""
    e = os.environ if env is None else env
    return e.get("ORCHESTRATA_CONVERSATION_CAPTURE", "1") != "0"


def project_slug(env: dict[str, str] | None = None) -> str:
    """``ORCHESTRATA_PROJECT_SLUG`` override, else the repo-directory basename."""
    e = os.environ if env is None else env
    slug = e.get("ORCHESTRATA_PROJECT_SLUG")
    if slug:
        return slug
    return Path(__file__).resolve().parent.parent.name


def _queue_max(env: dict[str, str] | None = None) -> int:
    e = os.environ if env is None else env
    try:
        return int(e.get("ORCHESTRATA_CONVERSATION_QUEUE_MAX", str(_DEFAULT_QUEUE_MAX)))
    except TypeError, ValueError:
        return _DEFAULT_QUEUE_MAX


#: The conversation queue's OWN Redis endpoint keys (same names as
#: ``knowledge_graph.conversation_consumer.CONVERSATION_REDIS_*_KEY``; repeated
#: here so this standalone script stays light). NO fallback to ``REDIS_*``.
_REDIS_KEYS = (
    "CONVERSATION_REDIS_HOST",
    "CONVERSATION_REDIS_PORT",
    "CONVERSATION_REDIS_DB",
)


def capture_log_path(env: dict[str, str] | None = None) -> Path:
    """Per-user drop log: ``$XDG_STATE_HOME/orchestrata/conversation_capture.log``."""
    e = os.environ if env is None else env
    state = e.get("XDG_STATE_HOME", "").strip()
    base = Path(state) if state else Path(e.get("HOME") or Path.home()) / ".local/state"
    return base / "orchestrata" / "conversation_capture.log"


def log_drop(reason: str, env: dict[str, str] | None = None) -> None:
    """Append ONE line (timestamp, reason, project) to the capture log.

    Fail-open: a log-write failure is swallowed (REQ-021-070). Makes a capture
    drop observable instead of silent.
    """
    try:
        path = capture_log_path(env)
        path.parent.mkdir(parents=True, exist_ok=True)
        line = f"{datetime.now(UTC).isoformat()} {project_slug(env)} {reason}\n"
        with path.open("a", encoding="utf-8") as fh:
            fh.write(line)
    except Exception as exc:
        print(f"[irc-conversation] capture log write failed: {exc}", file=sys.stderr)


def conversation_redis_config(
    env: dict[str, str] | None = None,
) -> tuple[str, int, int]:
    """``(host, port, db)`` from ``CONVERSATION_REDIS_*``; raises ``ValueError``
    when any is unset/blank or port/db is not an integer."""
    e = os.environ if env is None else env
    values = {k: (e.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]]),
    )


def build_redis_client(env: dict[str, str] | None = None) -> redis.Redis | None:
    """Lazily build a short-timeout Redis client. Returns ``None`` on failure.

    Resolves ``CONVERSATION_REDIS_HOST/PORT/DB`` only (never ``REDIS_*``); unset
    or malformed keys log one line to the capture log and return ``None``.

    Every caller in this module treats ``None`` the same way as a raised
    exception from a later Redis call: log to stderr, skip capture, keep
    going (REQ-021-070) — the IRC bus and the watchdog's liveness probe must
    never be affected by a capture hiccup.
    """
    try:
        import redis
        from redis.backoff import NoBackoff
        from redis.retry import Retry
    except Exception as exc:  # pragma: no cover - redis is a declared dep
        print(f"[irc-conversation] redis import failed: {exc}", file=sys.stderr)
        return None
    try:
        host, port, db = conversation_redis_config(env)
    except ValueError as exc:
        print(f"[irc-conversation] conversation redis unset: {exc}", file=sys.stderr)
        log_drop(f"redis-config: {exc}", env)
        return None
    try:
        return redis.Redis(
            host=host,
            port=port,
            db=db,
            decode_responses=True,
            socket_connect_timeout=_CONNECT_TIMEOUT,
            socket_timeout=_SOCKET_TIMEOUT,
            # redis-py's own default Retry (3 exponential-backoff attempts)
            # turns one dead Redis into several seconds of delay — exactly
            # what "must never delay message handling perceptibly" forbids.
            # A single fast-fail attempt is the whole point of the short
            # connect/socket timeouts above.
            retry=Retry(NoBackoff(), 0),
        )
    except Exception as exc:
        print(
            f"[irc-conversation] redis client construction failed: {exc}",
            file=sys.stderr,
        )
        return None


def _session_seq_key(project: str) -> str:
    return f"orchestrata:conversation:session_seq:{project}:{PLATFORM_SLUG}"


def _current_key(project: str) -> str:
    return f"orchestrata:conversation:current:{project}:{PLATFORM_SLUG}"


def _turn_seq_key(project: str, conversation_id: str) -> str:
    return f"orchestrata:conversation:seq:{project}:{conversation_id}"


def conversation_id_for(session_seq: int) -> str:
    """``kimi-{session_seq}`` — the natural conversation id (design doc)."""
    return f"{PLATFORM_SLUG}-{session_seq}"


_RETRY_PAUSE = 0.2


def _with_one_retry[T](
    op: Callable[[], T], *, what: str, env: dict[str, str] | None = None
) -> T | None:
    """Run ``op`` with ONE bounded retry, still fail-open (REQ-021-070).

    ``build_redis_client`` deliberately builds a fast-fail client --
    ``Retry(NoBackoff(), 0)`` behind 0.3s connect/socket deadlines -- so a
    DEAD Redis can never stall the IRC loop. Those same deadlines also fire
    against a HEALTHY Redis when this process is scheduler-starved, which is
    not a hypothetical: ``enqueue_message`` grew this exact retry on
    2026-08-05 after a 14-worker gate run dropped a turn that way.

    That fix hardened ONE of the two Redis touchpoints on the send path. The
    send path reads the current session key FIRST and skips capture when that
    read yields nothing, so a starved read discards the turn before the
    hardened enqueue is ever reached -- the retry could not help, because the
    call never happened. Measured: ``test_send_side_capture_joins_current_
    conversation`` recovered on solo retry SIX times across gate runs, always
    with an empty queue (capture skipped), always green when run alone.

    Worst case stays well inside the bound the fail-open test asserts:
    2 attempts x 2 socket deadlines (connect + one command) x 0.3s + 0.2s
    pause = ~1.4s against the test's 3.0s ceiling.
    """
    last_exc: Exception | None = None
    for attempt in (0, 1):
        try:
            return op()
        except Exception as exc:
            last_exc = exc
            if attempt == 0:
                time.sleep(_RETRY_PAUSE)
    print(f"[irc-conversation] {what} failed: {last_exc}", file=sys.stderr)
    log_drop(f"{what} failed: {last_exc}", env)
    return None


def start_conversation_session(
    *,
    env: dict[str, str] | None = None,
    client: redis.Redis | None = None,
) -> int | None:
    """Allocate a new ``session_seq`` for one listener connection+join.

    Called once per listener connect (design doc: "one listener connection =
    one conversation session"). INCRs the per-project session counter and
    writes the result to the CURRENT key so the separate, short-lived
    ``_irc_send.py`` process can join the same conversation (REQ-021-065).

    Fail-open: returns ``None`` (and logs to stderr) on any Redis error —
    the caller must keep running the IRC loop regardless.
    """
    project = project_slug(env)
    r = client if client is not None else build_redis_client(env)
    if r is None:
        return None

    def _allocate() -> int:
        # A retried attempt may INCR twice when the first INCR landed and a
        # later op timed out, so a session_seq number can be skipped. The
        # property this key carries is monotonicity, not density -- the same
        # trade enqueue_message already makes for turn_seq.
        seq = int(cast(int, r.incr(_session_seq_key(project))))
        r.set(_current_key(project), str(seq))
        return seq

    return _with_one_retry(_allocate, what="start_conversation_session", env=env)


def read_current_session_seq(
    *,
    env: dict[str, str] | None = None,
    client: redis.Redis | None = None,
) -> int | None:
    """Read the ``session_seq`` the (separate) listener process is on.

    Returns ``None`` if unreadable/absent — Redis down, or no listener has
    connected yet in this project. Callers (``_irc_send.py``) must still
    perform their primary action (sending the IRC line) and simply skip
    capture when this returns ``None``.
    """
    project = project_slug(env)
    r = client if client is not None else build_redis_client(env)
    if r is None:
        return None

    def _read() -> int | None:
        # An ABSENT key is not an error and must not be retried: "no listener
        # has connected yet" is a real answer, and the caller is required to
        # act on it (send the IRC line, skip capture). Only a raised Redis
        # error reaches the retry.
        raw = cast("str | None", r.get(_current_key(project)))
        return None if raw is None else int(raw)

    return _with_one_retry(_read, what="read_current_session_seq", env=env)


def enqueue_message(
    *,
    role: str,
    platform: str,
    content: str,
    session_seq: int,
    conversation_id: str,
    env: dict[str, str] | None = None,
    client: redis.Redis | None = None,
    queue_key: str = _DEFAULT_QUEUE_KEY,
) -> bool:
    """Build + LPUSH one message-v2 conversation turn.

    Wire shape matches ``plans/contracts/spec-021-ontology.contract.json``
    ``queue_message_schema.message_v2``, validated on the other end by
    ``knowledge_graph.conversation_consumer.ConversationTurnMessage.parse``.
    ``turn_seq`` is a producer-side Redis INCR on the per-conversation
    counter (REQ-021-066 — the ordering property); ``turn_id`` is a fresh
    UUIDv4 (IRC lines have no stable id); ``content`` is truncated to
    :data:`_MAX_CONTENT_LEN` (REQ-021-068). After LPUSH the queue is LTRIM'd
    to ``ORCHESTRATA_CONVERSATION_QUEUE_MAX`` (default 10000) so an
    un-consumed queue cannot grow without bound (drop-oldest).

    Returns ``False`` (never raises) when the kill switch is off or on any
    Redis error — fail-open (REQ-021-070).
    """
    if not capture_enabled(env):
        return False
    project = project_slug(env)
    r = client if client is not None else build_redis_client(env)
    if r is None:
        return False
    # Two bounded attempts, still fail-open (REQ-021-070): the 0.3s socket
    # timeouts exist so a DEAD Redis can never stall the IRC loop, but they
    # also fire against a HEALTHY Redis when this process is scheduler-starved
    # (live 2026-08-05, 14-worker gate: one op missed the deadline and the
    # turn silently dropped). One retry after a short pause absorbs that
    # transient without changing either guarantee — worst case ~2.6s bounded
    # (2 attempts × 4 socket deadlines [connect+INCR+LPUSH+LTRIM] × 0.3s +
    # 0.2s pause), never raises. A retried INCR may skip a turn_seq number;
    # the ordering property (REQ-021-066) needs monotonicity, not density.
    #
    # turn_id and timestamp are minted ONCE, before the loop: a lost-reply
    # retry (server applied the LPUSH, this process timed out waiting) would
    # otherwise enqueue the same turn under a second uuid4, and the consumer
    # — which MERGEs on turn_id — would write it to the KG twice. With a
    # stable turn_id the duplicate LPUSH is idempotent at the KG layer.
    turn_id = str(uuid.uuid4())
    timestamp = datetime.now(UTC).isoformat()
    last_exc: Exception | None = None
    for attempt in (0, 1):
        try:
            turn_seq = int(cast(int, r.incr(_turn_seq_key(project, conversation_id))))
            message: dict[str, Any] = {
                "schema": _SCHEMA_VERSION,
                "project": project,
                "conversation_id": conversation_id,
                "session_seq": session_seq,
                "turn_id": turn_id,
                "turn_seq": turn_seq,
                "role": role,
                "content": content[:_MAX_CONTENT_LEN],
                "timestamp": timestamp,
                "platform": platform,
                "tool_call_ids": [],
            }
            raw = json.dumps(message)
            r.lpush(queue_key, raw)
            r.ltrim(queue_key, 0, _queue_max(env) - 1)
            return True
        except Exception as exc:
            last_exc = exc
            if attempt == 0:
                time.sleep(0.2)
    print(f"[irc-conversation] enqueue_message failed: {last_exc}", file=sys.stderr)
    log_drop(f"enqueue_message failed: {last_exc}", env)
    return False


def platform_for_nick(nick: str, nick_map: dict[str, str] | None = None) -> str | None:
    """The ``platform`` value to capture ``nick``'s PRIVMSG under, or ``None``.

    ``nick_map`` defaults to :data:`PLATFORM_FOR_NICK` (the production fleet
    mapping); tests inject a test-scoped map so they never need to log in as
    the literal ``kimi``/``codewhale`` nicks (which could collide with the
    real fleet on the shared IRC server).
    """
    m = PLATFORM_FOR_NICK if nick_map is None else nick_map
    return m.get(nick)