fix(sse_client): address Volva code-vs-contract drift (issue #1)
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.
This commit is contained in:
@@ -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."""
|
||||
|
||||
Reference in New Issue
Block a user