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.

tests/test_no_worldtree_imports.py

@@ -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"

tests/test_sse_client.py

@@ -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)