feat(#20 ): rewire the web turn surface onto the wt adapter (slice-2, part 2b-ii)

The Starlette endpoints (create / stream / cancel / tools / messages) now go through ratatoskr.wt over the worldtree-sdk; the browser contract is preserved. This is the last consumer of the hand-rolled turn-stream family — after this, stream_turn* / cancel_turn are orphaned and get deleted in part 2b-iii (with the live smoke). - _wt_client wraps a client_factory transport as the adapter's WorldtreeClient (INV-CUT-1), reading base_url + bearer off the transport (a no-auth test transport falls back to a placeholder key). The hand-rolled endpoints (persona / agents / admin / bifrost) keep using the raw transport until their slices. - _event_to_browser_payload derives the browser payload from the SDK's `raw` (the wire body) minus the redundant `type`, plus the composite `sse_id` string — the SAME shape the old dataclasses produced, so the presentation fixture + browser JS are unchanged; the browser event_type is the wire `type`, not the SDK class name. - The stream endpoint captures the upstream cancel target from the composite sse_id (the SDK's top-level turn_id is body-derived, absent on text frames); create reads the SDK's open create dict; cancel reads CancelResult.cancelled and surfaces a generic 502 for CancelFailed (the SDK abstracts the upstream cancel HTTP status). - test_web_presentation_contract builds SDK events via build_event; two cancel tests adopt the SDK's (status, error_code) race pairs + the 502. Suite 570 green; web/server.py + presentation test ruff-clean, mypy unchanged (same pre-existing errors). Patch (internal; browser contract preserved).
feat(#20 ): rewire the CLI turn path onto the wt adapter (slice-2, part 2b-i)
2026-07-19 06:22:13 -07:00 · 2026-07-19 06:07:50 -07:00
8 changed files with 396 additions and 212 deletions
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"

 [project]
 name = "ratatoskr"
-version = "0.21.5"
+version = "0.21.7"
 description = "Worldtree Conversation API debug console (web + headless CLI) — multi-pane observability"
 readme = "README.md"
 requires-python = ">=3.12"
@@ -11,12 +11,30 @@ import hashlib
 import os
 import signal
 import sys
+from collections.abc import Mapping
 from dataclasses import dataclass, field
 from importlib.metadata import PackageNotFoundError, version
 from typing import Any, TextIO

 import httpx
+from worldtree_sdk import (
+    AffectUpdateEvent,
+    AwaitingLlmFirstTokenEvent,
+    CancelledEvent,
+    ConnectFailed,
+    DoneEvent,
+    ErrorEvent,
+    TextBoundaryEvent,
+    TextEvent,
+    ThinkingEvent,
+    ToolResultEvent,
+    ToolStartEvent,
+    TurnEvent,
+    WorkerPhaseEvent,
+    WorldtreeClient,
+)

+from ratatoskr import wt
 from ratatoskr.first_message import seed_preset_first_message
 from ratatoskr.sessions import (
    AgentNotFound,
@@ -37,29 +55,21 @@ from ratatoskr.sessions import (
    set_persona_state,
    write_authored_history,
 )
+
+# The turn path (create / stream / cancel) is served by the worldtree-sdk adapter
+# (`wt.*`); these caller-semantic exceptions are what the adapter raises, so the
+# presenter keeps catching ratatoskr's own types (DEC-2). The hand-rolled probes
+# (--whoami / --characters / --set-persona / --seed-first-message) stay on the
+# `sessions` wrappers until their own slices.
 from ratatoskr.sse_client import (
-    AffectUpdate,
-    AwaitingLlmFirstToken,
    CancelAlreadyCompleted,
    CancelFailed,
-    Cancelled,
    CancelTurnNotFound,
-    Done,
-    Error,
-    Event,
    MalformedSseData,
    MalformedSseId,
    SseConnectFailed,
    SseConnectionDropped,
-    Text,
-    TextBoundary,
-    Thinking,
-    ToolResult,
-    ToolStart,
    TurnIdFlip,
-    WorkerPhase,
-    cancel_turn,
-    stream_turn_resilient,
 )


@@ -337,6 +347,29 @@ def _format_usage(usage: dict[str, int], *, arrow: str) -> str:
    return f"{p} in {arrow} {c} out ({t} total, {ci} cached)"


+def _format_usage_safe(usage: Mapping[str, int] | None) -> str:
+    """Tolerant wrapper over `_format_usage` for the SDK's open-world
+    `DoneEvent.usage` (typed optional): the canonical four-key usage formats;
+    anything absent or malformed degrades to `(n/a)` rather than crashing the
+    presenter (same posture as `_format_whoami`)."""
+    keys = ("prompt_tokens", "completion_tokens", "total_tokens", "cached_input_tokens")
+    if usage is not None and all(k in usage for k in keys):
+        return _format_usage(dict(usage), arrow="->")
+    return "(n/a)"
+
+
+def _turn_id_from_sse_id(sse_id: str) -> int | None:
+    """The turn component of the SDK's composite sse_id (`"{turn}:{seq}"`). This is
+    the mid-stream cancel target: it is present on EVERY frame, unlike the SDK's
+    top-level `turn_id`, which is the body field (absent on text/thinking events)."""
+    head, _, _ = sse_id.partition(":")
+    try:
+        turn = int(head)
+    except ValueError:
+        return None
+    return turn if turn > 0 else None
+
+
@dataclass(slots=True)
 class CliPresenterState:
    """Per-turn presenter state for `--send` mode (issue #12).
@@ -348,24 +381,31 @@ class CliPresenterState:
    thinking_open: bool = False
    text_written_since_newline: bool = False

-    def render(self, event: Event, *, stdout: TextIO, stderr: TextIO) -> None:
-        """Render one Worldtree SSE event with editorial hierarchy + coalescing."""
+    def render(self, event: TurnEvent, *, stdout: TextIO, stderr: TextIO) -> None:
+        """Render one Worldtree SSE event with editorial hierarchy + coalescing.
+
+        Consumes the worldtree-sdk `TurnEvent` union. The SDK types the de-facto
+        fields as OPTIONAL (open-world), so every read is hardened: a
+        malformed/partial event degrades to a placeholder rather than crashing the
+        presenter — the same posture as `_format_whoami`.
+        """
        assert isinstance(
            event,
            (
-                WorkerPhase, Thinking, Text, TextBoundary,
-                ToolStart, ToolResult, Done, Error, Cancelled,
-                AffectUpdate, AwaitingLlmFirstToken,
+                WorkerPhaseEvent, ThinkingEvent, TextEvent, TextBoundaryEvent,
+                ToolStartEvent, ToolResultEvent, DoneEvent, ErrorEvent, CancelledEvent,
+                AffectUpdateEvent, AwaitingLlmFirstTokenEvent,
            ),
        )
        # Thinking events accumulate into the open run.
-        if isinstance(event, Thinking):
+        if isinstance(event, ThinkingEvent):
+            content = event.content or ""
            if not self.thinking_open:
                stderr.write(". thinking: ")
                self.thinking_open = True
-            stderr.write(event.content)
+            stderr.write(content)
            stderr.flush()
-            self.thinking_buffer.append(event.content)
+            self.thinking_buffer.append(content)
            return
        # Non-thinking event: close any open thinking run first.
        if self.thinking_open:
@@ -374,59 +414,60 @@ class CliPresenterState:
            self.thinking_open = False
            self.thinking_buffer.clear()
        # Now render the new event.
-        if isinstance(event, Text):
-            stdout.write(event.content)
+        if isinstance(event, TextEvent):
+            content = event.content or ""
+            stdout.write(content)
            stdout.flush()
            # POST-003: only set if cursor is mid-line (no trailing newline).
-            self.text_written_since_newline = not event.content.endswith("\n")
+            self.text_written_since_newline = not content.endswith("\n")
            return
-        if isinstance(event, (Done, Error, Cancelled)):
+        if isinstance(event, (DoneEvent, ErrorEvent, CancelledEvent)):
            # INV-005: ensure stdout newline boundary before stderr terminal label.
            if self.text_written_since_newline:
                stdout.write("\n")
                stdout.flush()
                self.text_written_since_newline = False
-        if isinstance(event, Done):
+        if isinstance(event, DoneEvent):
            stderr.write(
-                f"[done] turn_id={event.sse_id.turn_id} model={event.model} "
-                f"duration={_format_duration_ms(event.duration_ms)} "
-                f"usage {_format_usage(event.usage, arrow='->')}\n"
+                f"[done] turn_id={event.turn_id} model={event.model} "
+                f"duration={_format_duration_ms(event.duration_ms or 0)} "
+                f"usage {_format_usage_safe(event.usage)}\n"
            )
            return
-        if isinstance(event, WorkerPhase):
+        if isinstance(event, WorkerPhaseEvent):
            stderr.write(
                f". worker_phase: phase={event.phase} turn_id={event.turn_id}\n"
            )
            return
-        if isinstance(event, Error):
+        if isinstance(event, ErrorEvent):
            stderr.write(
-                f"[error] turn_id={event.sse_id.turn_id} code={event.error_code} "
+                f"[error] turn_id={event.turn_id} code={event.error_code} "
                f"message={event.message!r}\n"
            )
            return
-        if isinstance(event, Cancelled):
+        if isinstance(event, CancelledEvent):
            stderr.write(
                f"[cancelled] turn_id={event.turn_id} reason={event.reason!r} "
                f"partial_message_id={event.partial_message_id}\n"
            )
            return
-        if isinstance(event, ToolStart):
+        if isinstance(event, ToolStartEvent):
            stderr.write(
                f". tool_start: name={event.name} args={event.arguments!r}\n"
            )
            return
-        if isinstance(event, ToolResult):
+        if isinstance(event, ToolResultEvent):
            stderr.write(
                f". tool_result: name={event.name} duration_ms={event.duration_ms} "
                f"result={event.result!r:.200}\n"
            )
            return
-        if isinstance(event, TextBoundary):
+        if isinstance(event, TextBoundaryEvent):
            stderr.write(
                f". text_boundary: kind={event.kind} char_offset={event.char_offset}\n"
            )
            return
-        if isinstance(event, AffectUpdate):
+        if isinstance(event, AffectUpdateEvent):
            # Worldtree #204 / v0.28.0. CLI surface is debug telemetry —
            # one line to stderr with status + (for current) dominant_emotion.
            if event.snapshot is not None:
@@ -440,10 +481,10 @@ class CliPresenterState:
                    f". affect_update: status={event.status} turn_id={event.turn_id}\n"
                )
            return
-        if isinstance(event, AwaitingLlmFirstToken):
+        if isinstance(event, AwaitingLlmFirstTokenEvent):
            # Worldtree #201 / v0.29.0. Heartbeat during BuildingPrompt →
            # CallingLLM gap. Stderr surface, one line per heartbeat.
-            secs = event.elapsed_ms_since_building_prompt / 1000.0
+            secs = (event.elapsed_ms_since_building_prompt or 0) / 1000.0
            stderr.write(
                f". awaiting_llm_first_token: turn_id={event.turn_id} elapsed={secs:.1f}s\n"
            )
@@ -451,23 +492,31 @@ class CliPresenterState:


 async def _cancel_and_log(
-    client: httpx.AsyncClient,
+    client: WorldtreeClient,
    session_id: str,
    turn_id: int,
    *,
    stderr: TextIO,
 ) -> None:
-    """Spawn-and-forget cancel that never raises (INV-009)."""
+    """Spawn-and-forget cancel that never raises (INV-009). The adapter maps the
+    cancel races onto ratatoskr's typed exceptions; a benign late-cancel (200,
+    cancelled=False) returns a result and logs nothing."""
    assert client is not None
    assert isinstance(turn_id, int) and turn_id > 0
    try:
-        await cancel_turn(client, session_id, turn_id)
-    except (CancelFailed, CancelTurnNotFound, CancelAlreadyCompleted, httpx.RequestError) as exc:
+        await wt.cancel_turn(client, session_id, turn_id)
+    except (
+        CancelFailed,
+        CancelTurnNotFound,
+        CancelAlreadyCompleted,
+        ConnectFailed,  # SDK normalizes a transport drop to ConnectFailed(status=0)
+        httpx.RequestError,
+    ) as exc:
        stderr.write(f"[cancel_failed] {type(exc).__name__}: {exc}\n")


 async def _run_turn(
-    client: httpx.AsyncClient,
+    client: WorldtreeClient,
    session_id: str,
    content: str,
    sigint_event: asyncio.Event,
@@ -493,7 +542,10 @@ async def _run_turn(
    cancelling = False
    sigint_task: asyncio.Task[bool] | None = None
    cancel_task: asyncio.Task[None] | None = None  # strong ref to fire-and-forget cancel
-    aiter_obj = stream_turn_resilient(client, session_id, content).__aiter__()
+    # wt.stream_turn is an async generator — it is already its own iterator, so no
+    # explicit __aiter__(); keeping the concrete type lets __anext__() type as a
+    # coroutine for asyncio.create_task.
+    aiter_obj = wt.stream_turn(client, session_id, content)

    try:
        while True:
@@ -537,17 +589,21 @@ async def _run_turn(
            except TurnIdFlip as exc:
                stderr.write(f"[turn_id_flip] expected={exc.established} got={exc.got}\n")
                return 22
-            last_turn_id = event.sse_id.turn_id
+            # The cancel target is the turn from the composite sse_id (present on
+            # every frame); the body's turn_id is absent on text/thinking events.
+            tid = _turn_id_from_sse_id(event.sse_id)
+            if tid is not None:
+                last_turn_id = tid
            state.render(event, stdout=stdout, stderr=stderr)
-            if isinstance(event, Done):
+            if isinstance(event, DoneEvent):
                if sigint_task is not None and not cancelling:
                    sigint_task.cancel()
                return 0
-            if isinstance(event, Error):
+            if isinstance(event, ErrorEvent):
                if sigint_task is not None and not cancelling:
                    sigint_task.cancel()
                return 2
-            if isinstance(event, Cancelled):
+            if isinstance(event, CancelledEvent):
                if sigint_task is not None and not cancelling:
                    sigint_task.cancel()
                return 3
@@ -564,6 +620,11 @@ async def _run_turn(
 async def _amain(args: ParsedArgs) -> int:
    """Async orchestrator: create-session (if --new) → SIGINT install → _run_turn → cleanup."""
    assert isinstance(args, ParsedArgs)
+    # ratatoskr owns the transport (INV-CUT-1): the SDK is injected with it and
+    # never closes it. The transport carries base_url / User-Agent / timeout AND the
+    # default bearer — the SDK overrides Authorization per request (so a bound create
+    # still uses its consumer_key), while the not-yet-migrated hand-rolled
+    # `seed_preset_first_message` reuses the transport's default bearer directly.
    async with httpx.AsyncClient(
        base_url=args.server_url,
        headers={
@@ -575,7 +636,8 @@ async def _amain(args: ParsedArgs) -> int:
        # connect/write/pool keep modest timeouts so true network failures
        # still surface promptly.
        timeout=httpx.Timeout(connect=10.0, read=None, write=10.0, pool=10.0),
-    ) as client:
+    ) as transport:
+        client = wt.build_client(args.server_url, api_key=args.api_key, transport=transport)
        if args.new:
            assert args.agent_id is not None
            try:
@@ -586,7 +648,7 @@ async def _amain(args: ParsedArgs) -> int:
                    if args.system_prompt is not None
                    else None
                )
-                info = await create_session(
+                info = await wt.create_session(
                    client,
                    args.agent_id,
                    end_user_id=args.end_user_id,
@@ -616,23 +678,32 @@ async def _amain(args: ParsedArgs) -> int:
                        "(RATATOSKR_BIFROST_CONSUMER_KEY), not WORLDTREE_API_KEY\n"
                    )
                return 23
-            except SessionApiFailed as exc:
+            except wt.SessionApiFailed as exc:
                sys.stderr.write(f"[session_api_failed] status={exc.status} body={exc.body!r}\n")
                return 20
-            except (httpx.ConnectError, httpx.ReadTimeout, httpx.TransportError) as exc:
+            except (
+                httpx.ConnectError,
+                httpx.ReadTimeout,
+                httpx.TransportError,
+                ConnectFailed,  # SDK normalizes a pre-response transport failure here
+            ) as exc:
                sys.stderr.write(f"[network_error] {type(exc).__name__}: {exc}\n")
                return 21
            # Issue #12: demoted lifecycle line — written directly here (NOT via
-            # state.render, which only accepts SSE Event variants per PRE-001).
-            kind_suffix = f" kind={info.kind}" if info.kind else ""
+            # state.render, which only accepts SSE Event variants per PRE-001). The
+            # adapter returns the SDK's open-world create dict; read it as a mapping.
+            session_id = info["session_id"]
+            kind = info.get("kind")
+            kind_suffix = f" kind={kind}" if kind else ""
            sys.stderr.write(
-                f". create_session: session_id={info.session_id} "
-                f"agent_id={info.agent_id}{kind_suffix}\n"
+                f". create_session: session_id={session_id} "
+                f"agent_id={info['agent_id']}{kind_suffix}\n"
            )
            # #347 authored first-message: seed the agent's preset opening (best-effort).
-            if await seed_preset_first_message(client, info.session_id, args.agent_id):
+            # Uses the transport directly — first_message is a slice-3 hand-rolled path.
+            if await seed_preset_first_message(transport, session_id, args.agent_id):
                sys.stderr.write(
-                    f". first_message: seeded preset opening for {info.agent_id}\n"
+                    f". first_message: seeded preset opening for {args.agent_id}\n"
                )
            # Issue #17 bound-state indicator: plane + endpoint + status, so the
            # operator sees WHICH identity/endpoint bound (not a bare boolean).
@@ -642,7 +713,7 @@ async def _amain(args: ParsedArgs) -> int:
                    f". bifrost: status=bound plane={plane} "
                    f"endpoint={args.bifrost.endpoint_url}\n"
                )
-            session_id = info.session_id
+            # session_id was bound above from the create dict.
        else:
            assert args.session_id is not None
            session_id = args.session_id
@@ -26,8 +26,10 @@ from starlette.responses import (
 )
 from starlette.routing import Mount, Route
 from starlette.staticfiles import StaticFiles
+from worldtree_sdk import CancelledEvent, DoneEvent, ErrorEvent, WorldtreeClient

 from ratatoskr import local_agents as _local_agents
+from ratatoskr import wt
 from ratatoskr.first_message import seed_preset_first_message
 from ratatoskr.sessions import (
    AgentNotAvailable,
@@ -38,33 +40,44 @@ from ratatoskr.sessions import (
    BifrostHandshakeFailed,
    PersonaNotConfigured,
    SessionApiFailed,
-    create_session,
    endpoint_for_plane,
    get_persona_state,
    get_session_bifrost,
-    get_session_messages,
-    get_session_tools,
    list_agents,
 )
+
+# The turn path (create / stream / cancel / tools / messages) is served by the
+# worldtree-sdk adapter (`wt.*`), which raises ratatoskr's caller-semantic
+# exceptions (DEC-2). The hand-rolled endpoints (persona / agents / admin /
+# bifrost) stay on the `sessions` / `sse_client` wrappers until their own slices.
 from ratatoskr.sse_client import (
    AdminEvent,
    CancelAlreadyCompleted,
    CancelFailed,
-    Cancelled,
    CancelTurnNotFound,
-    Done,
-    Error,
    MalformedSseData,
    MalformedSseId,
    SseConnectFailed,
    SseConnectionDropped,
    TurnIdFlip,
-    cancel_turn,
    stream_admin_events,
-    stream_turn_resilient,
 )


+def _wt_client(client: httpx.AsyncClient, *, max_reconnects: int = 5) -> WorldtreeClient:
+    """Wrap a client_factory transport as the adapter's WorldtreeClient (INV-CUT-1:
+    the SDK never closes it). base_url + bearer are read off the transport (the
+    factory bakes them in); the SDK re-applies auth per request, so the extracted
+    key just mirrors the transport's default. A no-auth test transport falls back to
+    a placeholder key (respx ignores auth)."""
+    base_url = str(client.base_url) or "http://localhost"
+    header = client.headers.get("Authorization", "")
+    api_key = header[len("Bearer "):].strip() if header.startswith("Bearer ") else ""
+    return wt.build_client(
+        base_url, api_key=api_key or "ratatoskr", transport=client, max_reconnects=max_reconnects
+    )
+
+
 def _static_dir() -> str:
    """Locate the bundled static/ directory inside the installed package.

@@ -163,16 +176,17 @@ async def _create_session_endpoint(request: Request) -> JSONResponse:

    try:
        async with client_factory() as client:
-            info = await create_session(
-                client,
+            info = await wt.create_session(
+                _wt_client(client),
                agent_id,
                end_user_id=end_user_id,
                bifrost=bifrost,
                consumer_key=consumer_key if bifrost else None,
            )
-            # #347 authored first-message: seed the agent's preset opening
-            # (best-effort; never blocks create — see first_message INV-001).
-            await seed_preset_first_message(client, info.session_id, agent_id)
+            # #347 authored first-message: seed the agent's preset opening (best-effort;
+            # never blocks create). first_message is a slice-3 hand-rolled path — it
+            # reuses the raw transport (its default bearer), not the adapter client.
+            await seed_preset_first_message(client, info["session_id"], agent_id)
    except AgentNotFound:
        return JSONResponse({"error_code": "agent_not_found"}, status_code=404)
    except BifrostConsumerKeyMissing:
@@ -188,12 +202,13 @@ async def _create_session_endpoint(request: Request) -> JSONResponse:
            },
            status_code=502,
        )
-    except SessionApiFailed as exc:
+    except wt.SessionApiFailed as exc:
        return JSONResponse(
            {"error_code": "session_api_failed", "status": exc.status},
            status_code=exc.status,
        )
-    payload = _as_dict(info)
+    # The adapter returns the SDK's open-world create dict; the browser reads it as-is.
+    payload = dict(info)
    if bifrost is not None:
        # Bound-state for the UI indicator — plane + endpoint only, never the key.
        payload["bifrost"] = {
@@ -251,25 +266,19 @@ async def _submit_turn_endpoint(request: Request) -> JSONResponse:


 def _event_to_browser_payload(event: object) -> tuple[str, dict]:
-    """Serialize an upstream Event dataclass to (browser_event_type, json_dict).
+    """Serialize an SDK `TurnEvent` to (browser_event_type, json_dict).

-    Per INV-008 + FN stream_turn_endpoint STEP 3. The dict shape is
-    locked by tests/fixtures/presentation_contract.json — one entry per
-    Event type. Implementation: snake_case class name as event_type;
-    asdict(event) with sse_id flattened to "T:S" string.
+    Per INV-008 + FN stream_turn_endpoint STEP 3. The browser contract
+    (tests/fixtures/presentation_contract.json) is preserved: the SDK's `raw` is
+    the wire body — the same per-type field set the old dataclasses carried — so the
+    payload is `raw` minus the redundant `type`, plus the composite `sse_id` string
+    (already "T:S"). The browser event_type is the wire `type` ("text" / "done" /
+    …), NOT the SDK class name. Open-world: additive server fields pass through.
    """
-    type_name = type(event).__name__
-    # CamelCase → snake_case
-    browser_type = "".join(
-        ("_" + c.lower() if c.isupper() and i else c.lower())
-        for i, c in enumerate(type_name)
-    )
-    data = asdict(event)  # type: ignore[arg-type]
-    sse_id = data.get("sse_id")
-    if isinstance(sse_id, (list, tuple)) and len(sse_id) == 2:
-        data["sse_id"] = f"{sse_id[0]}:{sse_id[1]}"
-    elif isinstance(sse_id, dict) and "turn_id" in sse_id and "seq" in sse_id:
-        data["sse_id"] = f"{sse_id['turn_id']}:{sse_id['seq']}"
+    browser_type = getattr(event, "type", "") or ""
+    raw = getattr(event, "raw", None) or {}
+    data = {k: v for k, v in dict(raw).items() if k != "type"}
+    data["sse_id"] = getattr(event, "sse_id", None)
    return browser_type, data


@@ -281,6 +290,20 @@ def _format_sse(event_type: str, data: dict) -> bytes:
    return f"event: {event_type}\ndata: {json.dumps(data)}\n\n".encode()


+def _turn_id_from_sse_id(sse_id: object) -> int | None:
+    """The turn component of the SDK's composite sse_id (`"{turn}:{seq}"`) — the
+    upstream cancel target, present on every frame (the SDK's top-level `turn_id` is
+    the body field, absent on text/thinking events)."""
+    if not isinstance(sse_id, str):
+        return None
+    head, _, _ = sse_id.partition(":")
+    try:
+        turn = int(head)
+    except ValueError:
+        return None
+    return turn if turn > 0 else None
+
+
 async def _stream_turn_endpoint(request: Request) -> StreamingResponse:
    """GET /api/turns/{session_id}/stream?turn_id=N → proxy upstream SSE.

@@ -302,21 +325,22 @@ async def _stream_turn_endpoint(request: Request) -> StreamingResponse:

    async def gen() -> AsyncIterator[bytes]:
        client = client_factory()
+        wt_client = _wt_client(client)
        try:
            handle.status = "streaming"
            try:
-                async for event in stream_turn_resilient(client, session_id, handle.content):
-                    # v0.16.0: capture the upstream (Worldtree-assigned)
-                    # turn_id from the first event so cancel paths target
-                    # the real upstream turn, not our local counter.
+                async for event in wt.stream_turn(wt_client, session_id, handle.content):
+                    # v0.16.0: capture the upstream (Worldtree-assigned) turn_id from
+                    # the first event so cancel paths target the real upstream turn,
+                    # not our local counter — parsed from the composite sse_id.
                    if handle.upstream_turn_id is None:
-                        sse_id = getattr(event, "sse_id", None)
-                        if sse_id is not None:
-                            handle.upstream_turn_id = sse_id.turn_id
+                        handle.upstream_turn_id = _turn_id_from_sse_id(
+                            getattr(event, "sse_id", None)
+                        )
                    event_type, data = _event_to_browser_payload(event)
                    yield _format_sse(event_type, data)
-                    if isinstance(event, (Done, Error, Cancelled)):
-                        handle.status = type(event).__name__.lower()
+                    if isinstance(event, (DoneEvent, ErrorEvent, CancelledEvent)):
+                        handle.status = event.type or "done"
                        break
            except (SseConnectFailed, SseConnectionDropped, MalformedSseId,
                    MalformedSseData, TurnIdFlip) as exc:
@@ -330,7 +354,7 @@ async def _stream_turn_endpoint(request: Request) -> StreamingResponse:
                # turn (if it started) — never the local turn_id.
                if handle.status == "streaming" and handle.upstream_turn_id is not None:
                    try:
-                        await cancel_turn(client, session_id, handle.upstream_turn_id)
+                        await wt.cancel_turn(wt_client, session_id, handle.upstream_turn_id)
                    except (CancelAlreadyCompleted, CancelTurnNotFound):
                        pass  # cooperative race — turn already terminal upstream
                    except Exception as exc:
@@ -377,15 +401,18 @@ async def _cancel_turn_endpoint(request: Request) -> JSONResponse:
    client_factory = request.app.state.client_factory
    try:
        async with client_factory() as client:
-            await cancel_turn(client, session_id, handle.upstream_turn_id)
-        body = {"cancelled": True}
+            result = await wt.cancel_turn(
+                _wt_client(client), session_id, handle.upstream_turn_id
+            )
+        body = {"cancelled": bool(result.cancelled)}
    except (CancelAlreadyCompleted, CancelTurnNotFound):
        body = {"cancelled": False, "reason": "race_or_completed"}
-    except CancelFailed as exc:
+    except CancelFailed:
+        # The SDK abstracts the upstream cancel HTTP status; surface a generic 502.
        registry.pop((session_id, turn_id), None)
        return JSONResponse(
-            {"error_code": "cancel_failed", "status": exc.status},
-            status_code=exc.status,
+            {"error_code": "cancel_failed"},
+            status_code=502,
        )
    registry.pop((session_id, turn_id), None)
    return JSONResponse(body, status_code=200)
@@ -465,13 +492,13 @@ async def _session_tools_endpoint(request: Request) -> JSONResponse:
    client_factory = request.app.state.client_factory
    try:
        async with client_factory() as client:
-            info = await get_session_tools(client, session_id)
-    except SessionApiFailed as exc:
+            info = await wt.get_session_tools(_wt_client(client), session_id)
+    except wt.SessionApiFailed as exc:
        return JSONResponse(
            {"error_code": "session_tools_unavailable", "status": exc.status},
            status_code=exc.status,
        )
-    return JSONResponse(info, status_code=200)
+    return JSONResponse(dict(info), status_code=200)


 async def _session_messages_endpoint(request: Request) -> JSONResponse:
@@ -485,13 +512,13 @@ async def _session_messages_endpoint(request: Request) -> JSONResponse:
    client_factory = request.app.state.client_factory
    try:
        async with client_factory() as client:
-            data = await get_session_messages(client, session_id)
-    except SessionApiFailed as exc:
+            data = await wt.get_session_messages(_wt_client(client), session_id)
+    except wt.SessionApiFailed as exc:
        return JSONResponse(
            {"error_code": "session_messages_unavailable", "status": exc.status},
            status_code=exc.status,
        )
-    return JSONResponse(data, status_code=200)
+    return JSONResponse(dict(data), status_code=200)


 async def _session_bifrost_endpoint(request: Request) -> JSONResponse:
@@ -609,14 +636,15 @@ def create_app(
        ]
        if in_flight:
            client = client_factory()
+            wt_client = _wt_client(client)
            try:
                task_to_handle = {
                    asyncio.create_task(
-                        cancel_turn(client, h.session_id, h.upstream_turn_id)
+                        wt.cancel_turn(wt_client, h.session_id, h.upstream_turn_id)
                    ): h
                    for h in in_flight
                }
-                done, pending = await asyncio.wait(task_to_handle, timeout=5.0)
+                _done, pending = await asyncio.wait(task_to_handle, timeout=5.0)
                # Per-pending session/turn detail (INV-006 logging fidelity).
                for task in pending:
                    h = task_to_handle[task]
@@ -28,7 +28,7 @@ carries the SDK's parsed `error_code`).
 from __future__ import annotations

 import json
-from collections.abc import AsyncIterator, Mapping
+from collections.abc import AsyncGenerator, Mapping
 from typing import Any

 import httpx
@@ -81,6 +81,7 @@ def build_client(
    api_key: AuthProvider,
    admin_key: AuthProvider | None = None,
    transport: httpx.AsyncClient,
+    max_reconnects: int = 5,
 ) -> WorldtreeClient:
    """Construct the adapter's `WorldtreeClient` over a ratatoskr-owned transport.

@@ -88,15 +89,18 @@ def build_client(
    `_owns_client=False`, so `WorldtreeClient.aclose()` never closes it — ratatoskr
    owns the lifecycle exactly as today (INV-CUT-1). ratatoskr's `api_key` /
    `admin_key` map to the SDK's per-request `auth` / `admin_auth` providers; the
-    injected transport carries ratatoskr's User-Agent / timeout (wired by the
-    caller in slice-2), NOT the Authorization header — the SDK adds auth per
-    request.
+    injected transport carries ratatoskr's User-Agent / timeout, and (transitionally)
+    the default bearer — the SDK adds auth per request, overriding it.
+
+    `max_reconnects` is the resilient turn-stream's reconnect budget (SDK default 5);
+    pass 0 to surface a transport drop immediately without auto-resume.
    """
    return WorldtreeClient(
        base_url,
        auth=api_key,
        admin_auth=admin_key,
        transport=transport,
+        max_reconnects=max_reconnects,
    )


@@ -249,7 +253,7 @@ async def get_session_tools(

 async def stream_turn(
    client: WorldtreeClient, session_id: str, content: str
-) -> AsyncIterator[wtsdk.TurnEvent]:
+) -> AsyncGenerator[wtsdk.TurnEvent, None]:
    """Drive the resilient turn stream (auto-resume; absorbs the old `reconnect_turn`)
    and yield the SDK's `TurnEvent`s, re-wrapping the stream's TERMINAL SDK errors
    into ratatoskr's caller-semantic exceptions (INV-CUT-2 / DEC-2 — the presenter
@@ -7,8 +7,10 @@ import json
 import httpx
 import pytest
 import respx
+from worldtree_sdk.events import build_event

 from ratatoskr import cli as cli_mod
+from ratatoskr import wt
 from ratatoskr.cli import (
    ParsedArgs,
    UsageError,
@@ -20,18 +22,7 @@ from ratatoskr.cli import (
    main,
 )
 from ratatoskr.sessions import BifrostBinding
-from ratatoskr.sse_client import (
-    Cancelled,
-    Done,
-    Error,
-    SseId,
-    Text,
-    TextBoundary,
-    Thinking,
-    ToolResult,
-    ToolStart,
-    WorkerPhase,
-)
+from ratatoskr.sse_client import SseId


 class _FlushCountingIO(io.StringIO):
@@ -333,6 +324,83 @@ SID = SseId(42, 5)
 SID42 = SseId(42, 1)


+# ── SDK-event factories ──────────────────────────────────────────────────────
+# The presenter now consumes worldtree-sdk `TurnEvent`s. These build them exactly
+# as the SDK's parser does (via `build_event` from the raw envelope), preserving
+# the old dataclass call shapes so the render-test bodies stay unchanged. `sse_id`
+# is a parsed `SseId` here purely to keep the terse SID42 idiom; the SDK carries the
+# composite id as a string and turn_id top-level.
+def _sid_str(sse_id: SseId) -> str:
+    return f"{sse_id.turn_id}:{sse_id.seq}"
+
+
+def Thinking(*, sse_id: SseId, content: str) -> object:
+    return build_event("thinking", _sid_str(sse_id), sse_id.turn_id, {"content": content})
+
+
+def Text(*, sse_id: SseId, content: str) -> object:
+    return build_event("text", _sid_str(sse_id), sse_id.turn_id, {"content": content})
+
+
+def WorkerPhase(*, sse_id: SseId, phase: str, turn_id: int) -> object:
+    return build_event("worker_phase", _sid_str(sse_id), turn_id, {"phase": phase})
+
+
+def TextBoundary(*, sse_id: SseId, kind: str, char_offset: int, ts: str) -> object:
+    return build_event(
+        "text_boundary", _sid_str(sse_id), sse_id.turn_id,
+        {"kind": kind, "char_offset": char_offset, "ts": ts},
+    )
+
+
+def ToolStart(*, sse_id: SseId, name: str, arguments: object) -> object:
+    return build_event(
+        "tool_start", _sid_str(sse_id), sse_id.turn_id, {"name": name, "arguments": arguments}
+    )
+
+
+def ToolResult(*, sse_id: SseId, name: str, result: object, duration_ms: int) -> object:
+    return build_event(
+        "tool_result", _sid_str(sse_id), sse_id.turn_id,
+        {"name": name, "result": result, "duration_ms": duration_ms},
+    )
+
+
+def Done(
+    *, sse_id: SseId, phase: str, response: str, model: str, duration_ms: int, usage: object
+) -> object:
+    return build_event(
+        "done", _sid_str(sse_id), sse_id.turn_id,
+        {"phase": phase, "response": response, "model": model,
+         "duration_ms": duration_ms, "usage": usage},
+    )
+
+
+def Error(*, sse_id: SseId, phase: str, message: str, error_code: str) -> object:
+    return build_event(
+        "error", _sid_str(sse_id), sse_id.turn_id,
+        {"phase": phase, "message": message, "error_code": error_code},
+    )
+
+
+def Cancelled(
+    *, sse_id: SseId, phase: str, turn_id: int, reason: object, partial_message_id: object
+) -> object:
+    return build_event(
+        "cancelled", _sid_str(sse_id), turn_id,
+        {"phase": phase, "reason": reason, "partial_message_id": partial_message_id},
+    )
+
+
+def _wtc(transport: httpx.AsyncClient) -> object:
+    """The adapter's WorldtreeClient over a respx-mocked transport. Reconnects are
+    disabled (max_reconnects=0) so a transport drop surfaces immediately instead of
+    burning the resilient retry budget with real backoff sleeps."""
+    return wt.build_client(
+        "https://w.example", api_key="k", transport=transport, max_reconnects=0
+    )
+
+
 class TestCliPresenterState:
    """Tests for the new CliPresenterState — per issue #12 contract."""

@@ -687,7 +755,7 @@ _USAGE_ZERO: dict[str, int] = {
 }


-def _make_done(*, duration_ms: int = 1, usage: dict[str, int] | None = None) -> Done:
+def _make_done(*, duration_ms: int = 1, usage: dict[str, int] | None = None) -> object:
    return Done(
        sse_id=SID42,
        phase="succeeded",
@@ -709,7 +777,8 @@ class TestCancelAndLog:
            )
        )
        stderr = io.StringIO()
-        async with httpx.AsyncClient(base_url="https://w.example") as client:
+        async with httpx.AsyncClient(base_url="https://w.example") as _tp:
+            client = _wtc(_tp)
            result = await _cancel_and_log(client, "s-1", 42, stderr=stderr)
        assert result is None
        assert stderr.getvalue() == ""
@@ -721,7 +790,8 @@ class TestCancelAndLog:
            return_value=httpx.Response(500, content=b"boom")
        )
        stderr = io.StringIO()
