vh
11ef6830ab
Two related fixes for the same user-reported bug pattern from a
running session against ratatoskr:sindra (qwen3.6-35-a3b-heretic):
## 1. Streaming text overlapping the transcript
Operator: "new text comes at the bottom and overwrites the existing
pane information instead of pushing it up naturally."
Root cause: the v0.6.0 `#current-text` Static was `dock: bottom`
with `height: auto`, sitting between the transcript RichLog (1fr)
and the prompt Input (dock: bottom). As text streamed, the Static
grew UPWARD but Textual didn't dynamically resize the 1fr transcript
to accommodate — the growing Static visually OVERLAPPED the
transcript's bottom rows. On Done, `current_text.update("")` snapped
it to height 0 and the transcript re-laid-out — "boom, everything
updates."
Fix: remove `#current-text` Static entirely. Apply the same
coalesce-on-newline pattern v0.7.1 used for thinking — Text deltas
accumulate in `TuiPresenterState.text_chunk_buffer`, flushing whole
lines (each `\n` boundary) directly to `log` (transcript). On Done:
flush remaining tail, then [done] label + Rule + Markdown body.
Trade-off accepted: streamed lines + post-Done Markdown body are
both in the transcript (some content duplication). The Markdown
body re-renders the same content with proper formatting (lists,
bold, code blocks). Acceptable — operator gets both the live-progress
streaming AND the canonical rendered version.
## 2. MalformedSseId raw='' crashing every turn
Operator: "current session is erroring on every turn with
[malformed_sse_id] raw=''"
Worldtree's qwen3.6-35-a3b-heretic provider emits some events
without `id:` lines (observed 2026-05-25 mid-stream). When the FIRST
such event arrives before any prior id has been seen, httpx_sse's
`ServerSentEvent.id` is `""`. `_parse_sse_id('')` raised ValueError
→ MalformedSseId → turn worker bailed → operator saw the label
every turn.
Per SSE RFC, events without `id:` are legitimate (they just don't
update Last-Event-ID). Issue #7 already covered the empty-DATA
keepalive case with skip-silently semantics. Empty-id is the same
shape of wire weirdness; same fix shape:
if sse.id == "":
continue # treat as keepalive
Ordered AFTER the empty-data branch so an empty-data + empty-id
event still gets skipped on the data check.
## Tests + smoke
287/287 GREEN (was 286, +1 for empty-id skip; +1 net Text-flow test
adjustments). Ruff clean.
Verified Worldtree alive when the user hit the empty-id bug
(/healthz returned ok in 18ms) — not a server-down issue, just
wire-format mid-stream.
## Caveats
The fix doesn't recover content from the dropped empty-id event.
If the event happened to carry meaningful data (not a true
keepalive), we silently lose it. Acceptable trade-off: pre-v0.8.1
EVERY turn died on the offending agent; post-v0.8.1 the turn
continues and any single dropped frame is recoverable from logs if
debugging. Worldtree-side fix (always emit ids) is the right
upstream answer; ratatoskr just stops panicking on wire weirdness.
Patch bump (v0.8.0 → v0.8.1) — both fixes are bug fixes; no public
API change. The `TuiPresenterState.render` signature loses the
`current_text` parameter (was added v0.6.0), but presenter is an
internal contract; no external callers.
880 lines
36 KiB
Python
880 lines
36 KiB
Python
"""Tests for ratatoskr.sse_client per docs/contracts/issues/1.contract.md."""
|
|
|
|
import httpx
|
|
import pytest
|
|
import respx
|
|
|
|
from ratatoskr.sse_client import (
|
|
CancelAlreadyCompleted,
|
|
Cancelled,
|
|
CancelResult,
|
|
CancelTurnNotFound,
|
|
Done,
|
|
Error,
|
|
InvalidLastEventId,
|
|
MalformedSseId,
|
|
ResumeBufferExpired,
|
|
ResumeTurnFinished,
|
|
SseConnectFailed,
|
|
SseId,
|
|
Text,
|
|
TurnIdFlip,
|
|
_parse_sse_id,
|
|
cancel_turn,
|
|
reconnect_turn,
|
|
stream_turn,
|
|
)
|
|
|
|
_DONE_42_6 = {
|
|
"type": "done",
|
|
"phase": "succeeded",
|
|
"response": "hello",
|
|
"model": "m",
|
|
"duration_ms": 1,
|
|
"usage": {
|
|
"prompt_tokens": 0,
|
|
"completion_tokens": 0,
|
|
"total_tokens": 0,
|
|
"cached_input_tokens": 0,
|
|
},
|
|
}
|
|
|
|
|
|
def _sse_chunk(sse_id: str, body: dict[str, object]) -> bytes:
|
|
"""Compose one SSE event in wire format. Trailing blank line per spec."""
|
|
import json
|
|
|
|
return f"id: {sse_id}\ndata: {json.dumps(body)}\n\n".encode()
|
|
|
|
|
|
class TestParseSseId:
|
|
def test_happy_simple(self) -> None:
|
|
"""happy_simple [happy,tracer]: '42:3' -> SseId(turn_id=42, seq=3)."""
|
|
assert _parse_sse_id("42:3") == SseId(turn_id=42, seq=3)
|
|
|
|
def test_happy_seq_one(self) -> None:
|
|
"""happy_seq_one: smallest valid id per spec — first event of first turn."""
|
|
assert _parse_sse_id("1:1") == SseId(turn_id=1, seq=1)
|
|
|
|
def test_empty(self) -> None:
|
|
"""empty [adversarial]: '' -> ValueError."""
|
|
with pytest.raises(ValueError):
|
|
_parse_sse_id("")
|
|
|
|
def test_no_colon(self) -> None:
|
|
"""no_colon [adversarial]: '42' -> ValueError with '42' in the message."""
|
|
with pytest.raises(ValueError, match="42"):
|
|
_parse_sse_id("42")
|
|
|
|
def test_too_many_colons(self) -> None:
|
|
"""too_many_colons [adversarial]: '42:3:7' -> ValueError."""
|
|
with pytest.raises(ValueError):
|
|
_parse_sse_id("42:3:7")
|
|
|
|
def test_alpha_turn_id(self) -> None:
|
|
"""alpha_turn_id [adversarial]: 'foo:3' -> ValueError."""
|
|
with pytest.raises(ValueError):
|
|
_parse_sse_id("foo:3")
|
|
|
|
def test_alpha_seq(self) -> None:
|
|
"""alpha_seq [adversarial]: '42:bar' -> ValueError."""
|
|
with pytest.raises(ValueError):
|
|
_parse_sse_id("42:bar")
|
|
|
|
def test_zero_turn_id(self) -> None:
|
|
"""zero_turn_id [adversarial]: '0:3' -> ValueError (turn_id ≥ 1)."""
|
|
with pytest.raises(ValueError):
|
|
_parse_sse_id("0:3")
|
|
|
|
def test_zero_seq(self) -> None:
|
|
"""zero_seq [adversarial]: '1:0' -> ValueError (seq ≥ 1, spec line 705)."""
|
|
with pytest.raises(ValueError):
|
|
_parse_sse_id("1:0")
|
|
|
|
def test_negative_turn_id(self) -> None:
|
|
"""negative_turn_id [adversarial]: '-1:3' -> ValueError."""
|
|
with pytest.raises(ValueError):
|
|
_parse_sse_id("-1:3")
|
|
|
|
def test_negative_seq(self) -> None:
|
|
"""negative_seq [adversarial]: '42:-1' -> ValueError."""
|
|
with pytest.raises(ValueError):
|
|
_parse_sse_id("42:-1")
|
|
|
|
def test_trailing_whitespace(self) -> None:
|
|
"""trailing_whitespace [adversarial]: '42:3 ' -> ValueError (strict; no strip)."""
|
|
with pytest.raises(ValueError):
|
|
_parse_sse_id("42:3 ")
|
|
|
|
def test_truncation(self) -> None:
|
|
"""truncation [security]: 5000-char no-colon -> ValueError msg contains only raw[:64]."""
|
|
raw = "a" * 5000
|
|
with pytest.raises(ValueError) as exc_info:
|
|
_parse_sse_id(raw)
|
|
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
|
|
async def test_happy_one_text_done(self) -> None:
|
|
"""happy_one_text_done: text then done; same turn_id; iter ends after done."""
|
|
stream = _sse_chunk("42:1", {"type": "text", "content": "hello"}) + _sse_chunk(
|
|
"42:2",
|
|
{
|
|
"type": "done",
|
|
"phase": "succeeded",
|
|
"response": "hello",
|
|
"model": "glm5-turbo",
|
|
"duration_ms": 100,
|
|
"usage": {
|
|
"prompt_tokens": 0,
|
|
"completion_tokens": 0,
|
|
"total_tokens": 0,
|
|
"cached_input_tokens": 0,
|
|
},
|
|
},
|
|
)
|
|
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:
|
|
events = [e async for e in stream_turn(client, "s1", "hi")]
|
|
assert len(events) == 2
|
|
assert isinstance(events[0], Text)
|
|
assert events[0].content == "hello"
|
|
assert events[0].sse_id == SseId(42, 1)
|
|
assert isinstance(events[1], Done)
|
|
assert events[1].sse_id == SseId(42, 2)
|
|
assert events[0].sse_id.turn_id == events[1].sse_id.turn_id
|
|
|
|
@respx.mock
|
|
async def test_full_event_vocab(self) -> None:
|
|
"""full_event_vocab: one of each variant; all carry parsed sse_id."""
|
|
stream = b"".join(
|
|
[
|
|
_sse_chunk(
|
|
"42:1",
|
|
{"type": "worker_phase", "phase": "BuildingPrompt", "turn_id": 42},
|
|
),
|
|
_sse_chunk("42:2", {"type": "thinking", "content": "thinking out loud"}),
|
|
_sse_chunk("42:3", {"type": "text", "content": "hello"}),
|
|
_sse_chunk(
|
|
"42:4",
|
|
{
|
|
"type": "text_boundary",
|
|
"kind": "sentence",
|
|
"char_offset": 5,
|
|
"ts": "2026-05-21T00:00:00Z",
|
|
},
|
|
),
|
|
_sse_chunk(
|
|
"42:5",
|
|
{"type": "tool_start", "name": "search", "arguments": {"q": "x"}},
|
|
),
|
|
_sse_chunk(
|
|
"42:6",
|
|
{
|
|
"type": "tool_result",
|
|
"name": "search",
|
|
"result": {"n": 1},
|
|
"duration_ms": 12,
|
|
},
|
|
),
|
|
_sse_chunk("42:7", {"type": "text", "content": " world"}),
|
|
_sse_chunk(
|
|
"42:8",
|
|
{
|
|
"type": "done",
|
|
"phase": "succeeded",
|
|
"response": "hello world",
|
|
"model": "m",
|
|
"duration_ms": 50,
|
|
"usage": {
|
|
"prompt_tokens": 0,
|
|
"completion_tokens": 0,
|
|
"total_tokens": 0,
|
|
"cached_input_tokens": 0,
|
|
},
|
|
},
|
|
),
|
|
]
|
|
)
|
|
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:
|
|
events = [e async for e in stream_turn(client, "s1", "hi")]
|
|
types_seen = [type(e).__name__ for e in events]
|
|
assert types_seen == [
|
|
"WorkerPhase",
|
|
"Thinking",
|
|
"Text",
|
|
"TextBoundary",
|
|
"ToolStart",
|
|
"ToolResult",
|
|
"Text",
|
|
"Done",
|
|
]
|
|
# INV-002: every event carries parsed sse_id with turn_id, seq >= 1
|
|
for e in events:
|
|
assert e.sse_id.turn_id == 42
|
|
assert e.sse_id.seq >= 1
|
|
|
|
@respx.mock
|
|
async def test_error_terminal(self) -> None:
|
|
"""error_terminal: text then error; iteration ends; error_code populated."""
|
|
stream = _sse_chunk("42:1", {"type": "text", "content": "x"}) + _sse_chunk(
|
|
"42:2",
|
|
{
|
|
"type": "error",
|
|
"phase": "failed",
|
|
"error_code": "llm_output_invalid",
|
|
"message": "boom",
|
|
},
|
|
)
|
|
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:
|
|
events = [e async for e in stream_turn(client, "s1", "hi")]
|
|
assert len(events) == 2
|
|
assert isinstance(events[1], Error)
|
|
assert events[1].error_code == "llm_output_invalid"
|
|
assert events[1].message == "boom"
|
|
|
|
@respx.mock
|
|
async def test_cancelled_terminal(self) -> None:
|
|
"""cancelled_terminal: cancelled with phase=cancelled, turn_id; iteration ends."""
|
|
stream = _sse_chunk(
|
|
"42:1",
|
|
{
|
|
"type": "cancelled",
|
|
"phase": "cancelled",
|
|
"turn_id": 42,
|
|
"reason": "user_cancel",
|
|
"partial_message_id": None,
|
|
},
|
|
)
|
|
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:
|
|
events = [e async for e in stream_turn(client, "s1", "hi")]
|
|
assert len(events) == 1
|
|
assert isinstance(events[0], Cancelled)
|
|
assert events[0].turn_id == 42
|
|
assert events[0].reason == "user_cancel"
|
|
|
|
@respx.mock
|
|
async def test_session_not_found(self) -> None:
|
|
"""session_not_found: 404 -> SseConnectFailed(status=404)."""
|
|
respx.post("https://w.example/sessions/missing/messages").mock(
|
|
return_value=httpx.Response(404, json={"error": "session_not_found"})
|
|
)
|
|
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, "missing", "hi")]
|
|
assert exc_info.value.status == 404
|
|
|
|
@respx.mock
|
|
async def test_malformed_id_no_seq(self) -> None:
|
|
"""malformed_id_no_seq: id `42` (missing :seq) -> MalformedSseId; no event yielded."""
|
|
stream = b"id: 42\ndata: {\"type\":\"text\",\"content\":\"x\"}\n\n"
|
|
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:
|
|
events: list[object] = []
|
|
with pytest.raises(MalformedSseId):
|
|
async for e in stream_turn(client, "s1", "hi"):
|
|
events.append(e)
|
|
assert events == []
|
|
|
|
@respx.mock
|
|
async def test_malformed_id_alpha(self) -> None:
|
|
"""malformed_id_alpha: id `foo:bar` -> MalformedSseId."""
|
|
stream = b"id: foo:bar\ndata: {\"type\":\"text\",\"content\":\"x\"}\n\n"
|
|
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:
|
|
with pytest.raises(MalformedSseId):
|
|
_ = [e async for e in stream_turn(client, "s1", "hi")]
|
|
|
|
@respx.mock
|
|
async def test_turn_id_flip(self) -> None:
|
|
"""turn_id_flip: 42:1 then 99:2 -> TurnIdFlip; only first yielded."""
|
|
stream = _sse_chunk("42:1", {"type": "text", "content": "a"}) + _sse_chunk(
|
|
"99:2", {"type": "text", "content": "b"}
|
|
)
|
|
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(TurnIdFlip) 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.established == 42
|
|
assert exc_info.value.got == 99
|
|
|
|
@respx.mock
|
|
async def test_connection_drop(self) -> None:
|
|
"""connection_drop: RemoteProtocolError after one text -> SseConnectionDropped((42,1))."""
|
|
from ratatoskr.sse_client import SseConnectionDropped
|
|
|
|
class _DropAfter(httpx.AsyncByteStream):
|
|
def __init__(self, chunks: list[bytes]) -> None:
|
|
self._chunks = chunks
|
|
|
|
async def __aiter__(self): # type: ignore[no-untyped-def]
|
|
for c in self._chunks:
|
|
yield c
|
|
raise httpx.RemoteProtocolError("simulated mid-stream drop")
|
|
|
|
async def aclose(self) -> None:
|
|
return None
|
|
|
|
first = _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"},
|
|
stream=_DropAfter([first]),
|
|
)
|
|
)
|
|
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_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."""
|
|
stream = (
|
|
_sse_chunk("42:1", {"type": "text", "content": "hello "})
|
|
+ _sse_chunk("42:2", {"type": "text", "content": "world"})
|
|
+ _sse_chunk(
|
|
"42:3",
|
|
{
|
|
"type": "done",
|
|
"phase": "succeeded",
|
|
"response": "hello world",
|
|
"model": "m",
|
|
"duration_ms": 1,
|
|
"usage": {
|
|
"prompt_tokens": 0,
|
|
"completion_tokens": 0,
|
|
"total_tokens": 0,
|
|
"cached_input_tokens": 0,
|
|
},
|
|
},
|
|
)
|
|
)
|
|
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:
|
|
events = [e async for e in stream_turn(client, "s1", "hi")]
|
|
texts = [e for e in events if isinstance(e, Text)]
|
|
assert [t.content for t in texts] == ["hello ", "world"]
|
|
|
|
|
|
class TestReconnectTurn:
|
|
@respx.mock
|
|
async def test_happy_resume_from_seq_3(self) -> None:
|
|
"""happy_resume_from_seq_3 [scenario,tracer]: replay 42:4,42:5 then live 42:6 done."""
|
|
stream = (
|
|
_sse_chunk("42:4", {"type": "text", "content": "d"})
|
|
+ _sse_chunk("42:5", {"type": "text", "content": "e"})
|
|
+ _sse_chunk("42:6", _DONE_42_6)
|
|
)
|
|
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:
|
|
events = [e async for e in reconnect_turn(client, "s1", "original", "42:3")]
|
|
assert len(events) == 3
|
|
assert [e.sse_id.seq for e in events] == [4, 5, 6]
|
|
assert all(e.sse_id.turn_id == 42 for e in events)
|
|
assert isinstance(events[-1], Done)
|
|
|
|
@respx.mock
|
|
async def test_header_verbatim(self) -> None:
|
|
"""header_verbatim: outbound carries Last-Event-ID: 42:3 exactly."""
|
|
stream = _sse_chunk("42:4", _DONE_42_6)
|
|
route = 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:
|
|
_ = [e async for e in reconnect_turn(client, "s1", "x", "42:3")]
|
|
req = route.calls[0].request
|
|
assert req.headers["Last-Event-ID"] == "42:3"
|
|
|
|
@respx.mock
|
|
async def test_body_threads_content(self) -> None:
|
|
"""body_threads_content: outbound JSON body is {'content': <content>} byte-for-byte."""
|
|
import json as _json
|
|
|
|
stream = _sse_chunk("42:4", _DONE_42_6)
|
|
route = 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:
|
|
_ = [
|
|
e async for e in reconnect_turn(client, "s1", "hello mimir", "42:3")
|
|
]
|
|
body = _json.loads(route.calls[0].request.content)
|
|
assert body == {"content": "hello mimir"}
|
|
|
|
@respx.mock
|
|
async def test_buffer_expired(self) -> None:
|
|
"""buffer_expired: 412 -> ResumeBufferExpired(turn_id=42, buffered_from_seq=10)."""
|
|
respx.post("https://w.example/sessions/s1/messages").mock(
|
|
return_value=httpx.Response(
|
|
412,
|
|
json={"error": "buffer_expired", "buffered_from_seq": 10, "turn_id": 42},
|
|
)
|
|
)
|
|
async with httpx.AsyncClient(base_url="https://w.example") as client:
|
|
with pytest.raises(ResumeBufferExpired) as exc_info:
|
|
_ = [e async for e in reconnect_turn(client, "s1", "x", "42:3")]
|
|
assert exc_info.value.turn_id == 42
|
|
assert exc_info.value.buffered_from_seq == 10
|
|
|
|
@respx.mock
|
|
async def test_turn_finished(self) -> None:
|
|
"""turn_finished: 410 -> ResumeTurnFinished(turn_id=42)."""
|
|
respx.post("https://w.example/sessions/s1/messages").mock(
|
|
return_value=httpx.Response(
|
|
410, json={"error": "turn_finished", "turn_id": 42}
|
|
)
|
|
)
|
|
async with httpx.AsyncClient(base_url="https://w.example") as client:
|
|
with pytest.raises(ResumeTurnFinished) as exc_info:
|
|
_ = [e async for e in reconnect_turn(client, "s1", "x", "42:3")]
|
|
assert exc_info.value.turn_id == 42
|
|
|
|
@respx.mock
|
|
async def test_invalid_last_event_id_server(self) -> None:
|
|
"""invalid_last_event_id_server: 400 -> InvalidLastEventId."""
|
|
respx.post("https://w.example/sessions/s1/messages").mock(
|
|
return_value=httpx.Response(400, json={"error": "invalid_last_event_id"})
|
|
)
|
|
async with httpx.AsyncClient(base_url="https://w.example") as client:
|
|
with pytest.raises(InvalidLastEventId):
|
|
_ = [e async for e in reconnect_turn(client, "s1", "x", "42:3")]
|
|
|
|
@respx.mock
|
|
async def test_malformed_input(self) -> None:
|
|
"""malformed_input: caller passes '42' (no seq) -> ValueError; no HTTP issued."""
|
|
route = respx.post("https://w.example/sessions/s1/messages").mock(
|
|
return_value=httpx.Response(200, content=b"")
|
|
)
|
|
async with httpx.AsyncClient(base_url="https://w.example") as client:
|
|
with pytest.raises(ValueError):
|
|
_ = [e async for e in reconnect_turn(client, "s1", "x", "42")]
|
|
assert route.call_count == 0
|
|
|
|
@respx.mock
|
|
async def test_turn_id_flip_on_first_event(self) -> None:
|
|
"""turn_id_flip_on_first_event: server's first event 99:4 -> TurnIdFlip pre-yield."""
|
|
stream = _sse_chunk("99:4", {"type": "text", "content": "wrong turn"})
|
|
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(TurnIdFlip) as exc_info:
|
|
async for e in reconnect_turn(client, "s1", "x", "42:3"):
|
|
yielded.append(e) # type: ignore[arg-type]
|
|
assert yielded == []
|
|
assert exc_info.value.established == 42
|
|
assert exc_info.value.got == 99
|
|
|
|
|
|
class TestCancelTurn:
|
|
@respx.mock
|
|
async def test_happy_cancel(self) -> None:
|
|
"""happy_cancel [happy,tracer]: 200 -> CancelResult(42, True, None, None)."""
|
|
respx.post("https://w.example/sessions/s1/turns/42/cancel").mock(
|
|
return_value=httpx.Response(
|
|
200,
|
|
json={
|
|
"turn_id": 42,
|
|
"cancelled": True,
|
|
"reason": None,
|
|
"partial_message_id": None,
|
|
},
|
|
)
|
|
)
|
|
async with httpx.AsyncClient(base_url="https://w.example") as client:
|
|
result = await cancel_turn(client, "s1", 42)
|
|
assert result == CancelResult(
|
|
turn_id=42, cancelled=True, reason=None, partial_message_id=None
|
|
)
|
|
|
|
@respx.mock
|
|
async def test_persist_partial_query(self) -> None:
|
|
"""persist_partial_query [trace]: persist_partial=True -> URL has ?persist_partial=true."""
|
|
route = respx.post("https://w.example/sessions/s1/turns/42/cancel").mock(
|
|
return_value=httpx.Response(
|
|
200,
|
|
json={
|
|
"turn_id": 42,
|
|
"cancelled": True,
|
|
"reason": None,
|
|
"partial_message_id": None,
|
|
},
|
|
)
|
|
)
|
|
async with httpx.AsyncClient(base_url="https://w.example") as client:
|
|
await cancel_turn(client, "s1", 42, persist_partial=True)
|
|
assert "persist_partial=true" in str(route.calls[0].request.url)
|
|
|
|
@respx.mock
|
|
async def test_default_no_persist(self) -> None:
|
|
"""default_no_persist [trace]: default call -> URL has no persist_partial param."""
|
|
route = respx.post("https://w.example/sessions/s1/turns/42/cancel").mock(
|
|
return_value=httpx.Response(
|
|
200,
|
|
json={
|
|
"turn_id": 42,
|
|
"cancelled": True,
|
|
"reason": None,
|
|
"partial_message_id": None,
|
|
},
|
|
)
|
|
)
|
|
async with httpx.AsyncClient(base_url="https://w.example") as client:
|
|
await cancel_turn(client, "s1", 42)
|
|
assert "persist_partial" not in str(route.calls[0].request.url)
|
|
|
|
@respx.mock
|
|
async def test_cancel_already_completed(self) -> None:
|
|
"""cancel_already_completed [error]: 409 -> CancelAlreadyCompleted(turn_id=42)."""
|
|
respx.post("https://w.example/sessions/s1/turns/42/cancel").mock(
|
|
return_value=httpx.Response(409, json={"error": "turn_already_completed"})
|
|
)
|
|
async with httpx.AsyncClient(base_url="https://w.example") as client:
|
|
with pytest.raises(CancelAlreadyCompleted) as exc_info:
|
|
await cancel_turn(client, "s1", 42)
|
|
assert exc_info.value.turn_id == 42
|
|
|
|
@respx.mock
|
|
async def test_cancel_turn_not_found(self) -> None:
|
|
"""cancel_turn_not_found [error]: 404 -> CancelTurnNotFound(turn_id=42)."""
|
|
respx.post("https://w.example/sessions/s1/turns/42/cancel").mock(
|
|
return_value=httpx.Response(404, json={"error": "turn_not_found"})
|
|
)
|
|
async with httpx.AsyncClient(base_url="https://w.example") as client:
|
|
with pytest.raises(CancelTurnNotFound) as exc_info:
|
|
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."""
|
|
respx.post("https://w.example/sessions/s1/turns/42/cancel").mock(
|
|
side_effect=[
|
|
httpx.Response(
|
|
200,
|
|
json={
|
|
"turn_id": 42,
|
|
"cancelled": True,
|
|
"reason": None,
|
|
"partial_message_id": None,
|
|
},
|
|
),
|
|
httpx.Response(409, json={"error": "turn_already_completed"}),
|
|
]
|
|
)
|
|
async with httpx.AsyncClient(base_url="https://w.example") as client:
|
|
first = await cancel_turn(client, "s1", 42)
|
|
assert first.cancelled is True
|
|
with pytest.raises(CancelAlreadyCompleted):
|
|
await cancel_turn(client, "s1", 42)
|
|
|
|
|
|
# ============================================================================
|
|
# Issue #7: empty-data skip + MalformedSseData raise
|
|
# ============================================================================
|
|
|
|
|
|
def _sse_empty_chunk(sse_id: str) -> bytes:
|
|
"""SSE frame with id but empty data (server-emitted keepalive shape)."""
|
|
return f"id: {sse_id}\ndata:\n\n".encode()
|
|
|
|
|
|
def _sse_raw_chunk(sse_id: str, raw_data: str) -> bytes:
|
|
"""SSE frame with id + arbitrary raw data (for testing malformed JSON)."""
|
|
return f"id: {sse_id}\ndata: {raw_data}\n\n".encode()
|
|
|
|
|
|
def _sse_no_id_chunk(data: str) -> bytes:
|
|
"""SSE frame with NO id line + arbitrary data (v0.8.1: keepalive shape)."""
|
|
return f"data: {data}\n\n".encode()
|
|
|
|
|
|
class TestEmptyIdSkipped:
|
|
@respx.mock
|
|
async def test_empty_id_on_first_event_skipped(self) -> None:
|
|
"""empty_id_on_first_event_skipped [v0.8.1]: stream starts with an
|
|
event carrying NO `id:` line → httpx_sse exposes sse.id == ''
|
|
(no prior id to inherit). Pre-v0.8.1: MalformedSseId raw='' crashed
|
|
the turn. v0.8.1: treat same as empty-data keepalive — skip silently.
|
|
|
|
Observed 2026-05-25 on Worldtree's qwen3.6-35-a3b-heretic provider:
|
|
the first stream frame had no id line, every turn died with
|
|
`[malformed_sse_id] raw=''`.
|
|
"""
|
|
from ratatoskr.sse_client import Done as _Done
|
|
from ratatoskr.sse_client import Text as _Text
|
|
|
|
# First frame: no id line (httpx_sse → sse.id = ""). Skip it.
|
|
# Subsequent frames have ids; normal processing resumes.
|
|
stream = (
|
|
_sse_no_id_chunk('{"type":"keepalive"}') # ← skipped (sse.id == "")
|
|
+ _sse_chunk("42:1", {"type": "text", "content": "first"})
|
|
+ _sse_chunk("42:2", _DONE_42_6)
|
|
)
|
|
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:
|
|
events = [e async for e in stream_turn(client, "s1", "hi")]
|
|
# 2 events — the no-id frame is invisible (no MalformedSseId crash).
|
|
assert len(events) == 2
|
|
assert isinstance(events[0], _Text)
|
|
assert events[0].content == "first"
|
|
assert isinstance(events[1], _Done)
|
|
|
|
|
|
class TestEmptyDataSkipped:
|
|
@respx.mock
|
|
async def test_empty_data_skipped(self) -> None:
|
|
"""empty_data_skipped [trace]: 4 frames in, 3 events out; skip preserves last_sse_id."""
|
|
from ratatoskr.sse_client import Done as _Done
|
|
from ratatoskr.sse_client import Text as _Text
|
|
|
|
stream = (
|
|
_sse_chunk("42:1", {"type": "text", "content": "first"})
|
|
+ _sse_empty_chunk("42:2") # ← skipped silently
|
|
+ _sse_chunk("42:3", {"type": "text", "content": "second"})
|
|
+ _sse_chunk("42:4", _DONE_42_6)
|
|
)
|
|
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:
|
|
events = [e async for e in stream_turn(client, "s1", "hi")]
|
|
# Exactly 3 events: Text, Text, Done — empty-data frame at 42:2 is invisible
|
|
assert len(events) == 3
|
|
assert isinstance(events[0], _Text) and events[0].sse_id == SseId(42, 1)
|
|
assert isinstance(events[1], _Text) and events[1].sse_id == SseId(42, 3)
|
|
assert isinstance(events[2], _Done) and events[2].sse_id == SseId(42, 4)
|
|
# Per INV-003: skip MUST NOT advance through 42:2. The second Text's sse_id
|
|
# is (42, 3) — directly verifies the skip didn't bookkeep 42:2.
|
|
assert events[1].sse_id.seq == 3, "skip advanced through 42:2"
|
|
|
|
@respx.mock
|
|
async def test_empty_data_skip_preserves_last_seen_sse_id(self) -> None:
|
|
"""empty_skip_does_not_advance [trace]: drop-after-empty → last_seen is last real event."""
|
|
from ratatoskr.sse_client import SseConnectionDropped
|
|
|
|
# Stream: text(42:1), empty(42:2), then drop. Per INV-003, the skipped
|
|
# 42:2 must NOT advance internal last_sse_id. If the consumer caught a
|
|
# drop, SseConnectionDropped.last_seen_sse_id should be (42, 1) — the
|
|
# last *real* event — NOT (42, 2).
|
|
first = _sse_chunk("42:1", {"type": "text", "content": "x"})
|
|
empty = _sse_empty_chunk("42:2")
|
|
|
|
class _DropAfterEmpty(httpx.AsyncByteStream):
|
|
async def __aiter__(self): # type: ignore[no-untyped-def]
|
|
yield first
|
|
yield empty
|
|
raise httpx.RemoteProtocolError("drop after skip")
|
|
|
|
async def aclose(self) -> None:
|
|
return None
|
|
|
|
respx.post("https://w.example/sessions/s1/messages").mock(
|
|
return_value=httpx.Response(
|
|
200,
|
|
headers={"content-type": "text/event-stream"},
|
|
stream=_DropAfterEmpty(),
|
|
)
|
|
)
|
|
async with httpx.AsyncClient(base_url="https://w.example") as client:
|
|
with pytest.raises(SseConnectionDropped) as exc_info:
|
|
async for _ in stream_turn(client, "s1", "hi"):
|
|
pass
|
|
assert exc_info.value.last_seen_sse_id == SseId(42, 1), (
|
|
f"skip advanced last_sse_id through 42:2; got {exc_info.value.last_seen_sse_id}"
|
|
)
|
|
|
|
@respx.mock
|
|
async def test_malformed_data_raises(self) -> None:
|
|
"""malformed_data_raises [error]: text + bad-JSON → yields Text then MalformedSseData."""
|
|
from ratatoskr.sse_client import MalformedSseData
|
|
from ratatoskr.sse_client import Text as _Text
|
|
|
|
stream = (
|
|
_sse_chunk("42:1", {"type": "text", "content": "hi"})
|
|
+ _sse_raw_chunk("42:2", "not-json")
|
|
)
|
|
respx.post("https://w.example/sessions/s1/messages").mock(
|
|
return_value=httpx.Response(
|
|
200, headers={"content-type": "text/event-stream"}, content=stream
|
|
)
|
|
)
|
|
yielded: list = []
|
|
async with httpx.AsyncClient(base_url="https://w.example") as client:
|
|
with pytest.raises(MalformedSseData) as exc_info:
|
|
async for e in stream_turn(client, "s1", "hi"):
|
|
yielded.append(e)
|
|
assert len(yielded) == 1
|
|
assert isinstance(yielded[0], _Text)
|
|
assert exc_info.value.raw == "not-json"
|
|
|
|
@respx.mock
|
|
async def test_whitespace_data_raises(self) -> None:
|
|
"""whitespace_data_raises [adv]: single-space data → MalformedSseData (NOT skipped)."""
|
|
from ratatoskr.sse_client import MalformedSseData
|
|
|
|
stream = _sse_raw_chunk("42:1", " ") # single space — non-empty, JSON-invalid
|
|
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:
|
|
with pytest.raises(MalformedSseData):
|
|
async for _ in stream_turn(client, "s1", "hi"):
|
|
pass
|
|
|
|
@respx.mock
|
|
async def test_malformed_data_truncation(self) -> None:
|
|
"""malformed_data_truncation [security]: 5000-char bad data → raw truncated to 200."""
|
|
from ratatoskr.sse_client import MalformedSseData
|
|
|
|
huge_bad = "x" * 5000 # not JSON; very long
|
|
stream = _sse_raw_chunk("42:1", huge_bad)
|
|
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:
|
|
with pytest.raises(MalformedSseData) as exc_info:
|
|
async for _ in stream_turn(client, "s1", "hi"):
|
|
pass
|
|
assert len(exc_info.value.raw) == 200
|
|
assert exc_info.value.raw == "x" * 200
|
|
# Exception message also only contains the truncated form
|
|
assert "x" * 5000 not in str(exc_info.value)
|