feat(#20): stream + cancel adapter routes complete the wt surface (slice-2, part 2a)

Completes the adapter's session/turn surface, still additive and non-breaking (no
surface rewired, no hand-rolled path deleted — the cli/web rewire + deletions +
live smoke are part 2b).

- stream_turn: drives the SDK's resilient stream (auto-resume absorbs the old
  reconnect_turn) and yields SDK TurnEvents, re-wrapping the stream's TERMINAL SDK
  errors into ratatoskr's caller-semantic exceptions per DEC-2 (SessionRetired →
  SessionApiFailed; AgentNotAvailable / TurnLaunchUnavailable / MalformedSse* /
  TurnIdFlip → ratatoskr's same-named types; ConnectionDropped → SseConnectionDropped;
  ConnectFailed / terminal ResumeError → SseConnectFailed). The presenter keeps
  catching ratatoskr types (part 2b aligns the except clauses).
- cancel_turn: returns the SDK CancelResult (a 200 cancelled=False is the benign
  late-cancel race, B-CAN-3), mapping the typed cancel races onto ratatoskr's
  CancelTurnNotFound / CancelAlreadyCompleted / CancelFailed.
- SseConnectionDropped.last_seen_sse_id widened to SseId | str | None: the SDK's
  resume cursor is a raw composite-id str (the cutover's target form); the
  hand-rolled path's SseId stays accepted until it is deleted. The one live reader
  (stream_turn_resilient) generalizes cleanly — a str cursor is already the id.

Suite 570 green (555 + 15); wt.py + sse_client.py mypy + ruff clean. Patch.

pyproject.toml

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

src/ratatoskr/sse_client.py

@@ -286,9 +286,15 @@ def _eager_failure_fields(body: bytes, status: int) -> tuple[str, str]:
 
 
 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
 
@@ -611,7 +617,8 @@ async def stream_turn_resilient(
                 client,
                 session_id,
                 content,
-                last_event_id=f"{seen.turn_id}:{seen.seq}",
+                # A str cursor is already the composite id; an SseId is formatted.
+                last_event_id=seen if isinstance(seen, str) else f"{seen.turn_id}:{seen.seq}",
             )
 
 

src/ratatoskr/wt.py

@@ -28,16 +28,17 @@ carries the SDK's parsed `error_code`).
 from __future__ import annotations
 
 import json
-from collections.abc import Mapping
+from collections.abc import AsyncIterator, 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` module; they relocate into this adapter
+# type still live in the retiring `sessions` / `sse_client` modules; they relocate
-# as their call-sites are rewired in later slice-2 commits. wt → sessions is one-way
+# into this adapter as their call-sites are rewired in later slice-2 commits. wt →
-# (sessions never imports wt), so there is no cycle.
+# sessions / sse_client is one-way (neither imports wt), so there is no cycle.
 from .sessions import (
     AgentNotFound,
     BifrostBinding,
@@ -45,6 +46,18 @@ from .sessions import (
     BifrostHandshakeFailed,
     InvalidCursor,
 )
+from .sse_client import (
+    AgentNotAvailable,
+    CancelAlreadyCompleted,
+    CancelFailed,
+    CancelTurnNotFound,
+    MalformedSseData,
+    MalformedSseId,
+    SseConnectFailed,
+    SseConnectionDropped,
+    TurnIdFlip,
+    TurnLaunchUnavailable,
+)
 
 
 class SessionApiFailed(Exception):
@@ -232,3 +245,77 @@ async def get_session_tools(
         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
+) -> AsyncIterator[wtsdk.TurnEvent]:
+    """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_wt.py

@@ -16,7 +16,8 @@ from typing import Any, cast
 
 import httpx
 import pytest
-from worldtree_sdk import AgentNotAvailable, ApiError, WorldtreeClient
+import worldtree_sdk as wtsdk
+from worldtree_sdk import ApiError, CancelResult, WorldtreeClient
 
 from ratatoskr.sessions import (
     AgentNotFound,
@@ -25,13 +26,27 @@ from ratatoskr.sessions import (
     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,
 )
 
@@ -41,9 +56,18 @@ class _FakeSessions:
     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) -> None:
+    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:
@@ -64,6 +88,19 @@ class _FakeSessions:
     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:
@@ -135,8 +172,9 @@ 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).
+        # type at the REST layer — translate_error passes them through by identity
-        exc = AgentNotAvailable("agent_not_available", "gone", status=409)
+        # (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:
@@ -260,3 +298,115 @@ class TestReadPassthroughs:
         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.4"
+version = "0.21.5"
 source = { editable = "." }
 dependencies = [
     { name = "httpx" },