feat(sse_client): implement issue #1 contract via TDD

Implements docs/contracts/issues/1.contract.md. Four entry points (stream_turn, reconnect_turn, cancel_turn, _parse_sse_id) + nine typed Event variants + ten domain exceptions. 37 tests covering every TESTS: entry verbatim, plus the boundary smoke test still passes. Tracer-bullet ordering per the contract's per-FN tracer tags: _parse_sse_id (foundation; happy_simple) → stream_turn (happy_one_text_done) → reconnect_turn (happy_resume_from_seq_3) → cancel_turn (happy_cancel). Each FN's tracer went RED then GREEN before its other tests landed. Shared SSE-iteration logic (INV-002 sse_id presence + INV-003 turn_id stability + terminal-break) lives in private _iter_events helper. expected_turn_id=None gives stream_turn's "establish from first event" semantics; expected_turn_id=N gives reconnect_turn's "first event is already a flip-candidate" semantics — the two-entry-point distinction Volva surfaced during the paraphrase round. A few implementation choices worth recording: - _parse_sse_id uses a `^-?\\d+$` regex pre-check to reject any whitespace before int() is called. Python's `int(" 3 ")` silently strips, which would have made the trailing_whitespace adversarial test pass for the wrong reason. - The connection_drop test uses a custom httpx.AsyncByteStream subclass (_DropAfter) that yields chunks then raises RemoteProtocolError mid-stream. respx alone can't simulate mid-stream HTTP errors. - ToolResult.result and ToolStart.arguments are typed as Any because the server's tool wire shape varies per tool; the spec doesn't pin a generic schema. - Boundary smoke test (no core.* / worldtree.* imports under src/ratatoskr/) still GREEN — INV-005 holds. Also: one E501 line-length fix in test_no_worldtree_imports.py that ruff flagged once the new tests pulled it into scope.
2026-05-20 21:25:20 -07:00
parent 1526f0bc8e
commit 02f2a04b37
4 changed files with 1049 additions and 7 deletions
@@ -30,8 +30,7 @@ separate dev team rather than an in-tree Worldtree tool.

 ## Current state / in-flight

-**Status: scaffold verified + first contract authored.** Design is locked;
-implementation begins next, TDD against `sse_client.contract.md`.
+**Status: `ratatoskr.sse_client` implemented via TDD against issue #1's contract.** 38/38 tests GREEN; ruff clean; boundary smoke (`tests/test_no_worldtree_imports.py`) still passes.

 What's in the repo:
 - `docs/design-brief.md` — the locked design (copy from `brokkr-smithy/docs/ratatoskr-design-brief.md`).
@@ -41,11 +40,12 @@ What's in the repo:
 - `docs/contracts/issues/1.contract.md` — **issue-scoped contract for issue #1** (https://gitea.phasefinal.com/vh/ratatoskr/issues/1). v2.1, complexity=high. `target_module: ratatoskr.sse_client`. `prd:` block pins to issue body SHA `abcbc49467e86f1d` at `2026-05-21T03:57:37+00:00`. Four FN blocks: `stream_turn`, `reconnect_turn`, `cancel_turn`, `_parse_sse_id`. Drift check (`scripts/contract_drift_check.py`) returns clean.
 - `pyproject.toml` — Python 3.12, hatchling, uv-managed. Deps: httpx, httpx-sse, textual. Dev deps: pytest, pytest-asyncio, respx, ruff, mypy, textual-dev, pyyaml (consumed by `docs/contracts/contract_parser.py` + `scripts/contract_drift_check.py`).
 - `src/ratatoskr/__init__.py` + `cli.py` — stubs.
+- `src/ratatoskr/sse_client.py` — **implemented 2026-05-21** per `docs/contracts/issues/1.contract.md`. Four public entry points + nine typed Event variants + ten custom exceptions. Shared SSE-iteration logic (INV-002 + INV-003 + terminal-break) lives in private `_iter_events(event_source, *, expected_turn_id)` helper consumed by both `stream_turn` and `reconnect_turn` — `expected_turn_id=None` triggers "establish from first event" semantics, `expected_turn_id=N` triggers "first event is already a flip-candidate" semantics (the two-entry-point distinction Volva surfaced).
+- `tests/test_sse_client.py` — 37 tests covering all four FN blocks' TESTS: entries verbatim (13 + 10 + 8 + 6). Real HTTP wire via respx mocks; SSE wire format constructed by helper `_sse_chunk`. Connection-drop test uses custom `httpx.AsyncByteStream` subclass that yields chunks then raises `RemoteProtocolError`.
 - `tests/test_no_worldtree_imports.py` — boundary smoke test (passes; verified 2026-05-20).
 - `tests/snapshots/README.md` — recording/replay convention for SSE snapshot tests.

 What's NOT in the repo yet:
- Implementation of `ratatoskr.sse_client` — next move, TDD per the contract's tracer tests.
 - Gitea remote — operator provided `git@gitea.phasefinal.com:vh/ratatoskr.git` on 2026-05-20; about to be added + first push at the same commit as this update.
 - CLAUDE.md customization — currently using the canonical template's
  generic CLAUDE.md. The dev team may want to add Ratatoskr-specific
@@ -54,9 +54,10 @@ What's NOT in the repo yet:
 **Branch:** `main`. Remote: `origin → git@gitea.phasefinal.com:vh/ratatoskr.git` (added 2026-05-20).

 **Next natural moves:**
-1. Implement `ratatoskr.sse_client` via TDD per `docs/contracts/sse_client.contract.md`. Vertical slice — start with `_parse_sse_id` (the foundation; tracer test `happy_simple`), then `stream_turn` (tracer test `happy_one_text_done`), then `reconnect_turn` (tracer test `happy_resume_from_seq_3`), then `cancel_turn` (tracer test `happy_cancel`). Each FN's tracer test gets RED → GREEN before any other test in that FN.
-2. Record real SSE snapshot fixtures from a running Worldtree (per `tests/snapshots/README.md`) once `stream_turn` is GREEN against mocks — gives version-skew detection per design-brief §2.
-3. Build presenters: `--send` stdout presenter first (simplest consumer of `stream_turn`); Textual TUI second. Both consume the same `Event` iterator.
+1. Record real SSE snapshot fixtures from a running Worldtree (per `tests/snapshots/README.md`). Current tests use respx mocks against hand-rolled SSE wire — recording against a real Worldtree exercises spec conformance and gives version-skew detection per design-brief §2.
+2. Build the `--send` stdout presenter — simplest consumer of `stream_turn`, exercises the API path without TUI machinery. Useful as a tracer for `ratatoskr.cli` work.
+3. Textual TUI app shell — second presenter; multi-pane observability dashboard per design-brief §5. Layout shape locked there (Horizontal split, left=chat, right=TabbedContent with persona/tools/admin/bifrost/server-log).
+4. Run `/sleipnir-preflight 1` if any AFK dispatch is wanted for these follow-ups, but most of this is hands-on dev work.

 ## Recent decisions

