refactor(#20): delete the orphaned hand-rolled turn-stream paths (slice-2, part 2b-iii)

DEC-4 live smoke PASSED first (personal :8081, b127/b128): create → streamed turn
that rendered (worker_phase/text/text_boundary/done with usage) → SIGINT cancel that
round-tripped to a cancelled terminal. With both CLI + web on the adapter, the
hand-rolled turn-stream family is fully orphaned — deleting it now.

- sse_client.py (714 → 224): removed stream_turn / reconnect_turn /
  stream_turn_resilient / cancel_turn + the Event dataclasses (Text/Done/…/Event
  union) + CancelResult + the SSE parse helpers (_iter_events / _envelope_for_type /
  _parse_sse_id / _eager_failure_fields / _INT_RE). KEPT: the caller-semantic
  exceptions (the adapter raises them, DEC-2), SseId, AdminEvent, stream_admin_events
  (slice-6 admin surface).
- sessions.py (677 → 608): removed list_sessions + get_session_tools (no surface
  users) + SessionPage. KEPT: create_session / get_session_messages (the
  --seed-first-message probe still uses them, slice-3) + all exceptions + SessionInfo.
- tests: test_sse_client pruned to TestStreamAdminEvents; test_sessions dropped the
  list_sessions + get_session_tools classes. The deleted turn-stream behavior is now
  covered by test_wt.py + the CLI/web integration tests + the live smoke.

Suite 490 green (570 − 80 deleted turn-stream tests); ruff clean on all touched
files; no new mypy errors. Patch (internal cleanup; behavior preserved).

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "ratatoskr"
-version = "0.21.3"
+version = "0.21.8"
 description = "Worldtree Conversation API debug console (web + headless CLI) — multi-pane observability"
 readme = "README.md"
 requires-python = ">=3.12"

src/ratatoskr/cli.py

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

src/ratatoskr/sessions.py

@@ -38,12 +38,6 @@ class SessionInfo:
     config: dict[str, Any] | None = None
 
 
-@dataclass(frozen=True)
-class SessionPage:
-    """One page of GET /sessions results. `next_cursor=None` on the last page."""
-
-    items: list[SessionInfo]
-    next_cursor: str | None
 
 
 @dataclass(frozen=True)
@@ -204,53 +198,6 @@ class AuthoredHistoryUnavailable(Exception):
         self.session_id = session_id
 
 
-async def list_sessions(
-    client: httpx.AsyncClient,
-    *,
-    include_archived: bool = False,
-    limit: int = 50,
-    cursor: str | None = None,
-) -> SessionPage:
-    """GET /sessions. See contract FN list_sessions."""
-    assert client is not None
-    assert 1 <= limit <= 200
-    assert cursor is None or (isinstance(cursor, str) and cursor)
-
-    params: dict[str, str] = {"limit": str(limit)}
-    if include_archived:
-        params["include_archived"] = "true"
-    if cursor is not None:
-        params["cursor"] = cursor
-
-    resp = await client.get("/sessions", params=params)
-    if resp.status_code == 422:
-        try:
-            err = resp.json()
-        except ValueError:
-            err = {}
-        if err.get("error_code") == "cursor_invalid":
-            raise InvalidCursor(raw=cursor)
-        raise SessionApiFailed(status=422, body=resp.content)
-    if resp.status_code != 200:
-        raise SessionApiFailed(status=resp.status_code, body=resp.content)
-    body = resp.json()
-    items = [
-        SessionInfo(
-            session_id=item["session_id"],
-            agent_id=item["agent_id"],
-            created_at=item["created_at"],
-            last_active=item["last_active"],
-            metadata=item.get("metadata", {}),
-            message_count=None,
-            name=item.get("name"),
-            archived=item.get("archived") or False,
-            tags=item.get("tags") or [],
-            kind=item.get("kind"),      # INV-002 amendment (#161): present on list items
-            config=item.get("config"),  # forward-compat passthrough; None today
-        )
-        for item in body["items"]
-    ]
-    return SessionPage(items=items, next_cursor=body.get("next_cursor"))
 
 
 def endpoint_for_plane(plane: str, base_host: str) -> str:
@@ -572,22 +519,6 @@ async def get_session_bifrost(
     raise SessionApiFailed(status=resp.status_code, body=resp.content)
 
 
-async def get_session_tools(client: httpx.AsyncClient, session_id: str) -> dict[str, Any]:
-    """GET /sessions/{session_id}/tools — owner-scoped tool inventory (spec #183).
-
-    Returns the merged tool list the LLM saw at turn-fire: `{agent_id,
-    builtin_tools: [...], bifrost_tools: [{name, description, parameters}, ...]}`.
-    Owner-scoped (`ctx.user_id == session.user_id`) — reachable with the consumer
-    key, NO admin scope. Cross-owner access returns 404 `session_not_found`
-    (existence-hiding); a revoked session returns 401 `auth_revoked`. Parsed dict
-    verbatim; any non-200 → SessionApiFailed (mirrors get_persona_state).
-    """
-    assert client is not None
-    assert session_id and isinstance(session_id, str)
-    resp = await client.get(f"/sessions/{session_id}/tools")
-    if resp.status_code == 200:
-        return resp.json()
-    raise SessionApiFailed(status=resp.status_code, body=resp.content)
 
 
 async def get_capabilities(client: httpx.AsyncClient) -> dict[str, Any]:

src/ratatoskr/sse_client.py

@@ -6,7 +6,6 @@ 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
@@ -14,8 +13,6 @@ 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."""
@@ -24,155 +21,6 @@ class SseId(NamedTuple):
     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
-
-
-@dataclass(frozen=True)
-class AwaitingLlmFirstToken:
-    """SSE event `awaiting_llm_first_token`: heartbeat during slow first-token.
-
-    Fires at the configured interval (default 5s) during the gap between
-    `worker_phase` phase=BuildingPrompt and phase=CallingLLM. Lets clients
-    render a live "thinking for Ns…" indicator instead of a frozen line
-    during legitimate-slow first-token latency. Stops the moment CallingLLM
-    fires (defense-in-depth at three sites); no heartbeat after Cancelled
-    or stalled terminal events. Tool round-trip re-entries do NOT re-fire
-    heartbeats — INV-201-5 scopes the mechanism to the FIRST gap only.
-
-    `elapsed_ms_since_building_prompt` is server-authoritative
-    `time.monotonic()`-based — independent of network latency or clock
-    skew, monotonically increasing across the heartbeat sequence.
-
-    See docs/conversation-api-spec.md § awaiting_llm_first_token
-    (Worldtree #201, v0.29.0).
-    """
-
-    sse_id: SseId
-    turn_id: int
-    elapsed_ms_since_building_prompt: float
-
-
-@dataclass(frozen=True)
-class AffectUpdate:
-    """SSE event `affect_update`: persona-state observability snapshot.
-
-    Two emissions per qualifying turn (persona-enabled agent on non-
-    ephemeral session): `status="current"` at turn start carrying the full
-    snapshot, `status="scheduled"` after post-turn appraisal kicks off
-    (lightweight — `snapshot` is None). Suppressed entirely for persona-
-    disabled agents (e.g. `domari`, `muninn`), Tier 3 consumer-defined
-    agents (Phase 2.0), and ephemeral sessions.
-
-    Bootstrap reads available via `GET /agents/{agent_id}/persona_state`
-    (same `snapshot` shape, requires `persona.read` scope).
-
-    See docs/conversation-api-spec.md § affect_update (Worldtree #204,
-    v0.28.0).
-    """
-
-    sse_id: SseId
-    status: str  # "current" | "scheduled"
-    turn_id: int
-    snapshot: dict[str, Any] | None  # None when status="scheduled"
-
-
-Event = (
-    WorkerPhase
-    | Thinking
-    | Text
-    | TextBoundary
-    | ToolStart
-    | ToolResult
-    | Done
-    | Error
-    | Cancelled
-    | AffectUpdate
-    | AwaitingLlmFirstToken
-)
 
 
 @dataclass(frozen=True)
@@ -258,37 +106,18 @@ class TurnLaunchUnavailable(SseConnectFailed):
         self.message = message
 
 
-# Canonical error_codes (Worldtree #331 / v1.0.0b2): 409 -> agent_not_available,
-# 503 -> not_ready (retryable; re-pinned from internal_error). Used only as a
-# fallback default when the body omits error_code — the real code is surfaced
-# verbatim from the {detail:{error_code,message}} envelope.
-_EAGER_TURN_FAILURE_CODE = {409: "agent_not_available", 503: "not_ready"}
-
-
-def _eager_failure_fields(body: bytes, status: int) -> tuple[str, str]:
-    """Extract (error_code, message) from an eager turn-launch failure body
-    (#331). Accepts the Worldtree `{"detail": {...}}` envelope OR a flat
-    `{error_code, message}`; falls back to a status-derived default code and a
-    generic message when the body is absent / non-JSON / malformed."""
-    try:
-        parsed: Any = json.loads(body)
-    except (json.JSONDecodeError, ValueError):
-        parsed = None
-    src: dict[str, Any] = {}
-    if isinstance(parsed, dict):
-        detail = parsed.get("detail")
-        src = detail if isinstance(detail, dict) else parsed
-    code = src.get("error_code") or _EAGER_TURN_FAILURE_CODE[status]
-    message = src.get("message")
-    if not isinstance(message, str):
-        message = f"turn launch failed (HTTP {status})"
-    return str(code), message
 
 
 class SseConnectionDropped(Exception):
-    """Raised when the HTTP/SSE connection dropped mid-stream."""
+    """Raised when the HTTP/SSE connection dropped mid-stream.
 
-    def __init__(self, *, last_seen_sse_id: SseId | None) -> None:
+    `last_seen_sse_id` is the resume cursor of the last frame seen. The
+    hand-rolled path carries a parsed `SseId`; the worldtree-sdk cutover carries
+    the SDK's raw composite-id `str` (the cutover's target form) — both accepted
+    during the migration.
+    """
+
+    def __init__(self, *, last_seen_sse_id: SseId | str | 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
 
@@ -346,274 +175,6 @@ class CancelFailed(Exception):
         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"),
-        )
-    if t == "awaiting_llm_first_token":
-        # Worldtree #201 / v0.29.0: top-level heartbeat during BuildingPrompt
-        # → CallingLLM gap. Lets clients render live elapsed-time indicators
-        # instead of frozen lines on legitimate-slow first-token latency.
-        return AwaitingLlmFirstToken(
-            sse_id=sse_id,
-            turn_id=body["turn_id"],
-            elapsed_ms_since_building_prompt=body["elapsed_ms_since_building_prompt"],
-        )
-    if t == "affect_update":
-        # Worldtree #204 / v0.28.0: persona-state observability event.
-        # status="current" carries full snapshot at turn start;
-        # status="scheduled" omits snapshot (lightweight post-appraisal-
-        # kickoff notification).
-        return AffectUpdate(
-            sse_id=sse_id,
-            status=body["status"],
-            turn_id=body["turn_id"],
-            snapshot=body.get("snapshot"),
-        )
-    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
-    terminal_seen = False
-    try:
-        async for sse in event_source.aiter_sse():
-            # Issue #7 INV-001: empty-data frames are keepalives — skip silently.
-            # ORDERING: this branch fires BEFORE _parse_sse_id; an empty-data event
-            # with a malformed id is silently swallowed (intentional — a keepalive
-            # with a bad id is still a keepalive). Don't reorder.
-            if sse.data == "":
-                continue
-            # v0.8.1: empty-id frames are also treated as keepalives. Worldtree
-            # SOMETIMES emits events without an `id:` line (observed mid-stream
-            # on the qwen3.6-35-a3b-heretic provider, 2026-05-25). Per the SSE
-            # RFC, events without ids are legitimate (they just don't update
-            # Last-Event-ID); the previous strict behavior crashed every turn
-            # on the offending agent. Treat same as empty-data: skip silently.
-            if sse.id == "":
-                continue
-            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)
-            try:
-                body = json.loads(sse.data)
-            except json.JSONDecodeError as exc:
-                raise MalformedSseData(raw=sse.data) from exc
-            event = _envelope_for_type(body, sse_id=sse_id)
-            yield event
-            last_sse_id = sse_id
-            if isinstance(event, (Done, Error, Cancelled)):
-                terminal_seen = True
-                return
-    except (httpx.ReadError, httpx.RemoteProtocolError, httpx.ReadTimeout) as exc:
-        # ReadTimeout covers idle gaps that exceed httpx's read timeout — the SSE
-        # stream went quiet long enough for httpx to give up. Treat the same as a
-        # raw read error: surface as SseConnectionDropped so the caller can decide
-        # whether to reconnect_turn. (Callers SHOULD configure a long-or-disabled
-        # read timeout on their AsyncClient for SSE; this is defense in depth.)
-        raise SseConnectionDropped(last_seen_sse_id=last_sse_id) from exc
-    if not terminal_seen:
-        # Clean EOF before terminal event — INV-001 says stream MUST NOT end
-        # without exactly one Done/Error/Cancelled. Surface as connection drop;
-        # caller may reconnect_turn if it holds last_sse_id.
-        raise SseConnectionDropped(last_seen_sse_id=last_sse_id)
-
-
-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:
-        # Worldtree v1.0.0b1 (#331): turn-launch failures arrive EAGERLY as a
-        # status before any stream — 409 agent_not_available (pre-b1 this was a
-        # 200 + in-stream `error` event), 503 a transient retryable launch
-        # failure. Surface them as typed SseConnectFailed subclasses carrying
-        # error_code; request-level non-2xx (404 session_not_found, etc.) stay
-        # generic SseConnectFailed.
-        status = event_source.response.status_code
-        if status in (409, 503):
-            body = await event_source.response.aread()
-            code, message = _eager_failure_fields(body, status)
-            if status == 409:
-                raise AgentNotAvailable(body=body, error_code=code, message=message)
-            raise TurnLaunchUnavailable(body=body, error_code=code, message=message)
-        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
-
-
-async def stream_turn_resilient(
-    client: httpx.AsyncClient,
-    session_id: str,
-    content: str,
-    *,
-    max_reconnects: int = 5,
-) -> AsyncIterator[Event]:
-    """Resume-orchestration wrapper over stream_turn + reconnect_turn.
-
-    Yields ONE continuous Event stream; on `SseConnectionDropped` (mid-stream
-    drop or clean EOF before a terminal), resumes from the last-seen `sse_id`
-    via `reconnect_turn`, up to `max_reconnects` times, until a terminal
-    Done/Error/Cancelled arrives. The single shared surface presenters consume
-    for resilient streaming (design-brief §8b: "share the consumer, branch the
-    presenter"). Cross-process resume stays deferred to v2 (§8d): `last_seen`
-    lives only in this generator's frame. See contract FN stream_turn_resilient
-    (amendment 2026-06-30).
-    """
-    assert client is not None
-    assert session_id and isinstance(session_id, str)
-    assert content and isinstance(content, str)
-    assert isinstance(max_reconnects, int) and max_reconnects >= 0
-
-    last_seen: SseId | None = None
-    reconnects = 0
-    gen = stream_turn(client, session_id, content)
-    while True:
-        try:
-            async for event in gen:
-                last_seen = event.sse_id
-                yield event
-            return  # generator completed cleanly → terminal event reached (INV-001)
-        except SseConnectionDropped as drop:
-            # Prefer the id we tracked from a yielded event; fall back to the one
-            # the drop carries (covers a drop on the very first frame). Non-drop
-            # reconnect failures (412/410/400/flip) are NOT caught here — they
-            # propagate per the contract's "surface, not recover" policy.
-            seen = last_seen or drop.last_seen_sse_id
-            if seen is None or reconnects >= max_reconnects:
-                raise
-            reconnects += 1
-            gen = reconnect_turn(
-                client,
-                session_id,
-                content,
-                last_event_id=f"{seen.turn_id}:{seen.seq}",
-            )
-
 
 async def stream_admin_events(
     client: httpx.AsyncClient,
@@ -660,48 +221,3 @@ async def stream_admin_events(
             raise SseConnectionDropped(last_seen_sse_id=None) from exc
 
 
-def _parse_sse_id(raw: str) -> SseId:
-    """Parse the SSE wire `id:` as composite `{turn_id}:{seq}`. See contract FN _parse_sse_id."""
-    assert isinstance(raw, str)
-    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"),
-    )

src/ratatoskr/web/server.py

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

src/ratatoskr/wt.py

@@ -27,8 +27,37 @@ carries the SDK's parsed `error_code`).
 
 from __future__ import annotations
 
+import json
+from collections.abc import AsyncGenerator, Mapping
+from typing import Any
+
 import httpx
-from worldtree_sdk import ApiError, AuthProvider, WorldtreeClient
+import worldtree_sdk as wtsdk
+from worldtree_sdk import ApiError, AuthProvider, CancelResult, WorldtreeClient
+
+# Transitional (slice-2): the caller-semantic exceptions + the BifrostBinding input
+# type still live in the retiring `sessions` / `sse_client` modules; they relocate
+# into this adapter as their call-sites are rewired in later slice-2 commits. wt →
+# sessions / sse_client is one-way (neither imports wt), so there is no cycle.
+from .sessions import (
+    AgentNotFound,
+    BifrostBinding,
+    BifrostConsumerKeyMissing,
+    BifrostHandshakeFailed,
+    InvalidCursor,
+)
+from .sse_client import (
+    AgentNotAvailable,
+    CancelAlreadyCompleted,
+    CancelFailed,
+    CancelTurnNotFound,
+    MalformedSseData,
+    MalformedSseId,
+    SseConnectFailed,
+    SseConnectionDropped,
+    TurnIdFlip,
+    TurnLaunchUnavailable,
+)
 
 
 class SessionApiFailed(Exception):
@@ -52,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.
 
@@ -59,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,
     )
 
 
@@ -90,3 +123,203 @@ def translate_error(exc: BaseException) -> BaseException:
             status=exc.status, error_code=exc.error_code, body=exc.body
         )
     return exc
+
+
+# ── slice-2: sessions/turn adapter routes ────────────────────────────────────
+# Ratatoskr-semantic call surfaces over `WorldtreeClient.sessions.*`. Each builds
+# the request from ratatoskr's domain params, delegates the HTTP to the SDK, and
+# maps the SDK's `ApiError` floor by ROUTE (INV-CUT-2) — route-specific rows first,
+# `translate_error`'s `SessionApiFailed` default otherwise. Open-world reads are
+# returned verbatim (the parity-pass posture: presenters read them as mappings,
+# tolerant of wire drift). The turn STREAM + cancel land alongside the presenter
+# rewire in the next slice-2 commit.
+
+
+def _bifrost_error_from_body(body: str | None) -> str | None:
+    """Pull the spec-level `bifrost_error` from a bound-create 502 body string.
+
+    Tolerates both the FastAPI-nested `{"detail": {"bifrost_error": …}}` shape (the
+    real wire form) and a flat top-level `bifrost_error` — the same both-shape
+    unwrap the hand-rolled path used, adapted to the SDK's already-parsed str body.
+    """
+    if not body:
+        return None
+    try:
+        err = json.loads(body)
+    except (json.JSONDecodeError, ValueError):
+        return None
+    if not isinstance(err, dict):
+        return None
+    bifrost_error = err.get("bifrost_error")
+    if bifrost_error is None and isinstance(err.get("detail"), dict):
+        bifrost_error = err["detail"].get("bifrost_error")
+    return bifrost_error
+
+
+async def create_session(
+    client: WorldtreeClient,
+    agent_id: str,
+    *,
+    end_user_id: str | None = None,
+    bifrost: BifrostBinding | None = None,
+    consumer_key: str | None = None,
+    config: Mapping[str, Any] | None = None,
+) -> Mapping[str, Any]:
+    """Create a session (POST /sessions), returning the open-world create result.
+
+    Body-building mirrors the hand-rolled path: `{agent_id}` plus `end_user_id` /
+    `config` / `bifrost` when set. A bound create authenticates with `consumer_key`
+    via the SDK's per-request auth (never a header, never the canary fallback —
+    INV-001); the key is required pre-HTTP. Error map (INV-CUT-2): 404 →
+    `AgentNotFound`; a bound 502 → `BifrostHandshakeFailed`; otherwise the
+    `SessionApiFailed` default.
+    """
+    assert agent_id and isinstance(agent_id, str)
+    assert end_user_id is None or (isinstance(end_user_id, str) and end_user_id)
+    assert config is None or isinstance(config, Mapping)
+    # Ephemeral config + Bifrost binding are mutually exclusive (server 422s).
+    assert not (config is not None and bifrost is not None)
+    # INV-001: a bound create REQUIRES a non-empty consumer key — enforced pre-HTTP
+    # so it never falls back to the canary bearer.
+    if bifrost is not None and not (isinstance(consumer_key, str) and consumer_key):
+        raise BifrostConsumerKeyMissing()
+
+    body: dict[str, Any] = {"agent_id": agent_id}
+    if end_user_id is not None:
+        body["end_user_id"] = end_user_id
+    if config is not None:
+        body["config"] = dict(config)
+    if bifrost is not None:
+        body["bifrost"] = {"endpoint_url": bifrost.endpoint_url, "scope": bifrost.scope}
+
+    try:
+        return await client.sessions.create(body, consumer_key=consumer_key)
+    except ApiError as exc:
+        if exc.status == 404:
+            raise AgentNotFound(agent_id=agent_id) from exc
+        if bifrost is not None and exc.status == 502:
+            raise BifrostHandshakeFailed(
+                bifrost_error=_bifrost_error_from_body(exc.body),
+                body=(exc.body or "").encode(),
+            ) from exc
+        raise translate_error(exc) from exc
+
+
+async def list_sessions(
+    client: WorldtreeClient,
+    *,
+    include_archived: bool = False,
+    limit: int = 50,
+    cursor: str | None = None,
+) -> Mapping[str, Any]:
+    """List sessions (GET /sessions), returning the open-world page verbatim. A 422
+    `cursor_invalid` → `InvalidCursor`; otherwise the `SessionApiFailed` default."""
+    assert 1 <= limit <= 200
+    assert cursor is None or (isinstance(cursor, str) and cursor)
+    try:
+        return await client.sessions.list(
+            limit=limit, cursor=cursor, include_archived=include_archived or None
+        )
+    except ApiError as exc:
+        if exc.status == 422 and exc.error_code == "cursor_invalid":
+            raise InvalidCursor(raw=cursor) from exc
+        raise translate_error(exc) from exc
+
+
+async def get_session_messages(
+    client: WorldtreeClient, session_id: str
+) -> Mapping[str, Any]:
+    """The session's message history (GET /sessions/{id}/messages), verbatim. Any
+    error → the `SessionApiFailed` default (owner-scoped; 404 hide-existence stays
+    generic here — messages is not a hide-existence-mapped route)."""
+    assert session_id and isinstance(session_id, str)
+    try:
+        return await client.sessions.messages(session_id)
+    except ApiError as exc:
+        raise translate_error(exc) from exc
+
+
+async def get_session_tools(
+    client: WorldtreeClient, session_id: str
+) -> Mapping[str, Any]:
+    """The owner-scoped tool inventory (GET /sessions/{id}/tools), verbatim. Any
+    error → the `SessionApiFailed` default."""
+    assert session_id and isinstance(session_id, str)
+    try:
+        return await client.sessions.tools(session_id)
+    except ApiError as exc:
+        raise translate_error(exc) from exc
+
+
+async def stream_turn(
+    client: WorldtreeClient, session_id: str, content: str
+) -> 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
+    keeps catching ratatoskr's types).
+
+    The SDK's `stream_turn` retries only the transport-drop class internally; a
+    resume failure / protocol violation / connect failure surfaces unchanged
+    (B-RES-6), and a drop that exhausts the reconnect budget surfaces as
+    `ConnectionDropped`. The eager launch failures (`AgentNotAvailable` 409,
+    `TurnLaunchUnavailable` 503) and `SessionRetired` 410 are subclasses of the
+    SDK's `ConnectFailed`, so they are caught before the generic `ConnectFailed`.
+    """
+    try:
+        async for event in client.sessions.stream_turn(session_id, content):
+            yield event
+    except wtsdk.SessionRetired as exc:
+        # Fresh-mode 410 → the session is gone server-side; a generic API failure.
+        raise SessionApiFailed(
+            status=exc.status, error_code=exc.error_code, body=exc.message
+        ) from exc
+    except wtsdk.AgentNotAvailable as exc:
+        raise AgentNotAvailable(
+            body=(exc.message or "").encode(),
+            error_code=exc.error_code,
+            message=exc.message or "",
+        ) from exc
+    except wtsdk.TurnLaunchUnavailable as exc:
+        raise TurnLaunchUnavailable(
+            body=(exc.message or "").encode(),
+            error_code=exc.error_code,
+            message=exc.message or "",
+        ) from exc
+    except wtsdk.ConnectFailed as exc:
+        raise SseConnectFailed(status=exc.status, body=(exc.message or "").encode()) from exc
+    except wtsdk.ResumeError as exc:
+        # A terminal resume failure (the resilient stream absorbs the retryable ones).
+        raise SseConnectFailed(status=exc.status, body=(exc.message or "").encode()) from exc
+    except wtsdk.ConnectionDropped as exc:
+        raise SseConnectionDropped(last_seen_sse_id=exc.last_seen_sse_id) from exc
+    except wtsdk.MalformedSseId as exc:
+        raise MalformedSseId(raw=exc.raw) from exc
+    except wtsdk.MalformedSseData as exc:
+        raise MalformedSseData(raw=exc.raw) from exc
+    except wtsdk.TurnIdFlip as exc:
+        raise TurnIdFlip(established=exc.established, got=exc.got) from exc
+
+
+async def cancel_turn(
+    client: WorldtreeClient, session_id: str, turn_id: int, *, persist_partial: bool = False
+) -> CancelResult:
+    """Cancel a running turn (POST /sessions/{id}/turns/{turn_id}/cancel). Returns the
+    SDK `CancelResult` (a 200 with `cancelled=False` is the benign late-cancel race,
+    not an error). The typed cancel races map onto ratatoskr's same-named exceptions
+    (DEC-2): 404 `turn_not_found` → `CancelTurnNotFound`, 409 `turn_finished` →
+    `CancelAlreadyCompleted`, any other cancel failure → `CancelFailed`."""
+    assert session_id and isinstance(session_id, str)
+    assert isinstance(turn_id, int) and turn_id > 0
+    try:
+        return await client.sessions.cancel_turn(
+            session_id, turn_id, persist_partial=persist_partial
+        )
+    except wtsdk.CancelTurnNotFound as exc:
+        raise CancelTurnNotFound(turn_id=turn_id) from exc
+    except wtsdk.CancelAlreadyCompleted as exc:
+        raise CancelAlreadyCompleted(turn_id=turn_id) from exc
+    except wtsdk.CancelError as exc:
+        raise CancelFailed(
+            status=0, body=(getattr(exc, "message", "") or str(exc)).encode()
+        ) from exc

tests/test_cli.py

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

tests/test_sessions.py

@@ -13,10 +13,8 @@ from ratatoskr.sessions import (
     BifrostBinding,
     BifrostConsumerKeyMissing,
     BifrostHandshakeFailed,
-    InvalidCursor,
     PersonaNotConfigured,
     SessionApiFailed,
-    SessionPage,
     create_character,
     create_session,
     delete_character,
@@ -27,10 +25,8 @@ from ratatoskr.sessions import (
     get_persona_state,
     get_session_bifrost,
     get_session_messages,
-    get_session_tools,
     list_agents,
     list_character_models,
-    list_sessions,
     set_persona_state,
     write_authored_history,
 )
@@ -551,223 +547,6 @@ class TestEndpointForPlane:
             endpoint_for_plane("persona", "10.100.10.50")
 
 
-def _list_item(
-    *,
-    session_id: str = "s1",
-    agent_id: str = "mimir",
-    created_at: str = "2026-04-15T12:00:00+00:00",
-    last_active: str = "2026-04-15T12:05:00+00:00",
-    metadata: dict[str, object] | None = None,
-    name: str | None = "Research session",
-    archived: bool = False,
-    tags: list[str] | None = None,
-) -> dict[str, object]:
-    """Build a GET /sessions list-item body for tests."""
-    item: dict[str, object] = {
-        "session_id": session_id,
-        "agent_id": agent_id,
-        "created_at": created_at,
-        "last_active": last_active,
-        "metadata": metadata if metadata is not None else {},
-        "name": name,
-        "archived": archived,
-        "tags": tags if tags is not None else ["work"],
-    }
-    return item
-
-
-class TestListSessions:
-    @respx.mock
-    async def test_explicit_null_list_defaults(self) -> None:
-        """INV-002: explicit-null archived -> False; explicit-null tags -> []."""
-        raw_item = {
-            "session_id": "s1",
-            "agent_id": "mimir",
-            "created_at": "2026-04-15T12:00:00+00:00",
-            "last_active": "2026-04-15T12:05:00+00:00",
-            "metadata": {},
-            "name": None,
-            "archived": None,
-            "tags": None,
-        }
-        respx.get("https://w.example/sessions").mock(
-            return_value=httpx.Response(
-                200, json={"items": [raw_item], "next_cursor": None}
-            )
-        )
-        async with httpx.AsyncClient(base_url="https://w.example") as client:
-            page = await list_sessions(client)
-        info = page.items[0]
-        assert info.archived is False, "explicit-null archived must default to False"
-        assert info.tags == [], "explicit-null tags must default to []"
-        assert info.name is None
-
-    @respx.mock
-    async def test_happy_first_page(self) -> None:
-        """happy_first_page [happy,tracer]: one item + next_cursor -> SessionPage shape."""
-        respx.get("https://w.example/sessions").mock(
-            return_value=httpx.Response(
-                200,
-                json={
-                    "items": [_list_item()],
-                    "next_cursor": "v1.eyJhYmMifQ",
-                },
-            )
-        )
-        async with httpx.AsyncClient(base_url="https://w.example") as client:
-            page = await list_sessions(client)
-        assert isinstance(page, SessionPage)
-        assert len(page.items) == 1
-        assert page.next_cursor == "v1.eyJhYmMifQ"
-        info = page.items[0]
-        assert info.session_id == "s1"
-        assert info.message_count is None  # INV-002: not in list response
-        assert info.name == "Research session"
-        assert info.archived is False
-        assert info.tags == ["work"]
-
-    @respx.mock
-    async def test_happy_last_page(self) -> None:
-        """happy_last_page: next_cursor=null -> SessionPage(next_cursor=None)."""
-        respx.get("https://w.example/sessions").mock(
-            return_value=httpx.Response(
-                200,
-                json={"items": [_list_item()], "next_cursor": None},
-            )
-        )
-        async with httpx.AsyncClient(base_url="https://w.example") as client:
-            page = await list_sessions(client)
-        assert page.next_cursor is None
-
-    @respx.mock
-    async def test_empty_results(self) -> None:
-        """empty_results: {items: [], next_cursor: null} -> SessionPage([], None)."""
-        respx.get("https://w.example/sessions").mock(
-            return_value=httpx.Response(200, json={"items": [], "next_cursor": None})
-        )
-        async with httpx.AsyncClient(base_url="https://w.example") as client:
-            page = await list_sessions(client)
-        assert page == SessionPage(items=[], next_cursor=None)
-
-    @respx.mock
-    async def test_include_archived_query(self) -> None:
-        """include_archived_query: True -> has param; default -> NO param at all."""
-        route = respx.get("https://w.example/sessions").mock(
-            return_value=httpx.Response(
-                200, json={"items": [], "next_cursor": None}
-            )
-        )
-        async with httpx.AsyncClient(base_url="https://w.example") as client:
-            await list_sessions(client, include_archived=True)
-            await list_sessions(client)  # default
-        url_with = str(route.calls[0].request.url)
-        url_default = str(route.calls[1].request.url)
-        assert "include_archived=true" in url_with
-        assert "include_archived" not in url_default
-
-    @respx.mock
-    async def test_cursor_threaded(self) -> None:
-        """cursor_threaded: cursor=opaque -> URL has cursor=opaque."""
-        route = respx.get("https://w.example/sessions").mock(
-            return_value=httpx.Response(
-                200, json={"items": [], "next_cursor": None}
-            )
-        )
-        async with httpx.AsyncClient(base_url="https://w.example") as client:
-            await list_sessions(client, cursor="opaque-from-prev-page")
-        assert "cursor=opaque-from-prev-page" in str(route.calls[0].request.url)
-
-    @respx.mock
-    async def test_limit_query(self) -> None:
-        """limit_query: limit=10 -> URL has limit=10."""
-        route = respx.get("https://w.example/sessions").mock(
-            return_value=httpx.Response(
-                200, json={"items": [], "next_cursor": None}
-            )
-        )
-        async with httpx.AsyncClient(base_url="https://w.example") as client:
-            await list_sessions(client, limit=10)
-        assert "limit=10" in str(route.calls[0].request.url)
-
-    @respx.mock
-    async def test_invalid_cursor_server(self) -> None:
-        """invalid_cursor_server: 422 cursor_invalid -> InvalidCursor(raw=<passed cursor>)."""
-        respx.get("https://w.example/sessions").mock(
-            return_value=httpx.Response(
-                422,
-                json={"error_code": "cursor_invalid", "message": "bad cursor"},
-            )
-        )
-        async with httpx.AsyncClient(base_url="https://w.example") as client:
-            with pytest.raises(InvalidCursor) as exc_info:
-                await list_sessions(client, cursor="bogus")
-        assert exc_info.value.raw == "bogus"
-
-    @respx.mock
-    async def test_other_validation_failed(self) -> None:
-        """other_validation_failed: 422 other error_code -> SessionApiFailed(422); truncated."""
-        respx.get("https://w.example/sessions").mock(
-            return_value=httpx.Response(
-                422,
-                json={"error_code": "validation_failed", "message": "limit out of range"},
-            )
-        )
-        async with httpx.AsyncClient(base_url="https://w.example") as client:
-            with pytest.raises(SessionApiFailed) as exc_info:
-                await list_sessions(client)
-        assert exc_info.value.status == 422
-        assert len(exc_info.value.body) <= 1024
-
-    @respx.mock
-    async def test_unexpected_status_truncates(self) -> None:
-        """unexpected_status_truncates: 500 + 5000-byte body -> SessionApiFailed; body == 1024."""
-        big = b"x" * 5000
-        respx.get("https://w.example/sessions").mock(
-            return_value=httpx.Response(500, content=big)
-        )
-        async with httpx.AsyncClient(base_url="https://w.example") as client:
-            with pytest.raises(SessionApiFailed) as exc_info:
-                await list_sessions(client)
-        assert exc_info.value.status == 500
-        assert exc_info.value.body == big[:1024]
-
-    @respx.mock
-    async def test_limit_below_one(self) -> None:
-        """limit_below_one [adversarial]: limit=0 -> AssertionError; no HTTP."""
-        route = respx.get("https://w.example/sessions").mock(
-            return_value=httpx.Response(200, json={"items": [], "next_cursor": None})
-        )
-        async with httpx.AsyncClient(base_url="https://w.example") as client:
-            with pytest.raises(AssertionError):
-                await list_sessions(client, limit=0)
-        assert route.call_count == 0
-
-    @respx.mock
-    async def test_limit_above_max(self) -> None:
-        """limit_above_max [adversarial]: limit=300 -> AssertionError; no HTTP."""
-        route = respx.get("https://w.example/sessions").mock(
-            return_value=httpx.Response(200, json={"items": [], "next_cursor": None})
-        )
-        async with httpx.AsyncClient(base_url="https://w.example") as client:
-            with pytest.raises(AssertionError):
-                await list_sessions(client, limit=300)
-        assert route.call_count == 0
-
-    @respx.mock
-    async def test_empty_cursor(self) -> None:
-        """empty_cursor [adversarial]: cursor='' -> AssertionError; no HTTP."""
-        route = respx.get("https://w.example/sessions").mock(
-            return_value=httpx.Response(200, json={"items": [], "next_cursor": None})
-        )
-        async with httpx.AsyncClient(base_url="https://w.example") as client:
-            with pytest.raises(AssertionError):
-                await list_sessions(client, cursor="")
-        assert route.call_count == 0
-
-
-# ---- Issue #8: list_agents + AgentInfo --------------------------------------
-
-
 class TestListAgents:
     @respx.mock
     async def test_happy_full_shape(self) -> None:
@@ -1145,53 +924,6 @@ class TestGetCapabilities:
         assert exc.value.status == 500
 
 
-class TestGetSessionTools:
-    """docs/contracts/issues/2.contract.md — get_session_tools (GET /sessions/{id}/tools, #183)."""
-
-    @respx.mock
-    async def test_happy(self) -> None:
-        """happy [happy,tracer]: 200 → merged tool inventory dict verbatim."""
-        respx.get("https://w.example/sessions/s1/tools").mock(
-            return_value=httpx.Response(
-                200,
-                json={
-                    "agent_id": "alice:wizard",
-                    "builtin_tools": [],
-                    "bifrost_tools": [
-                        {"name": "bifrost.alice.set_field", "description": "d", "parameters": {}}
-                    ],
-                },
-            )
-        )
-        async with httpx.AsyncClient(base_url="https://w.example") as client:
-            tools = await get_session_tools(client, "s1")
-        assert tools["agent_id"] == "alice:wizard"
-        assert tools["builtin_tools"] == []
-        assert tools["bifrost_tools"][0]["name"] == "bifrost.alice.set_field"
-
-    @respx.mock
-    async def test_cross_owner_404_raises(self) -> None:
-        """cross_owner_404 [error]: 404 session_not_found → SessionApiFailed(404)."""
-        respx.get("https://w.example/sessions/s1/tools").mock(
-            return_value=httpx.Response(404, json={"error_code": "session_not_found"})
-        )
-        async with httpx.AsyncClient(base_url="https://w.example") as client:
-            with pytest.raises(SessionApiFailed) as exc:
-                await get_session_tools(client, "s1")
-        assert exc.value.status == 404
-
-    @respx.mock
-    async def test_empty_session_id_asserts(self) -> None:
-        """empty_session_id [adversarial]: '' → AssertionError; no HTTP issued."""
-        route = respx.get("https://w.example/sessions//tools").mock(
-            return_value=httpx.Response(200, json={})
-        )
-        async with httpx.AsyncClient(base_url="https://w.example") as client:
-            with pytest.raises(AssertionError):
-                await get_session_tools(client, "")
-        assert route.call_count == 0
-
-
 class TestGetSessionBifrost:
     """#2 contract — get_session_bifrost (GET /admin/sessions/{id}/bifrost, #176)."""
 

tests/test_web_presentation_contract.py

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

tests/test_web_server.py

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

tests/test_wt.py

@@ -12,10 +12,106 @@ slice 2. These tests hit no network (WorldtreeClient does no I/O at construction
 
 from __future__ import annotations
 
-import httpx
-from worldtree_sdk import AgentNotAvailable, ApiError, WorldtreeClient
+from typing import Any, cast
 
-from ratatoskr.wt import SessionApiFailed, build_client, translate_error
+import httpx
+import pytest
+import worldtree_sdk as wtsdk
+from worldtree_sdk import ApiError, CancelResult, WorldtreeClient
+
+from ratatoskr.sessions import (
+    AgentNotFound,
+    BifrostBinding,
+    BifrostConsumerKeyMissing,
+    BifrostHandshakeFailed,
+    InvalidCursor,
+)
+from ratatoskr.sse_client import (
+    AgentNotAvailable,
+    CancelAlreadyCompleted,
+    CancelFailed,
+    CancelTurnNotFound,
+    MalformedSseData,
+    MalformedSseId,
+    SseConnectFailed,
+    SseConnectionDropped,
+    TurnIdFlip,
+    TurnLaunchUnavailable,
+)
+from ratatoskr.wt import (
+    SessionApiFailed,
+    build_client,
+    cancel_turn,
+    create_session,
+    get_session_messages,
+    get_session_tools,
+    list_sessions,
+    stream_turn,
+    translate_error,
+)
+
+
+class _FakeSessions:
+    """A stand-in for `WorldtreeClient.sessions` — records the last call and
+    returns a canned result or raises a canned error. Lets the adapter's
+    body-building + error-mapping be unit-tested without any SDK HTTP."""
+
+    def __init__(
+        self,
+        *,
+        result: Any = None,
+        error: BaseException | None = None,
+        events: list[Any] | None = None,
+        stream_error: BaseException | None = None,
+    ) -> None:
+        self._result = result
+        self._error = error
+        self._events = events or []
+        self._stream_error = stream_error
+        self.calls: list[tuple[str, tuple[Any, ...], dict[str, Any]]] = []
+
+    async def _dispatch(self, name: str, *args: Any, **kwargs: Any) -> Any:
+        self.calls.append((name, args, kwargs))
+        if self._error is not None:
+            raise self._error
+        return self._result
+
+    async def create(self, *args: Any, **kwargs: Any) -> Any:
+        return await self._dispatch("create", *args, **kwargs)
+
+    async def list(self, *args: Any, **kwargs: Any) -> Any:
+        return await self._dispatch("list", *args, **kwargs)
+
+    async def messages(self, *args: Any, **kwargs: Any) -> Any:
+        return await self._dispatch("messages", *args, **kwargs)
+
+    async def tools(self, *args: Any, **kwargs: Any) -> Any:
+        return await self._dispatch("tools", *args, **kwargs)
+
+    def stream_turn(self, *args: Any, **kwargs: Any) -> Any:
+        self.calls.append(("stream_turn", args, kwargs))
+        return self._astream()
+
+    async def _astream(self) -> Any:
+        for event in self._events:
+            yield event
+        if self._stream_error is not None:
+            raise self._stream_error
+
+    async def cancel_turn(self, *args: Any, **kwargs: Any) -> Any:
+        return await self._dispatch("cancel_turn", *args, **kwargs)
+
+
+class _FakeClient:
+    def __init__(self, sessions: _FakeSessions) -> None:
+        self.sessions = sessions
+
+
+def _wt(sessions: _FakeSessions) -> WorldtreeClient:
+    """Cast the structural fake to the nominal client type the adapter is typed
+    against — the route functions only touch `client.sessions.*`, which the fake
+    provides. (No network; construction does no I/O.)"""
+    return cast(WorldtreeClient, _FakeClient(sessions))
 
 
 class TestBuildClient:
@@ -76,10 +172,241 @@ class TestTranslateError:
 
     def test_discriminated_subclass_passes_through_unchanged(self) -> None:
         # Discriminated WorldtreeError subclasses are already the right semantic
-        # type — the adapter passes them through by identity (no re-wrap).
-        exc = AgentNotAvailable("agent_not_available", "gone", status=409)
+        # type at the REST layer — translate_error passes them through by identity
+        # (the stream routes re-wrap them; that is stream_turn's job, not this one).
+        exc = wtsdk.AgentNotAvailable("agent_not_available", "gone", status=409)
         assert translate_error(exc) is exc
 
     def test_non_worldtree_error_passes_through_unchanged(self) -> None:
         exc = ValueError("unrelated")
         assert translate_error(exc) is exc
+
+
+class TestCreateSession:
+    async def test_happy_returns_sdk_dict_and_builds_body(self) -> None:
+        info = {"session_id": "s-1", "agent_id": "mimir", "created_at": "t", "last_active": "t"}
+        fake = _FakeSessions(result=info)
+        client = _wt(fake)
+        out = await create_session(client, "mimir", end_user_id="u-9")
+        assert out is info  # open-world passthrough — no re-shaping
+        name, args, kwargs = fake.calls[-1]
+        assert name == "create"
+        assert args[0] == {"agent_id": "mimir", "end_user_id": "u-9"}
+        assert kwargs["consumer_key"] is None
+
+    async def test_config_passthrough(self) -> None:
+        fake = _FakeSessions(result={"session_id": "s"})
+        await create_session(
+            _wt(fake), "echo", config={"system_prompt": "be terse"}
+        )
+        assert fake.calls[-1][1][0] == {
+            "agent_id": "echo",
+            "config": {"system_prompt": "be terse"},
+        }
+
+    async def test_bifrost_bound_body_and_consumer_key(self) -> None:
+        fake = _FakeSessions(result={"session_id": "s"})
+        binding = BifrostBinding(endpoint_url="http://h:8391", scope=None)
+        await create_session(
+            _wt(fake), "sindra", bifrost=binding, consumer_key="ck-real"
+        )
+        _name, args, kwargs = fake.calls[-1]
+        assert args[0] == {
+            "agent_id": "sindra",
+            "bifrost": {"endpoint_url": "http://h:8391", "scope": None},
+        }
+        # INV-CUT: the consumer key rides the SDK's per-request auth, NOT a header.
+        assert kwargs["consumer_key"] == "ck-real"
+
+    async def test_bifrost_without_consumer_key_rejected_pre_http(self) -> None:
+        fake = _FakeSessions(result={"session_id": "s"})
+        binding = BifrostBinding(endpoint_url="http://h:8391", scope=None)
+        with pytest.raises(BifrostConsumerKeyMissing):
+            await create_session(_wt(fake), "sindra", bifrost=binding)
+        assert fake.calls == []  # never reached the SDK
+
+    async def test_404_maps_to_agent_not_found(self) -> None:
+        fake = _FakeSessions(error=ApiError("agent_not_found", "no", status=404))
+        with pytest.raises(AgentNotFound) as ei:
+            await create_session(_wt(fake), "ghost")
+        assert ei.value.agent_id == "ghost"
+
+    async def test_bound_502_maps_to_bifrost_handshake_failed(self) -> None:
+        body = '{"detail": {"bifrost_error": "bifrost.auth_rejected"}}'
+        fake = _FakeSessions(
+            error=ApiError("bifrost_handshake_failed", "boom", status=502, body=body)
+        )
+        binding = BifrostBinding(endpoint_url="http://h:8391", scope=None)
+        with pytest.raises(BifrostHandshakeFailed) as ei:
+            await create_session(
+                _wt(fake), "sindra", bifrost=binding, consumer_key="ck"
+            )
+        assert ei.value.bifrost_error == "bifrost.auth_rejected"
+
+    async def test_unbound_502_stays_session_api_failed(self) -> None:
+        fake = _FakeSessions(error=ApiError("upstream", "boom", status=502, body="x"))
+        with pytest.raises(SessionApiFailed) as ei:
+            await create_session(_wt(fake), "mimir")
+        assert ei.value.status == 502
+
+    async def test_default_error_maps_to_session_api_failed(self) -> None:
+        fake = _FakeSessions(error=ApiError("weird", "boom", status=418, body="teapot"))
+        with pytest.raises(SessionApiFailed) as ei:
+            await create_session(_wt(fake), "mimir")
+        assert ei.value.status == 418
+        assert ei.value.error_code == "weird"
+
+
+class TestListSessions:
+    async def test_passes_params_and_returns_dict(self) -> None:
+        page: dict[str, Any] = {"items": [], "next_cursor": None}
+        fake = _FakeSessions(result=page)
+        out = await list_sessions(_wt(fake), limit=10, cursor="c1", include_archived=True)
+        assert out is page
+        kwargs = fake.calls[-1][2]
+        assert kwargs["limit"] == 10
+        assert kwargs["cursor"] == "c1"
+        assert kwargs["include_archived"] is True
+
+    async def test_422_cursor_invalid_maps_to_invalid_cursor(self) -> None:
+        fake = _FakeSessions(error=ApiError("cursor_invalid", "bad", status=422))
+        with pytest.raises(InvalidCursor) as ei:
+            await list_sessions(_wt(fake), cursor="bogus")
+        assert ei.value.raw == "bogus"
+
+    async def test_other_422_stays_session_api_failed(self) -> None:
+        fake = _FakeSessions(error=ApiError("validation_failed", "x", status=422))
+        with pytest.raises(SessionApiFailed):
+            await list_sessions(_wt(fake))
+
+
+class TestReadPassthroughs:
+    async def test_messages_returns_dict(self) -> None:
+        data = {"session_id": "s", "items": []}
+        fake = _FakeSessions(result=data)
+        assert await get_session_messages(_wt(fake), "s") is data
+        assert fake.calls[-1][0] == "messages"
+
+    async def test_tools_returns_dict(self) -> None:
+        data = {"agent_id": "mimir", "builtin_tools": []}
+        fake = _FakeSessions(result=data)
+        assert await get_session_tools(_wt(fake), "s") is data
+        assert fake.calls[-1][0] == "tools"
+
+    async def test_messages_error_maps_to_session_api_failed(self) -> None:
+        fake = _FakeSessions(error=ApiError("auth_revoked", "no", status=401))
+        with pytest.raises(SessionApiFailed) as ei:
+            await get_session_messages(_wt(fake), "s")
+        assert ei.value.status == 401
+
+
+async def _drain(aiter: Any) -> list[Any]:
+    out: list[Any] = []
+    async for ev in aiter:
+        out.append(ev)
+    return out
+
+
+class TestStreamTurn:
+    async def test_yields_events_verbatim(self) -> None:
+        e1, e2 = object(), object()
+        fake = _FakeSessions(events=[e1, e2])
+        got = await _drain(stream_turn(_wt(fake), "s-1", "hello"))
+        assert got == [e1, e2]
+        assert fake.calls[-1] == ("stream_turn", ("s-1", "hello"), {})
+
+    async def test_session_retired_maps_to_session_api_failed(self) -> None:
+        fake = _FakeSessions(
+            stream_error=wtsdk.SessionRetired("session_retired", "gone", status=410)
+        )
+        with pytest.raises(SessionApiFailed) as ei:
+            await _drain(stream_turn(_wt(fake), "s", "hi"))
+        assert ei.value.status == 410
+        assert ei.value.error_code == "session_retired"
+
+    async def test_agent_not_available_rewraps_to_ratatoskr(self) -> None:
+        fake = _FakeSessions(
+            stream_error=wtsdk.AgentNotAvailable("agent_not_available", "no agent", status=409)
+        )
+        with pytest.raises(AgentNotAvailable) as ei:
+            await _drain(stream_turn(_wt(fake), "s", "hi"))
+        assert ei.value.error_code == "agent_not_available"
+        assert ei.value.status == 409
+
+    async def test_turn_launch_unavailable_rewraps_to_ratatoskr(self) -> None:
+        fake = _FakeSessions(
+            stream_error=wtsdk.TurnLaunchUnavailable("not_ready", "busy", status=503)
+        )
+        with pytest.raises(TurnLaunchUnavailable) as ei:
+            await _drain(stream_turn(_wt(fake), "s", "hi"))
+        assert ei.value.retryable is True
+
+    async def test_generic_connect_failed_maps_to_sse_connect_failed(self) -> None:
+        fake = _FakeSessions(stream_error=wtsdk.ConnectFailed("connect_failed", "boom", status=500))
+        with pytest.raises(SseConnectFailed) as ei:
+            await _drain(stream_turn(_wt(fake), "s", "hi"))
+        assert ei.value.status == 500
+
+    async def test_connection_dropped_maps_and_carries_cursor(self) -> None:
+        fake = _FakeSessions(stream_error=wtsdk.ConnectionDropped("12:3"))
+        with pytest.raises(SseConnectionDropped) as ei:
+            await _drain(stream_turn(_wt(fake), "s", "hi"))
+        assert ei.value.last_seen_sse_id == "12:3"
+
+    async def test_resume_error_maps_to_sse_connect_failed(self) -> None:
+        fake = _FakeSessions(stream_error=wtsdk.ResumeError("resume_failed", "boom", status=412))
+        with pytest.raises(SseConnectFailed) as ei:
+            await _drain(stream_turn(_wt(fake), "s", "hi"))
+        assert ei.value.status == 412
+
+    async def test_malformed_sse_id_passes_through_as_ratatoskr(self) -> None:
+        fake = _FakeSessions(stream_error=wtsdk.MalformedSseId("bad-id"))
+        with pytest.raises(MalformedSseId) as ei:
+            await _drain(stream_turn(_wt(fake), "s", "hi"))
+        assert ei.value.raw == "bad-id"
+
+    async def test_malformed_sse_data_passes_through_as_ratatoskr(self) -> None:
+        fake = _FakeSessions(stream_error=wtsdk.MalformedSseData("not json"))
+        with pytest.raises(MalformedSseData):
+            await _drain(stream_turn(_wt(fake), "s", "hi"))
+
+    async def test_turn_id_flip_carries_established_and_got(self) -> None:
+        fake = _FakeSessions(stream_error=wtsdk.TurnIdFlip(5, 7))
+        with pytest.raises(TurnIdFlip) as ei:
+            await _drain(stream_turn(_wt(fake), "s", "hi"))
+        assert (ei.value.established, ei.value.got) == (5, 7)
+
+
+class TestCancelTurn:
+    async def test_happy_returns_cancel_result(self) -> None:
+        res = CancelResult(turn_id=42, cancelled=True, reason=None, partial_message_id=None)
+        fake = _FakeSessions(result=res)
+        out = await cancel_turn(_wt(fake), "s-1", 42, persist_partial=True)
+        assert out is res
+        assert fake.calls[-1] == ("cancel_turn", ("s-1", 42), {"persist_partial": True})
+
+    async def test_late_cancel_race_is_a_result_not_an_error(self) -> None:
+        # B-CAN-3: a 200 with cancelled=False is the benign late-cancel no-op.
+        res = CancelResult(turn_id=42, cancelled=False, reason=None, partial_message_id=None)
+        out = await cancel_turn(_wt(_FakeSessions(result=res)), "s", 42)
+        assert out.cancelled is False
+
+    async def test_turn_not_found_maps_to_ratatoskr(self) -> None:
+        fake = _FakeSessions(
+            error=wtsdk.CancelTurnNotFound(42, error_code="turn_not_found", message="gone")
+        )
+        with pytest.raises(CancelTurnNotFound) as ei:
+            await cancel_turn(_wt(fake), "s", 42)
+        assert ei.value.turn_id == 42
+
+    async def test_turn_finished_maps_to_ratatoskr(self) -> None:
+        fake = _FakeSessions(
+            error=wtsdk.CancelAlreadyCompleted(42, error_code="turn_finished", message="done")
+        )
+        with pytest.raises(CancelAlreadyCompleted):
+            await cancel_turn(_wt(fake), "s", 42)
+
+    async def test_other_cancel_failure_maps_to_cancel_failed(self) -> None:
+        fake = _FakeSessions(error=wtsdk.CancelFailed(42, error_code="boom", message="failed"))
+        with pytest.raises(CancelFailed):
+            await cancel_turn(_wt(fake), "s", 42)

uv.lock

@@ -472,7 +472,7 @@ wheels = [
 
 [[package]]
 name = "ratatoskr"
-version = "0.21.3"
+version = "0.21.8"
 source = { editable = "." }
 dependencies = [
     { name = "httpx" },