feat(web): address Heid code-review findings — issue #16 (v0.16.0)
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).
This commit is contained in:
+318
-11
@@ -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)"
|
||||
|
||||
Reference in New Issue
Block a user