-        async with httpx.AsyncClient(base_url="https://w.example") as client:
+        async with httpx.AsyncClient(base_url="https://w.example") as _tp:
+            client = _wtc(_tp)
            await _cancel_and_log(client, "s-1", 42, stderr=stderr)
        out = stderr.getvalue()
        assert "[cancel_failed]" in out
@@ -730,11 +800,14 @@ class TestCancelAndLog:
    @respx.mock
    async def test_cancel_already_completed(self) -> None:
        """cancel_already_completed [scenario]: …"""
+        # SDK gates the race on the (status, error_code) PAIR (B-CAN-3): 409 alone is
+        # a generic CancelFailed; 409 + turn_finished is the double-cancel race.
        respx.post("https://w.example/sessions/s-1/turns/42/cancel").mock(
-            return_value=httpx.Response(409)
+            return_value=httpx.Response(409, json={"error_code": "turn_finished"})
        )
        stderr = io.StringIO()
-        async with httpx.AsyncClient(base_url="https://w.example") as client:
+        async with httpx.AsyncClient(base_url="https://w.example") as _tp:
+            client = _wtc(_tp)
            await _cancel_and_log(client, "s-1", 42, stderr=stderr)
        out = stderr.getvalue()
        assert "[cancel_failed]" in out
@@ -743,11 +816,13 @@ class TestCancelAndLog:
    @respx.mock
    async def test_cancel_turn_not_found(self) -> None:
        """cancel_turn_not_found [scenario]: 404 → returns None; stderr CancelTurnNotFound."""
