feat(cli): implement issue #3 contract via TDD

54 contract-listed tests authored + GREEN per the vertical-slice
ordering (_parse_args → _render_event → _cancel_and_log → _run_turn
→ _amain → main). 117/117 tests GREEN suite-wide; ruff clean.

The _run_turn race-loop is the load-bearing piece. Per iteration,
the await on the next event is raced against sigint_event.wait()
when NOT cancelling. Once SIGINT fires (with last_turn_id known),
_cancel_and_log is spawned, cancelling=True flips, and subsequent
iterations skip wait()-task creation entirely — the bug Volva
flagged in contract review would otherwise busy-wake on the
already-set event each iteration.

Implementation notes:
- _UsageErrorParser subclasses argparse.ArgumentParser and overrides
  error() to raise _ArgparseError instead of calling sys.exit;
  _parse_args catches and re-raises as UsageError per the contract's
  ERROR_ROUTING.
- _GatedStream test helper (custom httpx.AsyncByteStream that pauses
  on asyncio.Event entries) makes SIGINT-mid-stream tests deterministic
  without sleep-based timing — gates release via side-channels (the
  cancel-mock sets an event when its endpoint is observed).
- _sse_resp test helper wraps respx Response with the
  text/event-stream content-type, dedupes the boilerplate across the
  13 _run_turn tests.
- Strong-ref cancel_task local in _run_turn holds the fire-and-forget
  cancel task to suppress RUF006 / asyncio GC warning.

One in-flight contract amendment during TDD: no_busy_loop_after_cancel
test description originally said "exactly ONE wait()-shaped task" but
the natural race-loop shape produces 2 (iter 1 raced w/ text, iter 2
raced w/ sigint → flipped cancelling; iter 3+ skipped). Amended to
"TWO total wait() coroutines" with rationale; the busy-loop check is
preserved (iter 3+ MUST skip).

Persistent-memory updated per the commit-along rule: new module
landed, recent-decisions log entries for #3 (contract + Volva
paraphrase + TDD), next natural moves rotated to /volva-code-review
on the implementation.

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=<from body>`; 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).
+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. 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.
+2. **Manual smoke against a local Worldtree**: `ratatoskr --send "hello" --new --agent <id>`. 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` is itself a recording probe — capture its outputs to `tests/snapshots/` for replay-based regression coverage.
+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.
+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
 

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
+Implements docs/contracts/issues/3.contract.md.
-shape (Textual app, `--send` non-interactive mode, `--session`, `--new`,
-`--agent`, `--api-key`, `--server-log <path>`, `--raw`, etc.).
 """
 
+from __future__ import annotations
 
-def main() -> int:
+import argparse
-    raise NotImplementedError(
+import asyncio
-        "Ratatoskr CLI is not implemented yet. "
+import os
-        "See docs/design-brief.md for the locked shape."
+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=<none>\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))

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