---
contract_version: "2.1"
module: "ratatoskr.sse_client"
purpose: "Worldtree Conversation API SSE consumer — turn streaming, in-process reconnect via Last-Event-ID, and server-side turn cancellation. Single shared consumer for TUI and stdout presenters."
depends_on:
  - "httpx"
  - "httpx-sse"
used_by:
  - "ratatoskr.cli"
  - "ratatoskr.tui"
language: "python"
complexity: "high"
estimated_loc: 320
confidence: 0.85
assumptions:
  - "Worldtree spec pin (`docs/conversation-api-spec.md` at v1.0, repo SHA `55101e909abcd2219833266b6f905c5bc956e0f0`) is the wire contract. Event names, JSON shapes, status codes, and `id:` format are read FROM the spec, not from any Worldtree source import."
  - "`httpx-sse` correctly parses the full SSE wire format including multi-line `data:`, `id:`, `event:`, `retry:`, and comment lines. Hand-rolled `data:`-only parsing is rejected (would silently drop the load-bearing composite `id:`)."
  - "Server emits exactly one terminal event (`done` | `error` | `cancelled`) per turn (spec INV at §SSE Event Types: 'Every stream terminates with exactly one done, error, or cancelled event')."
  - "Composite SSE wire `id:` format is `{turn_id}:{seq}` where both parts are decimal integers (spec INV-014). No other separators, no quoting."
  - "Cancel takes effect at the next agent_turn event boundary; in-flight tool calls run to completion. The cancel HTTP call is fast (200 OK with envelope) and decoupled from the SSE stream's `cancelled` terminal event."
open_questions:
  - "Reconnect-by-replay semantics: should `reconnect()` yield ONLY the post-`Last-Event-ID` events (server-replayed buffered + live), or also re-yield the historical events the caller already saw? Current draft: only replayed+live (the server already only sends those). Caller is responsible for de-dup if it re-displays."
  - "412 `buffer_expired` recovery policy: caller-driven? The module surfaces the error; caller decides whether to abandon-and-restart or fall back to non-resume display. Draft picks 'surface, not recover.'"
  - "Should the typed Event union live in a separate `events.py` module? Draft keeps it co-located in `sse_client.py` for v0; split if a non-consumer module ever needs to type-import Event without pulling httpx."
---

# SSE Client — Worldtree Conversation API turn streaming

## Context

`ratatoskr.sse_client` is Ratatoskr's reference implementation of the Worldtree Conversation API's per-turn SSE consumption pattern. Three entry points: `stream_turn` (POST a user message + consume the response stream), `reconnect_turn` (re-attach to an in-flight turn after disconnect, using `Last-Event-ID`), and `cancel_turn` (server-side cancellation of an in-flight turn via the cancel endpoint).

The module is the **single source of API-consumption code** for both presenters (TUI in `ratatoskr.tui`, stdout in the `--send` non-interactive mode). The presenters consume typed `Event` objects yielded by `stream_turn` and `reconnect_turn`; they share zero parsing code.

Why this module is the natural smallest unit to TDD against: it's the integration seam between Ratatoskr and Worldtree. Every higher feature (multi-pane TUI, --send mode, two-stage Ctrl-C, scripted CI probes) lands ON TOP of this module's contract. Getting the wire-level discipline right here is what makes Ratatoskr "the reference Python SSE-resume implementation" (per `docs/design-brief.md` §3).

## Data flow

**Input:**
- `httpx.AsyncClient` (caller-owned, carries base URL + `Authorization: Bearer <api_key>` header; module does not own client lifecycle).
- `session_id: str` — opaque session identifier from `POST /sessions` or `GET /sessions`.
- `content: str` — user message body (for `stream_turn`).
- `last_event_id: str | None` — composite `{turn_id}:{seq}` to resume from (for `reconnect_turn`); the same string the server emitted as the `id:` line of the last successfully-consumed event.
- `turn_id: int` — for `cancel_turn`; obtained from any prior event's parsed `id:` or `turn_id` body field.

