From 02f2a04b37c277807fd2aeb946e0d515ca6f8f67 Mon Sep 17 00:00:00 2001 From: Vuong Hoang Date: Wed, 20 May 2026 21:25:20 -0700 Subject: [PATCH] feat(sse_client): implement issue #1 contract via TDD MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implements docs/contracts/issues/1.contract.md. Four entry points (stream_turn, reconnect_turn, cancel_turn, _parse_sse_id) + nine typed Event variants + ten domain exceptions. 37 tests covering every TESTS: entry verbatim, plus the boundary smoke test still passes. Tracer-bullet ordering per the contract's per-FN tracer tags: _parse_sse_id (foundation; happy_simple) → stream_turn (happy_one_text_done) → reconnect_turn (happy_resume_from_seq_3) → cancel_turn (happy_cancel). Each FN's tracer went RED then GREEN before its other tests landed. Shared SSE-iteration logic (INV-002 sse_id presence + INV-003 turn_id stability + terminal-break) lives in private _iter_events helper. expected_turn_id=None gives stream_turn's "establish from first event" semantics; expected_turn_id=N gives reconnect_turn's "first event is already a flip-candidate" semantics — the two-entry-point distinction Volva surfaced during the paraphrase round. A few implementation choices worth recording: - _parse_sse_id uses a `^-?\\d+$` regex pre-check to reject any whitespace before int() is called. Python's `int(" 3 ")` silently strips, which would have made the trailing_whitespace adversarial test pass for the wrong reason. - The connection_drop test uses a custom httpx.AsyncByteStream subclass (_DropAfter) that yields chunks then raises RemoteProtocolError mid-stream. respx alone can't simulate mid-stream HTTP errors. - ToolResult.result and ToolStart.arguments are typed as Any because the server's tool wire shape varies per tool; the spec doesn't pin a generic schema. - Boundary smoke test (no core.* / worldtree.* imports under src/ratatoskr/) still GREEN — INV-005 holds. Also: one E501 line-length fix in test_no_worldtree_imports.py that ruff flagged once the new tests pulled it into scope. --- persistent-memory.md | 14 +- src/ratatoskr/sse_client.py | 413 +++++++++++++++++++ tests/test_no_worldtree_imports.py | 4 +- tests/test_sse_client.py | 625 +++++++++++++++++++++++++++++ 4 files changed, 1049 insertions(+), 7 deletions(-) create mode 100644 src/ratatoskr/sse_client.py create mode 100644 tests/test_sse_client.py diff --git a/persistent-memory.md b/persistent-memory.md index 3dc2d3a..04ea0a0 100644 --- a/persistent-memory.md +++ b/persistent-memory.md @@ -30,8 +30,7 @@ separate dev team rather than an in-tree Worldtree tool. ## Current state / in-flight -**Status: scaffold verified + first contract authored.** Design is locked; -implementation begins next, TDD against `sse_client.contract.md`. +**Status: `ratatoskr.sse_client` implemented via TDD against issue #1's contract.** 38/38 tests GREEN; ruff clean; boundary smoke (`tests/test_no_worldtree_imports.py`) still passes. What's in the repo: - `docs/design-brief.md` — the locked design (copy from `brokkr-smithy/docs/ratatoskr-design-brief.md`). @@ -41,11 +40,12 @@ What's in the repo: - `docs/contracts/issues/1.contract.md` — **issue-scoped contract for issue #1** (https://gitea.phasefinal.com/vh/ratatoskr/issues/1). v2.1, complexity=high. `target_module: ratatoskr.sse_client`. `prd:` block pins to issue body SHA `abcbc49467e86f1d` at `2026-05-21T03:57:37+00:00`. Four FN blocks: `stream_turn`, `reconnect_turn`, `cancel_turn`, `_parse_sse_id`. Drift check (`scripts/contract_drift_check.py`) returns clean. - `pyproject.toml` — Python 3.12, hatchling, uv-managed. Deps: httpx, httpx-sse, textual. Dev deps: pytest, pytest-asyncio, respx, ruff, mypy, textual-dev, pyyaml (consumed by `docs/contracts/contract_parser.py` + `scripts/contract_drift_check.py`). - `src/ratatoskr/__init__.py` + `cli.py` — stubs. +- `src/ratatoskr/sse_client.py` — **implemented 2026-05-21** per `docs/contracts/issues/1.contract.md`. Four public entry points + nine typed Event variants + ten custom exceptions. Shared SSE-iteration logic (INV-002 + INV-003 + terminal-break) lives in private `_iter_events(event_source, *, expected_turn_id)` helper consumed by both `stream_turn` and `reconnect_turn` — `expected_turn_id=None` triggers "establish from first event" semantics, `expected_turn_id=N` triggers "first event is already a flip-candidate" semantics (the two-entry-point distinction Volva surfaced). +- `tests/test_sse_client.py` — 37 tests covering all four FN blocks' TESTS: entries verbatim (13 + 10 + 8 + 6). Real HTTP wire via respx mocks; SSE wire format constructed by helper `_sse_chunk`. Connection-drop test uses custom `httpx.AsyncByteStream` subclass that yields chunks then raises `RemoteProtocolError`. - `tests/test_no_worldtree_imports.py` — boundary smoke test (passes; verified 2026-05-20). - `tests/snapshots/README.md` — recording/replay convention for SSE snapshot tests. What's NOT in the repo yet: -- Implementation of `ratatoskr.sse_client` — next move, TDD per the contract's tracer tests. - Gitea remote — operator provided `git@gitea.phasefinal.com:vh/ratatoskr.git` on 2026-05-20; about to be added + first push at the same commit as this update. - CLAUDE.md customization — currently using the canonical template's generic CLAUDE.md. The dev team may want to add Ratatoskr-specific @@ -54,9 +54,10 @@ What's NOT in the repo yet: **Branch:** `main`. Remote: `origin → git@gitea.phasefinal.com:vh/ratatoskr.git` (added 2026-05-20). **Next natural moves:** -1. Implement `ratatoskr.sse_client` via TDD per `docs/contracts/sse_client.contract.md`. Vertical slice — start with `_parse_sse_id` (the foundation; tracer test `happy_simple`), then `stream_turn` (tracer test `happy_one_text_done`), then `reconnect_turn` (tracer test `happy_resume_from_seq_3`), then `cancel_turn` (tracer test `happy_cancel`). Each FN's tracer test gets RED → GREEN before any other test in that FN. -2. Record real SSE snapshot fixtures from a running Worldtree (per `tests/snapshots/README.md`) once `stream_turn` is GREEN against mocks — gives version-skew detection per design-brief §2. -3. Build presenters: `--send` stdout presenter first (simplest consumer of `stream_turn`); Textual TUI second. Both consume the same `Event` iterator. +1. Record real SSE snapshot fixtures from a running Worldtree (per `tests/snapshots/README.md`). Current tests use respx mocks against hand-rolled SSE wire — recording against a real Worldtree exercises spec conformance and gives version-skew detection per design-brief §2. +2. Build the `--send` stdout presenter — simplest consumer of `stream_turn`, exercises the API path without TUI machinery. Useful as a tracer for `ratatoskr.cli` work. +3. Textual TUI app shell — second presenter; multi-pane observability dashboard per design-brief §5. Layout shape locked there (Horizontal split, left=chat, right=TabbedContent with persona/tools/admin/bifrost/server-log). +4. Run `/sleipnir-preflight 1` if any AFK dispatch is wanted for these follow-ups, but most of this is hands-on dev work. ## Recent decisions @@ -77,6 +78,7 @@ decision. Captures rationale that won't be obvious from code alone. - `[2026-05-20]` **First contract: `ratatoskr.sse_client`.** Bundles `stream_turn` + `reconnect_turn` + `cancel_turn` + private `_parse_sse_id` into one module — the SSE-resume flow is coupled (cancel needs `turn_id` from the SSE wire `id:`, reconnect re-uses the same parsed `SseId`), so they share a contract. Hard invariant INV-002 makes the composite `{turn_id}:{seq}` `id:` parsing load-bearing — closes the foot-gun the design-brief §3 names (hand-rolled `data:`-only parsing silently drops the `id:`). v2.1 test categories `adversarial`/`scenario`/`trace` used freely; parser warns but format spec §2.1.E permits them. - `[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 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 new file mode 100644 index 0000000..59075d8 --- /dev/null +++ b/src/ratatoskr/sse_client.py @@ -0,0 +1,413 @@ +"""SSE consumer for the Worldtree Conversation API. + +Implements docs/contracts/issues/1.contract.md. +""" + +from __future__ import annotations + +import json +import re +from collections.abc import AsyncIterator +from dataclasses import dataclass +from typing import Any, NamedTuple + +import httpx +import httpx_sse + +_INT_RE = re.compile(r"^-?\d+$") + + +class SseId(NamedTuple): + """Parsed composite SSE wire `id:` per spec §SSE id format.""" + + turn_id: int + seq: int + + +@dataclass(frozen=True) +class WorkerPhase: + """SSE event `worker_phase`: agent entered a new processing phase.""" + + sse_id: SseId + phase: str + turn_id: int + + +@dataclass(frozen=True) +class Thinking: + """SSE event `thinking`: incremental thinking content from thinking-enabled models.""" + + sse_id: SseId + content: str + + +@dataclass(frozen=True) +class Text: + """SSE event `text`: an incremental response-text delta.""" + + sse_id: SseId + content: str + + +@dataclass(frozen=True) +class TextBoundary: + """SSE event `text_boundary`: speakable breakpoint after a `text` event.""" + + sse_id: SseId + kind: str + char_offset: int + ts: str + + +@dataclass(frozen=True) +class ToolStart: + """SSE event `tool_start`: agent is about to execute a tool.""" + + sse_id: SseId + name: str + arguments: dict[str, Any] + + +@dataclass(frozen=True) +class ToolResult: + """SSE event `tool_result`: a tool call completed.""" + + sse_id: SseId + name: str + result: Any + duration_ms: int + + +@dataclass(frozen=True) +class Done: + """Terminal SSE event `done`: turn succeeded.""" + + sse_id: SseId + phase: str + response: str + model: str + duration_ms: int + usage: dict[str, int] + + +@dataclass(frozen=True) +class Error: + """Terminal SSE event `error`: turn failed.""" + + sse_id: SseId + phase: str + message: str + error_code: str | None + + +@dataclass(frozen=True) +class Cancelled: + """Terminal SSE event `cancelled`: turn was cancelled server-side.""" + + sse_id: SseId + phase: str + turn_id: int + reason: str | None + partial_message_id: int | None + + +Event = ( + WorkerPhase + | Thinking + | Text + | TextBoundary + | ToolStart + | ToolResult + | Done + | Error + | Cancelled +) + + +class MalformedSseId(Exception): + """Raised when an SSE event's `id:` wire field is missing or non-composite.""" + + def __init__(self, raw: str) -> None: + super().__init__(f"malformed SSE id: {raw[:64]!r}") + self.raw = raw[:64] + + +class TurnIdFlip(Exception): + """Raised when an SSE event's turn_id doesn't match the established/expected turn.""" + + def __init__(self, *, established: int, got: int) -> None: + super().__init__(f"turn_id flip: established={established}, got={got}") + self.established = established + self.got = got + + +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: + super().__init__(f"SSE connect failed: status={status}, body={body[:128]!r}") + self.status = status + self.body = body + + +class SseConnectionDropped(Exception): + """Raised when the HTTP/SSE connection dropped mid-stream.""" + + def __init__(self, *, last_seen_sse_id: SseId | None) -> None: + super().__init__(f"SSE connection dropped; last_seen_sse_id={last_seen_sse_id}") + self.last_seen_sse_id = last_seen_sse_id + + +class InvalidLastEventId(Exception): + """Raised on HTTP 400 from a reconnect request — caller's last_event_id was rejected.""" + + def __init__(self, *, raw: str) -> None: + super().__init__(f"server rejected Last-Event-ID: {raw!r}") + self.raw = raw + + +class ResumeTurnFinished(Exception): + """Raised on HTTP 410 from a reconnect — turn finished while disconnected.""" + + def __init__(self, *, turn_id: int) -> None: + super().__init__(f"resume target finished: turn_id={turn_id}") + self.turn_id = turn_id + + +class ResumeBufferExpired(Exception): + """Raised on HTTP 412 from a reconnect — Last-Event-ID older than replay buffer.""" + + def __init__(self, *, turn_id: int, buffered_from_seq: int) -> None: + super().__init__( + f"replay buffer expired: turn_id={turn_id}, buffered_from_seq={buffered_from_seq}" + ) + self.turn_id = turn_id + self.buffered_from_seq = buffered_from_seq + + +class CancelTurnNotFound(Exception): + """Raised on HTTP 404 from cancel — turn finished before cancel arrived (race).""" + + def __init__(self, *, turn_id: int) -> None: + super().__init__(f"cancel target not found: turn_id={turn_id}") + self.turn_id = turn_id + + +class CancelAlreadyCompleted(Exception): + """Raised on HTTP 409 from cancel — turn already terminated server-side (race).""" + + def __init__(self, *, turn_id: int) -> None: + super().__init__(f"cancel: turn already completed: turn_id={turn_id}") + self.turn_id = turn_id + + +class CancelFailed(Exception): + """Raised on unexpected cancel response status.""" + + def __init__(self, *, status: int, body: bytes) -> None: + super().__init__(f"cancel failed: status={status}, body={body[:128]!r}") + self.status = status + self.body = body + + +@dataclass(frozen=True) +class CancelResult: + """Response envelope from POST /sessions/{id}/turns/{turn_id}/cancel.""" + + turn_id: int + cancelled: bool + reason: str | None + partial_message_id: int | None + + +def _envelope_for_type(body: dict[str, Any], sse_id: SseId) -> Event: + """Dispatch a parsed JSON body to its typed Event variant.""" + t = body["type"] + if t == "text": + return Text(sse_id=sse_id, content=body["content"]) + if t == "worker_phase": + return WorkerPhase(sse_id=sse_id, phase=body["phase"], turn_id=body["turn_id"]) + if t == "thinking": + return Thinking(sse_id=sse_id, content=body["content"]) + if t == "text_boundary": + return TextBoundary( + sse_id=sse_id, + kind=body["kind"], + char_offset=body["char_offset"], + ts=body["ts"], + ) + if t == "tool_start": + return ToolStart(sse_id=sse_id, name=body["name"], arguments=body["arguments"]) + if t == "tool_result": + return ToolResult( + sse_id=sse_id, + name=body["name"], + result=body["result"], + duration_ms=body["duration_ms"], + ) + if t == "done": + return Done( + sse_id=sse_id, + phase=body["phase"], + response=body["response"], + model=body["model"], + duration_ms=body["duration_ms"], + usage=body["usage"], + ) + if t == "error": + return Error( + sse_id=sse_id, + phase=body.get("phase", "failed"), + message=body.get("message", ""), + error_code=body.get("error_code"), + ) + if t == "cancelled": + return Cancelled( + sse_id=sse_id, + phase=body["phase"], + turn_id=body["turn_id"], + reason=body.get("reason"), + partial_message_id=body.get("partial_message_id"), + ) + raise ValueError(f"unknown SSE event type: {t!r}") + + +async def _iter_events( + event_source: httpx_sse.EventSource, + *, + expected_turn_id: int | None, +) -> AsyncIterator[Event]: + """Apply INV-002 (sse_id present + in range) and INV-003 (turn_id stable) per event. + + `expected_turn_id=None` means "establish from the first event" (stream_turn semantics). + `expected_turn_id=N` means "every event must match N" (reconnect_turn semantics — the + first event is already a flip-candidate per INV-003). + """ + established = expected_turn_id + last_sse_id: SseId | None = None + try: + async for sse in event_source.aiter_sse(): + try: + sse_id = _parse_sse_id(sse.id) + except ValueError as exc: + raise MalformedSseId(raw=sse.id) from exc + if established is None: + established = sse_id.turn_id + elif sse_id.turn_id != established: + raise TurnIdFlip(established=established, got=sse_id.turn_id) + event = _envelope_for_type(json.loads(sse.data), sse_id=sse_id) + yield event + last_sse_id = sse_id + if isinstance(event, (Done, Error, Cancelled)): + return + except (httpx.ReadError, httpx.RemoteProtocolError) as exc: + raise SseConnectionDropped(last_seen_sse_id=last_sse_id) from exc + + +async def stream_turn( + client: httpx.AsyncClient, session_id: str, content: str +) -> AsyncIterator[Event]: + """POST a message and yield typed Events. See contract FN stream_turn.""" + assert client is not None + assert session_id and isinstance(session_id, str) + assert content and isinstance(content, str) + + async with httpx_sse.aconnect_sse( + client, + "POST", + f"/sessions/{session_id}/messages", + json={"content": content}, + ) as event_source: + try: + event_source.response.raise_for_status() + except httpx.HTTPStatusError as exc: + body = await exc.response.aread() + raise SseConnectFailed(status=exc.response.status_code, body=body) from exc + async for event in _iter_events(event_source, expected_turn_id=None): + yield event + + +async def reconnect_turn( + client: httpx.AsyncClient, + session_id: str, + content: str, + last_event_id: str, +) -> AsyncIterator[Event]: + """Re-POST with Last-Event-ID to resume. See contract FN reconnect_turn.""" + assert client is not None + assert session_id and isinstance(session_id, str) + assert isinstance(content, str) + expected = _parse_sse_id(last_event_id) + + async with httpx_sse.aconnect_sse( + client, + "POST", + f"/sessions/{session_id}/messages", + json={"content": content}, + headers={"Last-Event-ID": last_event_id}, + ) as event_source: + status = event_source.response.status_code + if status != 200: + body_bytes = await event_source.response.aread() + try: + body = json.loads(body_bytes) + except json.JSONDecodeError: + body = {} + if status == 400: + raise InvalidLastEventId(raw=last_event_id) + if status == 410: + raise ResumeTurnFinished(turn_id=body.get("turn_id", expected.turn_id)) + if status == 412: + raise ResumeBufferExpired( + turn_id=body.get("turn_id", expected.turn_id), + buffered_from_seq=body.get("buffered_from_seq", 0), + ) + raise SseConnectFailed(status=status, body=body_bytes) + async for event in _iter_events(event_source, expected_turn_id=expected.turn_id): + yield event + + +def _parse_sse_id(raw: str) -> SseId: + """Parse the SSE wire `id:` as composite `{turn_id}:{seq}`. See contract FN _parse_sse_id.""" + parts = raw.split(":") + if len(parts) != 2: + raise ValueError(f"expected '{{turn_id}}:{{seq}}', got: {raw[:64]!r}") + turn_id_str, seq_str = parts + if not _INT_RE.match(turn_id_str) or not _INT_RE.match(seq_str): + raise ValueError(f"expected '{{turn_id}}:{{seq}}' with decimal ints, got: {raw[:64]!r}") + turn_id = int(turn_id_str) + seq = int(seq_str) + if turn_id < 1 or seq < 1: + raise ValueError(f"expected both ints >= 1 per spec, got: {raw[:64]!r}") + return SseId(turn_id=turn_id, seq=seq) + + +async def cancel_turn( + client: httpx.AsyncClient, + session_id: str, + turn_id: int, + *, + persist_partial: bool = False, +) -> CancelResult: + """POST /sessions/{id}/turns/{turn_id}/cancel. See contract FN cancel_turn.""" + assert client is not None + assert session_id and isinstance(session_id, str) + assert isinstance(turn_id, int) and turn_id > 0 + + params = {"persist_partial": "true"} if persist_partial else None + resp = await client.post( + f"/sessions/{session_id}/turns/{turn_id}/cancel", params=params + ) + if resp.status_code == 404: + raise CancelTurnNotFound(turn_id=turn_id) + if resp.status_code == 409: + raise CancelAlreadyCompleted(turn_id=turn_id) + if resp.status_code != 200: + raise CancelFailed(status=resp.status_code, body=resp.content) + body = resp.json() + return CancelResult( + turn_id=body["turn_id"], + cancelled=body["cancelled"], + reason=body.get("reason"), + partial_message_id=body.get("partial_message_id"), + ) diff --git a/tests/test_no_worldtree_imports.py b/tests/test_no_worldtree_imports.py index d6749d8..7c04076 100644 --- a/tests/test_no_worldtree_imports.py +++ b/tests/test_no_worldtree_imports.py @@ -36,7 +36,9 @@ def test_no_worldtree_imports() -> None: line = text[: match.start()].count("\n") + 1 violations.append((path, f"line {line}: {match.group(0).strip()}")) if violations: - report = "\n".join(f" {p.relative_to(_SRC_ROOT.parent.parent)}: {v}" for p, v in violations) + report = "\n".join( + f" {p.relative_to(_SRC_ROOT.parent.parent)}: {v}" for p, v in violations + ) raise AssertionError( "Ratatoskr must not import from Worldtree source. Violations:\n" f"{report}\n\n" diff --git a/tests/test_sse_client.py b/tests/test_sse_client.py new file mode 100644 index 0000000..5a27e59 --- /dev/null +++ b/tests/test_sse_client.py @@ -0,0 +1,625 @@ +"""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" + + +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_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': } 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_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)