orchestrata_ultimate.kernel.cli_watch

Reference › orchestrata_ultimate › kernel › cli_watch

The streaming client behind orchestrata watch, described in its own docstring as the Phase 0 proof consumer. It follows one pipeline thread: it replays the backlog of events and then tails the live ones until the run ends. cli_main imports it. It uses cli_client for the gateway address and token, and pipeline_events for the topic name. It serves SPEC-000 (REQ-000-097, story US-006).

The design rule is that the gateway WebSocket is the single runtime ingress: the watcher never reads Redis itself. stream_thread_events connects, authenticates, subscribes to the thread's pipeline topic, asks gateway.replay for up to count past events (100 by default), yields each replayed event, and then yields live frames from _live_frames. The handshake calls run under a timeout (15 seconds by default) and the live loop waits without one. _normalize_replay_entry and _normalize_live_payload convert both shapes to one event dictionary, extract_live_event pulls the event out of a live message for the topic, and _mark_new drops duplicates that appear in both the backlog and the live stream. is_terminal_event stops the watch at pipeline.complete or pipeline.cancelled, and render_event_line formats an event as one line with its timestamp, type, volk, driver and data. Errors (authentication failure, an unreachable gateway, a closed connection) surface as click errors with clean messages.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

API

Functions

Source

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

"""Streaming client for ``orchestrata watch`` -- the Phase 0 proof consumer.

Phase 0 (``plans/contracts/volkdriver-phase0.contract.json``, deliverable
P0-B) proves the pipeline event stream is actually consumable end to end:
a real thread's backlog plus its live tail, rendered one line per event,
through the gateway's WebSocket JSON-RPC surface -- never Redis directly
(C3). ``kernel/cli_main.py``'s ``watch`` command is the thin click wrapper
over :func:`stream_thread_events` below.

Protocol notes (measured 2026-08-12; corrected from a refuted first draft
-- see the contract's ``refuted_and_corrected`` block):

1. ``auth.authenticate`` MUST be the first frame -- the gateway
   authenticates before registering the connection and before the
   receive loop starts.
2. SUBSCRIBE FIRST, THEN REPLAY. The live tail starts at ``last_id="$"``
   (new-only), so replaying first would leave a gap.
3. The subscribe method is the BARE ``subscribe`` -- ``gateway.subscribe``
   is not registered and dispatching it raises RPCError -32601. Replay is
   the namespaced ``gateway.replay``.
4. Live events arrive as JSON-RPC NOTIFICATIONS -- no ``id`` field. A
   client that correlates on response ``id`` sees nothing; events are
   matched on ``method`` (the topic) instead.
5. Replay entries and live payloads have asymmetric shapes: replay
   entries are stringly-typed and carry ``entry_id``; live payloads are
   already decoded (``data`` a dict, ``iteration`` an int, no
   ``entry_id``). Both are normalized to one render shape before yielding.

Author: Mohamed
"""

# Spec: 000 - Orchestrata CLI

from __future__ import annotations

import asyncio
import json
from collections import deque
from collections.abc import AsyncGenerator, AsyncIterator, Mapping
from typing import Any, cast

import click

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.kernel.cli_client import _mint_token, gateway_uri
from orchestrata_ultimate.kernel.pipeline_events import (
    PipelineEventType,
    pipeline_topic,
)

# C6: this set names two of the taxonomy's exactly-13 members; it does not
# add any.
TERMINAL_EVENT_TYPES = frozenset(
    {PipelineEventType.COMPLETE, PipelineEventType.CANCELLED}
)


def is_terminal_event(event: Mapping[str, Any]) -> bool:
    """True when *event* ends a watch (``pipeline.complete``/``.cancelled``)."""
    return event.get("event_type", "") in TERMINAL_EVENT_TYPES


def render_event_line(event: Mapping[str, Any]) -> str:
    """Render one normalized event as a single human-readable line.

    Generic on purpose (REQ-000-096's counterpart for human output): only
    ``event_type``/``timestamp`` are assumed always present; ``volk`` and
    ``driver`` are shown when non-empty so this never hard-depends on the
    P0-A ``driver`` field landing first.
    """
    parts = [
        str(event.get("timestamp", "") or "?"),
        str(event.get("event_type", "") or "?"),
    ]
    volk = event.get("volk", "")
    if volk:
        parts.append(f"volk={volk}")
    driver = event.get("driver", "")
    if driver:
        parts.append(f"driver={driver}")
    data = event.get("data")
    if data:
        parts.append(json.dumps(data, default=str, sort_keys=True))
    return "  ".join(parts)