**Output:**
- `stream_turn` / `reconnect_turn`: `AsyncIterator[Event]` — typed discriminated union of `WorkerPhase | Thinking | Text | TextBoundary | ToolStart | ToolResult | Done | Error | Cancelled`. Each event carries the parsed JSON fields per the spec's §SSE Event Types, PLUS the parsed `sse_id: SseId = (turn_id, seq)` lifted from the SSE wire `id:` field. Always-final event is exactly one of `Done | Error | Cancelled` (spec invariant).
- `cancel_turn`: `CancelResult` dataclass — `{turn_id: int, cancelled: bool, reason: str | None, partial_message_id: int | None}`. Returned on 200 OK from the cancel endpoint.

**Side effects:** outbound HTTP only; no disk I/O, no global state, no cross-process resume bookkeeping.

**On disk:** none. Per design-brief §8d, cross-process resume is explicitly deferred to v2. `Last-Event-ID` lives in caller-owned process memory.

## Invariants

- **INV-001 [hard]**: Every successful `stream_turn` / `reconnect_turn` iteration terminates with exactly one terminal event of type `done`, `error`, or `cancelled`. The iterator MUST NOT raise StopAsyncIteration before a terminal event arrives unless the underlying HTTP connection drops mid-stream — in which case it raises `SseConnectionDropped` (caller may invoke `reconnect_turn` if it has the last seen `sse_id`).
- **INV-002 [hard]**: Every yielded `Event` carries its parsed `sse_id: SseId = (turn_id: int, seq: int)`. Events whose SSE `id:` wire field is absent, malformed, or non-composite (`123` instead of `123:4`) MUST raise `MalformedSseId` before yielding. This is the foot-gun that hand-rolled `data:`-only parsing silently drops; the module's value is making it impossible to silently drop. (See `docs/design-brief.md` §3.)
- **INV-003 [hard]**: The composite `sse_id`'s `turn_id` component is stable across all events of a single turn (spec INV-014). The module MUST detect a turn_id flip mid-stream and raise `TurnIdFlip` rather than yield events from a different turn.
- **INV-004 [hard]**: `reconnect_turn` sets the `Last-Event-ID` HTTP header with the verbatim composite string `{turn_id}:{seq}` (not the integer parts separately). The caller passes the same string format the server emitted.
- **INV-005 [hard]**: No `core.*` or `worldtree.*` imports. Boundary verified by the existing `tests/test_no_worldtree_imports.py`.
- **INV-006 [soft, recovery_window=1]**: The `Event` yielded immediately AFTER a `worker_phase` event MAY belong to the next phase (e.g., a `text` event after `worker_phase Streaming`). Phase ordering invariant is server-side, not client-side; the consumer just forwards what arrives.
- **INV-007 [hard]**: `cancel_turn` is decoupled from the SSE stream — the HTTP 200 from the cancel endpoint does NOT mean the stream has emitted its `cancelled` terminal event yet. Callers driving two-stage Ctrl-C must keep iterating the stream after `cancel_turn` returns until the `Cancelled` event arrives (or the stream closes).

## Resume semantics

Per design-brief §8d: in-process reconnect only. `reconnect_turn` is the resume entry point; cross-process resume (persisting `Last-Event-ID` to disk under `~/.config/ratatoskr/`) is explicitly deferred to v2.

The module's resume contract:

1. The caller holds the most recently yielded event's `sse_id`. (Easy to thread: `last_seen = event.sse_id` on every iteration.)
2. On any `SseConnectionDropped`, the caller MAY invoke `reconnect_turn(client, session_id, last_event_id=f"{last_seen.turn_id}:{last_seen.seq}")`.
3. The server replays buffered events with `seq > last_seen.seq` for that `turn_id`, then resumes live streaming. The module yields them as it received them.
4. If the server returns 412 `buffer_expired`, the module raises `ResumeBufferExpired(buffered_from_seq=X)`. Caller's policy.
5. If the server returns 410 `turn_finished`, the module raises `ResumeTurnFinished(turn_id=N)`. The turn completed while the connection was dropped; caller can `GET /sessions/{id}/messages` to fetch the final assistant message.

