From b907a7b8a500d7ac208f7bfc80981a7bf8468437 Mon Sep 17 00:00:00 2001 From: Vuong Hoang Date: Sun, 19 Jul 2026 00:31:16 -0700 Subject: [PATCH] feat(#20): stream + cancel adapter routes complete the wt surface (slice-2, part 2a) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 | 2 +- src/ratatoskr/sse_client.py | 13 ++- src/ratatoskr/wt.py | 97 ++++++++++++++++++++-- tests/test_wt.py | 158 +++++++++++++++++++++++++++++++++++- uv.lock | 2 +- 5 files changed, 258 insertions(+), 14 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 45c8a25..614a6f8 100644 --- a/pyproject.toml +++ b/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" diff --git a/src/ratatoskr/sse_client.py b/src/ratatoskr/sse_client.py index d8e3c79..b4d44ff 100644 --- a/src/ratatoskr/sse_client.py +++ b/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}", ) diff --git a/src/ratatoskr/wt.py b/src/ratatoskr/wt.py index 8e4142f..469135e 100644 --- a/src/ratatoskr/wt.py +++ b/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 -# as their call-sites are rewired in later slice-2 commits. wt → sessions is one-way -# (sessions never imports wt), so there is no cycle. +# 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, @@ -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 diff --git a/tests/test_wt.py b/tests/test_wt.py index 2da352a..9815b0d 100644 --- a/tests/test_wt.py +++ b/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). - 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: @@ -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) diff --git a/uv.lock b/uv.lock index ecc9aa1..c243697 100644 --- a/uv.lock +++ b/uv.lock @@ -472,7 +472,7 @@ wheels = [ [[package]] name = "ratatoskr" -version = "0.21.4" +version = "0.21.5" source = { editable = "." } dependencies = [ { name = "httpx" },