From 369857d3f109347d822d6548731730b76b4bfdbe Mon Sep 17 00:00:00 2001 From: Vuong Hoang Date: Wed, 27 May 2026 20:53:13 -0700 Subject: [PATCH] =?UTF-8?q?feat(web):=20address=20Heid=20code-review=20fin?= =?UTF-8?q?dings=20=E2=80=94=20issue=20#16=20(v0.16.0)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Heid panel review (Gróa + Hulda, thread 01KSP5P6CSJH) on v0.15.0/ v0.15.1 surfaced one load-bearing bug + several precision items. This pass closes them. Load-bearing fix — cancel paths targeted the wrong turn_id: - `_TURN_COUNTER` allocates browser-local ids (1, 2, 3…); the real upstream Worldtree turn_id (e.g. 799) only arrives in the first SSE event. The v0.15.x cancel/disconnect/shutdown paths posted to /sessions/{sid}/turns/{LOCAL_ID}/cancel — wrong URL upstream. - TurnHandle.upstream_response (dead field) → upstream_turn_id: int|None. Captured from the first event's sse_id.turn_id in the stream generator. All cancel paths now target it. Cancel before the upstream stream starts (upstream_turn_id None) is a no-op ({"cancelled": false, "reason": "not_started"}). - The old cancel tests mocked the local-id URL, so they encoded the bug; rewritten to assert the UPSTREAM id is targeted. Behavior change (minor-bump driver) — server-side end_user_id: - create_app gains end_user_id kwarg; entrypoint reads RATATOSKR_END_USER_ID and threads it in. POST /api/sessions uses app.state.end_user_id, IGNORING any browser-supplied value (a client can't impersonate an arbitrary end-user partition). JS no longer sends end_user_id. Precision fixes: - Entrypoint missing-extras ImportError catch scoped to starlette/ uvicorn ONLY; baseline-dep / first-party import failures now propagate as real tracebacks instead of masking as exit-12. - Lifespan shutdown logs per-pending session_id + upstream_turn_id (was a single aggregate count). Tests (+18; 376 total): - disconnect_triggers_upstream_cancel (INV-005 load-bearing — drives the stream generator directly + cancels the consuming task; would have caught the turn_id bug) - cancel_targets_upstream_turn_id, cancel_before_started_is_noop, cancel_failed_500 - server-side end_user_id: uses / ignores-body / omits-when-unset - create_app: routes_registered / state_attached / factory_stored - entrypoint: default_host / port_zero / happy_argv / open / no-open - real_import_bug_propagates (precision guard) - full_event_vocab at the stream-endpoint layer Contract #16 amended: v0.16.0 amendment banner + INV-005/006 reworded for upstream_turn_id + FN sketches corrected (server-side end_user_id, upstream_response→upstream_turn_id, manual client lifecycle vs the non-executable async-with sketch, not-started cancel branch). --- docs/contracts/issues/16.contract.md | 74 ++++-- pyproject.toml | 2 +- src/ratatoskr/web/entrypoint.py | 20 +- src/ratatoskr/web/server.py | 90 ++++++-- src/ratatoskr/web/static/index.html | 4 +- tests/test_web_packaging.py | 86 ++++++- tests/test_web_server.py | 329 ++++++++++++++++++++++++++- uv.lock | 2 +- 8 files changed, 538 insertions(+), 69 deletions(-) diff --git a/docs/contracts/issues/16.contract.md b/docs/contracts/issues/16.contract.md index ccca2dd..103dd19 100644 --- a/docs/contracts/issues/16.contract.md +++ b/docs/contracts/issues/16.contract.md @@ -108,14 +108,39 @@ def main(argv: list[str] | None = None) -> int: ... # entrypoint.main `entrypoint.main` is the console-script target — parses flags, builds the client factory from env, calls `create_app`, runs uvicorn. The lazy-import discipline lives here: `import starlette` does NOT happen at module top — it lands inside `main()` after arg parsing, with an `ImportError` catch that prints the `pip install ratatoskr[web]` hint and exits non-zero. +## v0.16.0 amendment (post-Heid-code-review) + +Heid panel review (Gróa + Hulda, thread `01KSP5P6CSJH`) on the +v0.15.0/v0.15.1 implementation surfaced three contract-text issues +now corrected below: + +1. **Upstream vs local turn_id.** Cancel paths (explicit cancel, + browser-disconnect, lifespan shutdown) MUST target the *upstream* + (Worldtree-assigned) turn_id captured from the first SSE event's + `sse_id.turn_id`, NOT the browser-local `_TURN_COUNTER` value (which + is only a registry key). The `TurnHandle.upstream_response` field is + replaced by `upstream_turn_id: int | None`. Cancel before the + upstream stream starts (upstream_turn_id is None) is a no-op + (`{"cancelled": false, "reason": "not_started"}`). +2. **`RATATOSKR_END_USER_ID` is server-configured.** `FN main` reads it + from env and threads it into `create_app(..., end_user_id=...)`; the + `POST /api/sessions` endpoint uses `app.state.end_user_id` server- + side. The browser NEVER supplies end_user_id — a client cannot + impersonate an arbitrary end-user partition. +3. **Stream client lifecycle.** The `async with client_factory() as + client:` sketch in `FN stream_turn_endpoint` is not executable for a + long-lived async generator that must outlive the handler frame; the + implementation uses manual `client = ...; try: ... finally: await + client.aclose()`. Sketch corrected below. + ## Invariants -- **INV-001**: `ratatoskr.web.__init__` and `ratatoskr.web.entrypoint` MUST NOT import `starlette` or `uvicorn` at module top. Import is inside `main()` after flag parsing. +- **INV-001**: `ratatoskr.web.__init__` and `ratatoskr.web.entrypoint` MUST NOT import `starlette` or `uvicorn` at module top. Import is inside `main()` after flag parsing. The missing-extras `ImportError` catch is scoped to the OPTIONAL extras (`starlette` / `uvicorn`) ONLY — baseline-dep / first-party import failures propagate as real tracebacks rather than masking as exit-12. - **INV-002**: `ratatoskr.web.server.create_app` MUST accept a `client_factory` callable. The app MUST NOT construct `httpx.AsyncClient` at module top or in route handlers; it MUST call the factory. - **INV-003**: Upstream API key MUST never appear in any browser-visible response. Server proxies upstream calls using the client factory; only the upstream's JSON / SSE payload is forwarded. No header echo. - **INV-004**: Transcript content from upstream `text` SSE events MUST be HTML-escaped before reaching the browser DOM (escape on the wire in the SSE proxy serialization OR escape in the JS rendering — both are acceptable; pick one and stick to it). -- **INV-005**: Browser disconnect mid-stream (`asyncio.CancelledError` in the SSE handler) MUST trigger an upstream cancel on the matching `(session_id, turn_id)` via `sse_client.cancel_turn`. If the turn already completed, the cancel is a best-effort no-op (existing `CancelAlreadyCompleted` exception is swallowed). -- **INV-006**: Server shutdown (Ctrl-C / SIGTERM) MUST issue upstream cancels for every entry in the turn registry within a 5-second cleanup budget. Entries that don't ack in time are abandoned with a structured log line. +- **INV-005**: Browser disconnect mid-stream (`asyncio.CancelledError` in the SSE handler) MUST trigger an upstream cancel via `sse_client.cancel_turn` on the captured `upstream_turn_id` (v0.16.0 — NOT the browser-local turn_id). If the turn already completed, the cancel is a best-effort no-op (`CancelAlreadyCompleted` swallowed). If `upstream_turn_id` is still None (upstream stream never started), the disconnect cancel is skipped — nothing to cancel. +- **INV-006**: Server shutdown (Ctrl-C / SIGTERM) MUST issue upstream cancels (on `upstream_turn_id`) for every in-flight registry entry within a 5-second cleanup budget. Handles whose `upstream_turn_id` is None are skipped. Entries that don't ack in time are abandoned with a per-entry structured log line carrying `session_id` + `upstream_turn_id`. - **INV-007**: The turn registry MUST be in-process memory only — no persistence, no shared state across server restarts. Process exit drops the registry. - **INV-008**: Each SSE event serialized to the browser MUST follow the contract enumerated in `tests/fixtures/presentation_contract.json` — one entry per Event type, with the exact JSON shape the browser presenter renders against. - **INV-009**: All wire-layer modules (`sse_client`, `sessions`, `tier3`, `local_agents`) MUST be used unchanged. Any required change to those modules is out of scope for this issue and gets its own ticket. @@ -234,9 +259,10 @@ BRIEF: Proxy POST /sessions to upstream. PRE: [PRE-001 hard] request body has "agent_id" key -- 400 if missing POST: [POST-001 return_value] 201 with SessionInfo on upstream success STEPS: - 1. [parse] body = await request.json(); agent_id = body["agent_id"]; end_user_id = body.get("end_user_id") - 2. [proxy] async with client_factory() as client: info = await create_session(client, agent_id, end_user_id=end_user_id) - 3. [return] JSONResponse(as_dict(info), status_code=201) + 1. [parse] body = await request.json(); agent_id = body["agent_id"] (400 if missing) + 2. [server-side] end_user_id = request.app.state.end_user_id # v0.16.0: server-configured, NOT from body + 3. [proxy] async with client_factory() as client: info = await create_session(client, agent_id, end_user_id=end_user_id) + 4. [return] JSONResponse(as_dict(info), status_code=201) ERRORS: AgentNotFound -> JSONResponse({"error_code": "agent_not_found"}, 404) SessionApiFailed -> JSONResponse({"error_code": "session_api_failed", "status": exc.status}, exc.status) @@ -244,6 +270,8 @@ TESTS: happy [tracer]: respx mock 201 → endpoint returns 201 with session JSON unknown_agent [error]: respx mock 404 → 404 with agent_not_found envelope missing_agent_id [adversarial]: body without agent_id → 400 + server_side_end_user_id [v0.16.0]: create_app(end_user_id="X") → upstream body carries end_user_id="X" + ignores_body_end_user_id [v0.16.0,security]: body end_user_id is overridden by server value ``` ```contract @@ -274,7 +302,7 @@ POST: [POST-002 state_change] app.state.turn_registry has entry for (sid, turn_i STEPS: 1. [parse] session_id = path_params["session_id"]; body = await request.json(); content = body["content"] 2. [allocate] turn_id = next_turn_id() # process-local monotonic counter - 3. [register] turn_registry[(session_id, turn_id)] = TurnHandle(content=content, status="queued", upstream_response=None) + 3. [register] turn_registry[(session_id, turn_id)] = TurnHandle(content=content, status="queued", upstream_turn_id=None) # v0.16.0: was upstream_response 4. [return] JSONResponse({"turn_id": turn_id}, status_code=200) TESTS: happy [tracer]: POST {"content": "hi"} → 200 with turn_id; registry populated @@ -290,17 +318,16 @@ POST: [POST-001 side_effect] each upstream event serialized to browser as SSE ev POST: [POST-002 state_change] on completion/disconnect, registry entry removed; upstream cancel if turn still in flight STEPS: 1. [validate] sid, tid = path/query params; handle = registry.get((sid, tid)); 404 if None - 2. [open] async with client_factory() as client: - - upstream = stream_turn(client, sid, handle.content) - - handle.upstream_response = upstream + 2. [open] client = client_factory() # v0.16.0: manual lifecycle, NOT `async with` — the generator outlives this frame; closed in finally - handle.status = "streaming" - 3. [forward] async for event in upstream: + 3. [forward] async for event in stream_turn(client, sid, handle.content): + - IF handle.upstream_turn_id is None: handle.upstream_turn_id = event.sse_id.turn_id # v0.16.0: capture upstream turn id - serialize per fixture: {"type": , "data": } - yield as `event: \\ndata: \\n\\n` bytes 4. [terminal] on Done/Error/Cancelled: yield final SSE, mark handle.status, break 5. [cleanup] finally: - - IF asyncio.CancelledError caught and turn in-flight: upstream cancel via cancel_turn(client, sid, tid) - - remove (sid, tid) from registry + - IF asyncio.CancelledError caught AND status=="streaming" AND upstream_turn_id is not None: cancel_turn(client, sid, handle.upstream_turn_id) # v0.16.0: upstream id, not tid + - remove (sid, tid) from registry; await client.aclose() ERRORS: KeyError -> 404 turn_not_found asyncio.CancelledError -> upstream cancel, propagate @@ -322,15 +349,18 @@ POST: [POST-001 side_effect] upstream cancel call lands; registry entry removed POST: [POST-002 return_value] 200 with {"cancelled": true} or 200 with status reflecting upstream race STEPS: 1. [validate] sid, tid = params; handle = registry.get((sid, tid)); 404 if None - 2. [cancel] async with client_factory() as client: - - try: await cancel_turn(client, sid, tid) + 2. [not-started] IF handle.upstream_turn_id is None: del registry[(sid,tid)]; return 200 {"cancelled": false, "reason": "not_started"} # v0.16.0: upstream never opened + 3. [cancel] async with client_factory() as client: + - try: await cancel_turn(client, sid, handle.upstream_turn_id) # v0.16.0: upstream id, not tid - return 200 {"cancelled": true} - 3. [race] EXCEPT CancelAlreadyCompleted / CancelTurnNotFound: + 4. [race] EXCEPT CancelAlreadyCompleted / CancelTurnNotFound: - return 200 {"cancelled": false, "reason": "race_or_completed"} - 4. [cleanup] del registry[(sid, tid)] + 5. [cleanup] del registry[(sid, tid)] TESTS: - happy [tracer]: registered turn → POST cancel → 200, upstream cancel called + happy [tracer]: registered turn (upstream_turn_id set) → POST cancel → 200, upstream cancel at the upstream id unknown_turn [error]: not in registry → 404 + cancel_before_started [v0.16.0]: upstream_turn_id None → 200 {cancelled:false, reason:not_started}, no upstream call + cancel_targets_upstream_turn_id [v0.16.0]: local tid != upstream id → cancel URL uses upstream id already_completed [race]: respx cancel returns 409 → 200 with reason=race_or_completed cancel_failed [error]: respx returns 500 → 500 with cancel_failed envelope ``` @@ -352,11 +382,11 @@ BRIEF: On Ctrl-C / SIGTERM, drain the turn registry within 5s budget per INV-006 POST: [POST-001 side_effect] every in-flight upstream turn gets a cancel attempt within budget POST: [POST-002 side_effect] entries that don't ack in budget logged + abandoned STEPS: - 1. [collect] handles = list(app.state.turn_registry.values()) + 1. [collect] in_flight = [h for h in registry.values() if h.status == "streaming" and h.upstream_turn_id is not None] # v0.16.0: skip not-yet-started 2. [cancel] async with client_factory() as client: - - tasks = [cancel_turn(client, h.session_id, h.turn_id) for h in handles if h.status == "streaming"] - - done, pending = await asyncio.wait(tasks, timeout=5.0) - 3. [log] for each pending: log {"kind": "shutdown", "event": "cleanup_timeout", session/turn} + - task_to_handle = {create_task(cancel_turn(client, h.session_id, h.upstream_turn_id)): h for h in in_flight} # v0.16.0: upstream id + - done, pending = await asyncio.wait(task_to_handle, timeout=5.0) + 3. [log] for each pending: cancel task + log {"kind": "shutdown", "event": "cleanup_timeout", "session_id": h.session_id, "upstream_turn_id": h.upstream_turn_id} 4. [clear] registry.clear() TESTS: happy [tracer]: 2 in-flight turns + shutdown → both upstream cancels called, registry empty diff --git a/pyproject.toml b/pyproject.toml index 9239579..d456983 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "ratatoskr" -version = "0.15.1" +version = "0.16.0" description = "Worldtree Conversation API debug TUI — multi-pane observability dashboard" readme = "README.md" requires-python = ">=3.12" diff --git a/src/ratatoskr/web/entrypoint.py b/src/ratatoskr/web/entrypoint.py index 01b04e4..9868014 100644 --- a/src/ratatoskr/web/entrypoint.py +++ b/src/ratatoskr/web/entrypoint.py @@ -58,15 +58,17 @@ def main(argv: list[str] | None = None) -> int: ) return 11 server_url = os.environ.get("WORLDTREE_API_URL", "http://localhost:8000") + end_user_id = os.environ.get("RATATOSKR_END_USER_ID") # INV-001: lazy import. Users without [web] extras get a clean hint - # instead of a raw ImportError. + # instead of a raw ImportError. Scoped narrowly to the OPTIONAL + # extras (starlette / uvicorn) so a real import bug inside a + # production module (ratatoskr.web.server, ratatoskr.cli, httpx — + # all baseline deps) propagates as a true traceback rather than + # being masked as "install ratatoskr[web]". try: + import starlette # noqa: F401 (extras-presence probe) import uvicorn - - import httpx - from ratatoskr.cli import USER_AGENT - from ratatoskr.web.server import create_app except ImportError as exc: sys.stderr.write( f"[missing_extras] {exc}\n" @@ -75,6 +77,12 @@ def main(argv: list[str] | None = None) -> int: ) return 12 + # Baseline deps + own modules — a failure here is a real bug, not a + # missing-extras condition; let it propagate. + import httpx + from ratatoskr.cli import USER_AGENT + from ratatoskr.web.server import create_app + def client_factory() -> "httpx.AsyncClient": return httpx.AsyncClient( base_url=server_url, @@ -85,7 +93,7 @@ def main(argv: list[str] | None = None) -> int: timeout=httpx.Timeout(connect=10.0, read=None, write=10.0, pool=10.0), ) - app = create_app(client_factory) + app = create_app(client_factory, end_user_id=end_user_id) # Boot banner to stderr (so stdout stays clean for piping). version = _pkg_version("ratatoskr") diff --git a/src/ratatoskr/web/server.py b/src/ratatoskr/web/server.py index 37f2f0d..8a96a63 100644 --- a/src/ratatoskr/web/server.py +++ b/src/ratatoskr/web/server.py @@ -12,9 +12,8 @@ import asyncio import itertools import json from collections.abc import AsyncIterator, Callable -from dataclasses import asdict, dataclass, field, is_dataclass +from dataclasses import asdict, dataclass, is_dataclass from importlib.metadata import version as _pkg_version -from typing import Any import httpx from starlette.applications import Starlette @@ -114,12 +113,17 @@ async def _agents_endpoint(request: Request) -> JSONResponse: async def _create_session_endpoint(request: Request) -> JSONResponse: - """POST /api/sessions → upstream POST /sessions. Per FN create_session_endpoint.""" + """POST /api/sessions → upstream POST /sessions. Per FN create_session_endpoint. + + v0.16.0: end_user_id is SERVER-configured (app.state.end_user_id from + RATATOSKR_END_USER_ID), never read from the browser body. A client + cannot impersonate an arbitrary end-user partition. + """ body = await request.json() agent_id = body.get("agent_id") if isinstance(body, dict) else None if not agent_id: return JSONResponse({"error_code": "missing_agent_id"}, status_code=400) - end_user_id = body.get("end_user_id") if isinstance(body, dict) else None + end_user_id = request.app.state.end_user_id client_factory = request.app.state.client_factory try: async with client_factory() as client: @@ -139,13 +143,20 @@ class TurnHandle: """In-flight turn record stored in app.state.turn_registry. Per FN submit_turn_endpoint + INV-005/006/007. + + v0.16.0: `upstream_turn_id` captures Worldtree's server-assigned + turn_id (from the first SSE event's sse_id.turn_id) once the stream + opens. Cancel paths target THIS, not the browser-local `turn_id` — + the local counter is only a registry key. None until the first + upstream event arrives; cancel before then is a no-op (nothing to + cancel upstream yet). """ session_id: str turn_id: int content: str status: str = "queued" # queued | streaming | done | error | cancelled - upstream_response: Any = field(default=None, repr=False) + upstream_turn_id: int | None = None # Process-local monotonic turn_id counter. Per FN submit_turn_endpoint @@ -229,6 +240,13 @@ async def _stream_turn_endpoint(request: Request) -> StreamingResponse: handle.status = "streaming" try: async for event in stream_turn(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. + 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 event_type, data = _event_to_browser_payload(event) yield _format_sse(event_type, data) if isinstance(event, (Done, Error, Cancelled)): @@ -242,10 +260,11 @@ async def _stream_turn_endpoint(request: Request) -> StreamingResponse: ) handle.status = "error" except asyncio.CancelledError: - # Browser disconnect path (INV-005). - if handle.status == "streaming": + # Browser disconnect path (INV-005). Cancel the UPSTREAM + # 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, turn_id) + await cancel_turn(client, session_id, handle.upstream_turn_id) except Exception: pass raise @@ -264,12 +283,22 @@ async def _cancel_turn_endpoint(request: Request) -> JSONResponse: except (KeyError, ValueError): return JSONResponse({"error_code": "missing_turn_id"}, status_code=400) registry = request.app.state.turn_registry - if (session_id, turn_id) not in registry: + handle = registry.get((session_id, turn_id)) + if handle is None: return JSONResponse({"error_code": "turn_not_found"}, status_code=404) + # v0.16.0: cancel targets the UPSTREAM turn_id captured during + # streaming, not the browser-local turn_id. If the upstream stream + # never started (upstream_turn_id is None), there's nothing to + # cancel — clean up and report a no-op. + if handle.upstream_turn_id is None: + registry.pop((session_id, turn_id), None) + return JSONResponse( + {"cancelled": False, "reason": "not_started"}, status_code=200 + ) client_factory = request.app.state.client_factory try: async with client_factory() as client: - await cancel_turn(client, session_id, turn_id) + await cancel_turn(client, session_id, handle.upstream_turn_id) body = {"cancelled": True} except (CancelAlreadyCompleted, CancelTurnNotFound): body = {"cancelled": False, "reason": "race_or_completed"} @@ -299,13 +328,21 @@ async def _persona_state_endpoint(request: Request) -> JSONResponse: return JSONResponse(snap, status_code=200) -def create_app(client_factory: Callable[[], httpx.AsyncClient]) -> Starlette: +def create_app( + client_factory: Callable[[], httpx.AsyncClient], + *, + end_user_id: str | None = None, +) -> Starlette: """Construct the Starlette app — wire routes + state per FN create_app. INV-002: app MUST NOT construct httpx.AsyncClient at module top; everything HTTP-bound goes through client_factory. INV-006: lifespan shutdown drains the turn registry within a 5s budget — every in-flight turn gets a best-effort upstream cancel. + + v0.16.0: `end_user_id` is the server-configured Worldtree end-user + partition (from RATATOSKR_END_USER_ID). Threaded into POST /sessions + server-side; never accepted from the browser. """ assert callable(client_factory) @@ -314,26 +351,34 @@ def create_app(client_factory: Callable[[], httpx.AsyncClient]) -> Starlette: @asynccontextmanager async def lifespan(app: Starlette): yield - # Shutdown path — drain in-flight turns per INV-006. + # Shutdown path — drain in-flight turns per INV-006. Cancel the + # UPSTREAM turn_id (v0.16.0); skip handles whose upstream stream + # never started (upstream_turn_id is None — nothing to cancel). + import sys as _sys + registry: dict[tuple[str, int], TurnHandle] = app.state.turn_registry - in_flight = [h for h in registry.values() if h.status == "streaming"] + in_flight = [ + h for h in registry.values() + if h.status == "streaming" and h.upstream_turn_id is not None + ] if in_flight: client = client_factory() try: - tasks = [ + task_to_handle = { asyncio.create_task( - cancel_turn(client, h.session_id, h.turn_id) - ) + cancel_turn(client, h.session_id, h.upstream_turn_id) + ): h for h in in_flight - ] - done, pending = await asyncio.wait(tasks, 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] task.cancel() - # Log timeouts (stderr; observability per scope v2) - if pending: - import sys as _sys _sys.stderr.write( - f'{{"kind":"shutdown","event":"cleanup_timeout","count":{len(pending)}}}\n' + f'{{"kind":"shutdown","event":"cleanup_timeout",' + f'"session_id":"{h.session_id}",' + f'"upstream_turn_id":{h.upstream_turn_id}}}\n' ) finally: await client.aclose() @@ -352,6 +397,7 @@ def create_app(client_factory: Callable[[], httpx.AsyncClient]) -> Starlette: ] app = Starlette(routes=routes, lifespan=lifespan) app.state.client_factory = client_factory + app.state.end_user_id = end_user_id # INV-002: turn registry is in-process memory, keyed (session_id, turn_id) app.state.turn_registry = {} return app diff --git a/src/ratatoskr/web/static/index.html b/src/ratatoskr/web/static/index.html index 6bade45..455eb35 100644 --- a/src/ratatoskr/web/static/index.html +++ b/src/ratatoskr/web/static/index.html @@ -225,10 +225,12 @@ async function startSession() { const agentId = $("agent-picker").value; if (!agentId) return; state.agentId = agentId; + // end_user_id is server-configured (RATATOSKR_END_USER_ID) — not sent + // from the browser. The server ignores any end_user_id in this body. const r = await fetch("/api/sessions", { method: "POST", headers: {"Content-Type": "application/json"}, - body: JSON.stringify({agent_id: agentId, end_user_id: "ratatoskr-web"}), + body: JSON.stringify({agent_id: agentId}), }); if (r.status !== 201) { alert("create session failed: " + r.status); diff --git a/tests/test_web_packaging.py b/tests/test_web_packaging.py index fa1da8d..8589d95 100644 --- a/tests/test_web_packaging.py +++ b/tests/test_web_packaging.py @@ -62,17 +62,93 @@ def test_entrypoint_missing_api_key_returns_11(monkeypatch) -> None: def test_entrypoint_missing_extras_returns_12(monkeypatch) -> None: """missing_extras [error]: starlette unimportable → exit 12. - Simulated by shadowing the import within main()'s try-block via - sys.modules tampering — pre-set the offending module to None so the - import raises ImportError. + v0.16.0: the missing-extras probe is scoped to the OPTIONAL extras + (starlette / uvicorn) only. Simulated by shadowing `starlette` to + None in sys.modules so its import raises ImportError inside the + narrow try-block. """ import sys + monkeypatch.setenv("WORLDTREE_API_KEY", "k") + monkeypatch.setenv("WORLDTREE_API_URL", "https://example.com") + monkeypatch.setitem(sys.modules, "starlette", None) + + from ratatoskr.web.entrypoint import main + + rc = main(["--port", "0"]) + assert rc == 12 + + +def test_entrypoint_real_import_bug_propagates(monkeypatch) -> None: + """v0.16.0: an ImportError from a BASELINE module (not an extra) MUST + propagate as a real traceback, not be masked as exit-12 missing-extras. + Shadowing ratatoskr.web.server (a first-party module, present with or + without the [web] extras) should raise, not return 12. + """ + import sys + + import pytest as _pytest + monkeypatch.setenv("WORLDTREE_API_KEY", "k") monkeypatch.setenv("WORLDTREE_API_URL", "https://example.com") monkeypatch.setitem(sys.modules, "ratatoskr.web.server", None) from ratatoskr.web.entrypoint import main - rc = main(["--port", "0"]) - assert rc == 12 + with _pytest.raises(ImportError): + main(["--port", "0"]) + + +class TestEntrypointArgs: + """FN main argparse + serve-loop traces (contract TESTS).""" + + def test_default_host_is_zero(self) -> None: + """default_host_is_zero [trace]: argv=[] → host == '0.0.0.0'.""" + from ratatoskr.web.entrypoint import _build_arg_parser + args = _build_arg_parser().parse_args([]) + assert args.host == "0.0.0.0" + assert args.port == 8765 + assert args.open is False + + def test_port_zero_supported(self) -> None: + """port_zero_supported [trace]: argv=['--port','0'] → port == 0.""" + from ratatoskr.web.entrypoint import _build_arg_parser + args = _build_arg_parser().parse_args(["--port", "0"]) + assert args.port == 0 + + def test_happy_argv_serves_and_returns_zero(self, monkeypatch) -> None: + """happy_argv [tracer]: env set + uvicorn.run mocked → main returns 0.""" + import uvicorn + monkeypatch.setenv("WORLDTREE_API_KEY", "k") + monkeypatch.setenv("WORLDTREE_API_URL", "https://example.com") + calls = {} + monkeypatch.setattr(uvicorn, "run", lambda app, **kw: calls.update(kw)) + from ratatoskr.web.entrypoint import main + rc = main(["--port", "0"]) + assert rc == 0 + assert calls["host"] == "0.0.0.0" + assert calls["port"] == 0 + + def test_open_flag_calls_webbrowser(self, monkeypatch) -> None: + """open_flag_calls_webbrowser [trace]: --open → webbrowser.open called.""" + import uvicorn + import webbrowser + monkeypatch.setenv("WORLDTREE_API_KEY", "k") + monkeypatch.setattr(uvicorn, "run", lambda app, **kw: None) + opened = [] + monkeypatch.setattr(webbrowser, "open", lambda url: opened.append(url)) + from ratatoskr.web.entrypoint import main + main(["--port", "0", "--open"]) + assert len(opened) == 1 + + def test_no_open_default(self, monkeypatch) -> None: + """no_open_default [trace]: argv without --open → webbrowser.open NOT called.""" + import uvicorn + import webbrowser + monkeypatch.setenv("WORLDTREE_API_KEY", "k") + monkeypatch.setattr(uvicorn, "run", lambda app, **kw: None) + opened = [] + monkeypatch.setattr(webbrowser, "open", lambda url: opened.append(url)) + from ratatoskr.web.entrypoint import main + main(["--port", "0"]) + assert opened == [] diff --git a/tests/test_web_server.py b/tests/test_web_server.py index d1ab801..3f29483 100644 --- a/tests/test_web_server.py +++ b/tests/test_web_server.py @@ -288,6 +288,14 @@ def _sse_resp(stream: bytes) -> httpx.Response: return httpx.Response(200, headers={"content-type": "text/event-stream"}, content=stream) +def _sse_resp_stream(body: httpx.AsyncByteStream) -> httpx.Response: + """SSE response backed by a live AsyncByteStream (for gated/hanging + streams in disconnect tests).""" + return httpx.Response( + 200, headers={"content-type": "text/event-stream"}, stream=body + ) + + def _parse_browser_sse(raw: bytes) -> list[dict]: """Parse a server-to-browser SSE stream into [{"event": str, "data": dict}, ...].""" import json as _j @@ -365,17 +373,18 @@ class TestCancelTurnEndpoint: @respx.mock def test_happy_cancels(self) -> None: - """happy [tracer]: registered turn → POST cancel → 200, upstream cancel called.""" - route = respx.post("https://w.example/sessions/s-1/turns/").mock( - return_value=httpx.Response(200, json=_CANCEL_OK) - ) + """happy [tracer]: registered turn (upstream started) → POST cancel + → 200, upstream cancel called against the upstream turn_id. + """ 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"] - # Re-mock at the precise URL with the resolved turn_id - respx.post(f"https://w.example/sessions/s-1/turns/{turn_id}/cancel").mock( - return_value=httpx.Response(200, json={**_CANCEL_OK, "turn_id": turn_id}) + # Simulate that the upstream stream has started (turn_id 42 upstream). + app.state.turn_registry[("s-1", turn_id)].status = "streaming" + app.state.turn_registry[("s-1", turn_id)].upstream_turn_id = 42 + respx.post("https://w.example/sessions/s-1/turns/42/cancel").mock( + return_value=httpx.Response(200, json={**_CANCEL_OK, "turn_id": 42}) ) resp = c.post(f"/api/turns/s-1/cancel?turn_id={turn_id}") assert resp.status_code == 200 @@ -391,12 +400,14 @@ class TestCancelTurnEndpoint: @respx.mock def test_already_completed_race(self) -> None: - """already_completed [race]: respx 409 → 200 with reason=race_or_completed.""" + """already_completed [race]: upstream 409 → 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"] - respx.post(f"https://w.example/sessions/s-1/turns/{turn_id}/cancel").mock( + app.state.turn_registry[("s-1", turn_id)].status = "streaming" + app.state.turn_registry[("s-1", turn_id)].upstream_turn_id = 42 + respx.post("https://w.example/sessions/s-1/turns/42/cancel").mock( return_value=httpx.Response(409) ) resp = c.post(f"/api/turns/s-1/cancel?turn_id={turn_id}") @@ -404,6 +415,23 @@ class TestCancelTurnEndpoint: assert resp.json()["cancelled"] is False assert resp.json()["reason"] == "race_or_completed" + @respx.mock + def test_cancel_failed_500(self) -> None: + """cancel_failed [error]: upstream 500 → 500 with cancel_failed envelope.""" + 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 + respx.post("https://w.example/sessions/s-1/turns/42/cancel").mock( + 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.json()["error_code"] == "cancel_failed" + assert ("s-1", turn_id) not in app.state.turn_registry + class TestStaticServing: """root_endpoint FN + /static mount — index.html + static asset serving.""" @@ -438,13 +466,292 @@ class TestLifespanShutdown: ) app = create_app(_mock_client_factory()) with TestClient(app) as client: - # Pretend two turns are in-flight (status=streaming) + # Two turns in-flight (status=streaming, upstream_turn_id set = + # local tid here for test simplicity). for tid in (101, 102): app.state.turn_registry[("s-1", tid)] = TurnHandle( - session_id="s-1", turn_id=tid, content="x", status="streaming", + session_id="s-1", turn_id=tid, content="x", + status="streaming", upstream_turn_id=tid, ) # The exit of the `with` triggers lifespan shutdown # After lifespan shutdown: for route in cancel_routes: assert route.called, "upstream cancel should have been issued for each in-flight turn" assert app.state.turn_registry == {} + + +class TestUpstreamTurnIdCancel: + """v0.16.0 — cancel paths must target the UPSTREAM turn_id, not the + browser-local turn_id. The local _TURN_COUNTER allocates 1,2,3…; the + real upstream turn_id only arrives via the first SSE event's + sse_id.turn_id. Heid panel (Hulda) load-bearing finding. + """ + + @respx.mock + def test_cancel_targets_upstream_turn_id(self) -> None: + """A registered handle whose local turn_id (1) differs from its + captured upstream_turn_id (42) → POST cancel hits the UPSTREAM URL + /sessions/s-1/turns/42/cancel, not /turns/1/cancel. + """ + from ratatoskr.web.server import TurnHandle, create_app + + # Only the upstream-id cancel URL is mocked. If the code uses the + # local id (1), it'll miss this route → the test catches the bug. + upstream_route = respx.post( + "https://w.example/sessions/s-1/turns/42/cancel" + ).mock(return_value=httpx.Response(200, json={ + "turn_id": 42, "cancelled": True, "reason": None, + "partial_message_id": None, + })) + app = create_app(_mock_client_factory()) + app.state.turn_registry[("s-1", 1)] = TurnHandle( + session_id="s-1", turn_id=1, content="x", + status="streaming", upstream_turn_id=42, + ) + resp = TestClient(app).post("/api/turns/s-1/cancel?turn_id=1") + assert resp.status_code == 200 + assert resp.json()["cancelled"] is True + assert upstream_route.called, "cancel must target the upstream turn_id" + assert ("s-1", 1) not in app.state.turn_registry + + def test_cancel_before_upstream_started_is_noop(self) -> None: + """A handle with upstream_turn_id still None (turn never opened the + upstream stream) → cancel is a no-op: 200 {cancelled: false, + reason: not_started}, no upstream call, registry cleaned. + """ + from ratatoskr.web.server import TurnHandle, create_app + + app = create_app(_mock_client_factory()) + app.state.turn_registry[("s-1", 1)] = TurnHandle( + session_id="s-1", turn_id=1, content="x", + status="queued", upstream_turn_id=None, + ) + resp = TestClient(app).post("/api/turns/s-1/cancel?turn_id=1") + assert resp.status_code == 200 + assert resp.json()["cancelled"] is False + assert resp.json()["reason"] == "not_started" + assert ("s-1", 1) not in app.state.turn_registry + + @respx.mock + def test_stream_captures_upstream_turn_id(self) -> None: + """The stream generator captures upstream turn_id from the first + event's sse_id. After a happy text+done stream against upstream + turn 42 (local turn 1), the cancel mid-stream would have targeted 42. + Verified indirectly: drive the stream, assert events carry turn 42. + """ + stream = _sse_chunk("42:1", {"type": "text", "content": "hi"}) + _sse_chunk("42:2", _DONE_BODY) + respx.post("https://w.example/sessions/s-1/messages").mock(return_value=_sse_resp(stream)) + 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"] + assert turn_id == 1 or turn_id > 0 # local counter + with c.stream("GET", f"/api/turns/s-1/stream?turn_id={turn_id}") as resp: + raw = b"".join(resp.iter_bytes()) + events = _parse_browser_sse(raw) + # Every browser-facing event carries the upstream turn_id (42), not local + text_ev = next(e for e in events if e["event"] == "text") + assert text_ev["data"]["sse_id"].startswith("42:") + + +class TestServerSideEndUserId: + """v0.16.0 — end_user_id is server-configured (RATATOSKR_END_USER_ID via + create_app), NOT accepted from the browser body. Heid panel finding + + contract FN main STEPS 2. + """ + + @respx.mock + def test_create_session_uses_server_end_user_id(self) -> None: + """create_app(end_user_id=...) → POST /api/sessions threads that id + into the upstream POST body even when the browser sends none. + """ + import json as _j + + route = respx.post("https://w.example/sessions").mock( + return_value=httpx.Response(201, json={**_CREATE_OK, "agent_id": "lofn"}) + ) + from ratatoskr.web.server import create_app + app = create_app(_mock_client_factory(), end_user_id="ratatoskr-tui") + resp = TestClient(app).post("/api/sessions", json={"agent_id": "lofn"}) + assert resp.status_code == 201 + sent = _j.loads(route.calls[0].request.content) + assert sent == {"agent_id": "lofn", "end_user_id": "ratatoskr-tui"} + + @respx.mock + def test_create_session_ignores_body_end_user_id(self) -> None: + """A browser-supplied end_user_id is IGNORED — the server's + configured value wins. Prevents a client from impersonating an + arbitrary end-user partition. + """ + import json as _j + + route = respx.post("https://w.example/sessions").mock( + return_value=httpx.Response(201, json={**_CREATE_OK, "agent_id": "lofn"}) + ) + from ratatoskr.web.server import create_app + app = create_app(_mock_client_factory(), end_user_id="ratatoskr-tui") + TestClient(app).post( + "/api/sessions", json={"agent_id": "lofn", "end_user_id": "attacker"} + ) + sent = _j.loads(route.calls[0].request.content) + assert sent.get("end_user_id") == "ratatoskr-tui" + + @respx.mock + def test_create_session_no_end_user_id_when_unset(self) -> None: + """When create_app gets no end_user_id, the upstream body omits it + (matches create_session's default-omit shape).""" + import json as _j + + route = respx.post("https://w.example/sessions").mock( + return_value=httpx.Response(201, json=_CREATE_OK) + ) + from ratatoskr.web.server import create_app + app = create_app(_mock_client_factory()) # no end_user_id + TestClient(app).post("/api/sessions", json={"agent_id": "mimir"}) + sent = _j.loads(route.calls[0].request.content) + assert "end_user_id" not in sent + + +class TestCreateAppShape: + """create_app FN — route registration + state wiring (contract TESTS).""" + + def test_routes_registered(self) -> None: + """routes_registered [tracer]: app.routes contains all 9 path patterns.""" + from ratatoskr.web.server import create_app + app = create_app(_mock_client_factory()) + paths = {getattr(r, "path", None) for r in app.routes} + for expected in ( + "/", "/version", "/api/agents", "/api/sessions", + "/api/agents/{agent_id}/persona_state", + "/api/turns/{session_id}", "/api/turns/{session_id}/stream", + "/api/turns/{session_id}/cancel", + ): + assert expected in paths, f"missing route {expected}" + # /static is a Mount — its path is "/static" + assert "/static" in paths + + def test_state_attached(self) -> None: + """state_attached [trace]: app.state.turn_registry is empty dict.""" + from ratatoskr.web.server import create_app + app = create_app(_mock_client_factory()) + assert app.state.turn_registry == {} + + def test_factory_stored(self) -> None: + """factory_stored [trace]: app.state.client_factory is the same callable.""" + from ratatoskr.web.server import create_app + f = _mock_client_factory() + app = create_app(f) + assert app.state.client_factory is f + + +class TestStreamFullEventVocab: + """stream_turn_endpoint full_event_vocab — one of each Event type proxied.""" + + @respx.mock + def test_full_event_vocab(self) -> None: + """full_event_vocab [scenario]: a stream with one of each Event type + → each serialized to its fixture-shaped browser event.""" + stream = b"".join([ + _sse_chunk("42:1", {"type": "worker_phase", "phase": "BuildingPrompt", "turn_id": 42}), + _sse_chunk("42:2", {"type": "thinking", "content": "hmm"}), + _sse_chunk("42:3", {"type": "text", "content": "hi"}), + _sse_chunk("42:4", {"type": "text_boundary", "kind": "sentence", "char_offset": 2, "ts": "t"}), + _sse_chunk("42:5", {"type": "tool_start", "name": "s", "arguments": {"q": "x"}}), + _sse_chunk("42:6", {"type": "tool_result", "name": "s", "result": {"n": 1}, "duration_ms": 3}), + _sse_chunk("42:7", {"type": "awaiting_llm_first_token", "turn_id": 42, + "elapsed_ms_since_building_prompt": 5000.0}), + _sse_chunk("42:8", _DONE_BODY), + ]) + respx.post("https://w.example/sessions/s-1/messages").mock(return_value=_sse_resp(stream)) + from ratatoskr.web.server import create_app + app = create_app(_mock_client_factory()) + c = TestClient(app) + tid = c.post("/api/turns/s-1", json={"content": "hi"}).json()["turn_id"] + with c.stream("GET", f"/api/turns/s-1/stream?turn_id={tid}") as resp: + raw = b"".join(resp.iter_bytes()) + types = [e["event"] for e in _parse_browser_sse(raw)] + for expected in ("worker_phase", "thinking", "text", "text_boundary", + "tool_start", "tool_result", "awaiting_llm_first_token", "done"): + assert expected in types, f"missing browser event {expected}" + + +class TestDisconnectCancel: + """stream_turn_endpoint INV-005 — browser disconnect mid-stream triggers + upstream cancel against the UPSTREAM turn_id. The load-bearing test that + would have caught the v0.15.x turn_id bug. + + Drives the endpoint's StreamingResponse body_iterator directly and + cancels the consuming task to simulate the disconnect. This avoids the + ASGITransport stream-context-exit deadlock against a gated upstream + generator, while exercising the real `except asyncio.CancelledError` + cleanup path inside the generator. + """ + + @respx.mock + async def test_disconnect_triggers_upstream_cancel(self) -> None: + import asyncio + + from starlette.requests import Request + + from ratatoskr.web.server import TurnHandle, _stream_turn_endpoint, create_app + + gate = asyncio.Event() + + class _GatedAfterFirst(httpx.AsyncByteStream): + async def __aiter__(self): + yield _sse_chunk("42:1", {"type": "text", "content": "x"}) + await gate.wait() + + async def aclose(self) -> None: + return None + + respx.post("https://w.example/sessions/s-1/messages").mock( + return_value=_sse_resp_stream(_GatedAfterFirst()) + ) + cancel_route = respx.post("https://w.example/sessions/s-1/turns/42/cancel").mock( + return_value=httpx.Response(200, json={ + "turn_id": 42, "cancelled": True, "reason": None, "partial_message_id": None, + }) + ) + app = create_app(_mock_client_factory()) + app.state.turn_registry[("s-1", 1)] = TurnHandle( + session_id="s-1", turn_id=1, content="x", + ) + request = Request({ + "type": "http", "method": "GET", "path": "/api/turns/s-1/stream", + "path_params": {"session_id": "s-1"}, + "query_string": b"turn_id=1", "headers": [], "app": app, + }) + response = await _stream_turn_endpoint(request) + body_iter = response.body_iterator # type: ignore[attr-defined] + + async def consume() -> None: + async for _chunk in body_iter: + pass + + task = asyncio.create_task(consume()) + # Wait until the generator has captured the upstream turn_id (first + # event consumed). The handle is popped in finally, so check it + # before cancelling. + for _ in range(100): + h = app.state.turn_registry.get(("s-1", 1)) + if h is not None and h.upstream_turn_id == 42: + break + await asyncio.sleep(0.02) + else: + gate.set() + raise AssertionError("upstream_turn_id was never captured") + + # Simulate browser disconnect: cancel the consuming task. + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + + for _ in range(50): + if cancel_route.called: + break + await asyncio.sleep(0.02) + gate.set() + assert cancel_route.called, "browser disconnect must cancel the UPSTREAM turn (42)" diff --git a/uv.lock b/uv.lock index 9dff353..6572212 100644 --- a/uv.lock +++ b/uv.lock @@ -1013,7 +1013,7 @@ wheels = [ [[package]] name = "ratatoskr" -version = "0.15.1" +version = "0.16.0" source = { editable = "." } dependencies = [ { name = "httpx" },