vh/ratatoskr
1
0
Fork 0
Code Issues 4 Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(#20): delete the orphaned hand-rolled turn-stream paths (slice-2, part 2b-iii)

Browse Source 
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).
This commit is contained in:
vh
2026-07-19 06:35:04 -07:00
parent 5c595b862d
commit 59602fe3ff
6 changed files with 10 additions and 2079 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pyproject.toml
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
[project] [project]
name = "ratatoskr" name = "ratatoskr"
version = "0.21.7" version = "0.21.8"
description = "Worldtree Conversation API debug console (web + headless CLI) — multi-pane observability" description = "Worldtree Conversation API debug console (web + headless CLI) — multi-pane observability"
readme = "README.md" readme = "README.md"
requires-python = ">=3.12" requires-python = ">=3.12"
src/ratatoskr/sessions.py
-69
View File
@@ -38,12 +38,6 @@ class SessionInfo:
config: dict[str, Any] | None = None 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) @dataclass(frozen=True)
@@ -204,53 +198,6 @@ class AuthoredHistoryUnavailable(Exception):
self.session_id = session_id 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: 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) 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]: async def get_capabilities(client: httpx.AsyncClient) -> dict[str, Any]:
src/ratatoskr/sse_client.py
-491
View File
@@ -6,7 +6,6 @@ Implements docs/contracts/issues/1.contract.md.
from __future__ import annotations from __future__ import annotations
import json import json
import re
from collections.abc import AsyncIterator from collections.abc import AsyncIterator
from dataclasses import dataclass from dataclasses import dataclass
from typing import Any, NamedTuple from typing import Any, NamedTuple
@@ -14,8 +13,6 @@ from typing import Any, NamedTuple
import httpx import httpx
import httpx_sse import httpx_sse
_INT_RE = re.compile(r"^-?\d+$")
class SseId(NamedTuple): class SseId(NamedTuple):
"""Parsed composite SSE wire `id:` per spec §SSE id format.""" """Parsed composite SSE wire `id:` per spec §SSE id format."""
@@ -24,155 +21,6 @@ class SseId(NamedTuple):
seq: 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
@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) @dataclass(frozen=True)
@@ -258,31 +106,6 @@ class TurnLaunchUnavailable(SseConnectFailed):
self.message = message 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): class SseConnectionDropped(Exception):
@@ -352,275 +175,6 @@ class CancelFailed(Exception):
self.body = body 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( async def stream_admin_events(
client: httpx.AsyncClient, client: httpx.AsyncClient,
@@ -667,48 +221,3 @@ async def stream_admin_events(
raise SseConnectionDropped(last_seen_sse_id=None) from exc 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"),
)
tests/test_sessions.py
-268
View File
@@ -13,10 +13,8 @@ from ratatoskr.sessions import (
BifrostBinding, BifrostBinding,
BifrostConsumerKeyMissing, BifrostConsumerKeyMissing,
BifrostHandshakeFailed, BifrostHandshakeFailed,
InvalidCursor,
PersonaNotConfigured, PersonaNotConfigured,
SessionApiFailed, SessionApiFailed,
SessionPage,
create_character, create_character,
create_session, create_session,
delete_character, delete_character,
@@ -27,10 +25,8 @@ from ratatoskr.sessions import (
get_persona_state, get_persona_state,
get_session_bifrost, get_session_bifrost,
get_session_messages, get_session_messages,
get_session_tools,
list_agents, list_agents,
list_character_models, list_character_models,
list_sessions,
set_persona_state, set_persona_state,
write_authored_history, write_authored_history,
) )
@@ -551,223 +547,6 @@ class TestEndpointForPlane:
endpoint_for_plane("persona", "10.100.10.50") 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=<passed cursor>)."""
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: class TestListAgents:
@respx.mock @respx.mock
async def test_happy_full_shape(self) -> None: async def test_happy_full_shape(self) -> None:
@@ -1145,53 +924,6 @@ class TestGetCapabilities:
assert exc.value.status == 500 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: class TestGetSessionBifrost:
"""#2 contract — get_session_bifrost (GET /admin/sessions/{id}/bifrost, #176).""" """#2 contract — get_session_bifrost (GET /admin/sessions/{id}/bifrost, #176)."""
tests/test_sse_client.py
+8 -1249
View File
File diff suppressed because it is too large Load Diff
uv.lock
Generated
+1 -1
View File
@@ -472,7 +472,7 @@ wheels = [
[[package]] [[package]]
name = "ratatoskr" name = "ratatoskr"
version = "0.21.7" version = "0.21.8"
source = { editable = "." } source = { editable = "." }
dependencies = [ dependencies = [
{ name = "httpx" }, { name = "httpx" },