The "reconnect, not resume-across-process" framing in the design-brief §8d MUST appear in the README too (consumer-facing).

## Constraints

- **[compatibility]** Module must work against the spec pin (`55101e909abcd2219833266b6f905c5bc956e0f0`, Worldtree v0.19.0). Spec bumps trigger an explicit re-record of the recorded-SSE snapshot fixtures (see `tests/snapshots/README.md`).
- **[performance]** Streaming MUST NOT buffer the full turn in memory — events are yielded as they arrive. The complete-response field on `Done` is what the server sends; the consumer does not re-aggregate from `text` events.
- **[security]** `Authorization` header lives on the caller's `httpx.AsyncClient`. Module does not log the header, does not log full event bodies (they contain user message content). Logs are limited to `(turn_id, seq, type)` triples.
- **[style]** Async-native. No sync entry points. The consumer is `async def` + `async for`; presenters are async too.

---

```contract
FN stream_turn(client: httpx.AsyncClient, session_id: str, content: str) -> AsyncIterator[Event]
BRIEF: POST a user message to /sessions/{session_id}/messages and yield typed Events as they arrive on the SSE stream, terminating at the first Done/Error/Cancelled event.
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)
POST: [POST-001 side_effect] exactly one POST to /sessions/{session_id}/messages was issued -- assert mock_router.calls.call_count == 1
POST: [POST-002 return_value] AsyncIterator yields ≥1 event ending in exactly one of (Done, Error, Cancelled) -- collect events; assert isinstance(events[-1], (Done, Error, Cancelled))
POST: [POST-003 state_change] every yielded Event has a populated sse_id with int turn_id and int seq -- assert all(isinstance(e.sse_id.turn_id, int) and isinstance(e.sse_id.seq, int) for e in events)
ERROR_ROUTING:
  httpx.HTTPStatusError:
    local_handling: re-raise as SseConnectFailed(status=resp.status_code, body=resp.read()[:1024]) — server returned non-2xx before stream started (e.g., 404 session_not_found)
    flow_control: abort
    state_recovery: none (no events yielded yet)
  httpx.ReadError | httpx.RemoteProtocolError:
    local_handling: re-raise as SseConnectionDropped(last_seen_sse_id=<last yielded event's sse_id or None>)
    flow_control: abort
    state_recovery: none (caller may reconnect_turn)
  MalformedSseId:
    local_handling: log truncated raw `id:` value; re-raise to caller
    flow_control: abort
    state_recovery: none (stream is corrupt; cannot resume safely)
  TurnIdFlip:
    local_handling: log both turn_ids; re-raise to caller
    flow_control: abort
    state_recovery: none (server-side bug; do not paper over)
STEPS:
  1. [setup, flexibility=prescriptive] Validate inputs per PRE-001..PRE-003
  2. [sequential, flexibility=prescriptive] Open SSE connection via httpx_sse.aconnect_sse with method="POST", url=f"/sessions/{session_id}/messages", json={"content": content}
     ON httpx.HTTPStatusError before stream opens:
       RAISE SseConnectFailed
  3. [loop, flexibility=prescriptive] FOR EACH sse_event in event_source.aiter_sse():
     a. [sequential] CALL _parse_sse_id(sse_event.id) → (turn_id, seq)
        ON ValueError:
          RAISE MalformedSseId(raw=sse_event.id[:64])
     b. [branch, flexibility=prescriptive] IF this is not the first event AND turn_id != established_turn_id:
        RAISE TurnIdFlip(established=established_turn_id, got=turn_id)
        ELSE:
          SET established_turn_id = turn_id (on first event)
     c. [sequential] Parse sse_event.data as JSON → body dict
     d. [sequential] CALL _envelope_for_type(body, sse_id=(turn_id, seq)) → typed Event
     e. [sequential] YIELD event
     f. [branch] IF isinstance(event, (Done, Error, Cancelled)):
        BREAK (terminal event reached)
  4. [cleanup, flexibility=prescriptive] Close the event_source context (handled by `async with httpx_sse.aconnect_sse(...)`)
TESTS:
  happy_one_text_done [happy,tracer]: POST /sessions/{id}/messages mock yields `text` then `done` with valid `{turn_id}:{seq}` ids → consumer yields Text then Done; Done.sse_id.turn_id matches Text.sse_id.turn_id; iteration ends after Done
  full_event_vocab [happy]: mock emits worker_phase → thinking → text → text_boundary → tool_start → tool_result → text → done → consumer yields one of each typed Event variant; all carry parsed sse_id
  error_terminal [error]: mock emits one `text` then `error` with `error_code: "llm_output_invalid"` → consumer yields Text then Error; iteration ends; Error.message and Error.error_code are populated
  cancelled_terminal [error]: mock emits `cancelled` with phase=cancelled → consumer yields Cancelled with turn_id; iteration ends
  session_not_found [error]: mock returns 404 before stream opens → consumer raises SseConnectFailed(status=404)
  malformed_id_no_seq [adversarial]: mock event has `id: 42` (missing `:seq`) → consumer raises MalformedSseId; no event yielded
  malformed_id_alpha [adversarial]: mock event has `id: foo:bar` (non-integer parts) → consumer raises MalformedSseId
  turn_id_flip [adversarial]: mock emits text events with ids `42:1` then `99:2` → consumer raises TurnIdFlip; only the first event was yielded
  connection_drop [error]: mock raises httpx.RemoteProtocolError after one text event → consumer raises SseConnectionDropped(last_seen_sse_id=(42, 1))
  no_text_aggregation [trace]: assert consumer does NOT concatenate text deltas internally — each text event yields as its own Text envelope; aggregation is the caller's job
```

