From c17af183513c016ea05fc5162ed043b3af4a4a46 Mon Sep 17 00:00:00 2001 From: Vuong Hoang Date: Wed, 20 May 2026 21:33:32 -0700 Subject: [PATCH] fix(sse_client): address Volva code-vs-contract drift (issue #1) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Volva's code-spec review (thread 01KS4CP6ZZ1F) surfaced four code-vs- contract drift findings on the TDD-passing implementation. All four addressed here; no contract amendments required. 1. _iter_events fell off the end of aiter_sse() normally on clean EOF before any Done/Error/Cancelled. Per INV-001 the iterator MUST NOT raise StopAsyncIteration before a terminal event unless the HTTP connection drops, in which case it raises SseConnectionDropped. Clean EOF before terminal is the same semantic — the stream ended without delivering its contracted invariant. Fix: track terminal_seen inside _iter_events; after the async-for completes, if not seen, raise SseConnectionDropped(last_seen_sse_id=...). Two new tests: test_clean_eof_before_terminal (one text then EOF) and test_zero_event_eof (empty stream — last_seen_sse_id is None). 2. SseConnectFailed and CancelFailed both store .body without truncation; ERROR_ROUTING specifies resp.read()[:1024]. Fix truncates in each exception's __init__ before storing. New test test_connect_failed_body_truncated (503 + 5000-byte body → 1024) and test_cancel_failed_truncates_body (same shape on cancel). 3. _parse_sse_id PRE-001 specifies `assert isinstance(raw, str)`. Previous code called raw.split(":") directly, which raises an incidental AttributeError on non-str inputs — not the contracted precondition path. Fix adds the assert. New test test_non_string_input covers int and None. 4. Cancel ERROR_ROUTING said httpx.HTTPStatusError other status → CancelFailed, but no test exercised the branch. test_cancel_failed_ truncates_body covers this (above) — single test double-covers findings 2 and 4. 43 tests GREEN (42 sse_client + boundary smoke); ruff clean. Meta-note from Volva: TDD caught the main happy/adversarial SSE shape, resume header/body, turn-id flip, and cancel races. The remaining misses were "negative space" cases (clean premature EOF, exception payload truncation, untested generic cancel branch). Calibration evidence that cross-model review pulls weight on what same-model TDD's hypothesis-space doesn't probe. --- persistent-memory.md | 1 + src/ratatoskr/sse_client.py | 10 ++++++ tests/test_sse_client.py | 71 +++++++++++++++++++++++++++++++++++++ 3 files changed, 82 insertions(+) diff --git a/persistent-memory.md b/persistent-memory.md index 04ea0a0..e4e29a3 100644 --- a/persistent-memory.md +++ b/persistent-memory.md @@ -79,6 +79,7 @@ decision. Captures rationale that won't be obvious from code alone. - `[2026-05-21]` **Contract converted to issue-scoped (issue #1).** Moved `docs/contracts/sse_client.contract.md` → `docs/contracts/issues/1.contract.md`. Frontmatter shape switched from module-scoped (`module:`/`purpose:`) to issue-scoped (`target_module:`/`scope:`/`prd:`) per CONTRACT-FORMAT §2.1.I. `prd:` block pins to issue #1's body hash (`abcbc49467e86f1d`). `scripts/contract_drift_check.py` returns clean. **Known parser stale-ness**: `contract_parser.py --validate` ERRORs on issue-scoped frontmatter (missing `module:`/`purpose:`) — this is CONTRACT-FORMAT §2.1.L H10, a documented Brokkr-side follow-up. Parser is a canonical sync, so we do NOT patch it locally (would drift from canonical). Treat parser ERROR-on-issue-scoped as expected until the canonical bumps. - `[2026-05-21]` **Default issue-tracker labels seeded** (17 total). Sleipnir gating (`ready-for-agent`, `blocked-needs-contract`, `blocked-needs-dependency`), triage (`needs-triage`, `needs-architect-decision`, `needs-info`), type (`bug`, `enhancement`, `task`, `documentation`), resolution (`duplicate`, `wontfix`, `invalid`), Ratatoskr-specific area (`sse-client`, `tui`, `cli`, `observability`). - `[2026-05-21]` **`ratatoskr.sse_client` implemented via TDD against issue #1's contract.** 37 contract-listed tests authored + GREEN per the tracer-bullet vertical-slice ordering (`_parse_sse_id` → `stream_turn` → `reconnect_turn` → `cancel_turn`). Refactor pass extracted `_iter_events` helper to dedupe INV-002 + INV-003 + terminal-break logic across `stream_turn` and `reconnect_turn`; `expected_turn_id=None` vs `expected_turn_id=N` distinguishes the two entry-point semantics Volva surfaced. Notable choices made during implementation: (a) regex `^-?\d+$` pre-check in `_parse_sse_id` to reject whitespace before `int()` (Python's `int(" 3 ")` would silently strip — this kept the strict-no-whitespace test honest); (b) `_DropAfter` AsyncByteStream subclass in tests to simulate mid-stream `RemoteProtocolError`; (c) ToolResult.result and ToolStart.arguments typed as `Any` (server JSON varies); (d) ruff line-length=100 (per pyproject) forced some test docstrings to be tighter than v0 draft. +- `[2026-05-21]` **Volva code-vs-contract review round on `ratatoskr.sse_client`.** Volva flagged 4 findings (3 drifts + 1 test-gap), all code-side "fix it" recommendations: (1) `_iter_events` fell off cleanly on EOF before terminal, violating INV-001 ("MUST NOT raise StopAsyncIteration before a terminal event arrives unless connection drops"); fix tracks `terminal_seen` flag and raises `SseConnectionDropped` on clean-EOF-without-terminal. (2) Both `SseConnectFailed.body` and `CancelFailed.body` stored full response bytes; ERROR_ROUTING specified truncation to `[:1024]`; fix truncates in `__init__` before storing. (3) `_parse_sse_id` PRE-001 specified `assert isinstance(raw, str)`, but code called `.split(":")` directly (incidental `AttributeError` on non-str); fix adds the assert. (4) Test-gap on cancel_turn's "other status → CancelFailed" branch; fix adds a 503 test with >1024-byte body that double-covers finding #2. Meta-note: Volva said TDD caught the main happy/adversarial shape; the misses were "negative space" cases (clean EOF, exception payload truncation, untested generic cancel branch) — calibration evidence that cross-model review pulls weight on the same-model author's blind spots. 43 tests GREEN post-fix (42 sse_client + 1 boundary), ruff clean. - `[2026-05-21]` **Volva paraphrase round on `docs/contracts/issues/1.contract.md`.** Volva flagged 5 ambiguities; operator approved amendments to 3 of them. (1) `reconnect_turn` STEP 2 punt resolved: signature now carries `content: str`; STEP 2 body is `json={"content": content}` matching spec §Reconnect flow example verbatim. Spec line 732 makes the agent's tools+LLM run "exactly once regardless of disconnects/reconnects" — the `content` is a wire-schema requirement, not re-processed server-side. (2) `_parse_sse_id` tightened: `turn_id ≥ 1` AND `seq ≥ 1` (was `≥ 0`); spec §SSE id format line 705 explicitly states `seq` starts at 1, and `turn_id` is SQLite autoincrement (≥1). Test `happy_zero_seq` flipped to `zero_seq [adversarial]`; new `zero_turn_id` + `negative_seq` adversarial tests added. (3) INV-003 clarified to spell out the two-entry-point semantics: `stream_turn` establishes `turn_id` from the first event (first event always yields); `reconnect_turn` parses the expected `turn_id` FROM `last_event_id` BEFORE the connection opens, so the first server event is already a flip-candidate and is NOT yielded on mismatch. Volva flags #3 (MalformedSseId-vs-ValueError split) and #5 (exactly-one-terminal as server-assumed) noted but kept as-is — deliberate distinctions. Drift check still clean against issue #1 (amending the contract doesn't touch the pinned issue body). ## Tried and abandoned diff --git a/src/ratatoskr/sse_client.py b/src/ratatoskr/sse_client.py index 59075d8..6a5d9d0 100644 --- a/src/ratatoskr/sse_client.py +++ b/src/ratatoskr/sse_client.py @@ -145,6 +145,7 @@ class SseConnectFailed(Exception): """Raised when the SSE endpoint returned a non-2xx status before the stream opened.""" def __init__(self, *, status: int, body: bytes) -> None: + body = body[:1024] super().__init__(f"SSE connect failed: status={status}, body={body[:128]!r}") self.status = status self.body = body @@ -205,6 +206,7 @@ class CancelFailed(Exception): """Raised on unexpected cancel response status.""" def __init__(self, *, status: int, body: bytes) -> None: + body = body[:1024] super().__init__(f"cancel failed: status={status}, body={body[:128]!r}") self.status = status self.body = body @@ -285,6 +287,7 @@ async def _iter_events( """ established = expected_turn_id last_sse_id: SseId | None = None + terminal_seen = False try: async for sse in event_source.aiter_sse(): try: @@ -299,9 +302,15 @@ async def _iter_events( yield event last_sse_id = sse_id if isinstance(event, (Done, Error, Cancelled)): + terminal_seen = True return except (httpx.ReadError, httpx.RemoteProtocolError) as exc: raise SseConnectionDropped(last_seen_sse_id=last_sse_id) from exc + if not terminal_seen: + # Clean EOF before terminal event — INV-001 says stream MUST NOT end + # without exactly one Done/Error/Cancelled. Surface as connection drop; + # caller may reconnect_turn if it holds last_sse_id. + raise SseConnectionDropped(last_seen_sse_id=last_sse_id) async def stream_turn( @@ -369,6 +378,7 @@ async def reconnect_turn( def _parse_sse_id(raw: str) -> SseId: """Parse the SSE wire `id:` as composite `{turn_id}:{seq}`. See contract FN _parse_sse_id.""" + assert isinstance(raw, str) parts = raw.split(":") if len(parts) != 2: raise ValueError(f"expected '{{turn_id}}:{{seq}}', got: {raw[:64]!r}") diff --git a/tests/test_sse_client.py b/tests/test_sse_client.py index 5a27e59..6d7148e 100644 --- a/tests/test_sse_client.py +++ b/tests/test_sse_client.py @@ -114,6 +114,13 @@ class TestParseSseId: assert raw not in str(exc_info.value), "full 5000-char input must not appear in message" assert raw[:64] in str(exc_info.value), "first 64 chars must appear in message" + def test_non_string_input(self) -> None: + """PRE-001 hard: raw is a string -- isinstance check before parse.""" + with pytest.raises(AssertionError): + _parse_sse_id(42) # type: ignore[arg-type] + with pytest.raises(AssertionError): + _parse_sse_id(None) # type: ignore[arg-type] + class TestStreamTurn: @respx.mock @@ -367,6 +374,54 @@ class TestStreamTurn: assert len(yielded) == 1 assert exc_info.value.last_seen_sse_id == SseId(42, 1) + @respx.mock + async def test_clean_eof_before_terminal(self) -> None: + """INV-001: clean EOF without Done/Error/Cancelled -> SseConnectionDropped.""" + from ratatoskr.sse_client import SseConnectionDropped + + # One text event, then stream ends cleanly (no terminal). + stream = _sse_chunk("42:1", {"type": "text", "content": "x"}) + respx.post("https://w.example/sessions/s1/messages").mock( + return_value=httpx.Response( + 200, headers={"content-type": "text/event-stream"}, content=stream + ) + ) + async with httpx.AsyncClient(base_url="https://w.example") as client: + yielded: list[Text] = [] + with pytest.raises(SseConnectionDropped) as exc_info: + async for e in stream_turn(client, "s1", "hi"): + yielded.append(e) # type: ignore[arg-type] + assert len(yielded) == 1 + assert exc_info.value.last_seen_sse_id == SseId(42, 1) + + @respx.mock + async def test_zero_event_eof(self) -> None: + """INV-001: empty stream (no events at all) -> SseConnectionDropped(None).""" + from ratatoskr.sse_client import SseConnectionDropped + + respx.post("https://w.example/sessions/s1/messages").mock( + return_value=httpx.Response( + 200, headers={"content-type": "text/event-stream"}, content=b"" + ) + ) + async with httpx.AsyncClient(base_url="https://w.example") as client: + with pytest.raises(SseConnectionDropped) as exc_info: + _ = [e async for e in stream_turn(client, "s1", "hi")] + assert exc_info.value.last_seen_sse_id is None + + @respx.mock + async def test_connect_failed_body_truncated(self) -> None: + """ERROR_ROUTING: SseConnectFailed.body is truncated to <= 1024 bytes.""" + big_body = b"x" * 5000 + respx.post("https://w.example/sessions/s1/messages").mock( + return_value=httpx.Response(503, content=big_body) + ) + async with httpx.AsyncClient(base_url="https://w.example") as client: + with pytest.raises(SseConnectFailed) as exc_info: + _ = [e async for e in stream_turn(client, "s1", "hi")] + assert len(exc_info.value.body) <= 1024 + assert exc_info.value.body == big_body[:1024] + @respx.mock async def test_no_text_aggregation(self) -> None: """no_text_aggregation: consumer yields each text event separately; no concat.""" @@ -601,6 +656,22 @@ class TestCancelTurn: await cancel_turn(client, "s1", 42) assert exc_info.value.turn_id == 42 + @respx.mock + async def test_cancel_failed_truncates_body(self) -> None: + """ERROR_ROUTING: 5xx -> CancelFailed; body truncated to <= 1024 bytes.""" + from ratatoskr.sse_client import CancelFailed + + big_body = b"x" * 5000 + respx.post("https://w.example/sessions/s1/turns/42/cancel").mock( + return_value=httpx.Response(503, content=big_body) + ) + async with httpx.AsyncClient(base_url="https://w.example") as client: + with pytest.raises(CancelFailed) as exc_info: + await cancel_turn(client, "s1", 42) + assert exc_info.value.status == 503 + assert len(exc_info.value.body) <= 1024 + assert exc_info.value.body == big_body[:1024] + @respx.mock async def test_cancel_idempotent(self) -> None: """cancel_idempotent: first 200 -> happy; second 409 -> CancelAlreadyCompleted."""