def _normalize_replay_entry(entry: Mapping[str, Any]) -> dict[str, Any]:
    """Normalize a ``gateway.replay`` entry to the shared render shape.

    Replay entries are stringly-typed and carry ``entry_id`` (measured
    protocol fact 5) -- ``data`` is a JSON string, ``iteration`` a
    stringified int.
    """
    data: Any = entry.get("data", "{}")
    if isinstance(data, str):
        try:
            data = json.loads(data) if data else {}
        except json.JSONDecodeError, TypeError:
            data = {}
    iteration: Any = entry.get("iteration", 0)
    if isinstance(iteration, str):
        try:
            iteration = int(iteration)
        except TypeError, ValueError:
            iteration = 0
    return {
        "event_type": entry.get("event_type", ""),
        "thread_id": entry.get("thread_id", ""),
        "volk": entry.get("volk", ""),
        "driver": entry.get("driver", ""),
        "data": data if isinstance(data, dict) else {},
        "timestamp": entry.get("timestamp", ""),
        "iteration": iteration if isinstance(iteration, int) else 0,
    }


def _normalize_live_payload(payload: Mapping[str, Any]) -> dict[str, Any]:
    """Normalize an already-decoded live notification payload to the same
    render shape as a replay entry (``_decode_pipeline_event``'s output:
    ``data`` a dict, ``iteration`` an int, no ``entry_id`` -- measured
    protocol fact 5)."""
    data = payload.get("data", {})
    iteration = payload.get("iteration", 0)
    return {
        "event_type": payload.get("event_type", ""),
        "thread_id": payload.get("thread_id", ""),
        "volk": payload.get("volk", ""),
        "driver": payload.get("driver", ""),
        "data": data if isinstance(data, dict) else {},
        "timestamp": payload.get("timestamp", ""),
        "iteration": iteration if isinstance(iteration, int) else 0,
    }


def extract_live_event(msg: Mapping[str, Any], topic: str) -> dict[str, Any] | None:
    """Extract a normalized pipeline event from a JSON-RPC frame IFF it is
    a live notification on *topic*.

    A8 regression guard: live events arrive as JSON-RPC NOTIFICATIONS --
    no ``id`` field (measured protocol fact 4). A client that correlates
    on response ``id`` would silently see nothing; this recognizes the
    frame by ``method`` alone and returns ``None`` for anything else (a
    response to one of our own requests, or a notification on a different
    topic).
    """
    if "id" in msg:
        return None
    if msg.get("method") != topic:
        return None
    params = msg.get("params")
    if not isinstance(params, dict):
        return None
    payload = params.get("payload")
    if not isinstance(payload, dict):
        return None
    return _normalize_live_payload(payload)


def _dedup_key(event: Mapping[str, Any]) -> tuple[str, str]:
    return (str(event.get("timestamp", "")), str(event.get("event_type", "")))


def _mark_new(event: Mapping[str, Any], seen: set[tuple[str, str]]) -> bool:
    """Record *event* as seen against *seen*; True iff this is the first time.

    Keyed on ``(timestamp, event_type)`` -- deliberately NOT ``entry_id``:
    a replay entry and its live-notification twin for the SAME logical
    event come from two different Redis streams (measured protocol fact
    5) and never share one, so comparing raw ``entry_id`` would silently
    defeat the dedup this exists for (the subscribe-before-replay overlap
    window).
    """
    key = _dedup_key(event)
    if key in seen:
        return False
    seen.add(key)
    return True


@traces(spec="000", story="US-006", reqs=["REQ-000-097"])
async def _recv_frame(ws: Any, *, timeout: float) -> dict[str, Any]:
    raw = await asyncio.wait_for(ws.recv(), timeout=timeout)
    return cast(dict[str, Any], json.loads(raw))


@traces(spec="000", story="US-006", reqs=["REQ-000-097"])
async def _call_rpc(
    ws: Any,
    msg_id: int,
    method: str,
    params: dict[str, Any],
    *,
    timeout: float,
    pending: deque[dict[str, Any]],
) -> dict[str, Any]:
    """Send one JSON-RPC request over *ws* and return its result.

    Any frame that arrives before the matching response -- a live
    notification delivered during the subscribe-before-replay overlap
    window (measured protocol fact 2) -- is queued onto *pending* rather
    than discarded, so the caller's live tail sees it once the handshake
    completes.

    Raises:
        click.ClickException: the gateway returned a JSON-RPC error for
            *method* (REQ-000-097: a clean message, never a raw traceback).
    """
    await ws.send(
        json.dumps({"jsonrpc": "2.0", "id": msg_id, "method": method, "params": params})
    )
    while True:
        msg = await _recv_frame(ws, timeout=timeout)
        if msg.get("id") == msg_id:
            if "error" in msg:
                err = msg["error"]
                raise click.ClickException(
                    f"{method} failed: {err.get('message', err)}"
                )
            result = msg.get("result", {})
            return result if isinstance(result, dict) else {}
        # A live notification arrived before our response -- queue it; the
        # caller's tail loop drains it once the backlog has rendered.
        if "id" not in msg:
            pending.append(msg)