```contract
FN reconnect_turn(client: httpx.AsyncClient, session_id: str, last_event_id: str) -> AsyncIterator[Event]
BRIEF: Re-POST /sessions/{session_id}/messages with `Last-Event-ID` header to resume an in-flight turn; server replays buffered events with seq > last_event_id.seq then streams live until terminal event.
PRE: [PRE-001 hard] client is not None -- assert client is not None
PRE: [PRE-002 hard] session_id is non-empty -- assert session_id
PRE: [PRE-003 hard] last_event_id matches `{int}:{int}` format -- _parse_sse_id(last_event_id) succeeds; raise ValueError otherwise (the caller is misusing the API; do not RAISE on the SSE stream's behalf)
POST: [POST-001 side_effect] exactly one POST issued with `Last-Event-ID: <last_event_id>` header verbatim -- assert mock_router.calls[0].request.headers["Last-Event-ID"] == last_event_id
POST: [POST-002 return_value] iterator yields the (server-replayed buffered events with seq > last_event_id.seq) followed by live events, terminating in Done/Error/Cancelled
POST: [POST-003 state_change] all yielded events share `turn_id == _parse_sse_id(last_event_id).turn_id` -- assert all(e.sse_id.turn_id == resume_turn_id for e in events)
ERROR_ROUTING:
  HTTP 400 invalid_last_event_id:
    local_handling: parse body for error_code; raise InvalidLastEventId(raw=last_event_id)
    flow_control: abort
    state_recovery: none (caller has a bug — bad last_event_id format or turn_id mismatch)
  HTTP 410 turn_finished:
    local_handling: parse body for `turn_id`; raise ResumeTurnFinished(turn_id=N)
    flow_control: abort
    state_recovery: caller can GET /sessions/{id}/messages for final assistant text
  HTTP 412 buffer_expired:
    local_handling: parse body for `buffered_from_seq` + `turn_id`; raise ResumeBufferExpired(turn_id=N, buffered_from_seq=X)
    flow_control: abort
    state_recovery: caller policy — abandon turn, or restart with a fresh stream_turn
  httpx.HTTPStatusError (other status):
    local_handling: re-raise as SseConnectFailed(status=resp.status_code, body=resp.read()[:1024])
    flow_control: abort
    state_recovery: none
STEPS:
  1. [setup, flexibility=prescriptive] Validate `_parse_sse_id(last_event_id)` succeeds (raise ValueError on malformed input — this is caller misuse, not stream corruption)
  2. [sequential, flexibility=prescriptive] Open SSE connection via httpx_sse.aconnect_sse with method="POST", url=f"/sessions/{session_id}/messages", json={"content": ""} OR with no body (TBD per spec §Reconnect & Resume — confirm during implementation), headers={"Last-Event-ID": last_event_id}
  3. [branch, flexibility=prescriptive] IF response status == 400: parse body and RAISE InvalidLastEventId
     ELIF response status == 410: parse body and RAISE ResumeTurnFinished
     ELIF response status == 412: parse body and RAISE ResumeBufferExpired
     ELIF response status != 200: RAISE SseConnectFailed
  4. [loop, flexibility=prescriptive] Iterate event_source as in stream_turn STEP 3 (INV-002 INV-003 apply equally; TurnIdFlip means caller passed a last_event_id whose turn_id no longer matches what the server thinks is the current turn)
  5. [cleanup] Close event_source context
TESTS:
  happy_resume_from_seq_3 [scenario,tracer]: mock previously streamed `42:1, 42:2, 42:3`; caller invokes reconnect_turn("...", "42:3"); mock replays `42:4, 42:5` then live `42:6 done` → consumer yields events 4-6; all turn_id==42
  header_verbatim [trace]: assert outbound request carries `Last-Event-ID: 42:3` header exactly (not `42` only, not `seq=3`)
  buffer_expired [error]: mock returns 412 with `{"error":"buffer_expired","buffered_from_seq":10,"turn_id":42}` → consumer raises ResumeBufferExpired(turn_id=42, buffered_from_seq=10)
  turn_finished [error]: mock returns 410 with `{"error":"turn_finished","turn_id":42}` → consumer raises ResumeTurnFinished(turn_id=42)
  invalid_last_event_id_server [error]: mock returns 400 with `{"error":"invalid_last_event_id"}` → consumer raises InvalidLastEventId
  malformed_input [adversarial]: caller passes last_event_id="42" (no seq) → consumer raises ValueError before opening any connection (no HTTP issued)
  turn_id_flip_on_resume [adversarial]: caller passes last_event_id="42:3" but server streams `99:4` → consumer raises TurnIdFlip(established=42, got=99); first event was not yielded
```

