Volva's contract paraphrase round (thread 01KS4B3B0Y62) surfaced three
real contract-time ambiguities — addressing each here before applying
ready-for-agent.
1) reconnect_turn body was a punt. STEP 2 literally said "json={'content':
''} OR with no body (TBD per spec — confirm during implementation)".
The spec §Reconnect flow example shows POST with Content-Type:
application/json and a body shaped {"content": "..."} — the wire schema
requires content; the server identifies the resume target via the
Last-Event-ID header and does NOT re-process content (spec line 732:
"agent's tools and LLM call run exactly once regardless of disconnects/
reconnects"). reconnect_turn now takes content: str explicitly; STEP 2
sends json={"content": content}. Caller convention: pass the original
content sent to stream_turn. POST-002 added to assert byte-for-byte
body shape; new test body_threads_content covers it.
2) _parse_sse_id allowed turn_id and seq ≥ 0 — too loose. Spec §SSE id
format line 705 says seq starts at 1 (resets per turn); turn_id is
from SQLite turns.id (autoincrement, ≥1). Tightened POST-001 to require
both ≥1; STEP 5 raises ValueError on either < 1. Test happy_zero_seq
flipped to adversarial zero_seq; added zero_turn_id and negative_seq.
INV-002 tightened to reflect the same range.
3) INV-003 (TurnIdFlip) had a subtle wording gap between stream_turn
(first event ESTABLISHES turn_id; cannot be a flip) and reconnect_turn
(expected turn_id parsed FROM last_event_id BEFORE connect; first event
is already a flip-candidate). Volva noticed the reconnect test said
"first event was not yielded" while stream_turn semantics depend on
the first event being yielded unconditionally. Spelled out both entry
points in INV-003 as a numbered sub-list. reconnect_turn STEP 4 and
test turn_id_flip_on_first_event reworded to match.
Volva flags #3 (MalformedSseId-vs-ValueError split) and #5 (exactly-
one-terminal as server-assumed, not client-verified) reviewed and kept
as-is — both intentional. Drift check unchanged: amending the contract
does not touch the pinned issue body, so prd: hashes remain valid.
27 KiB
contract_version, target_module, scope, depends_on, used_by, language, complexity, estimated_loc, confidence, prd, assumptions, open_questions
| contract_version | target_module | scope | depends_on | used_by | language | complexity | estimated_loc | confidence | prd | assumptions | open_questions | ||||||||||||||||||||||||||
|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|
| 2.1 | ratatoskr.sse_client | 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). |
|
|
python | high | 320 | 0.85 |
|
|
|
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 fromPOST /sessionsorGET /sessions.content: str— user message body (forstream_turn).last_event_id: str | None— composite{turn_id}:{seq}to resume from (forreconnect_turn); the same string the server emitted as theid:line of the last successfully-consumed event.turn_id: int— forcancel_turn; obtained from any prior event's parsedid:orturn_idbody field.
Output:
stream_turn/reconnect_turn:AsyncIterator[Event]— typed discriminated union ofWorkerPhase | Thinking | Text | TextBoundary | ToolStart | ToolResult | Done | Error | Cancelled. Each event carries the parsed JSON fields per the spec's §SSE Event Types, PLUS the parsedsse_id: SseId = (turn_id, seq)lifted from the SSE wireid:field. Always-final event is exactly one ofDone | Error | Cancelled(spec invariant).cancel_turn:CancelResultdataclass —{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_turniteration terminates with exactly one terminal event of typedone,error, orcancelled. The iterator MUST NOT raise StopAsyncIteration before a terminal event arrives unless the underlying HTTP connection drops mid-stream — in which case it raisesSseConnectionDropped(caller may invokereconnect_turnif it has the last seensse_id). - INV-002 [hard]: Every yielded
Eventcarries its parsedsse_id: SseId = (turn_id: int, seq: int)withturn_id ≥ 1andseq ≥ 1(server-side spec invariants). Events whose SSEid:wire field is absent, malformed, non-composite (123instead of123:4), or carries an out-of-range integer MUST raiseMalformedSseIdbefore yielding. This is the foot-gun that hand-rolleddata:-only parsing silently drops; the module's value is making it impossible to silently drop. (Seedocs/design-brief.md§3.) - INV-003 [hard]: The composite
sse_id'sturn_idcomponent is stable across all events of a single turn (spec INV-014). The module MUST detect a turn_id flip and raiseTurnIdFliprather 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 ofturn_id. The first event yielded establishesestablished_turn_id; subsequent events must match it. The first event is always yielded (it cannot be a flip). - In
reconnect_turn,last_event_idcarries the caller's prior expectation.established_turn_idis parsed FROMlast_event_idBEFORE the connection opens. The first event from the server is therefore already a flip-candidate — if itsturn_iddiffers from the expected one, the module raisesTurnIdFlipBEFORE yielding it.
- In
- INV-004 [hard]:
reconnect_turnsets theLast-Event-IDHTTP 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.*orworldtree.*imports. Boundary verified by the existingtests/test_no_worldtree_imports.py. - INV-006 [soft, recovery_window=1]: The
Eventyielded immediately AFTER aworker_phaseevent MAY belong to the next phase (e.g., atextevent afterworker_phase Streaming). Phase ordering invariant is server-side, not client-side; the consumer just forwards what arrives. - INV-007 [hard]:
cancel_turnis decoupled from the SSE stream — the HTTP 200 from the cancel endpoint does NOT mean the stream has emitted itscancelledterminal event yet. Callers driving two-stage Ctrl-C must keep iterating the stream aftercancel_turnreturns until theCancelledevent 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:
- The caller holds the most recently yielded event's
sse_idAND the originalcontentit sent tostream_turn. (Easy to thread:last_seen = event.sse_idon every iteration;contentstays in the caller's frame.) - On any
SseConnectionDropped, the caller MAY invokereconnect_turn(client, session_id, content, last_event_id=f"{last_seen.turn_id}:{last_seen.seq}"). Thecontentargument is sent verbatim in the JSON body (spec wire requirement) but the server identifies the resume target byLast-Event-IDand does NOT re-processcontent(spec §Reconnect flow). - The server replays buffered events with
seq > last_seen.seqfor thatturn_id, then resumes live streaming. The module yields them as it received them. - If the server returns 412
buffer_expired, the module raisesResumeBufferExpired(buffered_from_seq=X). Caller's policy. - If the server returns 410
turn_finished, the module raisesResumeTurnFinished(turn_id=N). The turn completed while the connection was dropped; caller canGET /sessions/{id}/messagesto 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 (seetests/snapshots/README.md). - [performance] Streaming MUST NOT buffer the full turn in memory — events are yielded as they arrive. The complete-response field on
Doneis what the server sends; the consumer does not re-aggregate fromtextevents. - [security]
Authorizationheader lives on the caller'shttpx.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.
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)
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
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)
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)
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)