@@ -77,6 +78,7 @@ decision. Captures rationale that won't be obvious from code alone.
 - `[2026-05-20]` **First contract: `ratatoskr.sse_client`.** Bundles `stream_turn` + `reconnect_turn` + `cancel_turn` + private `_parse_sse_id` into one module — the SSE-resume flow is coupled (cancel needs `turn_id` from the SSE wire `id:`, reconnect re-uses the same parsed `SseId`), so they share a contract. Hard invariant INV-002 makes the composite `{turn_id}:{seq}` `id:` parsing load-bearing — closes the foot-gun the design-brief §3 names (hand-rolled `data:`-only parsing silently drops the `id:`). v2.1 test categories `adversarial`/`scenario`/`trace` used freely; parser warns but format spec §2.1.E permits them.
 - `[2026-05-21]` **Contract converted to issue-scoped (issue #1).** Moved `docs/contracts/sse_client.contract.md` → `docs/contracts/issues/1.contract.md`. Frontmatter shape switched from module-scoped (`module:`/`purpose:`) to issue-scoped (`target_module:`/`scope:`/`prd:`) per CONTRACT-FORMAT §2.1.I. `prd:` block pins to issue #1's body hash (`abcbc49467e86f1d`). `scripts/contract_drift_check.py` returns clean. **Known parser stale-ness**: `contract_parser.py --validate` ERRORs on issue-scoped frontmatter (missing `module:`/`purpose:`) — this is CONTRACT-FORMAT §2.1.L H10, a documented Brokkr-side follow-up. Parser is a canonical sync, so we do NOT patch it locally (would drift from canonical). Treat parser ERROR-on-issue-scoped as expected until the canonical bumps.
 - `[2026-05-21]` **Default issue-tracker labels seeded** (17 total). Sleipnir gating (`ready-for-agent`, `blocked-needs-contract`, `blocked-needs-dependency`), triage (`needs-triage`, `needs-architect-decision`, `needs-info`), type (`bug`, `enhancement`, `task`, `documentation`), resolution (`duplicate`, `wontfix`, `invalid`), Ratatoskr-specific area (`sse-client`, `tui`, `cli`, `observability`).
+- `[2026-05-21]` **`ratatoskr.sse_client` implemented via TDD against issue #1's contract.** 37 contract-listed tests authored + GREEN per the tracer-bullet vertical-slice ordering (`_parse_sse_id` → `stream_turn` → `reconnect_turn` → `cancel_turn`). Refactor pass extracted `_iter_events` helper to dedupe INV-002 + INV-003 + terminal-break logic across `stream_turn` and `reconnect_turn`; `expected_turn_id=None` vs `expected_turn_id=N` distinguishes the two entry-point semantics Volva surfaced. Notable choices made during implementation: (a) regex `^-?\d+$` pre-check in `_parse_sse_id` to reject whitespace before `int()` (Python's `int(" 3 ")` would silently strip — this kept the strict-no-whitespace test honest); (b) `_DropAfter` AsyncByteStream subclass in tests to simulate mid-stream `RemoteProtocolError`; (c) ToolResult.result and ToolStart.arguments typed as `Any` (server JSON varies); (d) ruff line-length=100 (per pyproject) forced some test docstrings to be tighter than v0 draft.
 - `[2026-05-21]` **Volva paraphrase round on `docs/contracts/issues/1.contract.md`.** Volva flagged 5 ambiguities; operator approved amendments to 3 of them. (1) `reconnect_turn` STEP 2 punt resolved: signature now carries `content: str`; STEP 2 body is `json={"content": content}` matching spec §Reconnect flow example verbatim. Spec line 732 makes the agent's tools+LLM run "exactly once regardless of disconnects/reconnects" — the `content` is a wire-schema requirement, not re-processed server-side. (2) `_parse_sse_id` tightened: `turn_id ≥ 1` AND `seq ≥ 1` (was `≥ 0`); spec §SSE id format line 705 explicitly states `seq` starts at 1, and `turn_id` is SQLite autoincrement (≥1). Test `happy_zero_seq` flipped to `zero_seq [adversarial]`; new `zero_turn_id` + `negative_seq` adversarial tests added. (3) INV-003 clarified to spell out the two-entry-point semantics: `stream_turn` establishes `turn_id` from the first event (first event always yields); `reconnect_turn` parses the expected `turn_id` FROM `last_event_id` BEFORE the connection opens, so the first server event is already a flip-candidate and is NOT yielded on mismatch. Volva flags #3 (MalformedSseId-vs-ValueError split) and #5 (exactly-one-terminal as server-assumed) noted but kept as-is — deliberate distinctions. Drift check still clean against issue #1 (amending the contract doesn't touch the pinned issue body).

 ## Tried and abandoned