```contract
FN cancel_turn(client: httpx.AsyncClient, session_id: str, turn_id: int, *, persist_partial: bool = False) -> CancelResult
BRIEF: POST /sessions/{session_id}/turns/{turn_id}/cancel to request server-side cancellation. Returns CancelResult on 200. Does NOT block on the SSE stream's `cancelled` terminal event — that is the caller's job to observe on the still-iterating stream.
PRE: [PRE-001 hard] client is not None -- assert client is not None
PRE: [PRE-002 hard] session_id is non-empty -- assert session_id
PRE: [PRE-003 hard] turn_id is positive int -- assert isinstance(turn_id, int) and turn_id > 0
POST: [POST-001 side_effect] exactly one POST to /sessions/{session_id}/turns/{turn_id}/cancel issued -- assert mock_router.calls.call_count == 1
POST: [POST-002 side_effect] query string carries `persist_partial=true` iff `persist_partial=True` -- assert ("persist_partial=true" in str(req.url)) == persist_partial
POST: [POST-003 return_value] returns CancelResult with `turn_id`, `cancelled: bool`, `reason: str | None`, `partial_message_id: int | None` -- assert all fields present
ERROR_ROUTING:
  HTTP 404 turn_not_found:
    local_handling: raise CancelTurnNotFound(turn_id=N)
    flow_control: abort
    state_recovery: none (race: turn finished before cancel arrived; caller should observe the terminal event on the still-iterating stream)
  HTTP 409 turn_already_completed:
    local_handling: raise CancelAlreadyCompleted(turn_id=N)
    flow_control: abort
    state_recovery: none (same race as 404; treat as benign)
  httpx.HTTPStatusError (other status):
    local_handling: re-raise as CancelFailed(status=resp.status_code, body=resp.read()[:1024])
    flow_control: abort
    state_recovery: none
STEPS:
  1. [setup, flexibility=prescriptive] Validate inputs per PRE-001..PRE-003
  2. [sequential, flexibility=prescriptive] Build query params: `{"persist_partial": "true"}` if persist_partial else `{}`
  3. [sequential, flexibility=prescriptive] CALL client.post(f"/sessions/{session_id}/turns/{turn_id}/cancel", params=params)
     tool: { destructive: true, idempotent: true, read_only: false, open_world: false }
  4. [branch, flexibility=prescriptive] IF resp.status_code == 404: RAISE CancelTurnNotFound
     ELIF resp.status_code == 409: RAISE CancelAlreadyCompleted
     ELIF resp.status_code != 200: RAISE CancelFailed
  5. [sequential] Parse resp.json() into CancelResult dataclass
  6. [cleanup] RETURN CancelResult
TESTS:
  happy_cancel [happy,tracer]: mock returns 200 with `{"turn_id":42,"cancelled":true,"reason":null,"partial_message_id":null}` → returns CancelResult(turn_id=42, cancelled=True, reason=None, partial_message_id=None)
  persist_partial_query [trace]: cancel_turn(..., persist_partial=True) → outbound request URL ends with `?persist_partial=true`
  default_no_persist [trace]: cancel_turn(...) (default) → outbound request URL has no `persist_partial` query param
  cancel_already_completed [error]: mock returns 409 → raises CancelAlreadyCompleted(turn_id=42)
  cancel_turn_not_found [error]: mock returns 404 → raises CancelTurnNotFound(turn_id=42)
  cancel_idempotent [scenario]: two back-to-back cancel_turn calls (same turn_id) — first returns 200 cancelled=True, second returns 409 → first happy, second raises CancelAlreadyCompleted (no client-side dedup; surface the race honestly)
```

