fix(web): close Heid pass-2 findings — stream vocab + disconnect catch (v0.16.1)
Second Heid panel pass (thread 01KSPBMFRRQE) on the v0.16.0 tree:
Gróa returned zero findings; Hulda surfaced two minor tightening
items, both closed here.
1. test-gap — TestStreamFullEventVocab drove only 8 of 11 Event types
through the stream endpoint (omitted Error, Cancelled, AffectUpdate).
Serialization for all 11 was already covered by the presentation-
contract fixture tests; this was a stream-integration coverage gap.
- Added AffectUpdate to the vocab stream (non-terminal, coexists
with done).
- Added dedicated test_error_terminal_event + test_cancelled_terminal_event
(terminal events are mutually exclusive with done, so they can't
share one stream).
2. precision — the disconnect-cancel path caught bare `except Exception:
pass`, silently swallowing real CancelFailed / transport errors. The
contract intent is to swallow only the cooperative race
(CancelAlreadyCompleted). Narrowed: swallow CancelAlreadyCompleted /
CancelTurnNotFound as the no-op race; log unexpected cancel failures
as a structured stderr line for diagnosability. Never re-raises (we're
unwinding the cancelled generator and must not mask CancelledError).
Tests: +2 (376 → 378). Patch per SemVer discipline — coverage +
diagnosability tightening, no behavior change observable to callers.
This commit is contained in:
+1
-1
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
|
||||
|
||||
[project]
|
||||
name = "ratatoskr"
|
||||
version = "0.16.0"
|
||||
version = "0.16.1"
|
||||
description = "Worldtree Conversation API debug TUI — multi-pane observability dashboard"
|
||||
readme = "README.md"
|
||||
requires-python = ">=3.12"
|
||||
|
||||
@@ -265,8 +265,21 @@ async def _stream_turn_endpoint(request: Request) -> StreamingResponse:
|
||||
if handle.status == "streaming" and handle.upstream_turn_id is not None:
|
||||
try:
|
||||
await cancel_turn(client, session_id, handle.upstream_turn_id)
|
||||
except Exception:
|
||||
pass
|
||||
except (CancelAlreadyCompleted, CancelTurnNotFound):
|
||||
pass # cooperative race — turn already terminal upstream
|
||||
except Exception as exc:
|
||||
# v0.16.1: unexpected cancel failure during disconnect
|
||||
# cleanup (e.g. CancelFailed, transport error) — log for
|
||||
# diagnosability instead of silently swallowing. Never
|
||||
# re-raise: we're already unwinding the cancelled
|
||||
# generator and must not mask the CancelledError.
|
||||
import sys as _sys
|
||||
_sys.stderr.write(
|
||||
f'{{"kind":"disconnect_cancel","event":"cancel_failed",'
|
||||
f'"session_id":"{session_id}",'
|
||||
f'"upstream_turn_id":{handle.upstream_turn_id},'
|
||||
f'"exc":"{type(exc).__name__}"}}\n'
|
||||
)
|
||||
raise
|
||||
finally:
|
||||
registry.pop((session_id, turn_id), None)
|
||||
|
||||
+57
-20
@@ -647,33 +647,70 @@ class TestCreateAppShape:
|
||||
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())
|
||||
def _drive_stream(self, app, stream: bytes) -> list[str]:
|
||||
respx.post("https://w.example/sessions/s-1/messages").mock(
|
||||
return_value=_sse_resp(stream)
|
||||
)
|
||||
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",
|
||||
return [e["event"] for e in _parse_browser_sse(raw)]
|
||||
|
||||
@respx.mock
|
||||
def test_full_event_vocab(self) -> None:
|
||||
"""full_event_vocab [scenario]: a stream with the non-terminal Event
|
||||
types (incl. AffectUpdate) + Done → each serialized to its fixture-
|
||||
shaped browser event. Error / Cancelled are terminal and exclusive
|
||||
with Done, so they get dedicated tests below.
|
||||
"""
|
||||
from ratatoskr.web.server import create_app
|
||||
|
||||
stream = b"".join([
|
||||
_sse_chunk("42:1", {"type": "affect_update", "status": "current", "turn_id": 42,
|
||||
"snapshot": {"agent_id": "mimir", "pad": {}, "dominant_emotion": "x"}}),
|
||||
_sse_chunk("42:2", {"type": "worker_phase", "phase": "BuildingPrompt", "turn_id": 42}),
|
||||
_sse_chunk("42:3", {"type": "thinking", "content": "hmm"}),
|
||||
_sse_chunk("42:4", {"type": "text", "content": "hi"}),
|
||||
_sse_chunk("42:5", {"type": "text_boundary", "kind": "sentence", "char_offset": 2, "ts": "t"}),
|
||||
_sse_chunk("42:6", {"type": "tool_start", "name": "s", "arguments": {"q": "x"}}),
|
||||
_sse_chunk("42:7", {"type": "tool_result", "name": "s", "result": {"n": 1}, "duration_ms": 3}),
|
||||
_sse_chunk("42:8", {"type": "awaiting_llm_first_token", "turn_id": 42,
|
||||
"elapsed_ms_since_building_prompt": 5000.0}),
|
||||
_sse_chunk("42:9", _DONE_BODY),
|
||||
])
|
||||
types = self._drive_stream(create_app(_mock_client_factory()), stream)
|
||||
for expected in ("affect_update", "worker_phase", "thinking", "text", "text_boundary",
|
||||
"tool_start", "tool_result", "awaiting_llm_first_token", "done"):
|
||||
assert expected in types, f"missing browser event {expected}"
|
||||
|
||||
@respx.mock
|
||||
def test_error_terminal_event(self) -> None:
|
||||
"""error terminal [scenario]: an SSE `error` event (distinct from the
|
||||
synthetic connection-error event) proxies to a browser `error` event.
|
||||
"""
|
||||
from ratatoskr.web.server import create_app
|
||||
|
||||
stream = _sse_chunk("42:1", {"type": "text", "content": "x"}) + _sse_chunk(
|
||||
"42:2", {"type": "error", "phase": "failed",
|
||||
"error_code": "llm_output_invalid", "message": "boom"},
|
||||
)
|
||||
types = self._drive_stream(create_app(_mock_client_factory()), stream)
|
||||
assert "error" in types
|
||||
|
||||
@respx.mock
|
||||
def test_cancelled_terminal_event(self) -> None:
|
||||
"""cancelled terminal [scenario]: an SSE `cancelled` event proxies to
|
||||
a browser `cancelled` event."""
|
||||
from ratatoskr.web.server import create_app
|
||||
|
||||
stream = _sse_chunk("42:1", {"type": "text", "content": "x"}) + _sse_chunk(
|
||||
"42:2", {"type": "cancelled", "phase": "cancelled", "turn_id": 42,
|
||||
"reason": "user_cancel", "partial_message_id": None},
|
||||
)
|
||||
types = self._drive_stream(create_app(_mock_client_factory()), stream)
|
||||
assert "cancelled" in types
|
||||
|
||||
|
||||
class TestDisconnectCancel:
|
||||
"""stream_turn_endpoint INV-005 — browser disconnect mid-stream triggers
|
||||
|
||||
Reference in New Issue
Block a user