Issue #3: non-interactive `--send` stdout presenter (design-brief §8b). Composes sessions.create_session (when --new) with sse_client.stream_turn + cancel_turn. First contract in the repo with a code-level dependencies: block (issues #1 and #2). Six FN blocks (main, _parse_args, _amain, _render_event, _run_turn, _cancel_and_log). 10 hard invariants codifying: no textual/rich imports; Text deltas + post-Done newline are the only stdout writes; --session xor --new + agent/api-key resolution chains; SIGINT semantics (cancel-with- known-turn-id; early-exit code 3 without; swallow cancel failures during drain). 9-bucket exit code table. Volva paraphrase round flagged 5 ambiguities; all 5 amended: - INV-001 wording: "no in-repo modules other than X/Y" (was strict-only) - INV-002 + _render_event POST-002/003: Done's stdout newline is part of the contract, restructured to avoid the "no other event writes stdout" contradiction - Data flow stderr list: added the pre-event [cancelled] (before any event arrived) label - _run_turn STEPS 3 race-loop: gate sigint_task creation behind `if not cancelling` to avoid busy-wake once event is set; new no_busy_loop_after_cancel trace test - assumptions list: added CancelTurnNotFound + CancelAlreadyCompleted prd: block pinned to issue #3 body SHA 206ef51709d43b2c at 2026-05-21T05:13:29+00:00. Drift check clean.
42 KiB
contract_version, target_module, scope, depends_on, used_by, language, complexity, estimated_loc, confidence, assumptions, open_questions, prd, dependencies
| contract_version | target_module | scope | depends_on | used_by | language | complexity | estimated_loc | confidence | assumptions | open_questions | prd | dependencies | |||||||||||||||||||||||||||||||||||||||
|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|
| 2.1 | ratatoskr.cli | Implement the non-interactive `--send` CLI presenter. One sync entry point (`main`) plus its async orchestrator (`_amain`) plus three helpers (`_parse_args`, `_render_event`, `_run_turn`). Composes `ratatoskr.sessions.create_session` (when `--new`) with `ratatoskr.sse_client.stream_turn` + `cancel_turn`. Renders the SSE event stream to stdout (agent message deltas only, raw text) and stderr (all other events, labeled) — the stdout/stderr split is the load-bearing scriptability contract. Handles SIGINT by issuing server-side `cancel_turn` then draining until the terminal `Cancelled` event arrives. No Textual import; no markdown rendering; no in-process reconnect (one-shot, `SseConnectionDropped` exits non-zero). First code-level consumer of issues #1 and #2 in the repo — the integration test the contract-mock unit tests cannot deliver. |
|
python | medium | 280 | 0.85 |
|
|
|
|
CLI — Non-interactive --send stdout presenter
Context
ratatoskr.cli is the non-interactive --send presenter described in docs/design-brief.md §8b. One entry point — main — registered as the ratatoskr console script in pyproject.toml. Streams a single turn against a Worldtree session and exits; useful for CI smoke tests, scripted dev probes, for i in {1..5}; do ratatoskr --send ...; done loops, and paste-into-shell debugging.
The CLI is the first code-level composition of ratatoskr.sse_client (issue #1) and ratatoskr.sessions (issue #2). The two upstream modules unit-test against respx HTTP mocks; the CLI is where their real-Worldtree integration discipline gets exercised end-to-end. The stdout/stderr split is intentional: agent message deltas (the user-visible payload) on stdout, all observability events (tool calls, persona affect, errors, lifecycle) labeled on stderr. This makes the CLI both pipeable (ratatoskr --send "..." | grep ...) and human-watchable (run without redirection, watch both streams interleave).
The Textual TUI is the eventual primary product (design-brief §5) — separate issue, larger surface. This contract deliberately ships ONLY the stdout presenter. ratatoskr.cli MUST NOT import textual (the TUI lands as a sibling module under ratatoskr.tui; the CLI dispatches to neither in this issue).
Data flow
Input:
argv: list[str] | None— command-line argv (None ⇒ usesys.argv[1:]).- Environment:
WORLDTREE_API_KEY(fallback for--api-key),WORLDTREE_API_URL(fallback for--server).
Parsed arguments (ParsedArgs frozen dataclass):
send_content: str— the user message text (--send <content>, required).session_id: str | None— existing session (--session <id>); mutex withnew.new: bool— mint a fresh session (--new); mutex withsession_id.agent_id: str | None— required iffnew=True.api_key: str— resolved from--api-keythen$WORLDTREE_API_KEY.server_url: str— resolved from--server, then$WORLDTREE_API_URL, thenhttp://localhost:8000.
Output (stdout):
- Raw text deltas from
Textevents, written without trailing newline per chunk. A single trailing newline is written after the terminalDoneevent so the next shell prompt lands on a fresh line.
Output (stderr):
- One labeled line per non-
Textevent. Format:[worker_phase] phase=streaming turn_id=42[thinking] <truncated content>(truncate to 200 chars; long thinking blobs are noisy)[text_boundary] kind=sentence char_offset=128[tool_start] name=read_file args={...}[tool_result] name=read_file duration_ms=42 result=<truncated>(truncate result repr to 200 chars)[done] turn_id=42 model=glm5-turbo duration_ms=1234 usage=...[error] turn_id=42 code=llm_output_invalid message=<...>[cancelled] turn_id=42 reason=user_request partial_message_id=N|None
- Plus pre-stream lifecycle:
[create_session] session_id=... agent_id=...(when--new)
- Plus the pre-first-event cancel label (INV-008):
[cancelled] (before any event arrived)— SIGINT fired before any event yielded a turn_id; CLI exits 3 without issuingcancel_turn. Distinct from the post-event[cancelled] turn_id=... reason=... partial_message_id=...line emitted by theCancelledterminal event.
- Plus failure paths (always to stderr):
[auth_error] no API key (set --api-key or WORLDTREE_API_KEY)[usage_error] --session and --new are mutually exclusive[agent_not_found] agent_id=...[session_api_failed] status=... body=<truncated>[sse_connect_failed] status=... body=<truncated>[connection_dropped] last_seen=<turn_id:seq | none>[malformed_sse_id] raw=...[turn_id_flip] expected=... got=...[cancel_failed] status=... body=<truncated>(during SIGINT-triggered cancel; informational, does not change the primary exit code)
Exit codes:
0— cleanDoneterminal event received.2—Errorterminal event received (server signaled turn-level failure).3—Cancelledterminal event received. Distinguishes user-cancel from server-cancel via stderr label only; the exit code is the same.10— usage error (bad argv, mutex violation, missing required).11— auth error (missing API key).12—AgentNotFoundfromcreate_session(unknownagent_id).20— server API failure:SessionApiFailedfromcreate_session, ORSseConnectFailedfromstream_turn(including HTTP 401 / 404 / 5xx before the stream opens). Status surfaced on stderr.21— network failure:SseConnectionDropped(no in-process reconnect in--send), generichttpx.ConnectError/httpx.ReadTimeoutraised before any HTTP call lands.22— protocol failure:MalformedSseIdorTurnIdFlip(server-side wire bug — surfaces them honestly, does not paper over).
Side effects: outbound HTTP only (via the modules); writes to stdout / stderr; one SIGINT handler installed on the running event loop, uninstalled in cleanup.
On disk: none. No config files, no state persistence. Cross-process resume is explicitly deferred to v2 per design-brief §8d.
Invariants
- INV-001 [hard]:
ratatoskr.cliimports no in-repo modules other thanratatoskr.sessionsandratatoskr.sse_client. Standard-library imports (argparse,asyncio,os,signal,sys,dataclasses,typing) and the already-declaredhttpxdependency are unrestricted. It does NOT importtextual, does NOT importrich(raw text on stdout — no markdown rendering per design-brief §6's "TUI-only" framing for--send). The Textual TUI lands as a sibling module in a later issue; the textual-boundary half of this invariant is enforced by an inline import-check test (test_no_textual_import_in_cli). - INV-002 [hard]: stdout is written to in exactly two cases: (a)
Textevent deltas (rawevent.content, no newline appended, flushed per chunk per INV-010); (b) a single trailing newline written immediately after the terminalDoneevent (so the next shell prompt lands on a fresh line). No other event variant —WorkerPhase,Thinking,TextBoundary,ToolStart,ToolResult,Error,Cancelled— writes anything to stdout. TheDonenewline is the only non-Textstdout write and is part of the contract, not an exception to it. - INV-003 [hard]: All observability output (every non-
Textevent, every error label, every lifecycle label) goes to stderr. Redirecting stderr to/dev/nullMUST leave a clean text-only stream on stdout suitable for piping. - INV-004 [hard]:
--sessionand--neware mutually exclusive AND exactly one is required. Passing both, or neither, OR passing--agenttogether with--session, exits with code10and a[usage_error] ...line to stderr BEFORE any HTTP call is issued. - INV-005 [hard]:
--api-keyresolution order: explicit flag wins; otherwise$WORLDTREE_API_KEY. If neither is set, exit with code11and[auth_error] ...to stderr BEFORE any HTTP call is issued. - INV-006 [hard]:
--serverresolution order: explicit flag, then$WORLDTREE_API_URL, then defaulthttp://localhost:8000. The default is a documented local-dev posture (design-brief §6); no error if all three are absent. - INV-007 [hard]: SIGINT during a streaming turn issues exactly one server-side
cancel_turn(client, session_id, last_seen_turn_id)call wherelast_seen_turn_idis theturn_idof the most recently yielded event. A SECOND SIGINT during the same drain re-issues nothing (the cancel is in-flight; subsequent SIGINTs are no-op until the terminal event arrives). The iterator continues to drain until theCancelledterminal event lands; only then does the CLI exit. - INV-008 [hard]: SIGINT BEFORE the first event arrives (turn_id unknown — connection not yet established, or open but no events yielded) skips
cancel_turnentirely and exits with code3immediately. The server's stall-watchdog handles the orphan turn (per spec). - INV-009 [hard]:
cancel_turnfailures during SIGINT handling (e.g.,CancelFailed,CancelTurnNotFound,CancelAlreadyCompleted) write a[cancel_failed] ...line to stderr and continue draining the stream. They do NOT raise out of_run_turn— the primary exit code is determined by the stream's terminal event, not the cancel attempt's outcome. - INV-010 [hard]: stdout writes are flushed after every
Textdelta (sys.stdout.flush()per chunk). The terminal-newline write is also flushed. Without per-chunk flush, streaming is invisible under pipes — the load-bearing scriptability contract dies.
Out of scope
- Textual TUI — separate issue. This contract ships ONLY the stdout presenter.
ratatoskr.clidoes NOT importtextual; INV-001 enforces. - Markdown rendering on stdout — design-brief §6 commits to markdown-default-on with
--rawopt-out, but ONLY for the TUI.--sendstreams raw deltas always; norich, no--rawflag in this contract. - Two-stage Ctrl-C (design-brief §8c) — one-shot mode has no idle state for the second-Ctrl-C-exits semantic. INV-007 lock simple cancel-then-drain.
--server-logpane — TUI-only observability surface (design-brief §5).reconnect_turnmid-process —--sendis one-shot;SseConnectionDroppedexits with code21. In-process reconnect is a TUI feature.- Recorded SSE snapshot fixtures — captured by a separate follow-up issue. The
--send --newinvocation IS the recording probe but the snapshot replay test infrastructure is out of scope here. - JSON output mode — deferred until a scripted caller actually needs it. The stdout/stderr split provides enough structure for most piping use cases.
--quiet— deferred. If the default stderr labeling is empirically noisy in real use, add a--quietflag in a follow-up.- Windows support — design-brief §6 commits to terminal-native POSIX.
asyncio.add_signal_handleris POSIX-only; no Windows shim.
Constraints
- [compatibility] Module must work against the spec pin (
55101e909abcd2219833266b6f905c5bc956e0f0, Worldtree v0.19.0). Spec bumps go through the issue #1 / #2 modules; the CLI is insulated from wire-level changes. - [performance] Streaming MUST NOT buffer the turn in memory. Per-chunk write + flush to stdout. The
Done.complete_responsefield is observable on stderr (truncated) but NOT used as the source of stdout output (which is delta-by-delta). - [security] Module does not log full request/response bodies, does not log the
Authorizationheader, does not log--api-keyvalue. Per-event stderr labels limit content to truncated reprs (200 chars forThinkingcontent andToolResult.result). - [style] Async-native. The sync entry point is a thin
asyncio.run(_amain(...))wrapper. No nested event loops; no thread pools. argparse-only (no Click / Typer) for stdlib-only dep posture.
Architecture
main(argv) [sync]
│
├─ _parse_args(argv) → ParsedArgs [argparse + env resolution + xor validation; raises UsageError on violation]
│
└─ asyncio.run(_amain(args)) → int [exit code]
│
├─ httpx.AsyncClient(base_url=args.server_url, headers={Auth: Bearer args.api_key}) as client:
│
├─ IF args.new:
│ info = await sessions.create_session(client, args.agent_id)
│ session_id = info.session_id
│ stderr ← f"[create_session] session_id={info.session_id} agent_id={info.agent_id}"
│ ELSE:
│ session_id = args.session_id
│
├─ sigint_event = asyncio.Event()
│ loop.add_signal_handler(SIGINT, sigint_event.set)
│
└─ return await _run_turn(client, session_id, args.send_content, sigint_event, stdout, stderr)
│
├─ async-iterate sse_client.stream_turn(client, session_id, content)
├─ race each __anext__() against sigint_event.wait() so a mid-stream SIGINT lands within one event boundary
├─ on each event: _render_event(event, stdout, stderr)
├─ on sigint_event set + last_turn_id known + not already cancelling:
│ asyncio.create_task(sse_client.cancel_turn(client, session_id, last_turn_id))
│ set cancelling=True (idempotent across repeated sigints)
├─ on terminal event (Done | Error | Cancelled): break + map to exit code
└─ on exception bubbled out of stream_turn: map to exit code per the table in Data flow
FN main(argv: list[str] | None = None) -> int
BRIEF: Sync entry point registered as the `ratatoskr` console script. Parses argv, runs the async orchestrator under asyncio.run, returns the exit code. Catches `UsageError` (from _parse_args) and `_AuthError` and maps them to exit codes 10 / 11 before reaching the event loop. Everything else propagates through _amain.
PRE: [PRE-001 hard] argv is None or a list of strings -- assert argv is None or all(isinstance(a, str) for a in argv)
POST: [POST-001 return_value] returns an int exit code in the documented range (0, 2, 3, 10, 11, 12, 20, 21, 22)
POST: [POST-002 side_effect] usage/auth errors are reported to stderr before the event loop runs -- assert capsys.readouterr().err includes "[usage_error]" or "[auth_error]"
ERROR_ROUTING:
UsageError:
local_handling: write `[usage_error] {msg}` to stderr
flow_control: abort
state_recovery: none (no resources acquired before _parse_args)
_AuthError:
local_handling: write `[auth_error] no API key (set --api-key or WORLDTREE_API_KEY)` to stderr
flow_control: abort
state_recovery: none
STEPS:
1. [setup, flexibility=prescriptive] TRY: args = _parse_args(argv)
ON UsageError as exc:
WRITE f"[usage_error] {exc}\n" to stderr
RETURN 10
ON _AuthError as exc:
WRITE f"[auth_error] {exc}\n" to stderr
RETURN 11
2. [sequential, flexibility=prescriptive] RETURN asyncio.run(_amain(args))
TESTS:
happy_returns_amain_exit_code [happy,tracer]: argv specifies a complete --send invocation; monkeypatch _amain to return 0 → main returns 0
usage_error_no_send [error]: argv=[] → main returns 10; stderr has "[usage_error]"; _amain never called
usage_error_both_session_and_new [error]: argv has both --session and --new → returns 10; stderr "[usage_error]"
auth_error_missing_key [error]: argv specifies --send/--new/--agent but neither --api-key nor WORLDTREE_API_KEY is set → returns 11; stderr "[auth_error]"; _amain never called
no_argv_uses_sys_argv [trace]: argv=None → _parse_args is called with sys.argv[1:] (monkeypatched argparse capture confirms)
FN _parse_args(argv: list[str] | None) -> ParsedArgs
BRIEF: argparse + env-fallback + xor-validation. Returns a frozen `ParsedArgs` on success. Raises `UsageError` on argument violations and `_AuthError` on missing API key (both caught by `main` and mapped to exit codes 10 / 11 BEFORE the event loop opens).
PRE: [PRE-001 hard] argv is None or a list of strings -- assert argv is None or all(isinstance(a, str) for a in argv)
POST: [POST-001 return_value] returns ParsedArgs with `send_content` non-empty, exactly one of `session_id` / `new` set, and `api_key` non-empty
POST: [POST-002 return_value] when args.new, args.agent_id is non-empty; when args.session_id is set, args.agent_id is None
POST: [POST-003 return_value] args.server_url is one of: explicit --server value, $WORLDTREE_API_URL value, or "http://localhost:8000" — in that resolution order (INV-006)
ERROR_ROUTING:
argparse.ArgumentError | SystemExit-from-argparse:
local_handling: catch and re-raise as UsageError(msg) so main() can format it consistently
flow_control: abort
state_recovery: none
xor violation (both --session and --new, or neither):
local_handling: raise UsageError("--session and --new are mutually exclusive; pass exactly one")
flow_control: abort
state_recovery: none
--agent passed with --session:
local_handling: raise UsageError("--agent is required with --new and forbidden with --session")
flow_control: abort
state_recovery: none
--new without --agent:
local_handling: raise UsageError("--agent is required when --new is passed")
flow_control: abort
state_recovery: none
api_key unresolved (no flag, no env):
local_handling: raise _AuthError("no API key (set --api-key or WORLDTREE_API_KEY)")
flow_control: abort
state_recovery: none
STEPS:
1. [setup, flexibility=prescriptive] Construct argparse.ArgumentParser:
--send <content> (required, str, non-empty)
--session <id> (str, optional)
--new (bool flag, default False)
--agent <id> (str, optional — required-with-validation in step 3)
--api-key <key> (str, optional — env fallback in step 4)
--server <url> (str, optional — env fallback + default in step 5)
2. [sequential, flexibility=prescriptive] Parse argv:
TRY: ns = parser.parse_args(argv if argv is not None else sys.argv[1:])
ON SystemExit:
Re-raise as UsageError with argparse's captured message
3. [branch, flexibility=prescriptive] Validate session/new/agent triad per INV-004:
IF ns.session and ns.new: RAISE UsageError("--session and --new are mutually exclusive")
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")
4. [sequential, flexibility=prescriptive] Resolve api_key per INV-005:
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)")
5. [sequential, flexibility=prescriptive] Resolve server_url per INV-006:
server_url = ns.server or os.environ.get("WORLDTREE_API_URL") or "http://localhost:8000"
6. [cleanup] 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,
)
TESTS:
happy_new [happy,tracer]: argv=["--send", "hi", "--new", "--agent", "mimir", "--api-key", "k"] → ParsedArgs(send_content="hi", session_id=None, new=True, agent_id="mimir", api_key="k", server_url="http://localhost:8000")
happy_existing_session [happy]: argv=["--send", "hi", "--session", "s-1", "--api-key", "k"] → ParsedArgs with session_id="s-1", new=False, agent_id=None
api_key_from_env [trace]: monkeypatch WORLDTREE_API_KEY="from-env"; argv omits --api-key → ParsedArgs.api_key=="from-env"
api_key_flag_beats_env [trace]: monkeypatch WORLDTREE_API_KEY="env"; argv has --api-key "flag" → ParsedArgs.api_key=="flag"
server_default [trace]: argv omits --server and WORLDTREE_API_URL is unset → ParsedArgs.server_url=="http://localhost:8000"
server_env_fallback [trace]: monkeypatch WORLDTREE_API_URL="http://t.local:9000"; argv omits --server → ParsedArgs.server_url=="http://t.local:9000"
server_flag_beats_env [trace]: monkeypatch WORLDTREE_API_URL="env"; argv has --server "flag" → ParsedArgs.server_url=="flag"
usage_no_send [error]: argv=["--new", "--agent", "mimir", "--api-key", "k"] → UsageError (argparse required-flag)
usage_both_session_and_new [adversarial]: argv has both --session and --new → UsageError("mutually exclusive")
usage_neither_session_nor_new [adversarial]: argv has --send but neither --session nor --new → UsageError("pass exactly one")
usage_new_without_agent [adversarial]: argv=["--send","hi","--new","--api-key","k"] → UsageError("--agent is required when --new")
usage_session_with_agent [adversarial]: argv=["--send","hi","--session","s-1","--agent","x","--api-key","k"] → UsageError("--agent is required with --new and forbidden with --session")
auth_missing [error]: argv lacks --api-key and WORLDTREE_API_KEY unset → _AuthError
empty_send [adversarial]: argv=["--send", "", "--new", "--agent", "x", "--api-key", "k"] → UsageError (non-empty enforced via argparse type=lambda or explicit check)
FN _amain(args: ParsedArgs) -> int
BRIEF: Async orchestrator. Opens an authenticated httpx.AsyncClient, optionally creates a fresh session, installs a SIGINT handler, drives the turn via _run_turn, returns the exit code. Maps top-level upstream exceptions (`AgentNotFound`, `SessionApiFailed`, generic httpx connection failures during create_session) to exit codes 12 / 20 / 21 before reaching _run_turn.
PRE: [PRE-001 hard] args is a ParsedArgs (post-validation; PRE-002/PRE-003 of _parse_args hold) -- assert isinstance(args, ParsedArgs)
POST: [POST-001 return_value] returns one of the documented exit codes (0, 2, 3, 12, 20, 21, 22)
POST: [POST-002 side_effect] when args.new is True, exactly one POST /sessions was issued -- assert respx tracked the call
POST: [POST-003 side_effect] when args.new is True, stderr contains "[create_session] session_id=... agent_id=..." before any stream events
POST: [POST-004 side_effect] the SIGINT handler is removed in cleanup (loop.remove_signal_handler called) -- verified via teardown probe in test fixtures
ERROR_ROUTING:
AgentNotFound:
local_handling: write `[agent_not_found] agent_id={exc.agent_id}` to stderr
flow_control: abort
state_recovery: none
SessionApiFailed:
local_handling: write `[session_api_failed] status={exc.status} body={exc.body!r}` to stderr
flow_control: abort
state_recovery: none
httpx.ConnectError | httpx.ReadTimeout | httpx.TransportError (from create_session):
local_handling: write `[network_error] {type(exc).__name__}: {exc}` to stderr
flow_control: abort
state_recovery: none
STEPS:
1. [setup, flexibility=prescriptive] OPEN httpx.AsyncClient(base_url=args.server_url, headers={"Authorization": f"Bearer {args.api_key}"}) via async-with
2. [branch, flexibility=prescriptive] IF args.new:
TRY: info = await sessions.create_session(client, args.agent_id)
ON AgentNotFound as exc:
WRITE stderr; RETURN 12
ON SessionApiFailed as exc:
WRITE stderr; RETURN 20
ON httpx.ConnectError | httpx.ReadTimeout | httpx.TransportError as exc:
WRITE stderr; RETURN 21
WRITE f"[create_session] session_id={info.session_id} agent_id={info.agent_id}\n" to stderr
SET session_id = info.session_id
ELSE:
SET session_id = args.session_id # pre-validated non-None
3. [sequential, flexibility=prescriptive] Install SIGINT handler:
sigint_event = asyncio.Event()
loop = asyncio.get_running_loop()
loop.add_signal_handler(signal.SIGINT, sigint_event.set)
4. [sequential, flexibility=prescriptive] TRY:
exit_code = await _run_turn(client, session_id, args.send_content, sigint_event, stdout=sys.stdout, stderr=sys.stderr)
FINALLY:
loop.remove_signal_handler(signal.SIGINT)
5. [cleanup] RETURN exit_code
TESTS:
happy_new_session_then_stream [happy,tracer]: respx mocks POST /sessions → 201 + the SSE POST → text+done; argv specifies --new --agent mimir → _amain returns 0; stderr has "[create_session]" before "[done]"
happy_existing_session [happy]: respx mocks the SSE POST only; argv specifies --session s-1 → _amain returns 0; respx tracked exactly 0 POST /sessions calls
agent_not_found_exits_12 [error]: respx mocks POST /sessions → 404; --new → returns 12; stderr "[agent_not_found]"; stream_turn never invoked
session_api_failed_exits_20 [error]: respx mocks POST /sessions → 500 with body → returns 20; stderr "[session_api_failed] status=500 body=..."
connect_error_exits_21 [error]: respx simulates httpx.ConnectError on POST /sessions → returns 21; stderr "[network_error]"
sigint_handler_installed_and_removed [trace]: probe asyncio loop signal handlers before/after _amain — handler present during _run_turn execution, absent after _amain returns
no_textual_import [scenario]: import ratatoskr.cli and assert "textual" not in sys.modules induced by that import (INV-001 verified at the import boundary)
FN _render_event(event: Event, *, stdout: TextIO, stderr: TextIO) -> None
BRIEF: Pure event-to-output renderer. Routes `Text` deltas to stdout (with per-chunk flush per INV-010); routes every other Event variant to stderr with a labeled line. No I/O outside the two passed TextIO objects; no side effects on the event itself.
PRE: [PRE-001 hard] event is an instance of one of the Event union variants -- assert isinstance(event, (WorkerPhase, Thinking, Text, TextBoundary, ToolStart, ToolResult, Done, Error, Cancelled))
POST: [POST-001 side_effect] for Text events: stdout received event.content (no newline appended) AND stdout was flushed -- assert stdout.getvalue().endswith(event.content) and stdout.flush.called
POST: [POST-002 side_effect] for non-Text-non-Done events (WorkerPhase, Thinking, TextBoundary, ToolStart, ToolResult, Error, Cancelled): stdout was NOT written to (INV-002); stderr received exactly one line ending in newline -- assert stdout.getvalue() == "" and stderr.getvalue().endswith("\n")
POST: [POST-003 side_effect] for Done: stdout receives a single newline AND is flushed; stderr receives a single labeled line including `turn_id` (from event.sse_id.turn_id) + `model` + `duration_ms` (INV-002 carve-out — Done is the one non-Text variant that writes to stdout) -- assert stdout.getvalue() == "\n" and stderr.getvalue().startswith("[done]")
ERROR_ROUTING:
(none — pure function over the typed union; if an instance doesn't match any branch, PRE-001 catches it as an assertion failure)
STEPS:
1. [setup, flexibility=prescriptive] Match on `type(event)`:
2. [branch, flexibility=prescriptive]
CASE Text:
stdout.write(event.content); stdout.flush()
CASE Done:
stdout.write("\n"); stdout.flush()
stderr.write(f"[done] turn_id={event.sse_id.turn_id} model={event.model} duration_ms={event.duration_ms} usage={event.usage!r}\n")
CASE Error:
stderr.write(f"[error] turn_id={event.sse_id.turn_id} code={event.error_code} message={event.message!r}\n")
CASE Cancelled:
stderr.write(f"[cancelled] turn_id={event.turn_id} reason={event.reason!r} partial_message_id={event.partial_message_id}\n")
CASE WorkerPhase:
stderr.write(f"[worker_phase] phase={event.phase} turn_id={event.turn_id}\n")
CASE Thinking:
stderr.write(f"[thinking] {event.content[:200]!r}\n")
CASE TextBoundary:
stderr.write(f"[text_boundary] kind={event.kind} char_offset={event.char_offset}\n")
CASE ToolStart:
stderr.write(f"[tool_start] name={event.name} args={event.arguments!r}\n")
CASE ToolResult:
stderr.write(f"[tool_result] name={event.name} duration_ms={event.duration_ms} result={event.result!r:.200}\n")
TESTS:
text_to_stdout_only [happy,tracer]: Text(content="hello", sse_id=...) → stdout=="hello"; stderr==""; stdout.flush called once
done_writes_newline_and_label [happy]: Done(sse_id=(42,5), model="glm5-turbo", duration_ms=1234, ...) → stdout=="\n"; stderr starts with "[done]" and contains "turn_id=42" + "model=glm5-turbo"
error_to_stderr_only [happy]: Error(sse_id=(42,5), error_code="llm_output_invalid", message="m", ...) → stdout==""; stderr starts with "[error]"; contains "code=llm_output_invalid"; turn_id from sse_id
cancelled_to_stderr_only [happy]: Cancelled(sse_id=(42,5), turn_id=42, reason="user", partial_message_id=7) → stderr starts with "[cancelled]" and contains "reason='user'" + "partial_message_id=7"; stdout==""
worker_phase_to_stderr [happy]: WorkerPhase(phase="streaming", turn_id=42, ...) → stderr starts with "[worker_phase]"; stdout==""
thinking_truncated [trace]: Thinking(content="a"*500, ...) → stderr line includes only first 200 chars of content
tool_start_to_stderr [happy]: ToolStart(name="read_file", arguments={"path": "/x"}, ...) → stderr starts with "[tool_start] name=read_file args="
tool_result_truncated [trace]: ToolResult(name="x", result="b"*500, duration_ms=42, ...) → stderr line repr truncated to ≤200 chars in result field
text_boundary_to_stderr [happy]: TextBoundary(kind="sentence", char_offset=128, ...) → stderr starts with "[text_boundary]"
invariant_inv003_stderr_only [scenario]: emit one of each non-Text variant in sequence; assert stdout buffer is empty after each (INV-003 verified by exhaustion of the non-Text union)
FN _run_turn(client: httpx.AsyncClient, session_id: str, content: str, sigint_event: asyncio.Event, *, stdout: TextIO, stderr: TextIO) -> int
BRIEF: Drive `stream_turn`, render events, race each `__anext__()` against `sigint_event.wait()` so a SIGINT lands within one event boundary. On first SIGINT (with last_turn_id known), spawn `cancel_turn` as a background task and keep draining until the `Cancelled` terminal event arrives. Map terminal events and uncaught exceptions to exit codes per the Data flow table.
PRE: [PRE-001 hard] client is not None -- assert client is not None
PRE: [PRE-002 hard] session_id is a non-empty string -- assert session_id and isinstance(session_id, str)
PRE: [PRE-003 hard] content is a non-empty string -- assert content and isinstance(content, str)
PRE: [PRE-004 hard] sigint_event is an asyncio.Event -- assert isinstance(sigint_event, asyncio.Event)
POST: [POST-001 return_value] returns one of (0, 2, 3, 20, 21, 22) — terminal-event-driven OR exception-mapped
POST: [POST-002 side_effect] each yielded event passed through _render_event exactly once -- spy on _render_event call count == event count
POST: [POST-003 side_effect] sigint mid-stream issues exactly one cancel_turn HTTP call -- assert respx tracked one POST /sessions/{id}/turns/{turn_id}/cancel
POST: [POST-004 side_effect] sigint before any event yields zero cancel_turn calls -- INV-008: turn_id is unknown so cancel cannot be issued
POST: [POST-005 side_effect] cancel_failed during sigint drains writes "[cancel_failed]" to stderr but does NOT raise -- INV-009: primary exit code is the stream's terminal-event code
ERROR_ROUTING:
SseConnectFailed:
local_handling: write `[sse_connect_failed] status={exc.status} body={exc.body!r}` to stderr
flow_control: abort
state_recovery: none
exit_code: 20
SseConnectionDropped:
local_handling: write `[connection_dropped] last_seen={exc.last_seen_sse_id}` to stderr
flow_control: abort
state_recovery: none (no in-process reconnect in --send per design-brief §8d)
exit_code: 21
MalformedSseId:
local_handling: write `[malformed_sse_id] raw={exc.raw!r}` to stderr
flow_control: abort
state_recovery: none (server-side wire bug; surface honestly)
exit_code: 22
TurnIdFlip:
local_handling: write `[turn_id_flip] expected={exc.established} got={exc.got}` to stderr
flow_control: abort
state_recovery: none (server-side wire bug)
exit_code: 22
CancelFailed | CancelTurnNotFound | CancelAlreadyCompleted (during sigint drain):
local_handling: write `[cancel_failed] {type(exc).__name__}: {exc}` to stderr
flow_control: resume — INV-009 keeps draining stream until terminal
state_recovery: none
STEPS:
1. [setup, flexibility=prescriptive] Validate inputs per PRE-001..PRE-004
2. [setup, flexibility=prescriptive] Initialize state:
last_turn_id = None
cancelling = False
iterator = sse_client.stream_turn(client, session_id, content).__aiter__()
3. [loop, flexibility=prescriptive] Race-loop. Each iteration awaits the next event; while NOT cancelling, the await races against sigint_event so a mid-stream SIGINT lands within one event boundary. Once cancelling=True, the loop no longer creates a sigint_task — sigint_event stays set permanently and racing against it would busy-wake on every iteration. After cancellation is in flight, the loop just drains the stream:
WHILE True:
next_task = asyncio.create_task(iterator.__anext__())
IF NOT cancelling:
sigint_task = asyncio.create_task(sigint_event.wait())
done, _pending = await asyncio.wait({next_task, sigint_task}, return_when=FIRST_COMPLETED)
IF sigint_task in done:
IF last_turn_id is not None:
# INV-007: cancel server-side; flip cancelling=True so future iterations stop racing the (now permanently set) sigint_event. Let the stream drain to the Cancelled terminal.
asyncio.create_task(_cancel_and_log(client, session_id, last_turn_id, stderr=stderr))
cancelling = True
# next_task may still be pending — fall through to the "if next_task in done" branch (it may also be done; both branches handle gracefully)
ELSE:
# INV-008: turn_id unknown — no cancel possible; exit immediately
next_task.cancel()
WRITE "[cancelled] (before any event arrived)\n" to stderr
RETURN 3
ELSE:
# cancelling=True: don't race sigint_event (it's permanently set); just await the next event
done = {next_task}
AWAIT next_task # block until the next event or stream-terminal exception
IF next_task in done:
TRY:
event = next_task.result()
CATCH StopAsyncIteration:
# stream exited without a terminal — INV-001 of issue #1 guarantees this only on connection drop, which raises before we get here. Defensive: surface as connection-dropped.
WRITE "[connection_dropped] last_seen=<none>\n" to stderr
RETURN 21
CATCH (the exceptions in ERROR_ROUTING above):
# exception type → stderr label + exit_code per ERROR_ROUTING
RETURN <mapped exit code>
last_turn_id = event.sse_id.turn_id
_render_event(event, stdout=stdout, stderr=stderr)
IF isinstance(event, Done):
IF NOT cancelling: sigint_task.cancel()
RETURN 0
IF isinstance(event, Error):
IF NOT cancelling: sigint_task.cancel()
RETURN 2
IF isinstance(event, Cancelled):
IF NOT cancelling: sigint_task.cancel()
RETURN 3
ELSE:
# next_task NOT in done means we hit the sigint-without-turn-id branch above and already RETURNed. This branch is unreachable; defensive guard.
CONTINUE
TESTS:
happy_text_then_done [happy,tracer]: respx mock yields text + done; sigint_event never set → _run_turn returns 0; stdout has the text delta; stderr has "[done]"
error_terminal [happy]: mock yields text + error → returns 2; stderr contains "[error]"
cancelled_terminal_server [happy]: mock yields text + cancelled → returns 3; stderr contains "[cancelled]"
sse_connect_failed_404 [error]: mock returns 404 before stream opens → returns 20; stderr "[sse_connect_failed] status=404 ..."
connection_dropped [error]: mock raises RemoteProtocolError mid-stream → returns 21; stderr "[connection_dropped]"
malformed_sse_id [error]: mock yields an event with `id: 42` (no seq) → returns 22; stderr "[malformed_sse_id]"
turn_id_flip [error]: mock yields events 42:1 then 99:2 → returns 22; stderr "[turn_id_flip] expected=42 got=99"
sigint_before_first_event [scenario]: sigint_event set BEFORE the mock has yielded anything → returns 3; respx tracked zero cancel_turn POSTs (INV-008)
sigint_mid_stream_drains_to_cancelled [scenario,tracer]: mock yields text(42:1) → caller sets sigint_event → mock yields cancelled(42:2) → returns 3; respx tracked exactly one POST /sessions/{id}/turns/42/cancel (INV-007)
sigint_twice_issues_one_cancel [scenario]: mock yields text(42:1) → sigint set → set again → mock yields cancelled(42:2) → returns 3; respx tracked exactly one cancel POST (INV-007 idempotence across repeated sigints)
no_busy_loop_after_cancel [trace]: mock yields text(42:1) → sigint set → mock yields cancelled(42:2); assert sigint_event.wait() is created at most once per pre-cancelling iteration and NEVER after cancelling=True flips. For this test scenario that's exactly TWO total wait() coroutines: iter 1 (raced with text) + iter 2 (raced with sigint, flipped cancelling). Iter 3+ MUST skip wait() creation — the busy-loop bug would make the count grow unbounded with each iteration. Verified by patching sigint_event.wait() and counting coroutine creations.
cancel_failed_drains_anyway [scenario]: as above but cancel_turn HTTP mock returns 500 → _cancel_and_log writes "[cancel_failed]" to stderr; stream still drains to cancelled; _run_turn returns 3 (INV-009)
render_called_once_per_event [trace]: spy on _render_event; mock yields N events; assert _render_event.call_count == N
FN _cancel_and_log(client: httpx.AsyncClient, session_id: str, turn_id: int, *, stderr: TextIO) -> None
BRIEF: Background task spawned by _run_turn on first SIGINT. Calls `sse_client.cancel_turn`; on exception, writes `[cancel_failed] ...` to stderr (INV-009). Never raises out — the primary exit code path is _run_turn's stream-terminal logic, not the cancel attempt's outcome.
PRE: [PRE-001 hard] client is not None -- assert client is not None
PRE: [PRE-002 hard] turn_id is a positive int -- assert isinstance(turn_id, int) and turn_id > 0
POST: [POST-001 side_effect] exactly one POST /sessions/{id}/turns/{turn_id}/cancel was issued -- respx tracked one call
POST: [POST-002 side_effect] on cancel exception, stderr received a "[cancel_failed]" line -- assert "[cancel_failed]" in stderr.getvalue() when mock returns non-200
POST: [POST-003 exception] never raises -- pytest-asyncio test does not record any uncaught exception in the task
ERROR_ROUTING:
CancelFailed | CancelTurnNotFound | CancelAlreadyCompleted:
local_handling: write `[cancel_failed] {type(exc).__name__}: {exc}` to stderr
flow_control: skip (swallow — INV-009)
state_recovery: none
httpx.RequestError (generic transport error during cancel):
local_handling: write `[cancel_failed] {type(exc).__name__}: {exc}` to stderr
flow_control: skip
state_recovery: none
STEPS:
1. [setup, flexibility=prescriptive] Validate inputs per PRE-001, PRE-002
2. [sequential, flexibility=prescriptive] TRY:
await sse_client.cancel_turn(client, session_id, turn_id)
CATCH (any of the routed exceptions above):
WRITE labeled line to stderr
(return None — never raise)
TESTS:
happy_cancel [happy,tracer]: respx mock returns 200 with cancelled=True body → _cancel_and_log returns None; stderr empty
cancel_failed_500 [error]: mock returns 500 → returns None; stderr has "[cancel_failed] CancelFailed: ..."
cancel_already_completed [scenario]: mock returns 409 → returns None; stderr has "[cancel_failed] CancelAlreadyCompleted: ..."
cancel_turn_not_found [scenario]: mock returns 404 → returns None; stderr has "[cancel_failed] CancelTurnNotFound: ..."
transport_error_swallowed [error]: respx simulates httpx.ConnectError → returns None; stderr "[cancel_failed]"; INV-009 verified — no exception escapes