From 59602fe3ff85af99bca2aa6becaacd94be62bc58 Mon Sep 17 00:00:00 2001 From: Vuong Hoang Date: Sun, 19 Jul 2026 06:35:04 -0700 Subject: [PATCH] refactor(#20): delete the orphaned hand-rolled turn-stream paths (slice-2, part 2b-iii) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit DEC-4 live smoke PASSED first (personal :8081, b127/b128): create → streamed turn that rendered (worker_phase/text/text_boundary/done with usage) → SIGINT cancel that round-tripped to a cancelled terminal. With both CLI + web on the adapter, the hand-rolled turn-stream family is fully orphaned — deleting it now. - sse_client.py (714 → 224): removed stream_turn / reconnect_turn / stream_turn_resilient / cancel_turn + the Event dataclasses (Text/Done/…/Event union) + CancelResult + the SSE parse helpers (_iter_events / _envelope_for_type / _parse_sse_id / _eager_failure_fields / _INT_RE). KEPT: the caller-semantic exceptions (the adapter raises them, DEC-2), SseId, AdminEvent, stream_admin_events (slice-6 admin surface). - sessions.py (677 → 608): removed list_sessions + get_session_tools (no surface users) + SessionPage. KEPT: create_session / get_session_messages (the --seed-first-message probe still uses them, slice-3) + all exceptions + SessionInfo. - tests: test_sse_client pruned to TestStreamAdminEvents; test_sessions dropped the list_sessions + get_session_tools classes. The deleted turn-stream behavior is now covered by test_wt.py + the CLI/web integration tests + the live smoke. Suite 490 green (570 − 80 deleted turn-stream tests); ruff clean on all touched files; no new mypy errors. Patch (internal cleanup; behavior preserved). --- pyproject.toml | 2 +- src/ratatoskr/sessions.py | 69 -- src/ratatoskr/sse_client.py | 491 -------------- tests/test_sessions.py | 268 -------- tests/test_sse_client.py | 1257 +---------------------------------- uv.lock | 2 +- 6 files changed, 10 insertions(+), 2079 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 481cde2..13396f7 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "ratatoskr" -version = "0.21.7" +version = "0.21.8" description = "Worldtree Conversation API debug console (web + headless CLI) — multi-pane observability" readme = "README.md" requires-python = ">=3.12" diff --git a/src/ratatoskr/sessions.py b/src/ratatoskr/sessions.py index 1a40c17..4adcfa3 100644 --- a/src/ratatoskr/sessions.py +++ b/src/ratatoskr/sessions.py @@ -38,12 +38,6 @@ class SessionInfo: config: dict[str, Any] | None = None -@dataclass(frozen=True) -class SessionPage: - """One page of GET /sessions results. `next_cursor=None` on the last page.""" - - items: list[SessionInfo] - next_cursor: str | None @dataclass(frozen=True) @@ -204,53 +198,6 @@ class AuthoredHistoryUnavailable(Exception): self.session_id = session_id -async def list_sessions( - client: httpx.AsyncClient, - *, - include_archived: bool = False, - limit: int = 50, - cursor: str | None = None, -) -> SessionPage: - """GET /sessions. See contract FN list_sessions.""" - assert client is not None - assert 1 <= limit <= 200 - assert cursor is None or (isinstance(cursor, str) and cursor) - - params: dict[str, str] = {"limit": str(limit)} - if include_archived: - params["include_archived"] = "true" - if cursor is not None: - params["cursor"] = cursor - - resp = await client.get("/sessions", params=params) - if resp.status_code == 422: - try: - err = resp.json() - except ValueError: - err = {} - if err.get("error_code") == "cursor_invalid": - raise InvalidCursor(raw=cursor) - raise SessionApiFailed(status=422, body=resp.content) - if resp.status_code != 200: - raise SessionApiFailed(status=resp.status_code, body=resp.content) - body = resp.json() - items = [ - SessionInfo( - session_id=item["session_id"], - agent_id=item["agent_id"], - created_at=item["created_at"], - last_active=item["last_active"], - metadata=item.get("metadata", {}), - message_count=None, - name=item.get("name"), - archived=item.get("archived") or False, - tags=item.get("tags") or [], - kind=item.get("kind"), # INV-002 amendment (#161): present on list items - config=item.get("config"), # forward-compat passthrough; None today - ) - for item in body["items"] - ] - return SessionPage(items=items, next_cursor=body.get("next_cursor")) def endpoint_for_plane(plane: str, base_host: str) -> str: @@ -572,22 +519,6 @@ async def get_session_bifrost( raise SessionApiFailed(status=resp.status_code, body=resp.content) -async def get_session_tools(client: httpx.AsyncClient, session_id: str) -> dict[str, Any]: - """GET /sessions/{session_id}/tools — owner-scoped tool inventory (spec #183). - - Returns the merged tool list the LLM saw at turn-fire: `{agent_id, - builtin_tools: [...], bifrost_tools: [{name, description, parameters}, ...]}`. - Owner-scoped (`ctx.user_id == session.user_id`) — reachable with the consumer - key, NO admin scope. Cross-owner access returns 404 `session_not_found` - (existence-hiding); a revoked session returns 401 `auth_revoked`. Parsed dict - verbatim; any non-200 → SessionApiFailed (mirrors get_persona_state). - """ - assert client is not None - assert session_id and isinstance(session_id, str) - resp = await client.get(f"/sessions/{session_id}/tools") - if resp.status_code == 200: - return resp.json() - raise SessionApiFailed(status=resp.status_code, body=resp.content) async def get_capabilities(client: httpx.AsyncClient) -> dict[str, Any]: diff --git a/src/ratatoskr/sse_client.py b/src/ratatoskr/sse_client.py index b4d44ff..3eef862 100644 --- a/src/ratatoskr/sse_client.py +++ b/src/ratatoskr/sse_client.py @@ -6,7 +6,6 @@ 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 @@ -14,8 +13,6 @@ 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.""" @@ -24,155 +21,6 @@ class SseId(NamedTuple): 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 - - -@dataclass(frozen=True) -class AwaitingLlmFirstToken: - """SSE event `awaiting_llm_first_token`: heartbeat during slow first-token. - - Fires at the configured interval (default 5s) during the gap between - `worker_phase` phase=BuildingPrompt and phase=CallingLLM. Lets clients - render a live "thinking for Ns…" indicator instead of a frozen line - during legitimate-slow first-token latency. Stops the moment CallingLLM - fires (defense-in-depth at three sites); no heartbeat after Cancelled - or stalled terminal events. Tool round-trip re-entries do NOT re-fire - heartbeats — INV-201-5 scopes the mechanism to the FIRST gap only. - - `elapsed_ms_since_building_prompt` is server-authoritative - `time.monotonic()`-based — independent of network latency or clock - skew, monotonically increasing across the heartbeat sequence. - - See docs/conversation-api-spec.md § awaiting_llm_first_token - (Worldtree #201, v0.29.0). - """ - - sse_id: SseId - turn_id: int - elapsed_ms_since_building_prompt: float - - -@dataclass(frozen=True) -class AffectUpdate: - """SSE event `affect_update`: persona-state observability snapshot. - - Two emissions per qualifying turn (persona-enabled agent on non- - ephemeral session): `status="current"` at turn start carrying the full - snapshot, `status="scheduled"` after post-turn appraisal kicks off - (lightweight — `snapshot` is None). Suppressed entirely for persona- - disabled agents (e.g. `domari`, `muninn`), Tier 3 consumer-defined - agents (Phase 2.0), and ephemeral sessions. - - Bootstrap reads available via `GET /agents/{agent_id}/persona_state` - (same `snapshot` shape, requires `persona.read` scope). - - See docs/conversation-api-spec.md § affect_update (Worldtree #204, - v0.28.0). - """ - - sse_id: SseId - status: str # "current" | "scheduled" - turn_id: int - snapshot: dict[str, Any] | None # None when status="scheduled" - - -Event = ( - WorkerPhase - | Thinking - | Text - | TextBoundary - | ToolStart - | ToolResult - | Done - | Error - | Cancelled - | AffectUpdate - | AwaitingLlmFirstToken -) @dataclass(frozen=True) @@ -258,31 +106,6 @@ class TurnLaunchUnavailable(SseConnectFailed): self.message = message -# Canonical error_codes (Worldtree #331 / v1.0.0b2): 409 -> agent_not_available, -# 503 -> not_ready (retryable; re-pinned from internal_error). Used only as a -# fallback default when the body omits error_code — the real code is surfaced -# verbatim from the {detail:{error_code,message}} envelope. -_EAGER_TURN_FAILURE_CODE = {409: "agent_not_available", 503: "not_ready"} - - -def _eager_failure_fields(body: bytes, status: int) -> tuple[str, str]: - """Extract (error_code, message) from an eager turn-launch failure body - (#331). Accepts the Worldtree `{"detail": {...}}` envelope OR a flat - `{error_code, message}`; falls back to a status-derived default code and a - generic message when the body is absent / non-JSON / malformed.""" - try: - parsed: Any = json.loads(body) - except (json.JSONDecodeError, ValueError): - parsed = None - src: dict[str, Any] = {} - if isinstance(parsed, dict): - detail = parsed.get("detail") - src = detail if isinstance(detail, dict) else parsed - code = src.get("error_code") or _EAGER_TURN_FAILURE_CODE[status] - message = src.get("message") - if not isinstance(message, str): - message = f"turn launch failed (HTTP {status})" - return str(code), message class SseConnectionDropped(Exception): @@ -352,275 +175,6 @@ class CancelFailed(Exception): 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"), - ) - if t == "awaiting_llm_first_token": - # Worldtree #201 / v0.29.0: top-level heartbeat during BuildingPrompt - # → CallingLLM gap. Lets clients render live elapsed-time indicators - # instead of frozen lines on legitimate-slow first-token latency. - return AwaitingLlmFirstToken( - sse_id=sse_id, - turn_id=body["turn_id"], - elapsed_ms_since_building_prompt=body["elapsed_ms_since_building_prompt"], - ) - if t == "affect_update": - # Worldtree #204 / v0.28.0: persona-state observability event. - # status="current" carries full snapshot at turn start; - # status="scheduled" omits snapshot (lightweight post-appraisal- - # kickoff notification). - return AffectUpdate( - sse_id=sse_id, - status=body["status"], - turn_id=body["turn_id"], - snapshot=body.get("snapshot"), - ) - 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 - terminal_seen = False - try: - async for sse in event_source.aiter_sse(): - # Issue #7 INV-001: empty-data frames are keepalives — skip silently. - # ORDERING: this branch fires BEFORE _parse_sse_id; an empty-data event - # with a malformed id is silently swallowed (intentional — a keepalive - # with a bad id is still a keepalive). Don't reorder. - if sse.data == "": - continue - # v0.8.1: empty-id frames are also treated as keepalives. Worldtree - # SOMETIMES emits events without an `id:` line (observed mid-stream - # on the qwen3.6-35-a3b-heretic provider, 2026-05-25). Per the SSE - # RFC, events without ids are legitimate (they just don't update - # Last-Event-ID); the previous strict behavior crashed every turn - # on the offending agent. Treat same as empty-data: skip silently. - if sse.id == "": - continue - 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) - try: - body = json.loads(sse.data) - except json.JSONDecodeError as exc: - raise MalformedSseData(raw=sse.data) from exc - event = _envelope_for_type(body, sse_id=sse_id) - yield event - last_sse_id = sse_id - if isinstance(event, (Done, Error, Cancelled)): - terminal_seen = True - return - except (httpx.ReadError, httpx.RemoteProtocolError, httpx.ReadTimeout) as exc: - # ReadTimeout covers idle gaps that exceed httpx's read timeout — the SSE - # stream went quiet long enough for httpx to give up. Treat the same as a - # raw read error: surface as SseConnectionDropped so the caller can decide - # whether to reconnect_turn. (Callers SHOULD configure a long-or-disabled - # read timeout on their AsyncClient for SSE; this is defense in depth.) - 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( - 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: - # Worldtree v1.0.0b1 (#331): turn-launch failures arrive EAGERLY as a - # status before any stream — 409 agent_not_available (pre-b1 this was a - # 200 + in-stream `error` event), 503 a transient retryable launch - # failure. Surface them as typed SseConnectFailed subclasses carrying - # error_code; request-level non-2xx (404 session_not_found, etc.) stay - # generic SseConnectFailed. - status = event_source.response.status_code - if status in (409, 503): - body = await event_source.response.aread() - code, message = _eager_failure_fields(body, status) - if status == 409: - raise AgentNotAvailable(body=body, error_code=code, message=message) - raise TurnLaunchUnavailable(body=body, error_code=code, message=message) - 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 - - -async def stream_turn_resilient( - client: httpx.AsyncClient, - session_id: str, - content: str, - *, - max_reconnects: int = 5, -) -> AsyncIterator[Event]: - """Resume-orchestration wrapper over stream_turn + reconnect_turn. - - Yields ONE continuous Event stream; on `SseConnectionDropped` (mid-stream - drop or clean EOF before a terminal), resumes from the last-seen `sse_id` - via `reconnect_turn`, up to `max_reconnects` times, until a terminal - Done/Error/Cancelled arrives. The single shared surface presenters consume - for resilient streaming (design-brief §8b: "share the consumer, branch the - presenter"). Cross-process resume stays deferred to v2 (§8d): `last_seen` - lives only in this generator's frame. See contract FN stream_turn_resilient - (amendment 2026-06-30). - """ - assert client is not None - assert session_id and isinstance(session_id, str) - assert content and isinstance(content, str) - assert isinstance(max_reconnects, int) and max_reconnects >= 0 - - last_seen: SseId | None = None - reconnects = 0 - gen = stream_turn(client, session_id, content) - while True: - try: - async for event in gen: - last_seen = event.sse_id - yield event - return # generator completed cleanly → terminal event reached (INV-001) - except SseConnectionDropped as drop: - # Prefer the id we tracked from a yielded event; fall back to the one - # the drop carries (covers a drop on the very first frame). Non-drop - # reconnect failures (412/410/400/flip) are NOT caught here — they - # propagate per the contract's "surface, not recover" policy. - seen = last_seen or drop.last_seen_sse_id - if seen is None or reconnects >= max_reconnects: - raise - reconnects += 1 - gen = reconnect_turn( - client, - session_id, - content, - # A str cursor is already the composite id; an SseId is formatted. - last_event_id=seen if isinstance(seen, str) else f"{seen.turn_id}:{seen.seq}", - ) - async def stream_admin_events( client: httpx.AsyncClient, @@ -667,48 +221,3 @@ async def stream_admin_events( raise SseConnectionDropped(last_seen_sse_id=None) from exc -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}") - 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_sessions.py b/tests/test_sessions.py index 0f96650..86e0cb2 100644 --- a/tests/test_sessions.py +++ b/tests/test_sessions.py @@ -13,10 +13,8 @@ from ratatoskr.sessions import ( BifrostBinding, BifrostConsumerKeyMissing, BifrostHandshakeFailed, - InvalidCursor, PersonaNotConfigured, SessionApiFailed, - SessionPage, create_character, create_session, delete_character, @@ -27,10 +25,8 @@ from ratatoskr.sessions import ( get_persona_state, get_session_bifrost, get_session_messages, - get_session_tools, list_agents, list_character_models, - list_sessions, set_persona_state, write_authored_history, ) @@ -551,223 +547,6 @@ class TestEndpointForPlane: endpoint_for_plane("persona", "10.100.10.50") -def _list_item( - *, - session_id: str = "s1", - agent_id: str = "mimir", - created_at: str = "2026-04-15T12:00:00+00:00", - last_active: str = "2026-04-15T12:05:00+00:00", - metadata: dict[str, object] | None = None, - name: str | None = "Research session", - archived: bool = False, - tags: list[str] | None = None, -) -> dict[str, object]: - """Build a GET /sessions list-item body for tests.""" - item: dict[str, object] = { - "session_id": session_id, - "agent_id": agent_id, - "created_at": created_at, - "last_active": last_active, - "metadata": metadata if metadata is not None else {}, - "name": name, - "archived": archived, - "tags": tags if tags is not None else ["work"], - } - return item - - -class TestListSessions: - @respx.mock - async def test_explicit_null_list_defaults(self) -> None: - """INV-002: explicit-null archived -> False; explicit-null tags -> [].""" - raw_item = { - "session_id": "s1", - "agent_id": "mimir", - "created_at": "2026-04-15T12:00:00+00:00", - "last_active": "2026-04-15T12:05:00+00:00", - "metadata": {}, - "name": None, - "archived": None, - "tags": None, - } - respx.get("https://w.example/sessions").mock( - return_value=httpx.Response( - 200, json={"items": [raw_item], "next_cursor": None} - ) - ) - async with httpx.AsyncClient(base_url="https://w.example") as client: - page = await list_sessions(client) - info = page.items[0] - assert info.archived is False, "explicit-null archived must default to False" - assert info.tags == [], "explicit-null tags must default to []" - assert info.name is None - - @respx.mock - async def test_happy_first_page(self) -> None: - """happy_first_page [happy,tracer]: one item + next_cursor -> SessionPage shape.""" - respx.get("https://w.example/sessions").mock( - return_value=httpx.Response( - 200, - json={ - "items": [_list_item()], - "next_cursor": "v1.eyJhYmMifQ", - }, - ) - ) - async with httpx.AsyncClient(base_url="https://w.example") as client: - page = await list_sessions(client) - assert isinstance(page, SessionPage) - assert len(page.items) == 1 - assert page.next_cursor == "v1.eyJhYmMifQ" - info = page.items[0] - assert info.session_id == "s1" - assert info.message_count is None # INV-002: not in list response - assert info.name == "Research session" - assert info.archived is False - assert info.tags == ["work"] - - @respx.mock - async def test_happy_last_page(self) -> None: - """happy_last_page: next_cursor=null -> SessionPage(next_cursor=None).""" - respx.get("https://w.example/sessions").mock( - return_value=httpx.Response( - 200, - json={"items": [_list_item()], "next_cursor": None}, - ) - ) - async with httpx.AsyncClient(base_url="https://w.example") as client: - page = await list_sessions(client) - assert page.next_cursor is None - - @respx.mock - async def test_empty_results(self) -> None: - """empty_results: {items: [], next_cursor: null} -> SessionPage([], None).""" - respx.get("https://w.example/sessions").mock( - return_value=httpx.Response(200, json={"items": [], "next_cursor": None}) - ) - async with httpx.AsyncClient(base_url="https://w.example") as client: - page = await list_sessions(client) - assert page == SessionPage(items=[], next_cursor=None) - - @respx.mock - async def test_include_archived_query(self) -> None: - """include_archived_query: True -> has param; default -> NO param at all.""" - route = respx.get("https://w.example/sessions").mock( - return_value=httpx.Response( - 200, json={"items": [], "next_cursor": None} - ) - ) - async with httpx.AsyncClient(base_url="https://w.example") as client: - await list_sessions(client, include_archived=True) - await list_sessions(client) # default - url_with = str(route.calls[0].request.url) - url_default = str(route.calls[1].request.url) - assert "include_archived=true" in url_with - assert "include_archived" not in url_default - - @respx.mock - async def test_cursor_threaded(self) -> None: - """cursor_threaded: cursor=opaque -> URL has cursor=opaque.""" - route = respx.get("https://w.example/sessions").mock( - return_value=httpx.Response( - 200, json={"items": [], "next_cursor": None} - ) - ) - async with httpx.AsyncClient(base_url="https://w.example") as client: - await list_sessions(client, cursor="opaque-from-prev-page") - assert "cursor=opaque-from-prev-page" in str(route.calls[0].request.url) - - @respx.mock - async def test_limit_query(self) -> None: - """limit_query: limit=10 -> URL has limit=10.""" - route = respx.get("https://w.example/sessions").mock( - return_value=httpx.Response( - 200, json={"items": [], "next_cursor": None} - ) - ) - async with httpx.AsyncClient(base_url="https://w.example") as client: - await list_sessions(client, limit=10) - assert "limit=10" in str(route.calls[0].request.url) - - @respx.mock - async def test_invalid_cursor_server(self) -> None: - """invalid_cursor_server: 422 cursor_invalid -> InvalidCursor(raw=).""" - respx.get("https://w.example/sessions").mock( - return_value=httpx.Response( - 422, - json={"error_code": "cursor_invalid", "message": "bad cursor"}, - ) - ) - async with httpx.AsyncClient(base_url="https://w.example") as client: - with pytest.raises(InvalidCursor) as exc_info: - await list_sessions(client, cursor="bogus") - assert exc_info.value.raw == "bogus" - - @respx.mock - async def test_other_validation_failed(self) -> None: - """other_validation_failed: 422 other error_code -> SessionApiFailed(422); truncated.""" - respx.get("https://w.example/sessions").mock( - return_value=httpx.Response( - 422, - json={"error_code": "validation_failed", "message": "limit out of range"}, - ) - ) - async with httpx.AsyncClient(base_url="https://w.example") as client: - with pytest.raises(SessionApiFailed) as exc_info: - await list_sessions(client) - assert exc_info.value.status == 422 - assert len(exc_info.value.body) <= 1024 - - @respx.mock - async def test_unexpected_status_truncates(self) -> None: - """unexpected_status_truncates: 500 + 5000-byte body -> SessionApiFailed; body == 1024.""" - big = b"x" * 5000 - respx.get("https://w.example/sessions").mock( - return_value=httpx.Response(500, content=big) - ) - async with httpx.AsyncClient(base_url="https://w.example") as client: - with pytest.raises(SessionApiFailed) as exc_info: - await list_sessions(client) - assert exc_info.value.status == 500 - assert exc_info.value.body == big[:1024] - - @respx.mock - async def test_limit_below_one(self) -> None: - """limit_below_one [adversarial]: limit=0 -> AssertionError; no HTTP.""" - route = respx.get("https://w.example/sessions").mock( - return_value=httpx.Response(200, json={"items": [], "next_cursor": None}) - ) - async with httpx.AsyncClient(base_url="https://w.example") as client: - with pytest.raises(AssertionError): - await list_sessions(client, limit=0) - assert route.call_count == 0 - - @respx.mock - async def test_limit_above_max(self) -> None: - """limit_above_max [adversarial]: limit=300 -> AssertionError; no HTTP.""" - route = respx.get("https://w.example/sessions").mock( - return_value=httpx.Response(200, json={"items": [], "next_cursor": None}) - ) - async with httpx.AsyncClient(base_url="https://w.example") as client: - with pytest.raises(AssertionError): - await list_sessions(client, limit=300) - assert route.call_count == 0 - - @respx.mock - async def test_empty_cursor(self) -> None: - """empty_cursor [adversarial]: cursor='' -> AssertionError; no HTTP.""" - route = respx.get("https://w.example/sessions").mock( - return_value=httpx.Response(200, json={"items": [], "next_cursor": None}) - ) - async with httpx.AsyncClient(base_url="https://w.example") as client: - with pytest.raises(AssertionError): - await list_sessions(client, cursor="") - assert route.call_count == 0 - - -# ---- Issue #8: list_agents + AgentInfo -------------------------------------- - - class TestListAgents: @respx.mock async def test_happy_full_shape(self) -> None: @@ -1145,53 +924,6 @@ class TestGetCapabilities: assert exc.value.status == 500 -class TestGetSessionTools: - """docs/contracts/issues/2.contract.md — get_session_tools (GET /sessions/{id}/tools, #183).""" - - @respx.mock - async def test_happy(self) -> None: - """happy [happy,tracer]: 200 → merged tool inventory dict verbatim.""" - respx.get("https://w.example/sessions/s1/tools").mock( - return_value=httpx.Response( - 200, - json={ - "agent_id": "alice:wizard", - "builtin_tools": [], - "bifrost_tools": [ - {"name": "bifrost.alice.set_field", "description": "d", "parameters": {}} - ], - }, - ) - ) - async with httpx.AsyncClient(base_url="https://w.example") as client: - tools = await get_session_tools(client, "s1") - assert tools["agent_id"] == "alice:wizard" - assert tools["builtin_tools"] == [] - assert tools["bifrost_tools"][0]["name"] == "bifrost.alice.set_field" - - @respx.mock - async def test_cross_owner_404_raises(self) -> None: - """cross_owner_404 [error]: 404 session_not_found → SessionApiFailed(404).""" - respx.get("https://w.example/sessions/s1/tools").mock( - return_value=httpx.Response(404, json={"error_code": "session_not_found"}) - ) - async with httpx.AsyncClient(base_url="https://w.example") as client: - with pytest.raises(SessionApiFailed) as exc: - await get_session_tools(client, "s1") - assert exc.value.status == 404 - - @respx.mock - async def test_empty_session_id_asserts(self) -> None: - """empty_session_id [adversarial]: '' → AssertionError; no HTTP issued.""" - route = respx.get("https://w.example/sessions//tools").mock( - return_value=httpx.Response(200, json={}) - ) - async with httpx.AsyncClient(base_url="https://w.example") as client: - with pytest.raises(AssertionError): - await get_session_tools(client, "") - assert route.call_count == 0 - - class TestGetSessionBifrost: """#2 contract — get_session_bifrost (GET /admin/sessions/{id}/bifrost, #176).""" diff --git a/tests/test_sse_client.py b/tests/test_sse_client.py index 7a1f652..8624e9d 100644 --- a/tests/test_sse_client.py +++ b/tests/test_sse_client.py @@ -1,4 +1,8 @@ -"""Tests for ratatoskr.sse_client per docs/contracts/issues/1.contract.md.""" +"""Tests for ratatoskr.sse_client — the admin-events stream (#11). + +The turn-stream + Event-model tests retired with the worldtree-sdk cutover (#20); +the turn path is now covered by tests/test_wt.py + the CLI/web integration tests. +This module keeps the still-hand-rolled admin-events surface (slice-6).""" import httpx import pytest @@ -6,47 +10,10 @@ import respx from ratatoskr.sse_client import ( AdminEvent, - AffectUpdate, - AgentNotAvailable, - AwaitingLlmFirstToken, - CancelAlreadyCompleted, - Cancelled, - CancelResult, - CancelTurnNotFound, - Done, - Error, - InvalidLastEventId, - MalformedSseId, - ResumeBufferExpired, - ResumeTurnFinished, SseConnectFailed, - SseConnectionDropped, - SseId, - Text, - TurnIdFlip, - TurnLaunchUnavailable, - _parse_sse_id, - cancel_turn, - reconnect_turn, stream_admin_events, - stream_turn, - stream_turn_resilient, ) -_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.""" @@ -55,1216 +22,6 @@ def _sse_chunk(sse_id: str, body: dict[str, object]) -> bytes: return f"id: {sse_id}\ndata: {json.dumps(body)}\n\n".encode() -_EVENT_STREAM = {"content-type": "text/event-stream"} - - -class _DropStream(httpx.AsyncByteStream): - """Yield the given chunks, then raise a mid-stream drop (RemoteProtocolError). - - Mirrors the inline `_DropAfter` used by TestStreamTurn.test_connection_drop; - hoisted to module scope because the resilient-wrapper tests reuse it. - """ - - 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 - - -def _drop_response(chunks: list[bytes]) -> httpx.Response: - return httpx.Response(200, headers=_EVENT_STREAM, stream=_DropStream(chunks)) - - -def _stream_response(content: bytes) -> httpx.Response: - return httpx.Response(200, headers=_EVENT_STREAM, content=content) - - -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 - # 500 (not 409/503 — those are now eager turn-launch carve-outs, #331). - respx.post("https://w.example/sessions/s1/messages").mock( - return_value=httpx.Response(500, 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_eager_409_agent_not_available(self) -> None: - """b1 #331: eager 409 -> AgentNotAvailable (SseConnectFailed subclass) with - typed error_code; the turn never streams.""" - respx.post("https://w.example/sessions/s1/messages").mock( - return_value=httpx.Response( - 409, - json={ - "detail": { - "error_code": "agent_not_available", - "message": "agent ratatoskr:sindra is unavailable", - } - }, - ) - ) - async with httpx.AsyncClient(base_url="https://w.example") as client: - with pytest.raises(AgentNotAvailable) as exc: - _ = [e async for e in stream_turn(client, "s1", "hi")] - assert exc.value.status == 409 - assert exc.value.error_code == "agent_not_available" - assert "unavailable" in exc.value.message - assert isinstance(exc.value, SseConnectFailed) # existing handlers still catch - - @respx.mock - async def test_eager_503_turn_launch_unavailable_retryable(self) -> None: - """b1 #331: eager 503 -> TurnLaunchUnavailable (retryable, SseConnectFailed subclass).""" - respx.post("https://w.example/sessions/s1/messages").mock( - return_value=httpx.Response( - 503, - json={"error_code": "turn_launch_failed", "message": "resource exhausted"}, - ) - ) - async with httpx.AsyncClient(base_url="https://w.example") as client: - with pytest.raises(TurnLaunchUnavailable) as exc: - _ = [e async for e in stream_turn(client, "s1", "hi")] - assert exc.value.status == 503 - assert exc.value.retryable is True - assert exc.value.error_code == "turn_launch_failed" - assert isinstance(exc.value, SseConnectFailed) - - @respx.mock - async def test_eager_409_non_json_body_defaults(self) -> None: - """b1 #331: eager 409 with a non-JSON body -> AgentNotAvailable with the - status-derived default error_code.""" - respx.post("https://w.example/sessions/s1/messages").mock( - return_value=httpx.Response(409, content=b"nope") - ) - async with httpx.AsyncClient(base_url="https://w.example") as client: - with pytest.raises(AgentNotAvailable) as exc: - _ = [e async for e in stream_turn(client, "s1", "hi")] - assert exc.value.error_code == "agent_not_available" - - @respx.mock - async def test_eager_503_non_json_body_defaults_not_ready(self) -> None: - """b2: eager 503 with a non-JSON body -> TurnLaunchUnavailable with the - canonical default error_code `not_ready`.""" - respx.post("https://w.example/sessions/s1/messages").mock( - return_value=httpx.Response(503, content=b"nope") - ) - async with httpx.AsyncClient(base_url="https://w.example") as client: - with pytest.raises(TurnLaunchUnavailable) as exc: - _ = [e async for e in stream_turn(client, "s1", "hi")] - assert exc.value.error_code == "not_ready" - assert exc.value.retryable is True - - @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_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) - - -class TestAffectUpdate: - """Worldtree #204 / v0.28.0 — persona-state observability SSE event. - - Two emissions per qualifying turn (persona-enabled agent, non-ephemeral - session): `status: "current"` at turn start with full snapshot, then - `status: "scheduled"` near turn end (lightweight, no snapshot). - - See docs/conversation-api-spec.md § affect_update. - """ - - @respx.mock - async def test_current_status_parsed_with_snapshot(self) -> None: - """current_status_parsed_with_snapshot [tracer]: status=current carries - the full snapshot dict; AffectUpdate.snapshot is populated with the - nested PAD / dominant_emotion / emotions_active fields. - """ - snapshot = { - "agent_id": "mimir", - "pad": {"pleasure": 0.52, "arousal": 0.47, "dominance": 0.50}, - "dominant_emotion": "curiosity", - "emotions_active": [ - {"type": "curiosity", "intensity": 0.6, "decay_remaining_s": 202.7} - ], - "baseline_pad": {"pleasure": 0.50, "arousal": 0.40, "dominance": 0.50}, - "mood_drift": {"valence_delta": 0.02, "arousal_delta": 0.07}, - "last_updated_at": "2026-05-25T22:30:18+00:00", - } - stream = _sse_chunk( - "42:1", - { - "type": "affect_update", - "status": "current", - "turn_id": 42, - "snapshot": snapshot, - }, - ) + _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")] - affect = events[0] - assert isinstance(affect, AffectUpdate) - assert affect.status == "current" - assert affect.turn_id == 42 - assert affect.snapshot == snapshot - assert affect.sse_id == SseId(42, 1) - - @respx.mock - async def test_scheduled_status_parsed_no_snapshot(self) -> None: - """scheduled_status_parsed_no_snapshot [trace]: status=scheduled carries - no snapshot field; AffectUpdate.snapshot is None. - """ - stream = ( - _sse_chunk("42:1", {"type": "text", "content": "x"}) - + _sse_chunk( - "42:2", - {"type": "affect_update", "status": "scheduled", "turn_id": 42}, - ) - + _sse_chunk("42:3", _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")] - affect = next(e for e in events if isinstance(e, AffectUpdate)) - assert affect.status == "scheduled" - assert affect.turn_id == 42 - assert affect.snapshot is None - assert affect.sse_id == SseId(42, 2) - - -class TestAwaitingLlmFirstToken: - """Worldtree #201 / v0.29.0 — `awaiting_llm_first_token` SSE heartbeat. - - Top-level event (not a worker_phase extension) fired during the - BuildingPrompt → CallingLLM gap at the configured interval (default - 5s). Server-authoritative elapsed_ms is time.monotonic()-based and - monotonically increasing across the heartbeat sequence. - - See docs/conversation-api-spec.md § awaiting_llm_first_token. - """ - - @respx.mock - async def test_single_heartbeat_parsed(self) -> None: - """single_heartbeat_parsed [tracer]: type=awaiting_llm_first_token → - AwaitingLlmFirstToken(turn_id, elapsed_ms_since_building_prompt). - """ - stream = _sse_chunk( - "42:1", - { - "type": "awaiting_llm_first_token", - "turn_id": 42, - "elapsed_ms_since_building_prompt": 5012.3, - }, - ) + _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")] - beat = events[0] - assert isinstance(beat, AwaitingLlmFirstToken) - assert beat.turn_id == 42 - assert beat.elapsed_ms_since_building_prompt == 5012.3 - assert beat.sse_id == SseId(42, 1) - - @respx.mock - async def test_heartbeat_sequence_monotonic(self) -> None: - """heartbeat_sequence_monotonic [scenario]: three consecutive heartbeats - in one turn — elapsed_ms_since_building_prompt monotonically increases, - all carry the same turn_id. - """ - stream = ( - _sse_chunk( - "42:1", - { - "type": "awaiting_llm_first_token", - "turn_id": 42, - "elapsed_ms_since_building_prompt": 5000.0, - }, - ) - + _sse_chunk( - "42:2", - { - "type": "awaiting_llm_first_token", - "turn_id": 42, - "elapsed_ms_since_building_prompt": 10005.4, - }, - ) - + _sse_chunk( - "42:3", - { - "type": "awaiting_llm_first_token", - "turn_id": 42, - "elapsed_ms_since_building_prompt": 15011.8, - }, - ) - + _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")] - beats = [e for e in events if isinstance(e, AwaitingLlmFirstToken)] - assert len(beats) == 3 - elapsed = [b.elapsed_ms_since_building_prompt for b in beats] - assert elapsed == sorted(elapsed) # monotonically increasing - assert all(b.turn_id == 42 for b in beats) - - -_URL = "https://w.example/sessions/s1/messages" - - -class TestStreamTurnResilient: - """docs/contracts/issues/1.contract.md FN stream_turn_resilient (amendment 2026-06-30).""" - - @respx.mock - async def test_happy_no_drop(self) -> None: - """happy_no_drop [happy]: clean stream passes through; no reconnect issued.""" - stream = _sse_chunk("42:1", {"type": "text", "content": "a"}) + _sse_chunk( - "42:2", _DONE_42_6 - ) - route = respx.post(_URL).mock(return_value=_stream_response(stream)) - async with httpx.AsyncClient(base_url="https://w.example") as client: - events = [e async for e in stream_turn_resilient(client, "s1", "hi")] - assert [e.sse_id for e in events] == [SseId(42, 1), SseId(42, 2)] - assert isinstance(events[-1], Done) - assert route.call_count == 1 # POST-001: no reconnect on a clean stream - - @respx.mock - async def test_resume_after_one_drop(self) -> None: - """resume_after_one_drop [tracer]: a mid-stream drop resumes via reconnect; one stream.""" - first = _drop_response([_sse_chunk("42:1", {"type": "text", "content": "a"})]) - resume = _stream_response( - _sse_chunk("42:2", {"type": "text", "content": "b"}) - + _sse_chunk("42:3", _DONE_42_6) - ) - route = respx.post(_URL).mock(side_effect=[first, resume]) - async with httpx.AsyncClient(base_url="https://w.example") as client: - events = [e async for e in stream_turn_resilient(client, "s1", "hi")] - assert [e.sse_id for e in events] == [SseId(42, 1), SseId(42, 2), SseId(42, 3)] - assert isinstance(events[-1], Done) - assert route.call_count == 2 - # POST-003: reconnect carries the last yielded pre-drop event's id. - assert route.calls[1].request.headers.get("Last-Event-ID") == "42:1" - # PRE/wire: first attempt does NOT carry a Last-Event-ID. - assert route.calls[0].request.headers.get("Last-Event-ID") is None - - @respx.mock - async def test_resume_after_clean_eof(self) -> None: - """resume_after_clean_eof: a clean EOF before terminal also triggers resume (INV-001).""" - first = _stream_response(_sse_chunk("42:1", {"type": "text", "content": "a"})) - resume = _stream_response(_sse_chunk("42:2", _DONE_42_6)) - route = respx.post(_URL).mock(side_effect=[first, resume]) - async with httpx.AsyncClient(base_url="https://w.example") as client: - events = [e async for e in stream_turn_resilient(client, "s1", "hi")] - assert [e.sse_id for e in events] == [SseId(42, 1), SseId(42, 2)] - assert isinstance(events[-1], Done) - assert route.calls[1].request.headers.get("Last-Event-ID") == "42:1" - - @respx.mock - async def test_two_drops_then_done(self) -> None: - """two_drops_then_done: two transient drops, third attempt completes; ids thread through.""" - a1 = _drop_response([_sse_chunk("42:1", {"type": "text", "content": "a"})]) - a2 = _drop_response([_sse_chunk("42:2", {"type": "text", "content": "b"})]) - a3 = _stream_response(_sse_chunk("42:3", _DONE_42_6)) - route = respx.post(_URL).mock(side_effect=[a1, a2, a3]) - async with httpx.AsyncClient(base_url="https://w.example") as client: - events = [e async for e in stream_turn_resilient(client, "s1", "hi")] - assert [e.sse_id for e in events] == [SseId(42, 1), SseId(42, 2), SseId(42, 3)] - assert route.call_count == 3 - assert route.calls[1].request.headers.get("Last-Event-ID") == "42:1" - assert route.calls[2].request.headers.get("Last-Event-ID") == "42:2" - - @respx.mock - async def test_unresumable_zero_event_drop(self) -> None: - """unresumable_zero_event_drop [adversarial]: drop before any event → propagate.""" - route = respx.post(_URL).mock(side_effect=[_drop_response([])]) - async with httpx.AsyncClient(base_url="https://w.example") as client: - with pytest.raises(SseConnectionDropped): - _ = [e async for e in stream_turn_resilient(client, "s1", "hi")] - assert route.call_count == 1 # no id to resume from → no reconnect - - @respx.mock - async def test_max_reconnects_exhausted(self) -> None: - """max_reconnects_exhausted [adversarial]: every attempt drops; budget caps reconnects.""" - side = [ - _drop_response([_sse_chunk(f"42:{n}", {"type": "text", "content": "x"})]) - for n in (1, 2, 3) - ] - route = respx.post(_URL).mock(side_effect=side) - async with httpx.AsyncClient(base_url="https://w.example") as client: - with pytest.raises(SseConnectionDropped): - _ = [ - e - async for e in stream_turn_resilient( - client, "s1", "hi", max_reconnects=2 - ) - ] - assert route.call_count == 3 # initial + 2 reconnects, then give up - - @respx.mock - async def test_zero_budget_no_resume(self) -> None: - """zero_budget_no_resume [adversarial]: max_reconnects=0 → first drop propagates.""" - first = _drop_response([_sse_chunk("42:1", {"type": "text", "content": "a"})]) - route = respx.post(_URL).mock(side_effect=[first]) - async with httpx.AsyncClient(base_url="https://w.example") as client: - with pytest.raises(SseConnectionDropped): - _ = [ - e - async for e in stream_turn_resilient( - client, "s1", "hi", max_reconnects=0 - ) - ] - assert route.call_count == 1 - - @respx.mock - async def test_buffer_expired_propagates(self) -> None: - """buffer_expired_propagates [error]: a 412 on reconnect surfaces, not retried.""" - first = _drop_response([_sse_chunk("42:1", {"type": "text", "content": "a"})]) - resume = httpx.Response(412, json={"turn_id": 42, "buffered_from_seq": 7}) - route = respx.post(_URL).mock(side_effect=[first, resume]) - async with httpx.AsyncClient(base_url="https://w.example") as client: - collected: list[object] = [] - with pytest.raises(ResumeBufferExpired): - async for e in stream_turn_resilient(client, "s1", "hi"): - collected.append(e) - assert [e.sse_id for e in collected] == [SseId(42, 1)] # type: ignore[attr-defined] - assert route.call_count == 2 - - class TestStreamAdminEvents: """docs/conversation-api-spec.md § Admin Event Stream — stream_admin_events (#11).""" @@ -1324,7 +81,9 @@ class TestStreamAdminEvents: """skips_malformed [adversarial]: a bad-JSON frame is skipped, not fatal.""" good = _sse_chunk("41", {"id": 41, "type": "session.created", "data": {"session_id": "s1"}}) bad = b"id: 42\ndata: not-json\n\n" - good2 = _sse_chunk("43", {"id": 43, "type": "session.deleted", "data": {"session_id": "s1"}}) + good2 = _sse_chunk( + "43", {"id": 43, "type": "session.deleted", "data": {"session_id": "s1"}} + ) respx.get("https://w.example/admin/events").mock( return_value=httpx.Response( 200, headers={"content-type": "text/event-stream"}, content=good + bad + good2 diff --git a/uv.lock b/uv.lock index 8584432..3ab7e6d 100644 --- a/uv.lock +++ b/uv.lock @@ -472,7 +472,7 @@ wheels = [ [[package]] name = "ratatoskr" -version = "0.21.7" +version = "0.21.8" source = { editable = "." } dependencies = [ { name = "httpx" },