```contract
FN _parse_sse_id(raw: str) -> SseId
BRIEF: Parse the SSE wire `id:` field as the composite `{turn_id}:{seq}` per spec INV-014. Returns SseId NamedTuple(turn_id: int, seq: int). Raises ValueError on any deviation (missing colon, non-integer parts, extra colons, empty parts).
PRE: [PRE-001 hard] raw is a string -- isinstance(raw, str)
POST: [POST-001 return_value] returns SseId with turn_id and seq both non-negative ints -- assert result.turn_id >= 0 and result.seq >= 0
POST: [POST-002 exception] on any malformed input, raises ValueError with `raw[:64]` in the message -- pytest.raises(ValueError, match=re.escape(raw[:64]))
STEPS:
  1. [setup, flexibility=prescriptive] Validate `raw` is non-empty; raise ValueError on empty
  2. [sequential, flexibility=prescriptive] SPLIT raw on ":" → parts
  3. [branch, flexibility=prescriptive] IF len(parts) != 2: RAISE ValueError(f"expected '{{turn_id}}:{{seq}}', got: {raw[:64]!r}")
  4. [sequential, flexibility=prescriptive] CALL int() on parts[0] and parts[1]
     ON ValueError:
       RE-RAISE with truncated raw in the message
  5. [branch, flexibility=prescriptive] IF either int is negative: RAISE ValueError(f"expected non-negative ints, got: {raw[:64]!r}")
  6. [cleanup] RETURN SseId(turn_id=parts[0], seq=parts[1])
TESTS:
  happy_simple [happy,tracer]: "42:3" → SseId(turn_id=42, seq=3)
  happy_zero_seq [happy]: "1:0" → SseId(turn_id=1, seq=0)
  empty [adversarial]: "" → ValueError
  no_colon [adversarial]: "42" → ValueError with "42" in the message
  too_many_colons [adversarial]: "42:3:7" → ValueError
  alpha_turn_id [adversarial]: "foo:3" → ValueError
  alpha_seq [adversarial]: "42:bar" → ValueError
  negative_turn_id [adversarial]: "-1:3" → ValueError
  trailing_whitespace [adversarial]: "42:3 " → ValueError (strict; do not strip; the server emits clean ids)
  truncation [security]: input is 5000-char string with no colon → ValueError message includes only `raw[:64]` (not the full 5000)
```