---
contract_version: "2.1"
target_module: "ratatoskr.sse_client"
scope: "Implement the Worldtree Conversation API SSE consumer for Ratatoskr. Single module bundling stream_turn (POST + SSE iteration), reconnect_turn (Last-Event-ID resume), cancel_turn (server-side cancel POST), and the private _parse_sse_id helper. The module is the shared API-consumption surface for both presenters (Textual TUI in ratatoskr.tui and stdout in --send mode via ratatoskr.cli); neither forks the SSE parsing. No core.* / worldtree.* imports — spec-only dependency (boundary enforced by tests/test_no_worldtree_imports.py)."
depends_on:
  - "httpx"
  - "httpx-sse"
used_by:
  - "ratatoskr.cli"
  - "ratatoskr.tui"
language: "python"
complexity: "high"
estimated_loc: 320
confidence: 0.85
prd:
  issue: 1
  issue_url: "https://gitea.phasefinal.com/vh/ratatoskr/issues/1"
  body_sha256_16: "abcbc49467e86f1d"
  lock_in_comment_id: null
  lock_in_sha256_16: null
  lock_in_at: null
  pinned_at: "2026-05-21T03:57:37+00:00"
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 ≥ 1 (spec §SSE id format: `turn_id` is SQLite `turns.id` (autoincrement, ≥ 1); `seq` is a per-turn monotonic integer starting at 1, resetting per turn). 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)` with `turn_id ≥ 1` and `seq ≥ 1` (server-side spec invariants). Events whose SSE `id:` wire field is absent, malformed, non-composite (`123` instead of `123:4`), or carries an out-of-range integer 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 and raise `TurnIdFlip` rather than yield an event from the wrong turn. The check has **two distinct entry points** depending on origin:
    - In `stream_turn`, the caller has no prior expectation of `turn_id`. The **first** event yielded establishes `established_turn_id`; subsequent events must match it. The first event is always yielded (it cannot be a flip).
    - In `reconnect_turn`, `last_event_id` carries the caller's prior expectation. `established_turn_id` is parsed FROM `last_event_id` BEFORE the connection opens. The **first** event from the server is therefore already a flip-candidate — if its `turn_id` differs from the expected one, the module raises `TurnIdFlip` BEFORE yielding it.
- **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` AND the original `content` it sent to `stream_turn`. (Easy to thread: `last_seen = event.sse_id` on every iteration; `content` stays in the caller's frame.)
2. On any `SseConnectionDropped`, the caller MAY invoke `reconnect_turn(client, session_id, content, last_event_id=f"{last_seen.turn_id}:{last_seen.seq}")`. The `content` argument is sent verbatim in the JSON body (spec wire requirement) but the server identifies the resume target by `Last-Event-ID` and does NOT re-process `content` (spec §Reconnect flow).
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.
- **[compatibility]** Callers MUST configure their `httpx.AsyncClient` with a long-or-disabled `read` timeout for SSE flows. LLM streaming has multi-second idle gaps between events (especially during prompt-building, thinking phases, and long completions); httpx's default 5s read timeout would kill the connection mid-stream. The recommended shape is `httpx.Timeout(connect=10.0, read=None, write=10.0, pool=10.0)` — disable read timeout, keep modest connect/write/pool timeouts so true network failures still surface promptly. As defense in depth, `stream_turn`'s ERROR_ROUTING also catches `httpx.ReadTimeout` and surfaces it as `SseConnectionDropped` — but the right place to configure is the caller-owned client.
- **[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 both fields >= 1 -- assert all(e.sse_id.turn_id >= 1 and e.sse_id.seq >= 1 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). EXCEPT the two eager turn-launch failures (Worldtree v1.0.0b1 #331), checked BEFORE raise_for_status and raised as typed SseConnectFailed SUBCLASSES carrying error_code: 409 -> AgentNotAvailable (agent unavailable; pre-b1 this was a 200 + in-stream `error` event), 503 -> TurnLaunchUnavailable (transient turn-launch failure; retryable=True). Subclassing keeps existing `except SseConnectFailed` handlers working with zero changes.
    flow_control: abort
    state_recovery: none (no events yielded yet)
  httpx.ReadError | httpx.RemoteProtocolError | httpx.ReadTimeout:
    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)
  MalformedSseData:
    local_handling: propagate to caller (no logging at sse_client layer — sse_client is a library, not a logging sink; presenters own observability). The exception carries `exc.raw` (already truncated to 200 chars at MalformedSseData.__init__) for presenters to render.
    flow_control: abort
    state_recovery: none (payload corruption; can't decode; consumer's call to reconnect)
    note: empty `sse.data == ''` is SKIPPED, not raised — see _iter_events STEP 3.0 (issue #7 INV-001). Only non-empty-but-malformed-JSON raises MalformedSseData.
  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}
     2a. [branch, flexibility=prescriptive] IF response.status_code in (409, 503) (b1 #331 eager turn-launch failures): read the body, parse (error_code, message) from the `{"detail": {...}}` envelope OR a flat `{error_code, message}` body (status-derived default code when absent), then RAISE AgentNotAvailable (409) / TurnLaunchUnavailable (503).
     ON httpx.HTTPStatusError before stream opens (any other non-2xx):
       RAISE SseConnectFailed
  3. [loop, flexibility=prescriptive] FOR EACH sse_event in event_source.aiter_sse():
     0. [branch, flexibility=prescriptive] IF sse_event.data == "":
        # Issue #7 INV-001: empty-data frames are treated as keepalives.
        # ORDERING: this check fires BEFORE _parse_sse_id; an empty-data event
        # with a malformed id is silently swallowed (a keepalive with a bad id
        # is still a keepalive). Skip means: don't yield, don't update last_sse_id,
        # don't set terminal_seen, don't raise. CONTINUE to next sse_event.
        CONTINUE
     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
        ON json.JSONDecodeError:
          # Issue #7 INV-002: non-empty data that fails json.loads is a protocol
          # error (server emitted malformed JSON). Surface as named exception.
          RAISE MalformedSseData(raw=sse_event.data)  # raw truncated to 200 in __init__
     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)
  eager_409_agent_not_available [error]: mock returns 409 {detail:{error_code:"agent_not_available", message}} before stream → consumer raises AgentNotAvailable(status=409, error_code="agent_not_available", retryable absent); isinstance SseConnectFailed
  eager_503_retryable [error]: mock returns 503 before stream → consumer raises TurnLaunchUnavailable(status=503, retryable=True); isinstance SseConnectFailed
  eager_409_non_json_body [adversarial]: mock returns 409 with a non-JSON body → consumer raises AgentNotAvailable with the status-derived default error_code "agent_not_available"
  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
  empty_data_skipped [trace,issue#7]: stream yields 4 frames — text(42:1), empty-data(42:2), text(42:3), done(42:4) — consumer yields exactly 3 events (Text, Text, Done); empty-data frame invisible. Intermediate state after the second Text: last_sse_id == (42, 3) (proves skip did not advance through 42:2).
  empty_data_skip_preserves_last_seen_sse_id [trace,issue#7]: stream yields text(42:1), empty-data(42:2), then RemoteProtocolError → consumer raises SseConnectionDropped with `last_seen_sse_id == SseId(42, 1)`. Directly probes INV-003 via the exception payload — if the skip had transiently advanced last_sse_id through 42:2, the exception would carry (42, 2) and the test would fail.
  malformed_data_raises [error,issue#7]: stream yields text(42:1) + event with `data: not-json` → consumer yields one Text then raises MalformedSseData(raw="not-json"); iteration aborts.
  whitespace_data_raises [adversarial,issue#7]: stream yields event with `data:` followed by single space → MalformedSseData (whitespace is non-empty per INV-001 exact-equality skip rule).
  malformed_data_truncation [security,issue#7]: stream yields event with 5000-char malformed data → MalformedSseData.raw is exactly first 200 chars; exception message truncated too.
```

```contract
FN reconnect_turn(client: httpx.AsyncClient, session_id: str, content: 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. `content` is sent verbatim in the JSON body to satisfy the endpoint's request schema; the server resolves the resume target by `Last-Event-ID` and does NOT re-process `content` or re-invoke the agent (spec §Reconnect flow: "the agent's tools and LLM call run exactly once regardless of how many disconnects/reconnects occur"). Caller convention: pass the same content originally sent to `stream_turn`.
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] content is a string -- assert isinstance(content, str) (non-empty NOT required for resume — server ignores it; caller convention is to pass the original content for clarity, but the wire-level requirement is only "valid JSON satisfying the request schema")
PRE: [PRE-004 hard] last_event_id matches `{int}:{int}` format with both ints ≥ 1 -- _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 side_effect] outbound JSON body is `{"content": <content arg verbatim>}` -- assert json.loads(mock_router.calls[0].request.content) == {"content": content}
POST: [POST-003 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-004 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); SET expected_turn_id = parsed.turn_id
  2. [sequential, flexibility=prescriptive] Open SSE connection via httpx_sse.aconnect_sse with method="POST", url=f"/sessions/{session_id}/messages", json={"content": content}, headers={"Last-Event-ID": last_event_id}. The body shape matches the spec §Reconnect flow example verbatim; the server resolves the resume target from the header, not the body.
  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. For EACH event: parse sse_id; if turn_id != expected_turn_id, RAISE TurnIdFlip BEFORE yielding (per INV-003: in reconnect_turn the first event is already a flip-candidate — there is no first-event-establishes allowance). Otherwise yield. INV-002 applies equally; malformed ids RAISE MalformedSseId.
  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("...", "original content", "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`)
  body_threads_content [trace]: reconnect_turn(..., content="hello mimir", last_event_id="42:3") → outbound JSON body is `{"content": "hello mimir"}` byte-for-byte
  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_first_event [adversarial]: caller passes last_event_id="42:3" but server's first replayed/live event is `99:4` → consumer raises TurnIdFlip(expected=42, got=99) BEFORE yielding (the first event in reconnect_turn IS a flip-candidate — unlike stream_turn which establishes turn_id from the first event)
```

```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 §SSE id format. Returns SseId NamedTuple(turn_id: int, seq: int). Strict: both parts MUST be decimal integers `≥ 1` (server invariant — `turn_id` from SQLite autoincrement, `seq` starts at 1 per spec line 705). Raises ValueError on any deviation (missing colon, non-integer parts, extra colons, empty parts, out-of-range ints).
PRE: [PRE-001 hard] raw is a string -- isinstance(raw, str)
POST: [POST-001 return_value] returns SseId with both fields >= 1 -- assert result.turn_id >= 1 and result.seq >= 1
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 turn_id < 1 OR seq < 1: RAISE ValueError(f"expected both ints >= 1 per spec, 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_seq_one [happy]: "1:1" → SseId(turn_id=1, seq=1) (the smallest valid id per spec — first event of the first turn)
  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
  zero_turn_id [adversarial]: "0:3" → ValueError (server invariant: turn_id from SQLite autoincrement starts at 1)
  zero_seq [adversarial]: "1:0" → ValueError (server invariant: seq starts at 1, per spec §SSE id format line 705)
  negative_turn_id [adversarial]: "-1:3" → ValueError
  negative_seq [adversarial]: "42:-1" → 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)
```

## Amendment 2026-06-30 — shared resume orchestration (v1 coverage-audit, slice b1)

The original contract specs resume as **caller-owned** (§Resume semantics: "the
caller MAY invoke `reconnect_turn`"). The v1 coverage-audit found `reconnect_turn`
had **no caller** — every presenter (cli/tui/web) let a mid-stream drop propagate
instead of resuming, so the "reference SSE-resume implementation" (design-brief
§3/§8d) was unreachable. Per design-brief §8b ("share the consumer, branch the
presenter") the resume loop is a **single shared orchestration surface**, not
duplicated per presenter. This adds `stream_turn_resilient` as that surface;
presenters call it instead of `stream_turn` when they want transparent reconnect.
`stream_turn` and `reconnect_turn` are unchanged (still the primitives); this is
purely additive.

```contract
FN stream_turn_resilient(client: httpx.AsyncClient, session_id: str, content: str, *, max_reconnects: int = 5) -> AsyncIterator[Event]
BRIEF: The shared resume-orchestration wrapper over stream_turn + reconnect_turn. Yields a SINGLE continuous typed Event stream; on SseConnectionDropped (mid-stream drop OR clean EOF before terminal), transparently resumes via reconnect_turn from the last-seen sse_id, up to max_reconnects times, until a terminal Done/Error/Cancelled arrives. The one surface all presenters consume for resilient streaming (design-brief §8b). Cross-process resume stays deferred to v2 (§8d): last-seen lives only in this generator's frame.
PRE: [PRE-001 hard] client is not None -- assert client is not None
PRE: [PRE-002 hard] session_id is non-empty str -- assert session_id and isinstance(session_id, str)
PRE: [PRE-003 hard] content is non-empty str -- assert content and isinstance(content, str)
PRE: [PRE-004 hard] max_reconnects is a non-negative int -- assert isinstance(max_reconnects, int) and max_reconnects >= 0
POST: [POST-001 return_value] yielded events are the concatenation of each attempt's events in wire order; the wrapper does NOT re-yield events it already saw (the server replays only seq>last_seen) -- assert seq is non-decreasing within a turn_id across the seam
POST: [POST-002 return_value] a fully-consumed stream terminates at exactly one Done/Error/Cancelled (INV-001 holds across reconnects) -- assert isinstance(events[-1], (Done, Error, Cancelled))
POST: [POST-003 state_change] reconnect_turn is invoked with last_event_id == f"{last_seen.turn_id}:{last_seen.seq}" of the most recently yielded event -- assert the Last-Event-ID header on attempt N+1 == the last sse_id yielded before the drop
ERROR_ROUTING:
  SseConnectionDropped (from stream_turn or reconnect_turn):
    local_handling: IF a last-seen sse_id exists AND reconnects < max_reconnects → increment reconnects, resume via reconnect_turn(last_event_id=f"{turn_id}:{seq}"); ELSE re-raise
    flow_control: continue (resume) | abort (re-raise when no last-seen id, or budget exhausted)
    state_recovery: server replays buffered events seq>last_seen then streams live (spec §Reconnect flow)
  ResumeBufferExpired | ResumeTurnFinished | InvalidLastEventId | TurnIdFlip | SseConnectFailed (from reconnect_turn):
    local_handling: propagate unchanged — NOT a transient drop; caller policy is abandon/restart (§Resume semantics "surface, not recover")
    flow_control: abort
    state_recovery: none
STEPS:
  1. [setup, flexibility=prescriptive] Validate PRE-001..PRE-004; SET last_seen=None, reconnects=0, gen=stream_turn(client, session_id, content)
  2. [loop, flexibility=prescriptive] async-for event in gen: SET last_seen=event.sse_id; YIELD event. On clean generator completion (terminal reached): RETURN.
  3. [branch, flexibility=prescriptive] ON SseConnectionDropped d: SET seen = last_seen or d.last_seen_sse_id. IF seen is None OR reconnects >= max_reconnects: RE-RAISE. ELSE: reconnects += 1; gen = reconnect_turn(client, session_id, content, last_event_id=f"{seen.turn_id}:{seen.seq}"); GOTO step 2.
  4. [error_handler, flexibility=prescriptive] Any non-drop exception from gen (ResumeBufferExpired/ResumeTurnFinished/InvalidLastEventId/TurnIdFlip/SseConnectFailed) is NOT caught — it propagates unchanged.
TESTS:
  happy_no_drop [happy]: stream yields text(42:1), done(42:2) cleanly → wrapper yields exactly those 2; endpoint hit ONCE (no reconnect).
  resume_after_one_drop [scenario,tracer]: attempt 1 yields text(42:1) then RemoteProtocolError; reconnect replays text(42:2)+done(42:3) → wrapper yields 42:1,42:2,42:3 as ONE stream; 2nd request carried Last-Event-ID "42:1".
  resume_after_clean_eof [scenario]: attempt 1 yields text(42:1) then clean EOF (no terminal); reconnect yields done(42:2) → continuous (resumes on the INV-001 clean-eof drop too).
  two_drops_then_done [scenario]: drops after 42:1 then after 42:2; third attempt yields done(42:3) → all 3 events; reconnects==2; Last-Event-ID headers "42:1" then "42:2".
  unresumable_zero_event_drop [adversarial]: attempt 1 drops with ZERO events seen (last_seen None) → SseConnectionDropped propagates; only 1 request issued.
  max_reconnects_exhausted [adversarial]: every attempt drops after one event; max_reconnects=2 → after initial + 2 reconnects (3 requests), SseConnectionDropped propagates.
  buffer_expired_propagates [error]: attempt 1 drops after 42:1; reconnect returns 412 → ResumeBufferExpired propagates (not retried as a transient drop).
  zero_budget_no_resume [adversarial]: max_reconnects=0; attempt 1 drops after 42:1 → SseConnectionDropped propagates immediately (no reconnect attempted).
```