async def _live_frames(
    ws: Any, pending: deque[dict[str, Any]]
) -> AsyncGenerator[dict[str, Any]]:
    """Yield every live JSON-RPC frame: first drains anything queued during
    the handshake (the subscribe-before-replay overlap window), then reads
    fresh frames off the socket with NO timeout by design -- watch blocks
    until the next event or Ctrl-C, however long that takes."""
    while pending:
        yield pending.popleft()
    while True:
        raw = await ws.recv()
        yield cast(dict[str, Any], json.loads(raw))


@traces(spec="000", story="US-006", reqs=["REQ-000-097"])
async def stream_thread_events(
    thread_id: str,
    *,
    count: int = 100,
    env: Mapping[str, str] | None = None,
    recv_timeout: float = 15.0,
) -> AsyncIterator[dict[str, Any]]:
    """Stream a pipeline thread's events: backlog then live, one event per
    yield, until a terminal event or the caller cancels iteration (Ctrl-C).

    C3: the gateway WebSocket is the SINGLE runtime ingress -- this talks
    only to the gateway's JSON-RPC surface, never Redis directly.

    Args:
        thread_id: CXP thread to watch.
        count: Replay backlog depth (default 100, matches
            ``gateway.replay``'s own default).
        env: Optional explicit environment mapping forwarded to
            :func:`orchestrata_ultimate.kernel.cli_client.gateway_uri` --
            used by tests to target a hermetic gateway.
        recv_timeout: Timeout for the bounded auth/subscribe/replay
            handshake calls. NOT applied to the live tail loop, which
            blocks indefinitely by design (a watch waits for the next
            event, however long that takes).

    Raises:
        click.ClickException: auth failure, an RPC error, or the gateway
            being unreachable/closed -- always a clean message, never a
            raw traceback (REQ-000-097).
    """
    import websockets.exceptions

    uri = gateway_uri(env=env)
    topic = pipeline_topic(thread_id)
    pending: deque[dict[str, Any]] = deque()
    seen: set[tuple[str, str]] = set()

    try:
        async with websockets.connect(uri) as ws:
            auth_result = await _call_rpc(
                ws,
                1,
                "auth.authenticate",
                {"token": _mint_token()},
                timeout=recv_timeout,
                pending=pending,
            )
            if not auth_result.get("authenticated", False):
                raise click.ClickException(
                    "gateway authentication failed: "
                    f"{auth_result.get('error', 'unknown')}"
                )

            # 2/3: SUBSCRIBE FIRST (bare `subscribe`), THEN `gateway.replay`.
            await _call_rpc(
                ws,
                2,
                "subscribe",
                {"topic": topic},
                timeout=recv_timeout,
                pending=pending,
            )
            replay_result = await _call_rpc(
                ws,
                3,
                "gateway.replay",
                {"topic": topic, "count": count},
                timeout=recv_timeout,
                pending=pending,
            )
            entries = replay_result.get("events", [])
            if not isinstance(entries, list):
                entries = []

            for entry in entries:
                event = _normalize_replay_entry(entry)
                if not _mark_new(event, seen):
                    continue
                yield event
                if is_terminal_event(event):
                    return

            # 4: the live tail -- notifications carry NO `id`.
            async for live_msg in _live_frames(ws, pending):
                live_event = extract_live_event(live_msg, topic)
                if live_event is None or not _mark_new(live_event, seen):
                    continue
                yield live_event
                if is_terminal_event(live_event):
                    return
    except click.ClickException:
        raise
    except (TimeoutError, OSError) as e:
        raise click.ClickException(
            f"gateway unreachable at {uri} ({e}) — is the runtime up? "
            f"Start it with `orchestrata start` or scripts/start_test_services.py."
        ) from e
    except websockets.exceptions.ConnectionClosed as e:
        raise click.ClickException(f"gateway connection closed: {e}") from e