+        # 404 + turn_not_found is the benign "finished before cancel arrived" race.
        respx.post("https://w.example/sessions/s-1/turns/42/cancel").mock(
-            return_value=httpx.Response(404)
+            return_value=httpx.Response(404, json={"error_code": "turn_not_found"})
        )
        stderr = io.StringIO()
-        async with httpx.AsyncClient(base_url="https://w.example") as client:
+        async with httpx.AsyncClient(base_url="https://w.example") as _tp:
+            client = _wtc(_tp)
            await _cancel_and_log(client, "s-1", 42, stderr=stderr)
        out = stderr.getvalue()
        assert "[cancel_failed]" in out
@@ -760,11 +835,14 @@ class TestCancelAndLog:
            side_effect=httpx.ConnectError("network down")
        )
        stderr = io.StringIO()
-        async with httpx.AsyncClient(base_url="https://w.example") as client:
+        async with httpx.AsyncClient(base_url="https://w.example") as _tp:
+            client = _wtc(_tp)
            await _cancel_and_log(client, "s-1", 42, stderr=stderr)
        out = stderr.getvalue()
+        # The SDK normalizes a transport drop to ConnectFailed(status=0); _cancel_and_log
+        # swallows it (INV-009) and logs the normalized type.
        assert "[cancel_failed]" in out
