From 5c595b862db862fb811b72179967e06f5ac98fd9 Mon Sep 17 00:00:00 2001 From: Vuong Hoang Date: Sun, 19 Jul 2026 06:22:13 -0700 Subject: [PATCH] feat(#20): rewire the web turn surface onto the wt adapter (slice-2, part 2b-ii) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 | 2 +- src/ratatoskr/web/server.py | 138 ++++++++++++++---------- tests/test_web_presentation_contract.py | 81 ++++++-------- tests/test_web_server.py | 13 ++- uv.lock | 2 +- 5 files changed, 126 insertions(+), 110 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 4a7cece..481cde2 100644 --- a/pyproject.toml +++ b/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" diff --git a/src/ratatoskr/web/server.py b/src/ratatoskr/web/server.py index 3c3ea4d..e96841d 100644 --- a/src/ratatoskr/web/server.py +++ b/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] diff --git a/tests/test_web_presentation_contract.py b/tests/test_web_presentation_contract.py index 3924b3e..01d95ce 100644 --- a/tests/test_web_presentation_contract.py +++ b/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, ), ) diff --git a/tests/test_web_server.py b/tests/test_web_server.py index e9a0f01..a12d66e 100644 --- a/tests/test_web_server.py +++ b/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 diff --git a/uv.lock b/uv.lock index bee1b9d..8584432 100644 --- a/uv.lock +++ b/uv.lock @@ -472,7 +472,7 @@ wheels = [ [[package]] name = "ratatoskr" -version = "0.21.6" +version = "0.21.7" source = { editable = "." } dependencies = [ { name = "httpx" },