@@ -0,0 +1,413 @@
+"""SSE consumer for the Worldtree Conversation API.
+
+Implements docs/contracts/issues/1.contract.md.
+"""
+
+from __future__ import annotations
+
+import json
+import re
+from collections.abc import AsyncIterator
+from dataclasses import dataclass
+from typing import Any, NamedTuple
+
+import httpx
+import httpx_sse
+
+_INT_RE = re.compile(r"^-?\d+$")
+
+
+class SseId(NamedTuple):
+    """Parsed composite SSE wire `id:` per spec §SSE id format."""
+
+    turn_id: int
+    seq: int
+
+
+@dataclass(frozen=True)
+class WorkerPhase:
+    """SSE event `worker_phase`: agent entered a new processing phase."""
+
+    sse_id: SseId
+    phase: str
+    turn_id: int
+
+
+@dataclass(frozen=True)
+class Thinking:
+    """SSE event `thinking`: incremental thinking content from thinking-enabled models."""
+
+    sse_id: SseId
+    content: str
+
+
+@dataclass(frozen=True)
+class Text:
+    """SSE event `text`: an incremental response-text delta."""
+
+    sse_id: SseId
+    content: str
+
+
+@dataclass(frozen=True)
+class TextBoundary:
+    """SSE event `text_boundary`: speakable breakpoint after a `text` event."""
+
+    sse_id: SseId
+    kind: str
+    char_offset: int
+    ts: str
+
+
+@dataclass(frozen=True)
+class ToolStart:
+    """SSE event `tool_start`: agent is about to execute a tool."""
+
+    sse_id: SseId
+    name: str
+    arguments: dict[str, Any]
+
+
+@dataclass(frozen=True)
+class ToolResult:
+    """SSE event `tool_result`: a tool call completed."""
+
+    sse_id: SseId
+    name: str
+    result: Any
+    duration_ms: int
+
+
+@dataclass(frozen=True)
+class Done:
+    """Terminal SSE event `done`: turn succeeded."""
+
+    sse_id: SseId
+    phase: str
+    response: str
+    model: str
+    duration_ms: int
+    usage: dict[str, int]
+
+
+@dataclass(frozen=True)
+class Error:
+    """Terminal SSE event `error`: turn failed."""
+
+    sse_id: SseId
+    phase: str
+    message: str
+    error_code: str | None
+
+
+@dataclass(frozen=True)
+class Cancelled:
+    """Terminal SSE event `cancelled`: turn was cancelled server-side."""
+
+    sse_id: SseId
+    phase: str
+    turn_id: int
+    reason: str | None
+    partial_message_id: int | None
+
+
+Event = (
+    WorkerPhase
+    | Thinking
+    | Text
+    | TextBoundary
+    | ToolStart
+    | ToolResult
+    | Done
+    | Error
+    | Cancelled
+)
+
+
+class MalformedSseId(Exception):
+    """Raised when an SSE event's `id:` wire field is missing or non-composite."""
+
+    def __init__(self, raw: str) -> None:
+        super().__init__(f"malformed SSE id: {raw[:64]!r}")
+        self.raw = raw[:64]
+
+
+class TurnIdFlip(Exception):
+    """Raised when an SSE event's turn_id doesn't match the established/expected turn."""
+
+    def __init__(self, *, established: int, got: int) -> None:
+        super().__init__(f"turn_id flip: established={established}, got={got}")
+        self.established = established
+        self.got = got
+
+
+class SseConnectFailed(Exception):
+    """Raised when the SSE endpoint returned a non-2xx status before the stream opened."""
+
+    def __init__(self, *, status: int, body: bytes) -> None:
+        super().__init__(f"SSE connect failed: status={status}, body={body[:128]!r}")
+        self.status = status
+        self.body = body
+
+
+class SseConnectionDropped(Exception):
+    """Raised when the HTTP/SSE connection dropped mid-stream."""
+
+    def __init__(self, *, last_seen_sse_id: SseId | None) -> None:
+        super().__init__(f"SSE connection dropped; last_seen_sse_id={last_seen_sse_id}")
+        self.last_seen_sse_id = last_seen_sse_id
+
+
+class InvalidLastEventId(Exception):
+    """Raised on HTTP 400 from a reconnect request — caller's last_event_id was rejected."""
+
+    def __init__(self, *, raw: str) -> None:
+        super().__init__(f"server rejected Last-Event-ID: {raw!r}")
+        self.raw = raw
+
+
+class ResumeTurnFinished(Exception):
+    """Raised on HTTP 410 from a reconnect — turn finished while disconnected."""
+
+    def __init__(self, *, turn_id: int) -> None:
+        super().__init__(f"resume target finished: turn_id={turn_id}")
+        self.turn_id = turn_id
+
+
+class ResumeBufferExpired(Exception):
+    """Raised on HTTP 412 from a reconnect — Last-Event-ID older than replay buffer."""
+
+    def __init__(self, *, turn_id: int, buffered_from_seq: int) -> None:
+        super().__init__(
+            f"replay buffer expired: turn_id={turn_id}, buffered_from_seq={buffered_from_seq}"
+        )
+        self.turn_id = turn_id
+        self.buffered_from_seq = buffered_from_seq
+
+
+class CancelTurnNotFound(Exception):
+    """Raised on HTTP 404 from cancel — turn finished before cancel arrived (race)."""
+
+    def __init__(self, *, turn_id: int) -> None:
+        super().__init__(f"cancel target not found: turn_id={turn_id}")
+        self.turn_id = turn_id
+
+
+class CancelAlreadyCompleted(Exception):
+    """Raised on HTTP 409 from cancel — turn already terminated server-side (race)."""
+
+    def __init__(self, *, turn_id: int) -> None:
+        super().__init__(f"cancel: turn already completed: turn_id={turn_id}")
+        self.turn_id = turn_id
+
+
+class CancelFailed(Exception):
+    """Raised on unexpected cancel response status."""
+
+    def __init__(self, *, status: int, body: bytes) -> None:
+        super().__init__(f"cancel failed: status={status}, body={body[:128]!r}")
+        self.status = status
+        self.body = body
+
+
+@dataclass(frozen=True)
+class CancelResult:
+    """Response envelope from POST /sessions/{id}/turns/{turn_id}/cancel."""
+
+    turn_id: int
+    cancelled: bool
+    reason: str | None
+    partial_message_id: int | None
+
+
+def _envelope_for_type(body: dict[str, Any], sse_id: SseId) -> Event:
+    """Dispatch a parsed JSON body to its typed Event variant."""
+    t = body["type"]
+    if t == "text":
+        return Text(sse_id=sse_id, content=body["content"])
+    if t == "worker_phase":
+        return WorkerPhase(sse_id=sse_id, phase=body["phase"], turn_id=body["turn_id"])
+    if t == "thinking":
+        return Thinking(sse_id=sse_id, content=body["content"])
+    if t == "text_boundary":
+        return TextBoundary(
+            sse_id=sse_id,
+            kind=body["kind"],
+            char_offset=body["char_offset"],
+            ts=body["ts"],
+        )
+    if t == "tool_start":
+        return ToolStart(sse_id=sse_id, name=body["name"], arguments=body["arguments"])
+    if t == "tool_result":
+        return ToolResult(
+            sse_id=sse_id,
+            name=body["name"],
+            result=body["result"],
+            duration_ms=body["duration_ms"],
+        )
+    if t == "done":
+        return Done(
+            sse_id=sse_id,
+            phase=body["phase"],
+            response=body["response"],
+            model=body["model"],
+            duration_ms=body["duration_ms"],
+            usage=body["usage"],
+        )
+    if t == "error":
+        return Error(
+            sse_id=sse_id,
+            phase=body.get("phase", "failed"),
+            message=body.get("message", ""),
+            error_code=body.get("error_code"),
+        )
+    if t == "cancelled":
+        return Cancelled(
+            sse_id=sse_id,
+            phase=body["phase"],
+            turn_id=body["turn_id"],
+            reason=body.get("reason"),
+            partial_message_id=body.get("partial_message_id"),
+        )
+    raise ValueError(f"unknown SSE event type: {t!r}")
+
+
+async def _iter_events(
+    event_source: httpx_sse.EventSource,
+    *,
+    expected_turn_id: int | None,
+) -> AsyncIterator[Event]:
+    """Apply INV-002 (sse_id present + in range) and INV-003 (turn_id stable) per event.
+
+    `expected_turn_id=None` means "establish from the first event" (stream_turn semantics).
+    `expected_turn_id=N` means "every event must match N" (reconnect_turn semantics — the
+    first event is already a flip-candidate per INV-003).
+    """
+    established = expected_turn_id
+    last_sse_id: SseId | None = None
+    try:
+        async for sse in event_source.aiter_sse():
+            try:
+                sse_id = _parse_sse_id(sse.id)
+            except ValueError as exc:
+                raise MalformedSseId(raw=sse.id) from exc
+            if established is None:
+                established = sse_id.turn_id
+            elif sse_id.turn_id != established:
+                raise TurnIdFlip(established=established, got=sse_id.turn_id)
+            event = _envelope_for_type(json.loads(sse.data), sse_id=sse_id)
+            yield event
+            last_sse_id = sse_id
+            if isinstance(event, (Done, Error, Cancelled)):
+                return
+    except (httpx.ReadError, httpx.RemoteProtocolError) as exc:
+        raise SseConnectionDropped(last_seen_sse_id=last_sse_id) from exc
+
+
+async def stream_turn(
+    client: httpx.AsyncClient, session_id: str, content: str
+) -> AsyncIterator[Event]:
+    """POST a message and yield typed Events. See contract FN stream_turn."""
+    assert client is not None
+    assert session_id and isinstance(session_id, str)
+    assert content and isinstance(content, str)
+
+    async with httpx_sse.aconnect_sse(
+        client,
+        "POST",
+        f"/sessions/{session_id}/messages",
+        json={"content": content},
+    ) as event_source:
+        try:
+            event_source.response.raise_for_status()
+        except httpx.HTTPStatusError as exc:
+            body = await exc.response.aread()
+            raise SseConnectFailed(status=exc.response.status_code, body=body) from exc
+        async for event in _iter_events(event_source, expected_turn_id=None):
+            yield event
+
+
+async def reconnect_turn(
+    client: httpx.AsyncClient,
+    session_id: str,
+    content: str,
+    last_event_id: str,
+) -> AsyncIterator[Event]:
+    """Re-POST with Last-Event-ID to resume. See contract FN reconnect_turn."""
+    assert client is not None
+    assert session_id and isinstance(session_id, str)
+    assert isinstance(content, str)
+    expected = _parse_sse_id(last_event_id)
+
+    async with httpx_sse.aconnect_sse(
+        client,
+        "POST",
+        f"/sessions/{session_id}/messages",
+        json={"content": content},
+        headers={"Last-Event-ID": last_event_id},
+    ) as event_source:
+        status = event_source.response.status_code
+        if status != 200:
+            body_bytes = await event_source.response.aread()
+            try:
+                body = json.loads(body_bytes)
+            except json.JSONDecodeError:
+                body = {}
+            if status == 400:
+                raise InvalidLastEventId(raw=last_event_id)
+            if status == 410:
+                raise ResumeTurnFinished(turn_id=body.get("turn_id", expected.turn_id))
+            if status == 412:
+                raise ResumeBufferExpired(
+                    turn_id=body.get("turn_id", expected.turn_id),
+                    buffered_from_seq=body.get("buffered_from_seq", 0),
+                )
+            raise SseConnectFailed(status=status, body=body_bytes)
+        async for event in _iter_events(event_source, expected_turn_id=expected.turn_id):
+            yield event
+
+
+def _parse_sse_id(raw: str) -> SseId:
+    """Parse the SSE wire `id:` as composite `{turn_id}:{seq}`. See contract FN _parse_sse_id."""
+    parts = raw.split(":")
+    if len(parts) != 2:
+        raise ValueError(f"expected '{{turn_id}}:{{seq}}', got: {raw[:64]!r}")
+    turn_id_str, seq_str = parts
+    if not _INT_RE.match(turn_id_str) or not _INT_RE.match(seq_str):
+        raise ValueError(f"expected '{{turn_id}}:{{seq}}' with decimal ints, got: {raw[:64]!r}")
+    turn_id = int(turn_id_str)
+    seq = int(seq_str)
+    if turn_id < 1 or seq < 1:
+        raise ValueError(f"expected both ints >= 1 per spec, got: {raw[:64]!r}")
+    return SseId(turn_id=turn_id, seq=seq)
+
+
+async def cancel_turn(
+    client: httpx.AsyncClient,
+    session_id: str,
+    turn_id: int,
+    *,
+    persist_partial: bool = False,
+) -> CancelResult:
+    """POST /sessions/{id}/turns/{turn_id}/cancel. See contract FN cancel_turn."""
+    assert client is not None
+    assert session_id and isinstance(session_id, str)
+    assert isinstance(turn_id, int) and turn_id > 0
+
+    params = {"persist_partial": "true"} if persist_partial else None
+    resp = await client.post(
+        f"/sessions/{session_id}/turns/{turn_id}/cancel", params=params
+    )
+    if resp.status_code == 404:
+        raise CancelTurnNotFound(turn_id=turn_id)
+    if resp.status_code == 409:
+        raise CancelAlreadyCompleted(turn_id=turn_id)
+    if resp.status_code != 200:
+        raise CancelFailed(status=resp.status_code, body=resp.content)
+    body = resp.json()
+    return CancelResult(
+        turn_id=body["turn_id"],
+        cancelled=body["cancelled"],
+        reason=body.get("reason"),
+        partial_message_id=body.get("partial_message_id"),
+    )
@@ -36,7 +36,9 @@ def test_no_worldtree_imports() -> None:
                line = text[: match.start()].count("\n") + 1
                violations.append((path, f"line {line}: {match.group(0).strip()}"))
    if violations:
-        report = "\n".join(f"  {p.relative_to(_SRC_ROOT.parent.parent)}: {v}" for p, v in violations)
+        report = "\n".join(
+            f"  {p.relative_to(_SRC_ROOT.parent.parent)}: {v}" for p, v in violations
+        )
        raise AssertionError(
            "Ratatoskr must not import from Worldtree source. Violations:\n"
            f"{report}\n\n"
@@ -0,0 +1,625 @@
+"""Tests for ratatoskr.sse_client per docs/contracts/issues/1.contract.md."""
+
+import httpx
+import pytest
+import respx
+
+from ratatoskr.sse_client import (
+    CancelAlreadyCompleted,
+    Cancelled,
+    CancelResult,
+    CancelTurnNotFound,
+    Done,
+    Error,
+    InvalidLastEventId,
+    MalformedSseId,
+    ResumeBufferExpired,
+    ResumeTurnFinished,
+    SseConnectFailed,
+    SseId,
+    Text,
+    TurnIdFlip,
+    _parse_sse_id,
+    cancel_turn,
+    reconnect_turn,
+    stream_turn,
+)
+
+_DONE_42_6 = {
+    "type": "done",
+    "phase": "succeeded",
+    "response": "hello",
+    "model": "m",
+    "duration_ms": 1,
+    "usage": {
+        "prompt_tokens": 0,
+        "completion_tokens": 0,
+        "total_tokens": 0,
+        "cached_input_tokens": 0,
+    },
+}
+
+
+def _sse_chunk(sse_id: str, body: dict[str, object]) -> bytes:
+    """Compose one SSE event in wire format. Trailing blank line per spec."""
+    import json
+
+    return f"id: {sse_id}\ndata: {json.dumps(body)}\n\n".encode()
+
+
+class TestParseSseId:
+    def test_happy_simple(self) -> None:
+        """happy_simple [happy,tracer]: '42:3' -> SseId(turn_id=42, seq=3)."""
+        assert _parse_sse_id("42:3") == SseId(turn_id=42, seq=3)
+
+    def test_happy_seq_one(self) -> None:
+        """happy_seq_one: smallest valid id per spec — first event of first turn."""
+        assert _parse_sse_id("1:1") == SseId(turn_id=1, seq=1)
+
+    def test_empty(self) -> None:
+        """empty [adversarial]: '' -> ValueError."""
+        with pytest.raises(ValueError):
+            _parse_sse_id("")
+
+    def test_no_colon(self) -> None:
+        """no_colon [adversarial]: '42' -> ValueError with '42' in the message."""
+        with pytest.raises(ValueError, match="42"):
+            _parse_sse_id("42")
+
+    def test_too_many_colons(self) -> None:
+        """too_many_colons [adversarial]: '42:3:7' -> ValueError."""
+        with pytest.raises(ValueError):
+            _parse_sse_id("42:3:7")
+
+    def test_alpha_turn_id(self) -> None:
+        """alpha_turn_id [adversarial]: 'foo:3' -> ValueError."""
+        with pytest.raises(ValueError):
+            _parse_sse_id("foo:3")
+
+    def test_alpha_seq(self) -> None:
+        """alpha_seq [adversarial]: '42:bar' -> ValueError."""
+        with pytest.raises(ValueError):
+            _parse_sse_id("42:bar")
+
+    def test_zero_turn_id(self) -> None:
+        """zero_turn_id [adversarial]: '0:3' -> ValueError (turn_id ≥ 1)."""
+        with pytest.raises(ValueError):
+            _parse_sse_id("0:3")
+
+    def test_zero_seq(self) -> None:
+        """zero_seq [adversarial]: '1:0' -> ValueError (seq ≥ 1, spec line 705)."""
+        with pytest.raises(ValueError):
+            _parse_sse_id("1:0")
+
+    def test_negative_turn_id(self) -> None:
+        """negative_turn_id [adversarial]: '-1:3' -> ValueError."""
+        with pytest.raises(ValueError):
+            _parse_sse_id("-1:3")
+
+    def test_negative_seq(self) -> None:
+        """negative_seq [adversarial]: '42:-1' -> ValueError."""
+        with pytest.raises(ValueError):
+            _parse_sse_id("42:-1")
+
+    def test_trailing_whitespace(self) -> None:
+        """trailing_whitespace [adversarial]: '42:3 ' -> ValueError (strict; no strip)."""
+        with pytest.raises(ValueError):
+            _parse_sse_id("42:3 ")
+
+    def test_truncation(self) -> None:
+        """truncation [security]: 5000-char no-colon -> ValueError msg contains only raw[:64]."""
+        raw = "a" * 5000
+        with pytest.raises(ValueError) as exc_info:
+            _parse_sse_id(raw)
+        assert raw not in str(exc_info.value), "full 5000-char input must not appear in message"
+        assert raw[:64] in str(exc_info.value), "first 64 chars must appear in message"
+
+
+class TestStreamTurn:
+    @respx.mock
+    async def test_happy_one_text_done(self) -> None:
+        """happy_one_text_done: text then done; same turn_id; iter ends after done."""
+        stream = _sse_chunk("42:1", {"type": "text", "content": "hello"}) + _sse_chunk(
+            "42:2",
+            {
+                "type": "done",
+                "phase": "succeeded",
+                "response": "hello",
+                "model": "glm5-turbo",
+                "duration_ms": 100,
+                "usage": {
+                    "prompt_tokens": 0,
+                    "completion_tokens": 0,
+                    "total_tokens": 0,
+                    "cached_input_tokens": 0,
+                },
+            },
+        )
+        respx.post("https://w.example/sessions/s1/messages").mock(
+            return_value=httpx.Response(
+                200, headers={"content-type": "text/event-stream"}, content=stream
+            )
+        )
+        async with httpx.AsyncClient(base_url="https://w.example") as client:
+            events = [e async for e in stream_turn(client, "s1", "hi")]
+        assert len(events) == 2
+        assert isinstance(events[0], Text)
+        assert events[0].content == "hello"
+        assert events[0].sse_id == SseId(42, 1)
+        assert isinstance(events[1], Done)
+        assert events[1].sse_id == SseId(42, 2)
+        assert events[0].sse_id.turn_id == events[1].sse_id.turn_id
+
+    @respx.mock
+    async def test_full_event_vocab(self) -> None:
+        """full_event_vocab: one of each variant; all carry parsed sse_id."""
+        stream = b"".join(
+            [
+                _sse_chunk(
+                    "42:1",
+                    {"type": "worker_phase", "phase": "BuildingPrompt", "turn_id": 42},
+                ),
+                _sse_chunk("42:2", {"type": "thinking", "content": "thinking out loud"}),
+                _sse_chunk("42:3", {"type": "text", "content": "hello"}),
+                _sse_chunk(
+                    "42:4",
+                    {
+                        "type": "text_boundary",
+                        "kind": "sentence",
+                        "char_offset": 5,
+                        "ts": "2026-05-21T00:00:00Z",
+                    },
+                ),
+                _sse_chunk(
+                    "42:5",
+                    {"type": "tool_start", "name": "search", "arguments": {"q": "x"}},
+                ),
+                _sse_chunk(
+                    "42:6",
+                    {
+                        "type": "tool_result",
+                        "name": "search",
+                        "result": {"n": 1},
+                        "duration_ms": 12,
+                    },
+                ),
+                _sse_chunk("42:7", {"type": "text", "content": " world"}),
+                _sse_chunk(
+                    "42:8",
+                    {
+                        "type": "done",
+                        "phase": "succeeded",
+                        "response": "hello world",
+                        "model": "m",
+                        "duration_ms": 50,
+                        "usage": {
+                            "prompt_tokens": 0,
+                            "completion_tokens": 0,
+                            "total_tokens": 0,
+                            "cached_input_tokens": 0,
+                        },
+                    },
+                ),
+            ]
+        )
+        respx.post("https://w.example/sessions/s1/messages").mock(
+            return_value=httpx.Response(
+                200, headers={"content-type": "text/event-stream"}, content=stream
+            )
+        )
+        async with httpx.AsyncClient(base_url="https://w.example") as client:
+            events = [e async for e in stream_turn(client, "s1", "hi")]
+        types_seen = [type(e).__name__ for e in events]
+        assert types_seen == [
+            "WorkerPhase",
+            "Thinking",
+            "Text",
+            "TextBoundary",
+            "ToolStart",
+            "ToolResult",
+            "Text",
+            "Done",
+        ]
+        # INV-002: every event carries parsed sse_id with turn_id, seq >= 1
+        for e in events:
+            assert e.sse_id.turn_id == 42
+            assert e.sse_id.seq >= 1
+
+    @respx.mock
+    async def test_error_terminal(self) -> None:
+        """error_terminal: text then error; iteration ends; error_code populated."""
+        stream = _sse_chunk("42:1", {"type": "text", "content": "x"}) + _sse_chunk(
+            "42:2",
+            {
+                "type": "error",
+                "phase": "failed",
+                "error_code": "llm_output_invalid",
+                "message": "boom",
+            },
+        )
+        respx.post("https://w.example/sessions/s1/messages").mock(
+            return_value=httpx.Response(
+                200, headers={"content-type": "text/event-stream"}, content=stream
+            )
+        )
+        async with httpx.AsyncClient(base_url="https://w.example") as client:
+            events = [e async for e in stream_turn(client, "s1", "hi")]
+        assert len(events) == 2
+        assert isinstance(events[1], Error)
+        assert events[1].error_code == "llm_output_invalid"
+        assert events[1].message == "boom"
+
+    @respx.mock
+    async def test_cancelled_terminal(self) -> None:
+        """cancelled_terminal: cancelled with phase=cancelled, turn_id; iteration ends."""
+        stream = _sse_chunk(
+            "42:1",
+            {
+                "type": "cancelled",
+                "phase": "cancelled",
+                "turn_id": 42,
+                "reason": "user_cancel",
+                "partial_message_id": None,
+            },
+        )
+        respx.post("https://w.example/sessions/s1/messages").mock(
+            return_value=httpx.Response(
+                200, headers={"content-type": "text/event-stream"}, content=stream
+            )
+        )
+        async with httpx.AsyncClient(base_url="https://w.example") as client:
+            events = [e async for e in stream_turn(client, "s1", "hi")]
+        assert len(events) == 1
+        assert isinstance(events[0], Cancelled)
+        assert events[0].turn_id == 42
+        assert events[0].reason == "user_cancel"
+
+    @respx.mock
+    async def test_session_not_found(self) -> None:
+        """session_not_found: 404 -> SseConnectFailed(status=404)."""
+        respx.post("https://w.example/sessions/missing/messages").mock(
+            return_value=httpx.Response(404, json={"error": "session_not_found"})
+        )
+        async with httpx.AsyncClient(base_url="https://w.example") as client:
+            with pytest.raises(SseConnectFailed) as exc_info:
+                _ = [e async for e in stream_turn(client, "missing", "hi")]
+        assert exc_info.value.status == 404
+
+    @respx.mock
+    async def test_malformed_id_no_seq(self) -> None:
+        """malformed_id_no_seq: id `42` (missing :seq) -> MalformedSseId; no event yielded."""
+        stream = b"id: 42\ndata: {\"type\":\"text\",\"content\":\"x\"}\n\n"
+        respx.post("https://w.example/sessions/s1/messages").mock(
+            return_value=httpx.Response(
+                200, headers={"content-type": "text/event-stream"}, content=stream
+            )
+        )
+        async with httpx.AsyncClient(base_url="https://w.example") as client:
+            events: list[object] = []
+            with pytest.raises(MalformedSseId):
+                async for e in stream_turn(client, "s1", "hi"):
+                    events.append(e)
+        assert events == []
+
+    @respx.mock
+    async def test_malformed_id_alpha(self) -> None:
+        """malformed_id_alpha: id `foo:bar` -> MalformedSseId."""
+        stream = b"id: foo:bar\ndata: {\"type\":\"text\",\"content\":\"x\"}\n\n"
+        respx.post("https://w.example/sessions/s1/messages").mock(
+            return_value=httpx.Response(
+                200, headers={"content-type": "text/event-stream"}, content=stream
+            )
+        )
+        async with httpx.AsyncClient(base_url="https://w.example") as client:
+            with pytest.raises(MalformedSseId):
+                _ = [e async for e in stream_turn(client, "s1", "hi")]
+
+    @respx.mock
+    async def test_turn_id_flip(self) -> None:
+        """turn_id_flip: 42:1 then 99:2 -> TurnIdFlip; only first yielded."""
+        stream = _sse_chunk("42:1", {"type": "text", "content": "a"}) + _sse_chunk(
+            "99:2", {"type": "text", "content": "b"}
+        )
+        respx.post("https://w.example/sessions/s1/messages").mock(
+            return_value=httpx.Response(
+                200, headers={"content-type": "text/event-stream"}, content=stream
+            )
+        )
+        async with httpx.AsyncClient(base_url="https://w.example") as client:
+            yielded: list[Text] = []
+            with pytest.raises(TurnIdFlip) as exc_info:
+                async for e in stream_turn(client, "s1", "hi"):
+                    yielded.append(e)  # type: ignore[arg-type]
+        assert len(yielded) == 1
+        assert exc_info.value.established == 42
+        assert exc_info.value.got == 99
+
+    @respx.mock
+    async def test_connection_drop(self) -> None:
+        """connection_drop: RemoteProtocolError after one text -> SseConnectionDropped((42,1))."""
+        from ratatoskr.sse_client import SseConnectionDropped
+
+        class _DropAfter(httpx.AsyncByteStream):
+            def __init__(self, chunks: list[bytes]) -> None:
+                self._chunks = chunks
+
+            async def __aiter__(self):  # type: ignore[no-untyped-def]
+                for c in self._chunks:
+                    yield c
+                raise httpx.RemoteProtocolError("simulated mid-stream drop")
+
+            async def aclose(self) -> None:
+                return None
+
+        first = _sse_chunk("42:1", {"type": "text", "content": "x"})
+        respx.post("https://w.example/sessions/s1/messages").mock(
+            return_value=httpx.Response(
+                200,
+                headers={"content-type": "text/event-stream"},
+                stream=_DropAfter([first]),
+            )
+        )
+        async with httpx.AsyncClient(base_url="https://w.example") as client:
+            yielded: list[Text] = []
+            with pytest.raises(SseConnectionDropped) as exc_info:
+                async for e in stream_turn(client, "s1", "hi"):
+                    yielded.append(e)  # type: ignore[arg-type]
+        assert len(yielded) == 1
+        assert exc_info.value.last_seen_sse_id == SseId(42, 1)
+
+    @respx.mock
+    async def test_no_text_aggregation(self) -> None:
+        """no_text_aggregation: consumer yields each text event separately; no concat."""
+        stream = (
+            _sse_chunk("42:1", {"type": "text", "content": "hello "})
+            + _sse_chunk("42:2", {"type": "text", "content": "world"})
+            + _sse_chunk(
+                "42:3",
+                {
+                    "type": "done",
+                    "phase": "succeeded",
+                    "response": "hello world",
+                    "model": "m",
+                    "duration_ms": 1,
+                    "usage": {
+                        "prompt_tokens": 0,
+                        "completion_tokens": 0,
+                        "total_tokens": 0,
+                        "cached_input_tokens": 0,
+                    },
+                },
+            )
+        )
+        respx.post("https://w.example/sessions/s1/messages").mock(
+            return_value=httpx.Response(
+                200, headers={"content-type": "text/event-stream"}, content=stream
+            )
+        )
+        async with httpx.AsyncClient(base_url="https://w.example") as client:
+            events = [e async for e in stream_turn(client, "s1", "hi")]
+        texts = [e for e in events if isinstance(e, Text)]
+        assert [t.content for t in texts] == ["hello ", "world"]
+
+
+class TestReconnectTurn:
+    @respx.mock
+    async def test_happy_resume_from_seq_3(self) -> None:
+        """happy_resume_from_seq_3 [scenario,tracer]: replay 42:4,42:5 then live 42:6 done."""
+        stream = (
+            _sse_chunk("42:4", {"type": "text", "content": "d"})
+            + _sse_chunk("42:5", {"type": "text", "content": "e"})
+            + _sse_chunk("42:6", _DONE_42_6)
+        )
+        respx.post("https://w.example/sessions/s1/messages").mock(
+            return_value=httpx.Response(
+                200, headers={"content-type": "text/event-stream"}, content=stream
+            )
+        )
+        async with httpx.AsyncClient(base_url="https://w.example") as client:
+            events = [e async for e in reconnect_turn(client, "s1", "original", "42:3")]
+        assert len(events) == 3
+        assert [e.sse_id.seq for e in events] == [4, 5, 6]
+        assert all(e.sse_id.turn_id == 42 for e in events)
+        assert isinstance(events[-1], Done)
+
+    @respx.mock
+    async def test_header_verbatim(self) -> None:
+        """header_verbatim: outbound carries Last-Event-ID: 42:3 exactly."""
+        stream = _sse_chunk("42:4", _DONE_42_6)
+        route = respx.post("https://w.example/sessions/s1/messages").mock(
+            return_value=httpx.Response(
+                200, headers={"content-type": "text/event-stream"}, content=stream
+            )
+        )
+        async with httpx.AsyncClient(base_url="https://w.example") as client:
+            _ = [e async for e in reconnect_turn(client, "s1", "x", "42:3")]
+        req = route.calls[0].request
+        assert req.headers["Last-Event-ID"] == "42:3"
+
+    @respx.mock
+    async def test_body_threads_content(self) -> None:
+        """body_threads_content: outbound JSON body is {'content': <content>} byte-for-byte."""
+        import json as _json
+
+        stream = _sse_chunk("42:4", _DONE_42_6)
+        route = respx.post("https://w.example/sessions/s1/messages").mock(
+            return_value=httpx.Response(
+                200, headers={"content-type": "text/event-stream"}, content=stream
+            )
+        )
+        async with httpx.AsyncClient(base_url="https://w.example") as client:
+            _ = [
+                e async for e in reconnect_turn(client, "s1", "hello mimir", "42:3")
+            ]
+        body = _json.loads(route.calls[0].request.content)
+        assert body == {"content": "hello mimir"}
+
+    @respx.mock
+    async def test_buffer_expired(self) -> None:
+        """buffer_expired: 412 -> ResumeBufferExpired(turn_id=42, buffered_from_seq=10)."""
+        respx.post("https://w.example/sessions/s1/messages").mock(
+            return_value=httpx.Response(
+                412,
+                json={"error": "buffer_expired", "buffered_from_seq": 10, "turn_id": 42},
+            )
+        )
+        async with httpx.AsyncClient(base_url="https://w.example") as client:
+            with pytest.raises(ResumeBufferExpired) as exc_info:
+                _ = [e async for e in reconnect_turn(client, "s1", "x", "42:3")]
+        assert exc_info.value.turn_id == 42
+        assert exc_info.value.buffered_from_seq == 10
+
+    @respx.mock
+    async def test_turn_finished(self) -> None:
+        """turn_finished: 410 -> ResumeTurnFinished(turn_id=42)."""
+        respx.post("https://w.example/sessions/s1/messages").mock(
+            return_value=httpx.Response(
+                410, json={"error": "turn_finished", "turn_id": 42}
+            )
+        )
+        async with httpx.AsyncClient(base_url="https://w.example") as client:
+            with pytest.raises(ResumeTurnFinished) as exc_info:
+                _ = [e async for e in reconnect_turn(client, "s1", "x", "42:3")]
+        assert exc_info.value.turn_id == 42
+
+    @respx.mock
+    async def test_invalid_last_event_id_server(self) -> None:
+        """invalid_last_event_id_server: 400 -> InvalidLastEventId."""
+        respx.post("https://w.example/sessions/s1/messages").mock(
+            return_value=httpx.Response(400, json={"error": "invalid_last_event_id"})
+        )
+        async with httpx.AsyncClient(base_url="https://w.example") as client:
+            with pytest.raises(InvalidLastEventId):
+                _ = [e async for e in reconnect_turn(client, "s1", "x", "42:3")]
+
+    @respx.mock
+    async def test_malformed_input(self) -> None:
+        """malformed_input: caller passes '42' (no seq) -> ValueError; no HTTP issued."""
+        route = respx.post("https://w.example/sessions/s1/messages").mock(
+            return_value=httpx.Response(200, content=b"")
+        )
+        async with httpx.AsyncClient(base_url="https://w.example") as client:
+            with pytest.raises(ValueError):
+                _ = [e async for e in reconnect_turn(client, "s1", "x", "42")]
+        assert route.call_count == 0
+
+    @respx.mock
+    async def test_turn_id_flip_on_first_event(self) -> None:
+        """turn_id_flip_on_first_event: server's first event 99:4 -> TurnIdFlip pre-yield."""
+        stream = _sse_chunk("99:4", {"type": "text", "content": "wrong turn"})
+        respx.post("https://w.example/sessions/s1/messages").mock(
+            return_value=httpx.Response(
+                200, headers={"content-type": "text/event-stream"}, content=stream
+            )
+        )
+        async with httpx.AsyncClient(base_url="https://w.example") as client:
+            yielded: list[Text] = []
+            with pytest.raises(TurnIdFlip) as exc_info:
+                async for e in reconnect_turn(client, "s1", "x", "42:3"):
+                    yielded.append(e)  # type: ignore[arg-type]
+        assert yielded == []
+        assert exc_info.value.established == 42
+        assert exc_info.value.got == 99
+
+
+class TestCancelTurn:
+    @respx.mock
+    async def test_happy_cancel(self) -> None:
+        """happy_cancel [happy,tracer]: 200 -> CancelResult(42, True, None, None)."""
+        respx.post("https://w.example/sessions/s1/turns/42/cancel").mock(
+            return_value=httpx.Response(
+                200,
+                json={
+                    "turn_id": 42,
+                    "cancelled": True,
+                    "reason": None,
+                    "partial_message_id": None,
+                },
+            )
+        )
+        async with httpx.AsyncClient(base_url="https://w.example") as client:
+            result = await cancel_turn(client, "s1", 42)
+        assert result == CancelResult(
+            turn_id=42, cancelled=True, reason=None, partial_message_id=None
+        )
+
+    @respx.mock
+    async def test_persist_partial_query(self) -> None:
+        """persist_partial_query [trace]: persist_partial=True -> URL has ?persist_partial=true."""
+        route = respx.post("https://w.example/sessions/s1/turns/42/cancel").mock(
+            return_value=httpx.Response(
+                200,
+                json={
+                    "turn_id": 42,
+                    "cancelled": True,
+                    "reason": None,
+                    "partial_message_id": None,
+                },
+            )
+        )
+        async with httpx.AsyncClient(base_url="https://w.example") as client:
+            await cancel_turn(client, "s1", 42, persist_partial=True)
+        assert "persist_partial=true" in str(route.calls[0].request.url)
+
+    @respx.mock
+    async def test_default_no_persist(self) -> None:
+        """default_no_persist [trace]: default call -> URL has no persist_partial param."""
+        route = respx.post("https://w.example/sessions/s1/turns/42/cancel").mock(
+            return_value=httpx.Response(
+                200,
+                json={
+                    "turn_id": 42,
+                    "cancelled": True,
+                    "reason": None,
+                    "partial_message_id": None,
+                },
+            )
+        )
+        async with httpx.AsyncClient(base_url="https://w.example") as client:
+            await cancel_turn(client, "s1", 42)
+        assert "persist_partial" not in str(route.calls[0].request.url)
+
+    @respx.mock
+    async def test_cancel_already_completed(self) -> None:
+        """cancel_already_completed [error]: 409 -> CancelAlreadyCompleted(turn_id=42)."""
+        respx.post("https://w.example/sessions/s1/turns/42/cancel").mock(
+            return_value=httpx.Response(409, json={"error": "turn_already_completed"})
+        )
+        async with httpx.AsyncClient(base_url="https://w.example") as client:
+            with pytest.raises(CancelAlreadyCompleted) as exc_info:
+                await cancel_turn(client, "s1", 42)
+        assert exc_info.value.turn_id == 42
+
+    @respx.mock
+    async def test_cancel_turn_not_found(self) -> None:
+        """cancel_turn_not_found [error]: 404 -> CancelTurnNotFound(turn_id=42)."""
+        respx.post("https://w.example/sessions/s1/turns/42/cancel").mock(
+            return_value=httpx.Response(404, json={"error": "turn_not_found"})
+        )
+        async with httpx.AsyncClient(base_url="https://w.example") as client:
+            with pytest.raises(CancelTurnNotFound) as exc_info:
+                await cancel_turn(client, "s1", 42)
+        assert exc_info.value.turn_id == 42
+
+    @respx.mock
+    async def test_cancel_idempotent(self) -> None:
+        """cancel_idempotent: first 200 -> happy; second 409 -> CancelAlreadyCompleted."""
+        respx.post("https://w.example/sessions/s1/turns/42/cancel").mock(
+            side_effect=[
+                httpx.Response(
+                    200,
+                    json={
+                        "turn_id": 42,
+                        "cancelled": True,
+                        "reason": None,
+                        "partial_message_id": None,
+                    },
+                ),
+                httpx.Response(409, json={"error": "turn_already_completed"}),
+            ]
+        )
+        async with httpx.AsyncClient(base_url="https://w.example") as client:
+            first = await cancel_turn(client, "s1", 42)
+            assert first.cancelled is True
+            with pytest.raises(CancelAlreadyCompleted):
+                await cancel_turn(client, "s1", 42)