-        assert "ConnectError" in out
+        assert "ConnectFailed" in out


 class _GatedStream(httpx.AsyncByteStream):
@@ -797,7 +875,8 @@ class TestRunTurn:
        sigint = asyncio.Event()
        stdout = io.StringIO()
        stderr = io.StringIO()
-        async with httpx.AsyncClient(base_url="https://w.example") as client:
+        async with httpx.AsyncClient(base_url="https://w.example") as _tp:
+            client = _wtc(_tp)
            exit_code = await _run_turn(client, "s-1", "hi", sigint, stdout=stdout, stderr=stderr)
        assert exit_code == 0
        assert stdout.getvalue() == "hello\n"
@@ -820,7 +899,8 @@ class TestRunTurn:
        )
        sigint = asyncio.Event()
        stdout, stderr = io.StringIO(), io.StringIO()
-        async with httpx.AsyncClient(base_url="https://w.example") as client:
+        async with httpx.AsyncClient(base_url="https://w.example") as _tp:
+            client = _wtc(_tp)
            exit_code = await _run_turn(client, "s-1", "hi", sigint, stdout=stdout, stderr=stderr)
        assert exit_code == 2
        assert "[error]" in stderr.getvalue()
@@ -836,7 +916,8 @@ class TestRunTurn:
        )
        sigint = asyncio.Event()
        stdout, stderr = io.StringIO(), io.StringIO()
