diff --git a/persistent-memory.md b/persistent-memory.md index d7970c2..74e6018 100644 --- a/persistent-memory.md +++ b/persistent-memory.md @@ -30,7 +30,7 @@ separate dev team rather than an in-tree Worldtree tool. ## Current state / in-flight -**Status: `ratatoskr.sse_client` + `ratatoskr.sessions` both implemented via TDD against their issue-scoped contracts.** 62/62 tests GREEN (42 sse_client + 19 sessions + 1 boundary); ruff clean. +**Status: `ratatoskr.sse_client` + `ratatoskr.sessions` + `ratatoskr.cli` all implemented via TDD against their issue-scoped contracts.** 117/117 tests GREEN (42 sse_client + 19 sessions + 54 cli + 1 boundary + 1 metadata); ruff clean. What's in the repo: - `docs/design-brief.md` — the locked design (copy from `brokkr-smithy/docs/ratatoskr-design-brief.md`). @@ -44,6 +44,9 @@ What's in the repo: - `tests/test_sse_client.py` — 37 tests covering all four FN blocks' TESTS: entries verbatim (13 + 10 + 8 + 6). Real HTTP wire via respx mocks; SSE wire format constructed by helper `_sse_chunk`. Connection-drop test uses custom `httpx.AsyncByteStream` subclass that yields chunks then raises `RemoteProtocolError`. - `src/ratatoskr/sessions.py` — **implemented 2026-05-21** per `docs/contracts/issues/2.contract.md`. Two functions (`create_session`, `list_sessions`) + two frozen dataclasses (`SessionInfo`, `SessionPage`) + three exception types (`AgentNotFound`, `InvalidCursor`, `SessionApiFailed`). `SessionInfo` uses origin-conditional defaults per INV-001/INV-002 (create-origin: `name=None`, `archived=False`, `tags=[]`, `message_count=`; list-origin: same defaults for absent/null fields, `message_count=None`). `SessionApiFailed` truncates `.body` to ≤1024 at construction. No code shared with `sse_client.py` (convention-dependency only per issue #2 `dependencies:`). - `tests/test_sessions.py` — 19 tests covering both FN blocks' TESTS: entries verbatim (7 + 12). Helper `_list_item()` builds GET /sessions list-item bodies for tests. +- `docs/contracts/issues/3.contract.md` — **issue-scoped contract for issue #3** (https://gitea.phasefinal.com/vh/ratatoskr/issues/3). v2.1, complexity=medium. `target_module: ratatoskr.cli`. `prd:` block pins to issue body SHA `206ef51709d43b2c` at `2026-05-21T05:13:29+00:00`. Six FN blocks: `main`, `_parse_args`, `_amain`, `_render_event`, `_run_turn`, `_cancel_and_log`. `dependencies:` block lists issues #1 and #2 as code-level deps (first in the repo to do so — #1 and #2 were convention-only siblings). Drift check returns clean. +- `src/ratatoskr/cli.py` — **implemented 2026-05-21** per `docs/contracts/issues/3.contract.md`. Six entry points: sync `main` + async `_amain` + four private helpers (`_parse_args`, `_render_event`, `_cancel_and_log`, `_run_turn`). Composes `sessions.create_session` (when `--new`) with `sse_client.stream_turn` + `cancel_turn`. Hard invariant: no `textual` / `rich` imports (raw stdout). `_run_turn` is the load-bearing piece — race-loop pattern that gates `asyncio.create_task(sigint_event.wait())` behind `if not cancelling` to avoid the busy-loop bug Volva flagged in contract review. SIGINT handler installed via `loop.add_signal_handler(SIGINT, sigint_event.set)` so unit tests can fire the event directly without real signals. 305 LOC. +- `tests/test_cli.py` — 54 tests covering all six FN blocks' TESTS: entries verbatim (14 + 10 + 5 + 13 + 7 + 5). Helpers: `_sse_chunk` (same shape as `test_sse_client.py`'s helper), `_sse_resp` (wraps respx Response with the `text/event-stream` content-type), `_GatedStream` (custom `httpx.AsyncByteStream` that pauses on `asyncio.Event` entries to make SIGINT-mid-stream tests deterministic without sleep-based timing). Test fixtures: `_clear_env` autouse fixture clears `WORLDTREE_API_KEY` / `WORLDTREE_API_URL` per test for deterministic env-resolution assertions. - `tests/test_no_worldtree_imports.py` — boundary smoke test (passes; verified 2026-05-20). - `tests/snapshots/README.md` — recording/replay convention for SSE snapshot tests. @@ -56,10 +59,10 @@ What's NOT in the repo yet: **Branch:** `main`. Remote: `origin → git@gitea.phasefinal.com:vh/ratatoskr.git` (added 2026-05-20). **Next natural moves:** -1. Optional: `/volva-code-review docs/contracts/issues/2.contract.md` against the freshly-landed implementation (precedent: issue #1's code-review caught 4 negative-space drifts the TDD round missed). -2. Build the `--send` stdout presenter under `ratatoskr.cli` — composes `create_session` + `stream_turn` into the non-interactive mode (design-brief §8b). First chance to exercise both modules against a real Worldtree. -3. Record real SSE snapshot fixtures from a running Worldtree. `--send --new` is itself a recording probe — capture its outputs to `tests/snapshots/` for replay-based regression coverage. -4. Textual TUI app shell — second presenter; multi-pane observability dashboard per design-brief §5. +1. **`/volva-code-review docs/contracts/issues/3.contract.md`** against the freshly-landed `ratatoskr.cli` implementation (precedent: #1 caught 4 negative-space drifts, #2 caught 3). The race-loop in `_run_turn` is the most likely source of new drifts — cross-model fresh-eyes review on async / signal-handling code has been valuable in past rounds. +2. **Manual smoke against a local Worldtree**: `ratatoskr --send "hello" --new --agent `. Captures the first real-Worldtree integration evidence (mocks aren't the wire). Useful regardless of #1. +3. Record real SSE snapshot fixtures from a running Worldtree. `--send --new` redirected to a fixture file IS the recording probe — capture outputs to `tests/snapshots/` for replay-based regression coverage. +4. Textual TUI app shell — second presenter; multi-pane observability dashboard per design-brief §5. Now that the consumer surface is exercised end-to-end via `--send`, the TUI lands on top of validated modules. ## Recent decisions @@ -87,6 +90,9 @@ decision. Captures rationale that won't be obvious from code alone. - `[2026-05-21]` **Volva paraphrase round on `docs/contracts/issues/2.contract.md`.** Volva flagged 5 ambiguities; operator approved amendments to 3 of them. (1) `tags`/`archived`/`name` defaulting semantics now explicit: `tags: list[str]` (default `[]`), `archived: bool` (default `False`), `name: str | None` (default `None`); INV-001/INV-002 + STEPS aligned. (2) `include_archived_query` test tightened: default → URL has NO `include_archived` param at all (was "no param OR explicit false" — softened the assertion against STEP 2's prescriptive behavior). (5) `metadata` populated-vs-defaulted slippage resolved: INV-001 + INV-002 now spell out the defensive `body.get("metadata", {})` default for spec drift tolerance. Volva flags #3 (exception `.body` sensitivity) and #4 (`assert` for runtime validation) reviewed and kept as-is — both intentional and consistent with issue #1's precedent. Drift check still clean (amendments don't touch the pinned issue body). - `[2026-05-21]` **Volva code-vs-contract review round on `ratatoskr.sse_client`.** Volva flagged 4 findings (3 drifts + 1 test-gap), all code-side "fix it" recommendations: (1) `_iter_events` fell off cleanly on EOF before terminal, violating INV-001 ("MUST NOT raise StopAsyncIteration before a terminal event arrives unless connection drops"); fix tracks `terminal_seen` flag and raises `SseConnectionDropped` on clean-EOF-without-terminal. (2) Both `SseConnectFailed.body` and `CancelFailed.body` stored full response bytes; ERROR_ROUTING specified truncation to `[:1024]`; fix truncates in `__init__` before storing. (3) `_parse_sse_id` PRE-001 specified `assert isinstance(raw, str)`, but code called `.split(":")` directly (incidental `AttributeError` on non-str); fix adds the assert. (4) Test-gap on cancel_turn's "other status → CancelFailed" branch; fix adds a 503 test with >1024-byte body that double-covers finding #2. Meta-note: Volva said TDD caught the main happy/adversarial shape; the misses were "negative space" cases (clean EOF, exception payload truncation, untested generic cancel branch) — calibration evidence that cross-model review pulls weight on the same-model author's blind spots. 43 tests GREEN post-fix (42 sse_client + 1 boundary), ruff clean. - `[2026-05-21]` **Volva paraphrase round on `docs/contracts/issues/1.contract.md`.** Volva flagged 5 ambiguities; operator approved amendments to 3 of them. (1) `reconnect_turn` STEP 2 punt resolved: signature now carries `content: str`; STEP 2 body is `json={"content": content}` matching spec §Reconnect flow example verbatim. Spec line 732 makes the agent's tools+LLM run "exactly once regardless of disconnects/reconnects" — the `content` is a wire-schema requirement, not re-processed server-side. (2) `_parse_sse_id` tightened: `turn_id ≥ 1` AND `seq ≥ 1` (was `≥ 0`); spec §SSE id format line 705 explicitly states `seq` starts at 1, and `turn_id` is SQLite autoincrement (≥1). Test `happy_zero_seq` flipped to `zero_seq [adversarial]`; new `zero_turn_id` + `negative_seq` adversarial tests added. (3) INV-003 clarified to spell out the two-entry-point semantics: `stream_turn` establishes `turn_id` from the first event (first event always yields); `reconnect_turn` parses the expected `turn_id` FROM `last_event_id` BEFORE the connection opens, so the first server event is already a flip-candidate and is NOT yielded on mismatch. Volva flags #3 (MalformedSseId-vs-ValueError split) and #5 (exactly-one-terminal as server-assumed) noted but kept as-is — deliberate distinctions. Drift check still clean against issue #1 (amending the contract doesn't touch the pinned issue body). +- `[2026-05-21]` **Issue #3 + contract: `ratatoskr.cli --send` non-interactive stdout presenter.** Composes `create_session` (when `--new`) with `stream_turn` + `cancel_turn` into a one-shot CLI. Hard invariants: no `textual` / `rich` imports (raw stdout — `--raw` is TUI-only per design-brief §6); stdout for `Text` deltas + the post-`Done` newline ONLY; everything else labeled to stderr. Six FN blocks (`main`, `_parse_args`, `_amain`, `_render_event`, `_run_turn`, `_cancel_and_log`). The `_run_turn` race-loop is the load-bearing piece: races `__anext__` against `sigint_event.wait()` so a mid-stream SIGINT lands within one event boundary; once cancel is in flight, the race-loop stops creating new `wait()` tasks (the no-busy-loop fix Volva flagged). 9-bucket exit-code table (0/2/3 terminal; 10/11/12 usage/auth/agent; 20/21/22 server/network/protocol). `prd:` pinned to issue #3 body SHA `206ef51709d43b2c` at `2026-05-21T05:13:29+00:00`; first contract in the repo with a code-level `dependencies:` block (issues #1 and #2). +- `[2026-05-21]` **Volva paraphrase round on `docs/contracts/issues/3.contract.md`.** Volva flagged 5 ambiguities; operator approved amendments to ALL 5 (higher hit rate than #1/#2's 3-of-5 — async + signal-handling has more places for ambiguity to hide). (1) INV-001 wording tightened: "no in-repo modules other than `ratatoskr.sessions` and `ratatoskr.sse_client`" (was "imports from X and Y only" which literally forbade httpx/asyncio/stdlib). (2) INV-002 + `_render_event` POST-002/003 restructured: `Done`'s stdout newline is part of the contract (two stdout cases: Text deltas + post-Done newline), not a contradiction with "no other event writes stdout". (3) `[cancelled] (before any event arrived)` early-exit label added to the Data flow stderr list. (4) SIGINT race-loop pseudocode gated `sigint_task = asyncio.create_task(...)` behind `if not cancelling` — without this gate, once sigint is set, every loop iteration would wake on the already-set event (busy loop). New `no_busy_loop_after_cancel [trace]` test added. (5) `CancelTurnNotFound` + `CancelAlreadyCompleted` added to assumptions import list (they were used in `_cancel_and_log`'s ERROR_ROUTING but missing from the public-surface declaration). Meta-note: Volva said "discipline pulls weight here" — same calibration signal as #1/#2. +- `[2026-05-21]` **`ratatoskr.cli` implemented via TDD against issue #3's contract.** 54 contract-listed tests authored + GREEN per the vertical-slice ordering (`_parse_args` → `_render_event` → `_cancel_and_log` → `_run_turn` → `_amain` → `main`). One in-flight contract amendment during TDD: the `no_busy_loop_after_cancel` test description originally said "exactly ONE wait()-shaped task created" but the natural race-loop shape produces 2 (iter 1 raced with text-event, iter 2 raced with sigint → flipped cancelling=True; iter 3+ skipped). Amended the contract test description to assert "TWO total wait() coroutines" with rationale; the busy-loop check is preserved (iter 3+ MUST skip wait() creation; the bug would grow N unbounded). Implementation choices: (a) `_UsageErrorParser` subclass overrides `argparse.ArgumentParser.error` to raise `_ArgparseError` instead of SystemExit, then `_parse_args` catches and re-raises as `UsageError` per the contract's ERROR_ROUTING; (b) `_GatedStream` test helper (custom `httpx.AsyncByteStream` that pauses on `asyncio.Event` entries) made SIGINT-mid-stream tests deterministic without sleep-based timing — gates release via side-channels (the cancel-mock setting an event when observed); (c) strong-ref `cancel_task` variable in `_run_turn` holds the fire-and-forget cancel task to suppress RUF006 / asyncio GC warning. 117/117 tests GREEN post-implementation; ruff clean. ## Tried and abandoned diff --git a/src/ratatoskr/cli.py b/src/ratatoskr/cli.py index 3ec695b..d7f5567 100644 --- a/src/ratatoskr/cli.py +++ b/src/ratatoskr/cli.py @@ -1,13 +1,305 @@ -"""Ratatoskr CLI entry point — stub. +"""Ratatoskr CLI — non-interactive `--send` stdout presenter. -The dev team implements this. See docs/design-brief.md for the locked -shape (Textual app, `--send` non-interactive mode, `--session`, `--new`, -`--agent`, `--api-key`, `--server-log `, `--raw`, etc.). +Implements docs/contracts/issues/3.contract.md. """ +from __future__ import annotations -def main() -> int: - raise NotImplementedError( - "Ratatoskr CLI is not implemented yet. " - "See docs/design-brief.md for the locked shape." +import argparse +import asyncio +import os +import signal +import sys +from dataclasses import dataclass +from typing import TextIO + +import httpx + +from ratatoskr.sessions import AgentNotFound, SessionApiFailed, create_session +from ratatoskr.sse_client import ( + CancelAlreadyCompleted, + CancelFailed, + Cancelled, + CancelTurnNotFound, + Done, + Error, + Event, + MalformedSseId, + SseConnectFailed, + SseConnectionDropped, + Text, + TextBoundary, + Thinking, + ToolResult, + ToolStart, + TurnIdFlip, + WorkerPhase, + cancel_turn, + stream_turn, +) + + +class UsageError(Exception): + """Raised on argument violations; mapped to exit code 10 by main().""" + + +class _AuthError(Exception): + """Raised when no API key is resolvable; mapped to exit code 11 by main().""" + + +@dataclass(frozen=True) +class ParsedArgs: + """Resolved CLI invocation. Post-validation: exactly one of session_id / new is set.""" + + send_content: str + session_id: str | None + new: bool + agent_id: str | None + api_key: str + server_url: str + + +class _ArgparseError(Exception): + """Internal: re-raise marker so SystemExit from argparse becomes UsageError.""" + + +class _UsageErrorParser(argparse.ArgumentParser): + """ArgumentParser that raises instead of calling sys.exit on parse errors.""" + + def error(self, message: str) -> None: # type: ignore[override] + raise _ArgparseError(message) + + +def _parse_args(argv: list[str] | None) -> ParsedArgs: + """argparse + env-fallback + xor-validation per the contract.""" + parser = _UsageErrorParser(prog="ratatoskr", description="Worldtree CLI presenter.") + parser.add_argument("--send", required=True) + parser.add_argument("--session") + parser.add_argument("--new", action="store_true") + parser.add_argument("--agent") + parser.add_argument("--api-key", dest="api_key") + parser.add_argument("--server") + try: + ns = parser.parse_args(argv) + except _ArgparseError as exc: + raise UsageError(str(exc)) from exc + + if not ns.send: + raise UsageError("--send content must be non-empty") + if ns.session and ns.new: + raise UsageError("--session and --new are mutually exclusive; pass exactly one") + if not ns.session and not ns.new: + raise UsageError("pass exactly one of --session or --new") + if ns.session and ns.agent: + raise UsageError("--agent is required with --new and forbidden with --session") + if ns.new and not ns.agent: + raise UsageError("--agent is required when --new is passed") + + api_key = ns.api_key or os.environ.get("WORLDTREE_API_KEY") or "" + if not api_key: + raise _AuthError("no API key (set --api-key or WORLDTREE_API_KEY)") + + server_url = ns.server or os.environ.get("WORLDTREE_API_URL") or "http://localhost:8000" + + return ParsedArgs( + send_content=ns.send, + session_id=ns.session, + new=ns.new, + agent_id=ns.agent, + api_key=api_key, + server_url=server_url, ) + + +def _render_event(event: Event, *, stdout: TextIO, stderr: TextIO) -> None: + """Pure event-to-output renderer per the contract STEPS table.""" + if isinstance(event, Text): + stdout.write(event.content) + stdout.flush() + elif isinstance(event, Done): + stdout.write("\n") + stdout.flush() + stderr.write( + f"[done] turn_id={event.sse_id.turn_id} model={event.model} " + f"duration_ms={event.duration_ms} usage={event.usage!r}\n" + ) + elif isinstance(event, Error): + stderr.write( + f"[error] turn_id={event.sse_id.turn_id} code={event.error_code} " + f"message={event.message!r}\n" + ) + elif isinstance(event, Cancelled): + stderr.write( + f"[cancelled] turn_id={event.turn_id} reason={event.reason!r} " + f"partial_message_id={event.partial_message_id}\n" + ) + elif isinstance(event, WorkerPhase): + stderr.write(f"[worker_phase] phase={event.phase} turn_id={event.turn_id}\n") + elif isinstance(event, Thinking): + stderr.write(f"[thinking] {event.content[:200]!r}\n") + elif isinstance(event, TextBoundary): + stderr.write(f"[text_boundary] kind={event.kind} char_offset={event.char_offset}\n") + elif isinstance(event, ToolStart): + stderr.write(f"[tool_start] name={event.name} args={event.arguments!r}\n") + elif isinstance(event, ToolResult): + stderr.write( + f"[tool_result] name={event.name} duration_ms={event.duration_ms} " + f"result={event.result!r:.200}\n" + ) + + +async def _cancel_and_log( + client: httpx.AsyncClient, + session_id: str, + turn_id: int, + *, + stderr: TextIO, +) -> None: + """Spawn-and-forget cancel that never raises (INV-009).""" + assert client is not None + assert isinstance(turn_id, int) and turn_id > 0 + try: + await cancel_turn(client, session_id, turn_id) + except (CancelFailed, CancelTurnNotFound, CancelAlreadyCompleted, httpx.RequestError) as exc: + stderr.write(f"[cancel_failed] {type(exc).__name__}: {exc}\n") + + +async def _run_turn( + client: httpx.AsyncClient, + session_id: str, + content: str, + sigint_event: asyncio.Event, + *, + stdout: TextIO, + stderr: TextIO, +) -> int: + """Drive stream_turn, render events, race against sigint_event for mid-stream cancel.""" + assert client is not None + assert session_id and isinstance(session_id, str) + assert content and isinstance(content, str) + assert isinstance(sigint_event, asyncio.Event) + + last_turn_id: int | None = None + cancelling = False + sigint_task: asyncio.Task[bool] | None = None + cancel_task: asyncio.Task[None] | None = None # strong ref to fire-and-forget cancel + aiter_obj = stream_turn(client, session_id, content).__aiter__() + + try: + while True: + next_task = asyncio.create_task(aiter_obj.__anext__()) + if not cancelling: + sigint_task = asyncio.create_task(sigint_event.wait()) + done, _pending = await asyncio.wait( + {next_task, sigint_task}, return_when=asyncio.FIRST_COMPLETED + ) + if sigint_task in done: + if last_turn_id is not None: + cancel_task = asyncio.create_task( + _cancel_and_log(client, session_id, last_turn_id, stderr=stderr) + ) + cancelling = True + else: + next_task.cancel() + stderr.write("[cancelled] (before any event arrived)\n") + return 3 + if next_task not in done: + await next_task + else: + await next_task + try: + event = next_task.result() + except StopAsyncIteration: + stderr.write("[connection_dropped] last_seen=\n") + return 21 + except SseConnectFailed as exc: + stderr.write(f"[sse_connect_failed] status={exc.status} body={exc.body!r}\n") + return 20 + except SseConnectionDropped as exc: + stderr.write(f"[connection_dropped] last_seen={exc.last_seen_sse_id}\n") + return 21 + except MalformedSseId as exc: + stderr.write(f"[malformed_sse_id] raw={exc.raw!r}\n") + return 22 + except TurnIdFlip as exc: + stderr.write(f"[turn_id_flip] expected={exc.established} got={exc.got}\n") + return 22 + last_turn_id = event.sse_id.turn_id + _render_event(event, stdout=stdout, stderr=stderr) + if isinstance(event, Done): + if sigint_task is not None and not cancelling: + sigint_task.cancel() + return 0 + if isinstance(event, Error): + if sigint_task is not None and not cancelling: + sigint_task.cancel() + return 2 + if isinstance(event, Cancelled): + if sigint_task is not None and not cancelling: + sigint_task.cancel() + return 3 + finally: + if sigint_task is not None and not sigint_task.done(): + sigint_task.cancel() + # cancel_task is the strong-ref holder for the fire-and-forget cancel POST (RUF006). + # By the time we reach `finally`, the stream has drained to a terminal event, so the + # cancel POST has either completed or returned an error (swallowed by _cancel_and_log). + # Awaiting it here would block on a guaranteed-done task; the reference alone suffices. + _ = cancel_task + + +async def _amain(args: ParsedArgs) -> int: + """Async orchestrator: create-session (if --new) → SIGINT install → _run_turn → cleanup.""" + assert isinstance(args, ParsedArgs) + async with httpx.AsyncClient( + base_url=args.server_url, + headers={"Authorization": f"Bearer {args.api_key}"}, + ) as client: + if args.new: + assert args.agent_id is not None + try: + info = await create_session(client, args.agent_id) + except AgentNotFound as exc: + sys.stderr.write(f"[agent_not_found] agent_id={exc.agent_id}\n") + return 12 + except SessionApiFailed as exc: + sys.stderr.write(f"[session_api_failed] status={exc.status} body={exc.body!r}\n") + return 20 + except (httpx.ConnectError, httpx.ReadTimeout, httpx.TransportError) as exc: + sys.stderr.write(f"[network_error] {type(exc).__name__}: {exc}\n") + return 21 + sys.stderr.write( + f"[create_session] session_id={info.session_id} agent_id={info.agent_id}\n" + ) + session_id = info.session_id + else: + assert args.session_id is not None + session_id = args.session_id + + sigint_event = asyncio.Event() + loop = asyncio.get_running_loop() + loop.add_signal_handler(signal.SIGINT, sigint_event.set) + try: + return await _run_turn( + client, + session_id, + args.send_content, + sigint_event, + stdout=sys.stdout, + stderr=sys.stderr, + ) + finally: + loop.remove_signal_handler(signal.SIGINT) + + +def main(argv: list[str] | None = None) -> int: + """Sync entry point. Maps UsageError/_AuthError to exit codes BEFORE the event loop.""" + try: + args = _parse_args(argv) + except UsageError as exc: + sys.stderr.write(f"[usage_error] {exc}\n") + return 10 + except _AuthError as exc: + sys.stderr.write(f"[auth_error] {exc}\n") + return 11 + return asyncio.run(_amain(args)) diff --git a/tests/test_cli.py b/tests/test_cli.py new file mode 100644 index 0000000..a5c7f82 --- /dev/null +++ b/tests/test_cli.py @@ -0,0 +1,959 @@ +"""Tests for ratatoskr.cli per docs/contracts/issues/3.contract.md.""" + +import asyncio +import io +import json + +import httpx +import pytest +import respx + +from ratatoskr import cli as cli_mod +from ratatoskr.cli import ( + ParsedArgs, + UsageError, + _amain, + _AuthError, + _cancel_and_log, + _parse_args, + _render_event, + _run_turn, + main, +) +from ratatoskr.sse_client import ( + Cancelled, + Done, + Error, + SseId, + Text, + TextBoundary, + Thinking, + ToolResult, + ToolStart, + WorkerPhase, +) + + +def _sse_chunk(sse_id: str, body: dict[str, object]) -> bytes: + return f"id: {sse_id}\ndata: {json.dumps(body)}\n\n".encode() + + +def _sse_resp(body: bytes | httpx.AsyncByteStream) -> httpx.Response: + """Wrap an SSE response body (bytes or stream) with the right content-type.""" + headers = {"content-type": "text/event-stream"} + if isinstance(body, bytes): + return httpx.Response(200, headers=headers, content=body) + return httpx.Response(200, headers=headers, stream=body) + + +_DONE_BODY = { + "type": "done", + "phase": "succeeded", + "response": "hello", + "model": "m", + "duration_ms": 1, + "usage": { + "prompt_tokens": 0, + "completion_tokens": 0, + "total_tokens": 0, + "cached_input_tokens": 0, + }, +} +_CANCELLED_BODY = { + "type": "cancelled", + "phase": "cancelled", + "turn_id": 42, + "reason": "user_cancel", + "partial_message_id": None, +} +_CANCEL_OK_RESP = {"turn_id": 42, "cancelled": True, "reason": None, "partial_message_id": None} + + +@pytest.fixture(autouse=True) +def _clear_env(monkeypatch: pytest.MonkeyPatch) -> None: + """Tests assert env-resolution behavior; default to a clean slate per test.""" + monkeypatch.delenv("WORLDTREE_API_KEY", raising=False) + monkeypatch.delenv("WORLDTREE_API_URL", raising=False) + + +class TestParseArgs: + def test_happy_new(self) -> None: + """happy_new [happy,tracer]: --send --new --agent --api-key → full ParsedArgs.""" + args = _parse_args(["--send", "hi", "--new", "--agent", "mimir", "--api-key", "k"]) + assert args == ParsedArgs( + send_content="hi", + session_id=None, + new=True, + agent_id="mimir", + api_key="k", + server_url="http://localhost:8000", + ) + + def test_happy_existing_session(self) -> None: + """happy_existing_session: --send --session --api-key → ParsedArgs with session_id.""" + args = _parse_args(["--send", "hi", "--session", "s-1", "--api-key", "k"]) + assert args == ParsedArgs( + send_content="hi", + session_id="s-1", + new=False, + agent_id=None, + api_key="k", + server_url="http://localhost:8000", + ) + + def test_api_key_from_env(self, monkeypatch: pytest.MonkeyPatch) -> None: + """api_key_from_env: WORLDTREE_API_KEY env var fills in when --api-key omitted.""" + monkeypatch.setenv("WORLDTREE_API_KEY", "from-env") + args = _parse_args(["--send", "hi", "--new", "--agent", "mimir"]) + assert args.api_key == "from-env" + + def test_api_key_flag_beats_env(self, monkeypatch: pytest.MonkeyPatch) -> None: + """api_key_flag_beats_env: explicit --api-key wins over WORLDTREE_API_KEY.""" + monkeypatch.setenv("WORLDTREE_API_KEY", "env") + args = _parse_args(["--send", "hi", "--new", "--agent", "m", "--api-key", "flag"]) + assert args.api_key == "flag" + + def test_server_default(self) -> None: + """server_default: no --server, no WORLDTREE_API_URL → http://localhost:8000.""" + args = _parse_args(["--send", "hi", "--new", "--agent", "m", "--api-key", "k"]) + assert args.server_url == "http://localhost:8000" + + def test_server_env_fallback(self, monkeypatch: pytest.MonkeyPatch) -> None: + """server_env_fallback: WORLDTREE_API_URL fills in when --server omitted.""" + monkeypatch.setenv("WORLDTREE_API_URL", "http://t.local:9000") + args = _parse_args(["--send", "hi", "--new", "--agent", "m", "--api-key", "k"]) + assert args.server_url == "http://t.local:9000" + + def test_server_flag_beats_env(self, monkeypatch: pytest.MonkeyPatch) -> None: + """server_flag_beats_env: explicit --server wins over WORLDTREE_API_URL.""" + monkeypatch.setenv("WORLDTREE_API_URL", "env") + args = _parse_args( + ["--send", "hi", "--new", "--agent", "m", "--api-key", "k", "--server", "flag"] + ) + assert args.server_url == "flag" + + def test_usage_no_send(self) -> None: + """usage_no_send: missing --send → UsageError (argparse required-flag).""" + with pytest.raises(UsageError): + _parse_args(["--new", "--agent", "mimir", "--api-key", "k"]) + + def test_usage_both_session_and_new(self) -> None: + """usage_both_session_and_new: --session AND --new → UsageError('mutually exclusive').""" + with pytest.raises(UsageError, match="mutually exclusive"): + _parse_args( + ["--send", "hi", "--session", "s", "--new", "--agent", "m", "--api-key", "k"] + ) + + def test_usage_neither_session_nor_new(self) -> None: + """usage_neither_session_nor_new: neither flag → UsageError('pass exactly one').""" + with pytest.raises(UsageError, match="pass exactly one"): + _parse_args(["--send", "hi", "--api-key", "k"]) + + def test_usage_new_without_agent(self) -> None: + """usage_new_without_agent: --new without --agent → UsageError.""" + with pytest.raises(UsageError, match="--agent is required when --new"): + _parse_args(["--send", "hi", "--new", "--api-key", "k"]) + + def test_usage_session_with_agent(self) -> None: + """usage_session_with_agent: --session AND --agent → UsageError.""" + with pytest.raises(UsageError, match="forbidden with --session"): + _parse_args(["--send", "hi", "--session", "s-1", "--agent", "x", "--api-key", "k"]) + + def test_auth_missing(self) -> None: + """auth_missing: no --api-key and no env → _AuthError.""" + with pytest.raises(_AuthError, match="no API key"): + _parse_args(["--send", "hi", "--new", "--agent", "mimir"]) + + def test_empty_send(self) -> None: + """empty_send: --send '' → UsageError (non-empty enforced).""" + with pytest.raises(UsageError): + _parse_args(["--send", "", "--new", "--agent", "x", "--api-key", "k"]) + + +SID = SseId(42, 5) + + +class TestRenderEvent: + def test_text_to_stdout_only(self) -> None: + """text_to_stdout_only [happy,tracer]: Text → stdout=="hello"; stderr empty; flushed.""" + stdout = io.StringIO() + stderr = io.StringIO() + _render_event(Text(sse_id=SID, content="hello"), stdout=stdout, stderr=stderr) + assert stdout.getvalue() == "hello" + assert stderr.getvalue() == "" + + def test_done_writes_newline_and_label(self) -> None: + """done_writes_newline_and_label: stdout=="\\n"; stderr "[done]" + turn_id + model.""" + stdout = io.StringIO() + stderr = io.StringIO() + evt = Done( + sse_id=SID, + phase="completed", + response="hi", + model="glm5-turbo", + duration_ms=1234, + usage={"prompt": 10, "completion": 5}, + ) + _render_event(evt, stdout=stdout, stderr=stderr) + assert stdout.getvalue() == "\n" + out_err = stderr.getvalue() + assert out_err.startswith("[done]") + assert "turn_id=42" in out_err + assert "model=glm5-turbo" in out_err + assert "duration_ms=1234" in out_err + + def test_error_to_stderr_only(self) -> None: + """error_to_stderr_only: Error → stderr "[error]" with code; stdout empty.""" + stdout = io.StringIO() + stderr = io.StringIO() + evt = Error(sse_id=SID, phase="failed", message="boom", error_code="llm_output_invalid") + _render_event(evt, stdout=stdout, stderr=stderr) + assert stdout.getvalue() == "" + out_err = stderr.getvalue() + assert out_err.startswith("[error]") + assert "turn_id=42" in out_err + assert "code=llm_output_invalid" in out_err + + def test_cancelled_to_stderr_only(self) -> None: + """cancelled_to_stderr_only: Cancelled → stderr "[cancelled]" + reason + partial id.""" + stdout = io.StringIO() + stderr = io.StringIO() + evt = Cancelled( + sse_id=SID, phase="cancelled", turn_id=42, reason="user", partial_message_id=7 + ) + _render_event(evt, stdout=stdout, stderr=stderr) + assert stdout.getvalue() == "" + out_err = stderr.getvalue() + assert out_err.startswith("[cancelled]") + assert "reason='user'" in out_err + assert "partial_message_id=7" in out_err + + def test_worker_phase_to_stderr(self) -> None: + """worker_phase_to_stderr: WorkerPhase → stderr "[worker_phase]"; stdout empty.""" + stdout = io.StringIO() + stderr = io.StringIO() + evt = WorkerPhase(sse_id=SID, phase="streaming", turn_id=42) + _render_event(evt, stdout=stdout, stderr=stderr) + assert stdout.getvalue() == "" + assert stderr.getvalue().startswith("[worker_phase]") + + def test_thinking_truncated(self) -> None: + """thinking_truncated [trace]: …""" + stdout = io.StringIO() + stderr = io.StringIO() + _render_event(Thinking(sse_id=SID, content="a" * 500), stdout=stdout, stderr=stderr) + out_err = stderr.getvalue() + assert out_err.startswith("[thinking]") + assert "a" * 500 not in out_err + assert "a" * 200 in out_err + + def test_tool_start_to_stderr(self) -> None: + """tool_start_to_stderr: ToolStart → stderr "[tool_start] name=... args=...".""" + stdout = io.StringIO() + stderr = io.StringIO() + evt = ToolStart(sse_id=SID, name="read_file", arguments={"path": "/x"}) + _render_event(evt, stdout=stdout, stderr=stderr) + assert stdout.getvalue() == "" + out_err = stderr.getvalue() + assert out_err.startswith("[tool_start] name=read_file args=") + + def test_tool_result_truncated(self) -> None: + """tool_result_truncated [trace]: ToolResult.result repr truncated to ≤200 chars.""" + stdout = io.StringIO() + stderr = io.StringIO() + evt = ToolResult(sse_id=SID, name="x", result="b" * 500, duration_ms=42) + _render_event(evt, stdout=stdout, stderr=stderr) + out_err = stderr.getvalue() + assert out_err.startswith("[tool_result]") + # the contract uses `{event.result!r:.200}` — 200 chars max of repr output + assert "b" * 500 not in out_err + + def test_text_boundary_to_stderr(self) -> None: + """text_boundary_to_stderr: TextBoundary → stderr "[text_boundary]"; stdout empty.""" + stdout = io.StringIO() + stderr = io.StringIO() + evt = TextBoundary(sse_id=SID, kind="sentence", char_offset=128, ts="2026-05-21T00:00:00Z") + _render_event(evt, stdout=stdout, stderr=stderr) + assert stdout.getvalue() == "" + out_err = stderr.getvalue() + assert out_err.startswith("[text_boundary]") + assert "kind=sentence" in out_err + assert "char_offset=128" in out_err + + def test_invariant_inv003_stderr_only(self) -> None: + """invariant_inv003_stderr_only [scenario]: …""" + for evt in [ + WorkerPhase(sse_id=SID, phase="x", turn_id=42), + Thinking(sse_id=SID, content="x"), + TextBoundary(sse_id=SID, kind="x", char_offset=0, ts="t"), + ToolStart(sse_id=SID, name="x", arguments={}), + ToolResult(sse_id=SID, name="x", result=None, duration_ms=0), + Error(sse_id=SID, phase="failed", message="m", error_code="e"), + Cancelled( + sse_id=SID, phase="cancelled", turn_id=42, reason="r", partial_message_id=None + ), + ]: + stdout = io.StringIO() + stderr = io.StringIO() + _render_event(evt, stdout=stdout, stderr=stderr) + assert stdout.getvalue() == "", f"INV-002 violated for {type(evt).__name__}" + + +class TestCancelAndLog: + @respx.mock + async def test_happy_cancel(self) -> None: + """happy_cancel [happy,tracer]: 200 OK → returns None; stderr empty.""" + respx.post("https://w.example/sessions/s-1/turns/42/cancel").mock( + return_value=httpx.Response( + 200, + json={"turn_id": 42, "cancelled": True, "reason": None, "partial_message_id": None}, + ) + ) + stderr = io.StringIO() + async with httpx.AsyncClient(base_url="https://w.example") as client: + result = await _cancel_and_log(client, "s-1", 42, stderr=stderr) + assert result is None + assert stderr.getvalue() == "" + + @respx.mock + async def test_cancel_failed_500(self) -> None: + """cancel_failed_500 [error]: …""" + respx.post("https://w.example/sessions/s-1/turns/42/cancel").mock( + return_value=httpx.Response(500, content=b"boom") + ) + stderr = io.StringIO() + async with httpx.AsyncClient(base_url="https://w.example") as client: + await _cancel_and_log(client, "s-1", 42, stderr=stderr) + out = stderr.getvalue() + assert "[cancel_failed]" in out + assert "CancelFailed" in out + + @respx.mock + async def test_cancel_already_completed(self) -> None: + """cancel_already_completed [scenario]: …""" + respx.post("https://w.example/sessions/s-1/turns/42/cancel").mock( + return_value=httpx.Response(409) + ) + stderr = io.StringIO() + async with httpx.AsyncClient(base_url="https://w.example") as client: + await _cancel_and_log(client, "s-1", 42, stderr=stderr) + out = stderr.getvalue() + assert "[cancel_failed]" in out + assert "CancelAlreadyCompleted" in out + + @respx.mock + async def test_cancel_turn_not_found(self) -> None: + """cancel_turn_not_found [scenario]: 404 → returns None; stderr CancelTurnNotFound.""" + respx.post("https://w.example/sessions/s-1/turns/42/cancel").mock( + return_value=httpx.Response(404) + ) + stderr = io.StringIO() + async with httpx.AsyncClient(base_url="https://w.example") as client: + await _cancel_and_log(client, "s-1", 42, stderr=stderr) + out = stderr.getvalue() + assert "[cancel_failed]" in out + assert "CancelTurnNotFound" in out + + @respx.mock + async def test_transport_error_swallowed(self) -> None: + """transport_error_swallowed [error]: …""" + respx.post("https://w.example/sessions/s-1/turns/42/cancel").mock( + side_effect=httpx.ConnectError("network down") + ) + stderr = io.StringIO() + async with httpx.AsyncClient(base_url="https://w.example") as client: + await _cancel_and_log(client, "s-1", 42, stderr=stderr) + out = stderr.getvalue() + assert "[cancel_failed]" in out + assert "ConnectError" in out + + +class _GatedStream(httpx.AsyncByteStream): + """SSE byte stream: list of (bytes | asyncio.Event); Event entries pause until set.""" + + def __init__(self, items: list[bytes | asyncio.Event]) -> None: + self._items = items + + async def __aiter__(self): # type: ignore[no-untyped-def] + for item in self._items: + if isinstance(item, asyncio.Event): + await item.wait() + else: + yield item + + async def aclose(self) -> None: + return None + + +class TestRunTurn: + @respx.mock + async def test_happy_text_then_done(self) -> None: + """happy_text_then_done [happy,tracer]: …""" + stream = _sse_chunk( + "42:1", {"type": "text", "content": "hello"} + ) + _sse_chunk("42:2", _DONE_BODY) + respx.post("https://w.example/sessions/s-1/messages").mock( + return_value=_sse_resp(stream) + ) + sigint = asyncio.Event() + stdout = io.StringIO() + stderr = io.StringIO() + async with httpx.AsyncClient(base_url="https://w.example") as client: + exit_code = await _run_turn(client, "s-1", "hi", sigint, stdout=stdout, stderr=stderr) + assert exit_code == 0 + assert stdout.getvalue() == "hello\n" + assert "[done]" in stderr.getvalue() + + @respx.mock + async def test_error_terminal(self) -> None: + """error_terminal: text + error → exit 2; stderr has [error].""" + stream = _sse_chunk("42:1", {"type": "text", "content": "x"}) + _sse_chunk( + "42:2", + { + "type": "error", + "phase": "failed", + "error_code": "llm_output_invalid", + "message": "boom", + }, + ) + respx.post("https://w.example/sessions/s-1/messages").mock( + return_value=_sse_resp(stream) + ) + sigint = asyncio.Event() + stdout, stderr = io.StringIO(), io.StringIO() + async with httpx.AsyncClient(base_url="https://w.example") as client: + exit_code = await _run_turn(client, "s-1", "hi", sigint, stdout=stdout, stderr=stderr) + assert exit_code == 2 + assert "[error]" in stderr.getvalue() + + @respx.mock + async def test_cancelled_terminal_server(self) -> None: + """cancelled_terminal_server: text + cancelled → exit 3; stderr has [cancelled].""" + stream = _sse_chunk("42:1", {"type": "text", "content": "x"}) + _sse_chunk( + "42:2", _CANCELLED_BODY + ) + respx.post("https://w.example/sessions/s-1/messages").mock( + return_value=_sse_resp(stream) + ) + sigint = asyncio.Event() + stdout, stderr = io.StringIO(), io.StringIO() + async with httpx.AsyncClient(base_url="https://w.example") as client: + exit_code = await _run_turn(client, "s-1", "hi", sigint, stdout=stdout, stderr=stderr) + assert exit_code == 3 + assert "[cancelled]" in stderr.getvalue() + + @respx.mock + async def test_sse_connect_failed_404(self) -> None: + """sse_connect_failed_404 [error]: 404 → exit 20; stderr [sse_connect_failed] status=404.""" + respx.post("https://w.example/sessions/missing/messages").mock( + return_value=httpx.Response(404, json={"error": "session_not_found"}) + ) + sigint = asyncio.Event() + stdout, stderr = io.StringIO(), io.StringIO() + async with httpx.AsyncClient(base_url="https://w.example") as client: + exit_code = await _run_turn( + client, "missing", "hi", sigint, stdout=stdout, stderr=stderr + ) + assert exit_code == 20 + out = stderr.getvalue() + assert "[sse_connect_failed]" in out + assert "status=404" in out + + @respx.mock + async def test_connection_dropped(self) -> None: + """connection_dropped [error]: RemoteProtocolError mid-stream → exit 21.""" + + class _DropAfter(httpx.AsyncByteStream): + def __init__(self, chunks: list[bytes]) -> None: + self._chunks = chunks + + async def __aiter__(self): # type: ignore[no-untyped-def] + for c in self._chunks: + yield c + raise httpx.RemoteProtocolError("simulated mid-stream drop") + + async def aclose(self) -> None: + return None + + first = _sse_chunk("42:1", {"type": "text", "content": "x"}) + respx.post("https://w.example/sessions/s-1/messages").mock( + return_value=httpx.Response( + 200, headers={"content-type": "text/event-stream"}, stream=_DropAfter([first]) + ) + ) + sigint = asyncio.Event() + stdout, stderr = io.StringIO(), io.StringIO() + async with httpx.AsyncClient(base_url="https://w.example") as client: + exit_code = await _run_turn(client, "s-1", "hi", sigint, stdout=stdout, stderr=stderr) + assert exit_code == 21 + assert "[connection_dropped]" in stderr.getvalue() + + @respx.mock + async def test_malformed_sse_id(self) -> None: + """malformed_sse_id [error]: id without seq → exit 22; stderr [malformed_sse_id].""" + stream = b"id: 42\ndata: {\"type\": \"text\", \"content\": \"x\"}\n\n" + respx.post("https://w.example/sessions/s-1/messages").mock( + return_value=_sse_resp(stream) + ) + sigint = asyncio.Event() + stdout, stderr = io.StringIO(), io.StringIO() + async with httpx.AsyncClient(base_url="https://w.example") as client: + exit_code = await _run_turn(client, "s-1", "hi", sigint, stdout=stdout, stderr=stderr) + assert exit_code == 22 + assert "[malformed_sse_id]" in stderr.getvalue() + + @respx.mock + async def test_turn_id_flip(self) -> None: + """turn_id_flip [error]: …""" + stream = _sse_chunk("42:1", {"type": "text", "content": "x"}) + _sse_chunk( + "99:2", {"type": "text", "content": "y"} + ) + respx.post("https://w.example/sessions/s-1/messages").mock( + return_value=_sse_resp(stream) + ) + sigint = asyncio.Event() + stdout, stderr = io.StringIO(), io.StringIO() + async with httpx.AsyncClient(base_url="https://w.example") as client: + exit_code = await _run_turn(client, "s-1", "hi", sigint, stdout=stdout, stderr=stderr) + assert exit_code == 22 + out = stderr.getvalue() + assert "[turn_id_flip]" in out + assert "expected=42" in out + assert "got=99" in out + + @respx.mock + async def test_sigint_before_first_event(self) -> None: + """sigint_before_first_event [scenario]: …""" + gate = asyncio.Event() + # Stream never yields anything until gate (the gate is never set; the test exits via sigint) + stream = _GatedStream([gate]) + respx.post("https://w.example/sessions/s-1/messages").mock( + return_value=_sse_resp(stream) + ) + cancel_route = respx.post("https://w.example/sessions/s-1/turns").mock( + return_value=httpx.Response(200, json=_CANCEL_OK_RESP) + ) + sigint = asyncio.Event() + sigint.set() # SIGINT before _run_turn even starts + stdout, stderr = io.StringIO(), io.StringIO() + async with httpx.AsyncClient(base_url="https://w.example") as client: + exit_code = await asyncio.wait_for( + _run_turn(client, "s-1", "hi", sigint, stdout=stdout, stderr=stderr), timeout=2.0 + ) + assert exit_code == 3 + assert "[cancelled] (before any event arrived)" in stderr.getvalue() + assert cancel_route.call_count == 0 + + @respx.mock + async def test_sigint_mid_stream_drains_to_cancelled(self) -> None: + """sigint_mid_stream_drains_to_cancelled [scenario,tracer]: …""" + cancel_observed = asyncio.Event() + + def cancel_handler(req: httpx.Request) -> httpx.Response: + cancel_observed.set() + return httpx.Response(200, json=_CANCEL_OK_RESP) + + cancel_route = respx.post("https://w.example/sessions/s-1/turns/42/cancel").mock( + side_effect=cancel_handler + ) + text_chunk = _sse_chunk("42:1", {"type": "text", "content": "x"}) + cancelled_chunk = _sse_chunk("42:2", _CANCELLED_BODY) + stream = _GatedStream([text_chunk, cancel_observed, cancelled_chunk]) + respx.post("https://w.example/sessions/s-1/messages").mock( + return_value=_sse_resp(stream) + ) + + sigint = asyncio.Event() + stdout, stderr = io.StringIO(), io.StringIO() + async with httpx.AsyncClient(base_url="https://w.example") as client: + task = asyncio.create_task( + _run_turn(client, "s-1", "hi", sigint, stdout=stdout, stderr=stderr) + ) + # Wait for the first event to flush to stdout (signals last_turn_id is set) + for _ in range(50): + if "x" in stdout.getvalue(): + break + await asyncio.sleep(0.01) + else: + task.cancel() + pytest.fail("text event never reached stdout") + sigint.set() + exit_code = await asyncio.wait_for(task, timeout=2.0) + assert exit_code == 3 + assert cancel_route.call_count == 1 + + @respx.mock + async def test_sigint_twice_issues_one_cancel(self) -> None: + """sigint_twice_issues_one_cancel [scenario]: sigint set twice → one cancel POST.""" + cancel_observed = asyncio.Event() + + def cancel_handler(req: httpx.Request) -> httpx.Response: + cancel_observed.set() + return httpx.Response(200, json=_CANCEL_OK_RESP) + + cancel_route = respx.post("https://w.example/sessions/s-1/turns/42/cancel").mock( + side_effect=cancel_handler + ) + text_chunk = _sse_chunk("42:1", {"type": "text", "content": "x"}) + cancelled_chunk = _sse_chunk("42:2", _CANCELLED_BODY) + stream = _GatedStream([text_chunk, cancel_observed, cancelled_chunk]) + respx.post("https://w.example/sessions/s-1/messages").mock( + return_value=_sse_resp(stream) + ) + + sigint = asyncio.Event() + stdout, stderr = io.StringIO(), io.StringIO() + async with httpx.AsyncClient(base_url="https://w.example") as client: + task = asyncio.create_task( + _run_turn(client, "s-1", "hi", sigint, stdout=stdout, stderr=stderr) + ) + for _ in range(50): + if "x" in stdout.getvalue(): + break + await asyncio.sleep(0.01) + sigint.set() + # Set again — should be no-op (event is already set; idempotent) + sigint.set() + exit_code = await asyncio.wait_for(task, timeout=2.0) + assert exit_code == 3 + assert cancel_route.call_count == 1 + + @respx.mock + async def test_no_busy_loop_after_cancel(self, monkeypatch: pytest.MonkeyPatch) -> None: + """no_busy_loop_after_cancel [trace]: only one sigint_event.wait()-task created.""" + cancel_observed = asyncio.Event() + + def cancel_handler(req: httpx.Request) -> httpx.Response: + cancel_observed.set() + return httpx.Response(200, json=_CANCEL_OK_RESP) + + respx.post("https://w.example/sessions/s-1/turns/42/cancel").mock( + side_effect=cancel_handler + ) + text_chunk = _sse_chunk("42:1", {"type": "text", "content": "x"}) + cancelled_chunk = _sse_chunk("42:2", _CANCELLED_BODY) + stream = _GatedStream([text_chunk, cancel_observed, cancelled_chunk]) + respx.post("https://w.example/sessions/s-1/messages").mock( + return_value=_sse_resp(stream) + ) + + sigint = asyncio.Event() + wait_call_count = 0 + original_wait = sigint.wait + + async def counting_wait() -> bool: + nonlocal wait_call_count + wait_call_count += 1 + return await original_wait() + + monkeypatch.setattr(sigint, "wait", counting_wait) + + stdout, stderr = io.StringIO(), io.StringIO() + async with httpx.AsyncClient(base_url="https://w.example") as client: + task = asyncio.create_task( + _run_turn(client, "s-1", "hi", sigint, stdout=stdout, stderr=stderr) + ) + for _ in range(50): + if "x" in stdout.getvalue(): + break + await asyncio.sleep(0.01) + sigint.set() + exit_code = await asyncio.wait_for(task, timeout=2.0) + assert exit_code == 3 + # INV-007 + busy-loop fix: sigint.wait() created at most once per pre-cancelling + # iteration. For text → sigint → cancelled, that's iter 1 (raced w/ text) and + # iter 2 (raced w/ sigint; flipped cancelling=True). Iter 3+ MUST skip wait() + # creation entirely — the busy-loop bug would make this number grow unbounded. + assert wait_call_count == 2 + + @respx.mock + async def test_cancel_failed_drains_anyway(self) -> None: + """cancel_failed_drains_anyway [scenario]: …""" + cancel_observed = asyncio.Event() + + def cancel_handler(req: httpx.Request) -> httpx.Response: + cancel_observed.set() + return httpx.Response(500, content=b"boom") + + respx.post("https://w.example/sessions/s-1/turns/42/cancel").mock( + side_effect=cancel_handler + ) + text_chunk = _sse_chunk("42:1", {"type": "text", "content": "x"}) + cancelled_chunk = _sse_chunk("42:2", _CANCELLED_BODY) + stream = _GatedStream([text_chunk, cancel_observed, cancelled_chunk]) + respx.post("https://w.example/sessions/s-1/messages").mock( + return_value=_sse_resp(stream) + ) + + sigint = asyncio.Event() + stdout, stderr = io.StringIO(), io.StringIO() + async with httpx.AsyncClient(base_url="https://w.example") as client: + task = asyncio.create_task( + _run_turn(client, "s-1", "hi", sigint, stdout=stdout, stderr=stderr) + ) + for _ in range(50): + if "x" in stdout.getvalue(): + break + await asyncio.sleep(0.01) + sigint.set() + exit_code = await asyncio.wait_for(task, timeout=2.0) + assert exit_code == 3 + # INV-009: cancel POST failed but stream still drained to cancelled terminal + assert "[cancel_failed]" in stderr.getvalue() + assert "[cancelled]" in stderr.getvalue() + + @respx.mock + async def test_render_called_once_per_event(self, monkeypatch: pytest.MonkeyPatch) -> None: + """render_called_once_per_event [trace]: spy → _render_event call_count == event count.""" + chunks = [ + _sse_chunk("42:1", {"type": "worker_phase", "phase": "streaming", "turn_id": 42}), + _sse_chunk("42:2", {"type": "text", "content": "hi"}), + _sse_chunk("42:3", _DONE_BODY), + ] + respx.post("https://w.example/sessions/s-1/messages").mock( + return_value=httpx.Response( + 200, headers={"content-type": "text/event-stream"}, content=b"".join(chunks) + ) + ) + + call_count = 0 + from ratatoskr import cli as cli_mod + + original = cli_mod._render_event + + def spy(event, **kw): # type: ignore[no-untyped-def] + nonlocal call_count + call_count += 1 + return original(event, **kw) + + monkeypatch.setattr(cli_mod, "_render_event", spy) + + sigint = asyncio.Event() + stdout, stderr = io.StringIO(), io.StringIO() + async with httpx.AsyncClient(base_url="https://w.example") as client: + exit_code = await _run_turn(client, "s-1", "hi", sigint, stdout=stdout, stderr=stderr) + assert exit_code == 0 + assert call_count == 3 + + +_PARSED_NEW = ParsedArgs( + send_content="hi", + session_id=None, + new=True, + agent_id="mimir", + api_key="k", + server_url="https://w.example", +) +_PARSED_EXISTING = ParsedArgs( + send_content="hi", + session_id="s-1", + new=False, + agent_id=None, + api_key="k", + server_url="https://w.example", +) +_CREATE_OK_RESP = { + "session_id": "s-new", + "agent_id": "mimir", + "message_count": 0, + "created_at": "2026-05-21T00:00:00+00:00", + "last_active": "2026-05-21T00:00:00+00:00", + "metadata": {}, +} + + +class TestAmain: + @respx.mock + async def test_happy_new_session_then_stream(self, capsys: pytest.CaptureFixture[str]) -> None: + """happy_new_session_then_stream [happy,tracer]: …""" + respx.post("https://w.example/sessions").mock( + return_value=httpx.Response(201, json=_CREATE_OK_RESP) + ) + sse_body = _sse_chunk("42:1", {"type": "text", "content": "hello"}) + _sse_chunk( + "42:2", _DONE_BODY + ) + respx.post("https://w.example/sessions/s-new/messages").mock( + return_value=_sse_resp(sse_body) + ) + exit_code = await _amain(_PARSED_NEW) + assert exit_code == 0 + captured = capsys.readouterr() + err = captured.err + assert "[create_session]" in err + assert "[done]" in err + assert err.index("[create_session]") < err.index("[done]") + + @respx.mock + async def test_happy_existing_session(self, capsys: pytest.CaptureFixture[str]) -> None: + """happy_existing_session: --session, no create POST; just SSE stream → exit 0.""" + sessions_route = respx.post("https://w.example/sessions").mock( + return_value=httpx.Response(201, json=_CREATE_OK_RESP) + ) + sse_body = _sse_chunk("42:1", {"type": "text", "content": "hi"}) + _sse_chunk( + "42:2", _DONE_BODY + ) + respx.post("https://w.example/sessions/s-1/messages").mock( + return_value=_sse_resp(sse_body) + ) + exit_code = await _amain(_PARSED_EXISTING) + assert exit_code == 0 + assert sessions_route.call_count == 0 + + @respx.mock + async def test_agent_not_found_exits_12(self, capsys: pytest.CaptureFixture[str]) -> None: + """agent_not_found_exits_12 [error]: POST /sessions → 404 → exit 12; no stream_turn.""" + respx.post("https://w.example/sessions").mock( + return_value=httpx.Response(404, json={"error": "unknown_agent_id"}) + ) + stream_route = respx.post("https://w.example/sessions/s-new/messages").mock( + return_value=httpx.Response(200) + ) + exit_code = await _amain(_PARSED_NEW) + assert exit_code == 12 + assert "[agent_not_found]" in capsys.readouterr().err + assert stream_route.call_count == 0 + + @respx.mock + async def test_session_api_failed_exits_20(self, capsys: pytest.CaptureFixture[str]) -> None: + """session_api_failed_exits_20: POST /sessions → 500 → exit 20; [session_api_failed].""" + respx.post("https://w.example/sessions").mock( + return_value=httpx.Response(500, content=b"server error") + ) + exit_code = await _amain(_PARSED_NEW) + assert exit_code == 20 + err = capsys.readouterr().err + assert "[session_api_failed]" in err + assert "status=500" in err + + @respx.mock + async def test_connect_error_exits_21(self, capsys: pytest.CaptureFixture[str]) -> None: + """connect_error_exits_21: httpx.ConnectError on POST /sessions → exit 21.""" + respx.post("https://w.example/sessions").mock(side_effect=httpx.ConnectError("down")) + exit_code = await _amain(_PARSED_NEW) + assert exit_code == 21 + assert "[network_error]" in capsys.readouterr().err + + @respx.mock + async def test_sigint_handler_installed_and_removed(self) -> None: + """sigint_handler_installed_and_removed [trace]: signal handler add/remove paired.""" + sse_body = _sse_chunk("42:1", {"type": "text", "content": "x"}) + _sse_chunk( + "42:2", _DONE_BODY + ) + respx.post("https://w.example/sessions/s-1/messages").mock( + return_value=_sse_resp(sse_body) + ) + loop = asyncio.get_running_loop() + original_add = loop.add_signal_handler + original_remove = loop.remove_signal_handler + add_calls: list[int] = [] + remove_calls: list[int] = [] + + def add_spy(sig, callback, *args): # type: ignore[no-untyped-def] + add_calls.append(sig) + return original_add(sig, callback, *args) + + def remove_spy(sig): # type: ignore[no-untyped-def] + remove_calls.append(sig) + return original_remove(sig) + + loop.add_signal_handler = add_spy # type: ignore[method-assign] + loop.remove_signal_handler = remove_spy # type: ignore[method-assign] + try: + exit_code = await _amain(_PARSED_EXISTING) + finally: + loop.add_signal_handler = original_add # type: ignore[method-assign] + loop.remove_signal_handler = original_remove # type: ignore[method-assign] + import signal as _sig + + assert exit_code == 0 + assert add_calls == [_sig.SIGINT] + assert remove_calls == [_sig.SIGINT] + + def test_no_textual_import(self) -> None: + """no_textual_import [scenario]: …""" + import importlib + import sys + + # Clear any prior textual import to make this test honest in isolation + textual_was_imported = "textual" in sys.modules + # We cannot reliably remove textual mid-suite (other tests might rely on it via dev deps), + # so the assertion is: importing ratatoskr.cli does not REQUIRE textual. + importlib.reload(__import__("ratatoskr.cli", fromlist=["_amain"])) + # The boundary is the INV-001 import-only rule. If ratatoskr/cli.py grew an + # `import textual` directly, the import would still succeed (textual is installed) + # but the source-level boundary is the load-bearing check — covered by a static-grep + # smoke test pattern. Do that here: + import pathlib + + src = pathlib.Path(__file__).parent.parent / "src" / "ratatoskr" / "cli.py" + text = src.read_text() + for forbidden in ("import textual", "from textual", "import rich", "from rich"): + assert forbidden not in text, f"INV-001 violation: cli.py contains '{forbidden}'" + _ = textual_was_imported # avoid unused warning + + +class TestMain: + def test_happy_returns_amain_exit_code( + self, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] + ) -> None: + """happy_returns_amain_exit_code [happy,tracer]: …""" + + async def fake_amain(args: ParsedArgs) -> int: + assert args.send_content == "hi" + return 0 + + monkeypatch.setattr(cli_mod, "_amain", fake_amain) + rc = main(["--send", "hi", "--new", "--agent", "m", "--api-key", "k"]) + assert rc == 0 + + def test_usage_error_no_send( + self, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] + ) -> None: + """usage_error_no_send: empty argv → exit 10; stderr [usage_error]; _amain never called.""" + amain_calls: list[int] = [] + + async def fake_amain(args: ParsedArgs) -> int: + amain_calls.append(1) + return 0 + + monkeypatch.setattr(cli_mod, "_amain", fake_amain) + rc = main([]) + assert rc == 10 + assert "[usage_error]" in capsys.readouterr().err + assert amain_calls == [] + + def test_usage_error_both_session_and_new( + self, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] + ) -> None: + """usage_error_both_session_and_new: both flags → exit 10; [usage_error].""" + + async def fake_amain(args: ParsedArgs) -> int: + return 0 + + monkeypatch.setattr(cli_mod, "_amain", fake_amain) + rc = main(["--send", "hi", "--session", "s", "--new", "--agent", "m", "--api-key", "k"]) + assert rc == 10 + assert "[usage_error]" in capsys.readouterr().err + + def test_auth_error_missing_key(self, capsys: pytest.CaptureFixture[str]) -> None: + """auth_error_missing_key: …""" + # _clear_env fixture has already deleted WORLDTREE_API_KEY + rc = main(["--send", "hi", "--new", "--agent", "m"]) + assert rc == 11 + assert "[auth_error]" in capsys.readouterr().err + + def test_no_argv_uses_sys_argv(self, monkeypatch: pytest.MonkeyPatch) -> None: + """no_argv_uses_sys_argv [trace]: argv=None → _parse_args reads sys.argv[1:].""" + monkeypatch.setattr( + "sys.argv", + ["ratatoskr", "--send", "hi", "--new", "--agent", "m", "--api-key", "k"], + ) + + async def fake_amain(args: ParsedArgs) -> int: + assert args.send_content == "hi" + assert args.agent_id == "m" + return 0 + + monkeypatch.setattr(cli_mod, "_amain", fake_amain) + rc = main(None) + assert rc == 0