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

The Starlette endpoints (create / stream / cancel / tools / messages) now go through
ratatoskr.wt over the worldtree-sdk; the browser contract is preserved. This is the
last consumer of the hand-rolled turn-stream family — after this, stream_turn* /
cancel_turn are orphaned and get deleted in part 2b-iii (with the live smoke).

- _wt_client wraps a client_factory transport as the adapter's WorldtreeClient
  (INV-CUT-1), reading base_url + bearer off the transport (a no-auth test transport
  falls back to a placeholder key). The hand-rolled endpoints (persona / agents /
  admin / bifrost) keep using the raw transport until their slices.
- _event_to_browser_payload derives the browser payload from the SDK's `raw` (the
  wire body) minus the redundant `type`, plus the composite `sse_id` string — the
  SAME shape the old dataclasses produced, so the presentation fixture + browser JS
  are unchanged; the browser event_type is the wire `type`, not the SDK class name.
- The stream endpoint captures the upstream cancel target from the composite sse_id
  (the SDK's top-level turn_id is body-derived, absent on text frames); create reads
  the SDK's open create dict; cancel reads CancelResult.cancelled and surfaces a
  generic 502 for CancelFailed (the SDK abstracts the upstream cancel HTTP status).
- test_web_presentation_contract builds SDK events via build_event; two cancel tests
  adopt the SDK's (status, error_code) race pairs + the 502.

Suite 570 green; web/server.py + presentation test ruff-clean, mypy unchanged
(same pre-existing errors). Patch (internal; browser contract preserved).

pyproject.toml

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

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(
+            info = await wt.create_session(
-                client,
+                _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
+            # #347 authored first-message: seed the agent's preset opening (best-effort;
-            # (best-effort; never blocks create — see first_message INV-001).
+            # never blocks create). first_message is a slice-3 hand-rolled path — it
-            await seed_preset_first_message(client, info.session_id, agent_id)
+            # 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
+    Per INV-008 + FN stream_turn_endpoint STEP 3. The browser contract
-    locked by tests/fixtures/presentation_contract.json — one entry per
+    (tests/fixtures/presentation_contract.json) is preserved: the SDK's `raw` is
-    Event type. Implementation: snake_case class name as event_type;
+    the wire body — the same per-type field set the old dataclasses carried — so the
-    asdict(event) with sse_id flattened to "T:S" string.
+    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__
+    browser_type = getattr(event, "type", "") or ""
-    # CamelCase → snake_case
+    raw = getattr(event, "raw", None) or {}
-    browser_type = "".join(
+    data = {k: v for k, v in dict(raw).items() if k != "type"}
-        ("_" + c.lower() if c.isupper() and i else c.lower())
+    data["sse_id"] = getattr(event, "sse_id", None)
-        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']}"
     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):
+                async for event in wt.stream_turn(wt_client, session_id, handle.content):
-                    # v0.16.0: capture the upstream (Worldtree-assigned)
+                    # v0.16.0: capture the upstream (Worldtree-assigned) turn_id from
-                    # turn_id from the first event so cancel paths target
+                    # the first event so cancel paths target the real upstream turn,
-                    # the real upstream turn, not our local counter.
+                    # not our local counter — parsed from the composite sse_id.
                     if handle.upstream_turn_id is None:
-                        sse_id = getattr(event, "sse_id", None)
+                        handle.upstream_turn_id = _turn_id_from_sse_id(
-                        if sse_id is not None:
+                            getattr(event, "sse_id", None)
-                            handle.upstream_turn_id = sse_id.turn_id
+                        )
                     event_type, data = _event_to_browser_payload(event)
                     yield _format_sse(event_type, data)
-                    if isinstance(event, (Done, Error, Cancelled)):
+                    if isinstance(event, (DoneEvent, ErrorEvent, CancelledEvent)):
-                        handle.status = type(event).__name__.lower()
+                        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)
+            result = await wt.cancel_turn(
-        body = {"cancelled": True}
+                _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},
+            {"error_code": "cancel_failed"},
-            status_code=exc.status,
+            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)
+            info = await wt.get_session_tools(_wt_client(client), session_id)
-    except SessionApiFailed as exc:
+    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)
+            data = await wt.get_session_messages(_wt_client(client), session_id)
-    except SessionApiFailed as exc:
+    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]

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 (
+from worldtree_sdk.events import build_event
-    AffectUpdate,
+
-    AwaitingLlmFirstToken,
-    Cancelled,
-    Done,
-    Error,
-    SseId,
-    Text,
-    TextBoundary,
-    Thinking,
-    ToolResult,
-    ToolStart,
-    WorkerPhase,
-)
 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(
+    _check("worker_phase", _ev("worker_phase", "42:3", phase="BuildingPrompt", turn_id=42))
-        "worker_phase",
-        WorkerPhase(sse_id=SseId(42, 3), phase="BuildingPrompt", turn_id=42),
-    )
 
 
 def test_thinking_matches_fixture() -> None:
-    _check(
+    _check("thinking", _ev("thinking", "42:5", content="Let me think..."))
-        "thinking",
-        Thinking(sse_id=SseId(42, 5), content="Let me think..."),
-    )
 
 
 def test_text_matches_fixture() -> None:
-    _check(
+    _check("text", _ev("text", "42:7", content="Hello there"))
-        "text",
-        Text(sse_id=SseId(42, 7), content="Hello there"),
-    )
 
 
 def test_text_boundary_matches_fixture() -> None:
     _check(
         "text_boundary",
-        TextBoundary(
+        _ev("text_boundary", "42:8", kind="sentence", char_offset=11, ts="2026-05-28T00:00:00Z"),
-            sse_id=SseId(42, 8), kind="sentence",
-            char_offset=11, ts="2026-05-28T00:00:00Z",
-        ),
     )
 
 
 def test_tool_start_matches_fixture() -> None:
-    _check(
+    _check("tool_start", _ev("tool_start", "42:9", name="search", arguments={"q": "ratatoskr"}))
-        "tool_start",
-        ToolStart(sse_id=SseId(42, 9), name="search", arguments={"q": "ratatoskr"}),
-    )
 
 
 def test_tool_result_matches_fixture() -> None:
     _check(
         "tool_result",
-        ToolResult(
+        _ev("tool_result", "42:10", name="search", result={"n": 1}, duration_ms=12),
-            sse_id=SseId(42, 10), name="search",
-            result={"n": 1}, duration_ms=12,
-        ),
     )
 
 
 def test_done_matches_fixture() -> None:
     _check(
         "done",
-        Done(
+        _ev(
-            sse_id=SseId(42, 11), phase="succeeded", response="Hello there",
+            "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(
+        _ev(
-            sse_id=SseId(42, 11), phase="failed",
+            "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(
+        _ev(
-            sse_id=SseId(42, 11), phase="cancelled", turn_id=42,
+            "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(
+        _ev(
-            sse_id=SseId(42, 1), status="current", turn_id=42,
+            "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(
+        _ev(
-            sse_id=SseId(42, 2), turn_id=42,
+            "awaiting_llm_first_token", "42:2",
-            elapsed_ms_since_building_prompt=5012.3,
+            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
 

uv.lock

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