-        async with httpx.AsyncClient(base_url="https://w.example") as client:
+        async with httpx.AsyncClient(base_url="https://w.example") as _tp:
+            client = _wtc(_tp)
            exit_code = await _run_turn(client, "s-1", "hi", sigint, stdout=stdout, stderr=stderr)
        assert exit_code == 3
        assert "[cancelled]" in stderr.getvalue()
@@ -849,7 +930,8 @@ class TestRunTurn:
        )
        sigint = asyncio.Event()
        stdout, stderr = io.StringIO(), io.StringIO()
-        async with httpx.AsyncClient(base_url="https://w.example") as client:
+        async with httpx.AsyncClient(base_url="https://w.example") as _tp:
+            client = _wtc(_tp)
            exit_code = await _run_turn(
                client, "missing", "hi", sigint, stdout=stdout, stderr=stderr
            )
@@ -882,7 +964,8 @@ class TestRunTurn:
        )
        sigint = asyncio.Event()
        stdout, stderr = io.StringIO(), io.StringIO()
-        async with httpx.AsyncClient(base_url="https://w.example") as client:
+        async with httpx.AsyncClient(base_url="https://w.example") as _tp:
+            client = _wtc(_tp)
            exit_code = await _run_turn(client, "s-1", "hi", sigint, stdout=stdout, stderr=stderr)
        assert exit_code == 21
        assert "[connection_dropped]" in stderr.getvalue()
@@ -896,7 +979,8 @@ class TestRunTurn:
        )
        sigint = asyncio.Event()
        stdout, stderr = io.StringIO(), io.StringIO()
-        async with httpx.AsyncClient(base_url="https://w.example") as client:
+        async with httpx.AsyncClient(base_url="https://w.example") as _tp:
+            client = _wtc(_tp)
            exit_code = await _run_turn(client, "s-1", "hi", sigint, stdout=stdout, stderr=stderr)
        assert exit_code == 22
        assert "[malformed_sse_id]" in stderr.getvalue()
@@ -913,7 +997,8 @@ class TestRunTurn:
        )
        sigint = asyncio.Event()
        stdout, stderr = io.StringIO(), io.StringIO()
-        async with httpx.AsyncClient(base_url="https://w.example") as client:
+        async with httpx.AsyncClient(base_url="https://w.example") as _tp:
+            client = _wtc(_tp)
            exit_code = await _run_turn(client, "s-1", "hi", sigint, stdout=stdout, stderr=stderr)
        assert exit_code == 22
        out = stderr.getvalue()
@@ -933,7 +1018,8 @@ class TestRunTurn:
        )
        sigint = asyncio.Event()
        stdout, stderr = io.StringIO(), io.StringIO()
-        async with httpx.AsyncClient(base_url="https://w.example") as client:
+        async with httpx.AsyncClient(base_url="https://w.example") as _tp:
+            client = _wtc(_tp)
            exit_code = await _run_turn(client, "s-1", "hi", sigint, stdout=stdout, stderr=stderr)
        assert exit_code == 22
        out = stderr.getvalue()
@@ -955,7 +1041,8 @@ class TestRunTurn:
        )
        sigint = asyncio.Event()
        stdout, stderr = io.StringIO(), io.StringIO()
-        async with httpx.AsyncClient(base_url="https://w.example") as client:
+        async with httpx.AsyncClient(base_url="https://w.example") as _tp:
+            client = _wtc(_tp)
            exit_code = await _run_turn(client, "s-1", "hi", sigint, stdout=stdout, stderr=stderr)
        assert exit_code == 22
        out = stderr.getvalue()
@@ -978,7 +1065,8 @@ class TestRunTurn:
        sigint = asyncio.Event()
        sigint.set()  # SIGINT before _run_turn even starts
        stdout, stderr = io.StringIO(), io.StringIO()
-        async with httpx.AsyncClient(base_url="https://w.example") as client:
+        async with httpx.AsyncClient(base_url="https://w.example") as _tp:
+            client = _wtc(_tp)
            exit_code = await asyncio.wait_for(
                _run_turn(client, "s-1", "hi", sigint, stdout=stdout, stderr=stderr), timeout=2.0
            )
@@ -1007,7 +1095,8 @@ class TestRunTurn:

        sigint = asyncio.Event()
        stdout, stderr = io.StringIO(), io.StringIO()
-        async with httpx.AsyncClient(base_url="https://w.example") as client:
+        async with httpx.AsyncClient(base_url="https://w.example") as _tp:
+            client = _wtc(_tp)
            task = asyncio.create_task(
                _run_turn(client, "s-1", "hi", sigint, stdout=stdout, stderr=stderr)
            )
@@ -1045,7 +1134,8 @@ class TestRunTurn:

        sigint = asyncio.Event()
        stdout, stderr = io.StringIO(), io.StringIO()
-        async with httpx.AsyncClient(base_url="https://w.example") as client:
+        async with httpx.AsyncClient(base_url="https://w.example") as _tp:
+            client = _wtc(_tp)
            task = asyncio.create_task(
                _run_turn(client, "s-1", "hi", sigint, stdout=stdout, stderr=stderr)
            )
@@ -1091,7 +1181,8 @@ class TestRunTurn:
        monkeypatch.setattr(sigint, "wait", counting_wait)

        stdout, stderr = io.StringIO(), io.StringIO()
-        async with httpx.AsyncClient(base_url="https://w.example") as client:
+        async with httpx.AsyncClient(base_url="https://w.example") as _tp:
+            client = _wtc(_tp)
            task = asyncio.create_task(
                _run_turn(client, "s-1", "hi", sigint, stdout=stdout, stderr=stderr)
            )
@@ -1129,7 +1220,8 @@ class TestRunTurn:

        sigint = asyncio.Event()
        stdout, stderr = io.StringIO(), io.StringIO()
-        async with httpx.AsyncClient(base_url="https://w.example") as client:
+        async with httpx.AsyncClient(base_url="https://w.example") as _tp:
+            client = _wtc(_tp)
            task = asyncio.create_task(
                _run_turn(client, "s-1", "hi", sigint, stdout=stdout, stderr=stderr)
            )
@@ -1176,7 +1268,8 @@ class TestRunTurn:

        sigint = asyncio.Event()
        stdout, stderr = io.StringIO(), io.StringIO()
-        async with httpx.AsyncClient(base_url="https://w.example") as client:
+        async with httpx.AsyncClient(base_url="https://w.example") as _tp:
+            client = _wtc(_tp)
            exit_code = await _run_turn(client, "s-1", "hi", sigint, stdout=stdout, stderr=stderr)
        assert exit_code == 0
        assert call_count == 3
@@ -7,6 +7,12 @@ type. Server-side serialization (`_event_to_browser_payload`) is
 unit-tested against the fixture. JS-side rendering in
 `src/ratatoskr/web/static/index.html` consumes the same shape — if
 this fixture changes, both sides update in lockstep.
+
+Post worldtree-sdk cutover (#20): the presenter consumes SDK `TurnEvent`s.
+`_event_to_browser_payload` derives the browser payload from the SDK's `raw`
+(the wire body) plus the composite `sse_id` string — the SAME shape the old
+dataclasses produced, so the fixture is unchanged. These events are built via
+the SDK's own `build_event` from the wire body.
 """

 from __future__ import annotations
@@ -14,20 +20,8 @@ from __future__ import annotations
 import json
 from pathlib import Path

-from ratatoskr.sse_client import (
-    AffectUpdate,
-    AwaitingLlmFirstToken,
-    Cancelled,
-    Done,
-    Error,
-    SseId,
-    Text,
-    TextBoundary,
-    Thinking,
-    ToolResult,
-    ToolStart,
-    WorkerPhase,
-)
+from worldtree_sdk.events import build_event
+
 from ratatoskr.web.server import _event_to_browser_payload


@@ -36,6 +30,13 @@ def _load_fixture() -> dict:
    return json.loads(path.read_text())


+def _ev(ev_type: str, sse_id: str, **fields: object) -> object:
+    """Build an SDK TurnEvent from its wire body (raw includes `type`); turn_id is
+    the turn component of the composite sse_id."""
+    turn = int(sse_id.split(":", 1)[0])
+    return build_event(ev_type, sse_id, turn, {"type": ev_type, **fields})
+
+
 def _check(name: str, event: object) -> None:
    """Assert (event_type, data) for `event` matches the fixture entry."""
    fixture = _load_fixture()
@@ -51,58 +52,40 @@ def _check(name: str, event: object) -> None:


 def test_worker_phase_matches_fixture() -> None:
-    _check(
-        "worker_phase",
-        WorkerPhase(sse_id=SseId(42, 3), phase="BuildingPrompt", turn_id=42),
-    )
+    _check("worker_phase", _ev("worker_phase", "42:3", phase="BuildingPrompt", turn_id=42))


 def test_thinking_matches_fixture() -> None:
-    _check(
-        "thinking",
-        Thinking(sse_id=SseId(42, 5), content="Let me think..."),
-    )
+    _check("thinking", _ev("thinking", "42:5", content="Let me think..."))


 def test_text_matches_fixture() -> None:
-    _check(
-        "text",
-        Text(sse_id=SseId(42, 7), content="Hello there"),
-    )
+    _check("text", _ev("text", "42:7", content="Hello there"))


 def test_text_boundary_matches_fixture() -> None:
    _check(
        "text_boundary",
-        TextBoundary(
-            sse_id=SseId(42, 8), kind="sentence",
-            char_offset=11, ts="2026-05-28T00:00:00Z",
-        ),
+        _ev("text_boundary", "42:8", kind="sentence", char_offset=11, ts="2026-05-28T00:00:00Z"),
    )


 def test_tool_start_matches_fixture() -> None:
-    _check(
-        "tool_start",
-        ToolStart(sse_id=SseId(42, 9), name="search", arguments={"q": "ratatoskr"}),
-    )
+    _check("tool_start", _ev("tool_start", "42:9", name="search", arguments={"q": "ratatoskr"}))


 def test_tool_result_matches_fixture() -> None:
    _check(
        "tool_result",
-        ToolResult(
-            sse_id=SseId(42, 10), name="search",
-            result={"n": 1}, duration_ms=12,
-        ),
+        _ev("tool_result", "42:10", name="search", result={"n": 1}, duration_ms=12),
    )


 def test_done_matches_fixture() -> None:
    _check(
        "done",
-        Done(
-            sse_id=SseId(42, 11), phase="succeeded", response="Hello there",
+        _ev(
+            "done", "42:11", phase="succeeded", response="Hello there",
            model="qwen3.6-35-a3b", duration_ms=1234,
            usage={
                "prompt_tokens": 100, "completion_tokens": 50,
@@ -115,8 +98,8 @@ def test_done_matches_fixture() -> None:
 def test_error_matches_fixture() -> None:
    _check(
        "error",
-        Error(
-            sse_id=SseId(42, 11), phase="failed",
+        _ev(
+            "error", "42:11", phase="failed",
            message="llm output invalid", error_code="llm_output_invalid",
        ),
    )
@@ -125,8 +108,8 @@ def test_error_matches_fixture() -> None:
 def test_cancelled_matches_fixture() -> None:
    _check(
        "cancelled",
-        Cancelled(
-            sse_id=SseId(42, 11), phase="cancelled", turn_id=42,
+        _ev(
+            "cancelled", "42:11", phase="cancelled", turn_id=42,
            reason="user_cancel", partial_message_id=None,
        ),
    )
@@ -135,8 +118,8 @@ def test_cancelled_matches_fixture() -> None:
 def test_affect_update_matches_fixture() -> None:
    _check(
        "affect_update",
-        AffectUpdate(
-            sse_id=SseId(42, 1), status="current", turn_id=42,
+        _ev(
+            "affect_update", "42:1", status="current", turn_id=42,
            snapshot={
                "agent_id": "mimir",
                "pad": {"pleasure": 0.52, "arousal": 0.47, "dominance": 0.50},
@@ -155,8 +138,8 @@ def test_affect_update_matches_fixture() -> None:
 def test_awaiting_llm_first_token_matches_fixture() -> None:
    _check(
        "awaiting_llm_first_token",
-        AwaitingLlmFirstToken(
-            sse_id=SseId(42, 2), turn_id=42,
-            elapsed_ms_since_building_prompt=5012.3,
+        _ev(
+            "awaiting_llm_first_token", "42:2",
+            turn_id=42, elapsed_ms_since_building_prompt=5012.3,
        ),
    )
@@ -456,15 +456,16 @@ class TestCancelTurnEndpoint:

    @respx.mock
    def test_already_completed_race(self) -> None:
-        """already_completed [race]: upstream 409 → 200 reason=race_or_completed."""
+        """already_completed [race]: upstream 409 turn_finished → 200 reason=race_or_completed."""
        from ratatoskr.web.server import create_app
        app = create_app(_mock_client_factory())
        c = TestClient(app)
        turn_id = c.post("/api/turns/s-1", json={"content": "hi"}).json()["turn_id"]
        app.state.turn_registry[("s-1", turn_id)].status = "streaming"
        app.state.turn_registry[("s-1", turn_id)].upstream_turn_id = 42
+        # SDK gates the race on the (status, error_code) pair (B-CAN-3).
        respx.post("https://w.example/sessions/s-1/turns/42/cancel").mock(
-            return_value=httpx.Response(409)
+            return_value=httpx.Response(409, json={"error_code": "turn_finished"})
        )
        resp = c.post(f"/api/turns/s-1/cancel?turn_id={turn_id}")
        assert resp.status_code == 200
@@ -473,7 +474,11 @@ class TestCancelTurnEndpoint:

    @respx.mock
    def test_cancel_failed_500(self) -> None:
-        """cancel_failed [error]: upstream 500 → 500 with cancel_failed envelope."""
+        """cancel_failed [error]: upstream 500 → 502 cancel_failed envelope.
+
+        Post-cutover: the SDK abstracts the upstream cancel HTTP status behind a
+        typed CancelFailed, so the endpoint surfaces a generic 502 (bad gateway)
+        rather than echoing the upstream 500."""
        from ratatoskr.web.server import create_app
        app = create_app(_mock_client_factory())
        c = TestClient(app)
@@ -484,7 +489,7 @@ class TestCancelTurnEndpoint:
            return_value=httpx.Response(500, content=b"boom")
        )
        resp = c.post(f"/api/turns/s-1/cancel?turn_id={turn_id}")
-        assert resp.status_code == 500
+        assert resp.status_code == 502
        assert resp.json()["error_code"] == "cancel_failed"
        assert ("s-1", turn_id) not in app.state.turn_registry

@@ -472,7 +472,7 @@ wheels = [

 [[package]]
 name = "ratatoskr"
-version = "0.21.5"
+version = "0.21.7"
 source = { editable = "." }
 dependencies = [
    { name = "httpx" },