diff --git a/docs/contracts/worldtree_sdk_cutover.contract.md b/docs/contracts/worldtree_sdk_cutover.contract.md index d0ab81c..b58ae31 100644 --- a/docs/contracts/worldtree_sdk_cutover.contract.md +++ b/docs/contracts/worldtree_sdk_cutover.contract.md @@ -295,6 +295,46 @@ re-anchor its coverage-map rows. web-server caller — only the `--whoami` and `--characters` CLI one-shot probes. The web surface is untouched this slice. +### Slice-6 notes (Admin: bifrost inspection + admin-events stream, decided at TDD) + +- **Admin auth moves from a per-call header override to the client's `admin_auth`.** + The SDK's `admin.*` methods authenticate with the client's `admin_auth` provider + (set via `build_client(admin_key=...)`), NOT a per-request `Authorization` header. So + the two web admin endpoints build their wt client WITH `admin_key` (`_wt_client(client, + admin_key=...)`, extended this slice); the hand-rolled per-call `admin_key=` + + header-override is retired. The web already guards `if not admin_key: 400` before the + call, so the SDK's pre-HTTP `ConfigurationError` (missing admin_auth, W-5) is + unreachable from the web surface. **CLI has no admin caller** — both routes are + web-only (the coverage-map's `tui.py` rows were stale; corrected to `web/server.py`). +- **`get_session_bifrost` — no new § Error map row.** `client.admin.sessions.bifrost` + returns the open-world `BifrostInspection` dict verbatim; any `ApiError` (notably 403 + `auth_scope_denied`, 404 `session_not_bifrost_bound`) → the `SessionApiFailed` default + — exact parity with the retired path (which mapped every non-200 → `SessionApiFailed`). +- **`stream_admin_events` re-wraps the SDK's `AdminEvent` → ratatoskr's `AdminEvent` + (chosen over yield-through).** The SDK's `AdminEvent` diverges from ratatoskr's: + `admin_id: int|float` (`nan` for an id-less envelope) vs ratatoskr's `id: int` (0 + default), and the SDK's `type`/`data` are None-able where ratatoskr's are a dotted-str + / a `{}`-default dict. The web filter + SSE formatter read `ev.id`/`ev.type`/`ev.data`. + The adapter re-wraps at the boundary — `id = admin_id if int else 0` (nan→0), + `type = type or ""` (None→"" so `.startswith` never crashes), `data = data or {}` — + degrading the SDK's open-world None/nan ONCE at the adapter, keeping the web endpoint + + `_admin_event_matches_web` + the ratatoskr `AdminEvent` domain type UNCHANGED (preserves + the web surface per § Out of scope). **Rejected alternative:** yield SDK `AdminEvent`s + through and rewire the web filter for `admin_id`/None/nan (the slice-2 turn-stream + precedent) — heavier web churn + scatters the None/nan hardening through the filter; + re-wrap localizes it. The ratatoskr `AdminEvent` dataclass stays in `sse_client.py` this + slice (imported by `wt` + the web); its home moves in slice-7 teardown if `sse_client.py` + is retired. +- **Admin-stream error mapping (reuses the § Error map stream rows).** The SDK admin + stream raises `ApiError("admin_stream_failed", status=…)` on a NON-200 open (NOT + `ConnectFailed` — a gotcha the web integration test caught that the unit fake could not) + → `SseConnectFailed`; and `ConnectionDropped` on a connect-time transport failure + (cursor None) OR a mid-stream drop / the long-lived stream's resumable EOF (cursor set) + → `SseConnectionDropped`. The SDK admin stream is best-effort (skips malformed frames — + no `Malformed*`), as was the retired hand-rolled path; the web endpoint's existing + `except (…, MalformedSseId, MalformedSseData)` stays a harmless defensive superset + (pre-existing, not introduced here). + ## Out of scope - Bifrost PROVIDER planes (memory/affect) — hand-rolled, ADR-0009, untouched. diff --git a/docs/coverage-map.md b/docs/coverage-map.md index f329f93..dead070 100644 --- a/docs/coverage-map.md +++ b/docs/coverage-map.md @@ -97,8 +97,8 @@ sub-gap). | `GET /me` | ✅ | `wt.py` `get_me` (SDK `me.get`) → `cli.py` `--whoami` | **wt-adapter re-anchored (slice-5, #20)** — open-world identity dict verbatim; any error→SessionApiFailed default (401 on a bad/absent key), transport→ConnectFailed→exit 21. **LIVE-SMOKE 2026-07-19** on personal :8081 (b128): identity rendered (user_id ratatoskr, tier user, scopes incl. `character.*`, key_id c990f0be) | | `GET /capabilities` | ✅ | `wt.py` `get_capabilities` (SDK `capabilities.get`) → `cli.py` `--whoami` | **wt-adapter re-anchored (slice-5, #20)** — open-world advertisement verbatim; `_format_whoami` reads `allowed_roles`/`default_role` and degrades on a null/non-mapping template (slice-4 hardening); matches conversation-api-spec **v1.1** (`b4a278c`). **LIVE-SMOKE 2026-07-19**: `ephemeral_template echo: default=echo max_bytes=32768 roles=[echo]` | | `GET /sessions/{id}/tools` | ✅ | `sessions.py:411` `get_session_tools` → `tui.py` `_hydrate_session_tools` | owner-scoped tool inventory in the TUI Tools pane (#183) | -| `GET /admin/sessions/{id}/bifrost` | ✅ | `sessions.py:428` `get_session_bifrost` → `tui.py` `_hydrate_bifrost_state` | admin-scoped BifrostState pane (#176); admin key (`RATATOSKR_ADMIN_API_KEY`); live-auth-proven | -| `GET /admin/events` (SSE) | ✅ | `sse_client.py` `stream_admin_events` → `tui.py` `_stream_admin_events` | admin lifecycle SSE stream (#11), session-filtered AdminEvents pane; admin key; live-auth-proven | +| `GET /admin/sessions/{id}/bifrost` | ✅ | `wt.py` `get_session_bifrost` (SDK `admin.sessions.bifrost`) → `web/server.py` `_session_bifrost_endpoint` | **wt-adapter re-anchored (slice-6, #20)** — admin-scoped BifrostState (#176); admin_auth rides on the wt client (`_wt_client(admin_key=…)`), NOT a per-call header; open-world dict verbatim, any error→SessionApiFailed default. **LIVE-SMOKE 2026-07-19** on :8081 (readonly-admin key): admin-authed end-to-end (404 `session_not_bifrost_bound` clean envelope — auth + route + mapping proven). (Consumer is `web/server.py`, not `tui.py` — the old row was stale.) | +| `GET /admin/events` (SSE) | ✅ | `wt.py` `stream_admin_events` (SDK `admin.stream_events`) → `web/server.py` `_admin_events_endpoint` | **wt-adapter re-anchored (slice-6, #20)** — admin lifecycle SSE (#11), session-filtered; admin_auth on the wt client; the adapter re-wraps the SDK's `AdminEvent`→ratatoskr's (nan `admin_id`→id 0, None type/data→`""`/`{}`), non-200 open `ApiError`→SseConnectFailed, `ConnectionDropped`→SseConnectionDropped. **LIVE-SMOKE 2026-07-19**: a real `session.created` event (id=32) re-wrapped cleanly on live wire. (Consumer is `web/server.py`, not `tui.py` — stale row corrected.) | | `GET /models/available-for-characters` | ✅ | `wt.py` `list_character_models` (SDK `models.available_for_characters`) → `cli.py` `--characters` | **wt-adapter re-anchored (slice-5, #20)** — open-world catalog verbatim; the probe reads `items` null-safe (`or []`); any error→SessionApiFailed default. **LIVE-SMOKE 2026-07-19**: `character models: char-rp` | | `POST /characters` | ✅ | `wt.py` `create_character` (SDK `characters.create`) → `cli.py` `--characters` | **wt-adapter re-anchored (slice-5, #20)** — body `{character}` (+`state` only when set — SDK-idiomatic, drops the redundant explicit null); open-world create ACK verbatim; the probe degrades on a missing `character_id` (no hard-index). **LIVE-SMOKE 2026-07-19**: `created char_8c00006e…` | | `GET /characters/{id}/state` | ✅ | `wt.py` `get_character_state` (SDK `characters.state`) → `cli.py` `--characters` | **wt-adapter re-anchored (slice-5, #20)** — open-world live PAD/emotions verbatim; TTL-refreshing read. **LIVE-SMOKE 2026-07-19**: `state pad=[0.234, -0.136, 0.065]` read back | diff --git a/pyproject.toml b/pyproject.toml index 5397696..7addd42 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "ratatoskr" -version = "0.21.18" +version = "0.21.19" 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 1136be1..ad1c51b 100644 --- a/src/ratatoskr/sessions.py +++ b/src/ratatoskr/sessions.py @@ -6,9 +6,6 @@ Implements docs/contracts/issues/2.contract.md. from __future__ import annotations from dataclasses import dataclass -from typing import Any - -import httpx @dataclass(frozen=True) @@ -228,28 +225,3 @@ def endpoint_for_plane(plane: str, base_host: str) -> str: if plane not in ports: raise ValueError(f"unknown plane: {plane!r} (expected 'memory', 'affect', or 'combined')") return f"http://{base_host}:{ports[plane]}" - - -async def get_session_bifrost( - client: httpx.AsyncClient, session_id: str, *, admin_key: str -) -> dict[str, Any]: - """GET /admin/sessions/{session_id}/bifrost — admin-scoped Bifrost dispatch state (#176). - - Returns the live Bifrost binding for a session: `{endpoint_url, consumer_id, - connected, capabilities_granted, tools: [{name, description}]}`. Requires the - `admin.sessions.read` scope (admin tier), so the request OVERRIDES the - Authorization header with `admin_key` (distinct from the client's default - consumer key). Read-only (audited server-side). Parsed dict verbatim; any - non-200 → SessionApiFailed — notably 403 `auth_scope_denied` (key lacks the - scope) and 404 `session_not_bifrost_bound` (session exists, no live client). - """ - assert client is not None - assert session_id and isinstance(session_id, str) - assert admin_key and isinstance(admin_key, str) - resp = await client.get( - f"/admin/sessions/{session_id}/bifrost", - headers={"Authorization": f"Bearer {admin_key}"}, - ) - if resp.status_code == 200: - return resp.json() - raise SessionApiFailed(status=resp.status_code, body=resp.content) diff --git a/src/ratatoskr/sse_client.py b/src/ratatoskr/sse_client.py index 3eef862..4aae17e 100644 --- a/src/ratatoskr/sse_client.py +++ b/src/ratatoskr/sse_client.py @@ -5,14 +5,9 @@ Implements docs/contracts/issues/1.contract.md. from __future__ import annotations -import json -from collections.abc import AsyncIterator from dataclasses import dataclass from typing import Any, NamedTuple -import httpx -import httpx_sse - class SseId(NamedTuple): """Parsed composite SSE wire `id:` per spec §SSE id format.""" @@ -173,51 +168,3 @@ class CancelFailed(Exception): super().__init__(f"cancel failed: status={status}, body={body[:128]!r}") self.status = status self.body = body - - - -async def stream_admin_events( - client: httpx.AsyncClient, - *, - admin_key: str, - last_event_id: int | None = None, -) -> AsyncIterator[AdminEvent]: - """GET /admin/events SSE — the admin-tier lifecycle broadcast stream (INV-046). - - Yields `AdminEvent` envelopes as they arrive. Admin-scoped (admin.events.read): - the request OVERRIDES Authorization with `admin_key` (distinct from the - client's default consumer bearer). `last_event_id` sets the `Last-Event-ID` - header for resume (plain decimal int). Long-lived — iterate until the caller - stops or the connection ends. Non-200 → SseConnectFailed; a mid-stream drop - → SseConnectionDropped (caller may reconnect from the last-seen `AdminEvent.id`). - Malformed frames are skipped (best-effort stream). - """ - assert client is not None - assert admin_key and isinstance(admin_key, str) - headers = {"Authorization": f"Bearer {admin_key}"} - if last_event_id is not None: - headers["Last-Event-ID"] = str(last_event_id) - async with httpx_sse.aconnect_sse( - client, "GET", "/admin/events", headers=headers - ) as event_source: - if event_source.response.status_code != 200: - body = await event_source.response.aread() - raise SseConnectFailed(status=event_source.response.status_code, body=body) - try: - async for sse in event_source.aiter_sse(): - if sse.data == "": - continue - try: - env = json.loads(sse.data) - except json.JSONDecodeError: - continue # skip a malformed admin frame (best-effort) - yield AdminEvent( - id=env.get("id", 0), - type=env["type"], - timestamp=env.get("timestamp"), - data=env.get("data", {}), - ) - except (httpx.ReadError, httpx.RemoteProtocolError, httpx.ReadTimeout) as exc: - raise SseConnectionDropped(last_seen_sse_id=None) from exc - - diff --git a/src/ratatoskr/web/server.py b/src/ratatoskr/web/server.py index 46efb5e..80518f2 100644 --- a/src/ratatoskr/web/server.py +++ b/src/ratatoskr/web/server.py @@ -45,15 +45,14 @@ from ratatoskr.sessions import ( BifrostConsumerKeyMissing, BifrostHandshakeFailed, PersonaNotConfigured, - SessionApiFailed, endpoint_for_plane, - get_session_bifrost, ) -# The turn path (create / stream / cancel / tools / messages) AND the agents / -# persona-state reads are served by the worldtree-sdk adapter (`wt.*`), which raises -# ratatoskr's caller-semantic exceptions (DEC-2). The remaining hand-rolled endpoint -# (admin bifrost) stays on the `sessions` / `sse_client` wrappers until slice 6. +# The turn path (create / stream / cancel / tools / messages), the agents / +# persona-state reads, AND the admin surface (bifrost inspection + admin-events stream) +# are all served by the worldtree-sdk adapter (`wt.*`), which raises ratatoskr's +# caller-semantic exceptions (DEC-2). `AdminEvent` is still ratatoskr's domain event +# type the adapter re-wraps into (imported from `sse_client` until slice-7 teardown). from ratatoskr.sse_client import ( AdminEvent, CancelAlreadyCompleted, @@ -64,16 +63,21 @@ from ratatoskr.sse_client import ( SseConnectFailed, SseConnectionDropped, TurnIdFlip, - stream_admin_events, ) -def _wt_client(client: httpx.AsyncClient, *, max_reconnects: int = 5) -> WorldtreeClient: +def _wt_client( + client: httpx.AsyncClient, *, admin_key: str | None = None, max_reconnects: int = 5 +) -> WorldtreeClient: """Wrap a client_factory transport as the adapter's WorldtreeClient (INV-CUT-1: the SDK never closes it). base_url + bearer are read off the transport (the factory bakes them in); the SDK re-applies auth per request, so the extracted key just mirrors the transport's default. A no-auth test transport falls back to - a placeholder key (respx ignores auth).""" + a placeholder key (respx ignores auth). + + `admin_key` is the SERVER-HELD admin credential (slice-6): the SDK's `admin.*` + routes authenticate with the client's `admin_auth`, NOT a per-call header, so an + admin endpoint passes it here. Omitted for the default-tier reads.""" base_url = str(client.base_url) or "http://localhost" header = client.headers.get("Authorization", "") # Case-insensitive scheme + tolerant of extra whitespace, so a valid bearer is @@ -81,7 +85,11 @@ def _wt_client(client: httpx.AsyncClient, *, max_reconnects: int = 5) -> Worldtr parts = header.split(None, 1) api_key = parts[1].strip() if len(parts) == 2 and parts[0].lower() == "bearer" else "" return wt.build_client( - base_url, api_key=api_key or "ratatoskr", transport=client, max_reconnects=max_reconnects + base_url, + api_key=api_key or "ratatoskr", + admin_key=admin_key, + transport=client, + max_reconnects=max_reconnects, ) @@ -573,14 +581,24 @@ async def _session_bifrost_endpoint(request: Request) -> JSONResponse: return JSONResponse({"error_code": "admin_key_not_configured"}, status_code=400) client_factory = request.app.state.client_factory try: - async with client_factory() as client: - bstate = await get_session_bifrost(client, session_id, admin_key=admin_key) - except SessionApiFailed as exc: + async with client_factory() as transport: + # slice-6: the SDK's admin.* routes use the client's admin_auth (built with + # admin_key), not a per-call header — so it rides on the wt client here. + client = _wt_client(transport, admin_key=admin_key) + bstate = await wt.get_session_bifrost(client, session_id) + except wt.SessionApiFailed as exc: return JSONResponse( {"error_code": "bifrost_state_unavailable", "status": exc.status}, status_code=exc.status, ) - return JSONResponse(bstate, status_code=200) + except (httpx.RequestError, ConnectFailed) as exc: + # SDK normalizes a transport failure to ConnectFailed(status=0), not a raw + # httpx error; both surface the same network envelope (cutover foot-gun). + return JSONResponse( + {"error_code": "network_error", "message": str(exc)}, + status_code=502, + ) + return JSONResponse(dict(bstate), status_code=200) def _admin_event_matches_web(ev: AdminEvent, session_id: str | None) -> bool: @@ -608,9 +626,13 @@ async def _admin_events_endpoint(request: Request) -> Response: client_factory = request.app.state.client_factory async def gen() -> AsyncIterator[bytes]: - client = client_factory() + transport = client_factory() + # slice-6: admin_auth rides on the wt client (built with admin_key); the adapter + # re-wraps the SDK's AdminEvent → ratatoskr's (id/type/data degraded) and the + # stream's terminal SDK errors (incl. ConnectFailed) → the Sse* types below. + client = _wt_client(transport, admin_key=admin_key) try: - async for ev in stream_admin_events(client, admin_key=admin_key): + async for ev in wt.stream_admin_events(client): if not _admin_event_matches_web(ev, session_id): continue # Fixed SSE event name so the browser renders EVERY admin type @@ -630,7 +652,9 @@ async def _admin_events_endpoint(request: Request) -> Response: except asyncio.CancelledError: raise # browser disconnect — let the generator unwind finally: - await client.aclose() + # ratatoskr owns the transport lifecycle (INV-CUT-1); close the injected + # httpx client, never the wt client (which would no-op the transport anyway). + await transport.aclose() return StreamingResponse(gen(), media_type="text/event-stream") diff --git a/src/ratatoskr/wt.py b/src/ratatoskr/wt.py index df9c417..dd29abc 100644 --- a/src/ratatoskr/wt.py +++ b/src/ratatoskr/wt.py @@ -60,6 +60,7 @@ from .sessions import ( Tier3UserIdUnsupported, ) from .sse_client import ( + AdminEvent, AgentNotAvailable, CancelAlreadyCompleted, CancelFailed, @@ -684,3 +685,66 @@ async def delete_character( return await client.characters.delete(character_id) except ApiError as exc: raise translate_error(exc) from exc + + +# ── slice-6: admin (bifrost inspection + admin-events stream) adapter routes ── +# The admin surface over `client.admin.*` — admin_auth-scoped (set via +# `build_client(admin_key=...)`, NOT a per-call `Authorization` header). Both are +# web-only. `get_session_bifrost` reads the open-world `BifrostInspection` verbatim +# (any error → the `SessionApiFailed` default); `stream_admin_events` drives the +# long-lived D2 admin-events SSE, re-wrapping the SDK's `AdminEvent` → ratatoskr's +# (degrading the SDK's `admin_id`-nan / None `type`/`data` at the boundary so the web +# filter never crashes) and re-wrapping the stream's terminal errors → ratatoskr's +# `Sse*` types (INV-CUT-2 stream rows). + + +async def get_session_bifrost( + client: WorldtreeClient, session_id: str +) -> Mapping[str, Any]: + """The admin-scoped Bifrost dispatch state for a session (GET + /admin/sessions/{id}/bifrost, #176), open-world dict verbatim. + + Admin-tier — the client MUST carry `admin_auth` (built with `admin_key`); the SDK + uses that provider, not a per-call header. Any error → the `SessionApiFailed` + default (notably 403 `auth_scope_denied`, 404 `session_not_bifrost_bound`) — the + retired hand-rolled path likewise mapped every non-200 generically. + """ + assert session_id and isinstance(session_id, str) + try: + return await client.admin.sessions.bifrost(session_id) + except ApiError as exc: + raise translate_error(exc) from exc + + +async def stream_admin_events( + client: WorldtreeClient, *, last_event_id: int | None = None +) -> AsyncGenerator[AdminEvent, None]: + """Drive the long-lived admin-events SSE (GET /admin/events, #11 / INV-046) and yield + ratatoskr `AdminEvent`s, re-wrapping the SDK's typed `AdminEvent` at the boundary. + + Admin-tier (the client MUST carry `admin_auth`). The SDK's `AdminEvent` is open-world + where ratatoskr's is stable: `admin_id` is `nan` for an id-less envelope (→ `id=0`), + and `type`/`data` may be None (→ `""` / `{}`) — normalized HERE so the web filter + + formatter (`ev.id` / `ev.type` / `ev.data`) never crash on a partial wire (chosen over + yielding SDK events through + rewiring the web filter). Error map (INV-CUT-2, stream + rows): SDK `ApiError` (a non-200 open — the admin stream raises `admin_stream_failed`, + NOT `ConnectFailed`) → `SseConnectFailed`; SDK `ConnectionDropped` (a connect-time + transport failure → cursor None, OR a mid-stream drop / the long-lived stream's + resumable EOF → cursor) → `SseConnectionDropped`. The SDK stream is best-effort (skips + malformed frames — no `Malformed*`). + """ + try: + async for ev in client.admin.stream_events(last_event_id=last_event_id): + yield AdminEvent( + id=ev.admin_id if isinstance(ev.admin_id, int) else 0, + type=ev.type or "", + timestamp=ev.timestamp, + data=dict(ev.data) if isinstance(ev.data, Mapping) else {}, + ) + except wtsdk.ConnectionDropped as exc: + raise SseConnectionDropped(last_seen_sse_id=exc.last_seen_sse_id) from exc + except ApiError as exc: + # The admin stream raises ApiError("admin_stream_failed", status=…) on a non-200 + # open (a connect-time transport failure instead surfaces as ConnectionDropped); + # map the non-200 → SseConnectFailed so the web's stream-error handler catches it. + raise SseConnectFailed(status=exc.status, body=(exc.body or "").encode()) from exc diff --git a/tests/test_sessions.py b/tests/test_sessions.py index 0213cc2..b730f37 100644 --- a/tests/test_sessions.py +++ b/tests/test_sessions.py @@ -1,14 +1,13 @@ -"""Tests for ratatoskr.sessions per docs/contracts/issues/2.contract.md.""" +"""Tests for ratatoskr.sessions per docs/contracts/issues/2.contract.md. + +Post worldtree-sdk cutover the `sessions` module is down to `endpoint_for_plane` +(the Bifrost provider-plane helper) + the caller-semantic exception classes the +`ratatoskr.wt` adapter raises; every wire wrapper has retired onto the SDK adapter +(the wrappers' tests live in `test_wt.py`).""" -import httpx import pytest -import respx -from ratatoskr.sessions import ( - SessionApiFailed, - endpoint_for_plane, - get_session_bifrost, -) +from ratatoskr.sessions import endpoint_for_plane class TestEndpointForPlane: @@ -30,65 +29,3 @@ class TestEndpointForPlane: """unknown_plane [adversarial]: any other plane → ValueError (PRE-001).""" with pytest.raises(ValueError): endpoint_for_plane("persona", "10.100.10.50") - - -class TestGetSessionBifrost: - """#2 contract — get_session_bifrost (GET /admin/sessions/{id}/bifrost, #176).""" - - @respx.mock - async def test_happy_uses_admin_bearer(self) -> None: - """happy [happy,tracer]: 200 → binding dict; request carries the ADMIN bearer (override).""" - route = respx.get("https://w.example/admin/sessions/s1/bifrost").mock( - return_value=httpx.Response( - 200, - json={ - "endpoint_url": "https://bifrost.example/mcp", - "consumer_id": "alice", - "connected": True, - "capabilities_granted": ["tools:call", "tools:read"], - "tools": [{"name": "bifrost.alice.echo", "description": "echo"}], - }, - ) - ) - async with httpx.AsyncClient( - base_url="https://w.example", - headers={"Authorization": "Bearer consumer-key"}, - ) as client: - state = await get_session_bifrost(client, "s1", admin_key="admin-xyz") - assert state["connected"] is True - assert state["tools"][0]["name"] == "bifrost.alice.echo" - # the request overrode the client's default consumer bearer with the admin key - assert route.calls[0].request.headers["Authorization"] == "Bearer admin-xyz" - - @respx.mock - async def test_403_scope_denied(self) -> None: - """403 [error]: admin key lacks admin.sessions.read → SessionApiFailed(403).""" - respx.get("https://w.example/admin/sessions/s1/bifrost").mock( - return_value=httpx.Response(403, json={"error_code": "auth_scope_denied"}) - ) - async with httpx.AsyncClient(base_url="https://w.example") as client: - with pytest.raises(SessionApiFailed) as exc: - await get_session_bifrost(client, "s1", admin_key="k") - assert exc.value.status == 403 - - @respx.mock - async def test_404_not_bound(self) -> None: - """404 [error]: session_not_bifrost_bound → SessionApiFailed(404).""" - respx.get("https://w.example/admin/sessions/s1/bifrost").mock( - return_value=httpx.Response(404, json={"error_code": "session_not_bifrost_bound"}) - ) - async with httpx.AsyncClient(base_url="https://w.example") as client: - with pytest.raises(SessionApiFailed) as exc: - await get_session_bifrost(client, "s1", admin_key="k") - assert exc.value.status == 404 - - @respx.mock - async def test_empty_admin_key_asserts(self) -> None: - """empty_admin_key [adversarial]: '' → AssertionError; no HTTP issued.""" - route = respx.get("https://w.example/admin/sessions/s1/bifrost").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_bifrost(client, "s1", admin_key="") - assert route.call_count == 0 diff --git a/tests/test_sse_client.py b/tests/test_sse_client.py deleted file mode 100644 index 8624e9d..0000000 --- a/tests/test_sse_client.py +++ /dev/null @@ -1,94 +0,0 @@ -"""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 -import respx - -from ratatoskr.sse_client import ( - AdminEvent, - SseConnectFailed, - stream_admin_events, -) - - -def _sse_chunk(sse_id: str, body: dict[str, object]) -> bytes: - """Compose one SSE event in wire format. Trailing blank line per spec.""" - import json - - return f"id: {sse_id}\ndata: {json.dumps(body)}\n\n".encode() - - -class TestStreamAdminEvents: - """docs/conversation-api-spec.md § Admin Event Stream — stream_admin_events (#11).""" - - @respx.mock - async def test_happy_multi_event_admin_bearer(self) -> None: - """happy [happy,tracer]: yields AdminEvent envelopes; request uses the ADMIN bearer.""" - env1 = { - "id": 41, "type": "session.created", "timestamp": "2026-05-06T10:00:00.000Z", - "data": {"session_id": "s1", "agent_id": "mimir", "user_id": None}, - } - env2 = { - "id": 42, "type": "turn.started", "timestamp": "2026-05-06T10:00:01.000Z", - "data": {"session_id": "s1", "turn_id": 7, "agent_id": "mimir", "user_id": None}, - } - stream = _sse_chunk("41", env1) + _sse_chunk("42", env2) - route = respx.get("https://w.example/admin/events").mock( - return_value=httpx.Response( - 200, headers={"content-type": "text/event-stream"}, content=stream - ) - ) - async with httpx.AsyncClient( - base_url="https://w.example", headers={"Authorization": "Bearer consumer"} - ) as client: - events = [e async for e in stream_admin_events(client, admin_key="admin-xyz")] - assert [e.type for e in events] == ["session.created", "turn.started"] - assert isinstance(events[0], AdminEvent) - assert events[0].id == 41 - assert events[1].data["turn_id"] == 7 - assert route.calls[0].request.headers["Authorization"] == "Bearer admin-xyz" - - @respx.mock - async def test_last_event_id_header(self) -> None: - """last_event_id_header [trace]: empty stream → []; Last-Event-ID header sent.""" - route = respx.get("https://w.example/admin/events").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: - events = [e async for e in stream_admin_events(client, admin_key="k", last_event_id=99)] - assert events == [] - assert route.calls[0].request.headers["Last-Event-ID"] == "99" - - @respx.mock - async def test_403_scope_denied(self) -> None: - """403 [error]: key lacks admin.events.read → SseConnectFailed(403).""" - respx.get("https://w.example/admin/events").mock( - return_value=httpx.Response(403, json={"error_code": "auth_scope_denied"}) - ) - async with httpx.AsyncClient(base_url="https://w.example") as client: - with pytest.raises(SseConnectFailed) as exc: - _ = [e async for e in stream_admin_events(client, admin_key="k")] - assert exc.value.status == 403 - - @respx.mock - async def test_skips_malformed_frame(self) -> None: - """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"}} - ) - respx.get("https://w.example/admin/events").mock( - return_value=httpx.Response( - 200, headers={"content-type": "text/event-stream"}, content=good + bad + good2 - ) - ) - async with httpx.AsyncClient(base_url="https://w.example") as client: - events = [e async for e in stream_admin_events(client, admin_key="k")] - assert [e.type for e in events] == ["session.created", "session.deleted"] diff --git a/tests/test_wt.py b/tests/test_wt.py index 2415f91..47814f7 100644 --- a/tests/test_wt.py +++ b/tests/test_wt.py @@ -38,6 +38,7 @@ from ratatoskr.sessions import ( Tier3UserIdUnsupported, ) from ratatoskr.sse_client import ( + AdminEvent, AgentNotAvailable, CancelAlreadyCompleted, CancelFailed, @@ -62,6 +63,7 @@ from ratatoskr.wt import ( get_character_state, get_me, get_persona_state, + get_session_bifrost, get_session_messages, get_session_tools, list_agents, @@ -69,6 +71,7 @@ from ratatoskr.wt import ( list_sessions, patch_agent, set_persona_state, + stream_admin_events, stream_turn, translate_error, write_authored_history, @@ -1066,3 +1069,153 @@ class TestDeleteCharacter: with pytest.raises(SessionApiFailed) as ei: await delete_character(_wtm(fake), "char_x") assert ei.value.status == 500 + + +# ── slice-6: admin (bifrost inspection + admin-events stream) adapter routes ── + + +class _SdkAdminEvent: + """Minimal stand-in for the SDK's `AdminEvent` — the adapter reads + `admin_id`/`type`/`timestamp`/`data`. `admin_id` may be `nan` (id-less); + `type`/`data` may be None (open-world).""" + + def __init__(self, admin_id: Any, type: Any, timestamp: Any = None, data: Any = None) -> None: + self.admin_id = admin_id + self.type = type + self.timestamp = timestamp + self.data = data + + +class _FakeAdminSessions: + def __init__(self, admin: _FakeAdmin) -> None: + self._admin = admin + + async def bifrost(self, *a: Any, **k: Any) -> Any: + return await self._admin._bifrost(*a, **k) + + +class _FakeAdmin: + """Stand-in for `client.admin` — `.sessions.bifrost(id)` (canned result/error) + + `.stream_events(...)` (canned events / terminal error). Same shape as `_FakeSessions`.""" + + def __init__( + self, + *, + result: Any = None, + error: BaseException | None = None, + events: list[Any] | None = None, + stream_error: BaseException | None = None, + ) -> None: + self._result = result + self._error = error + self._events = events or [] + self._stream_error = stream_error + self.calls: list[tuple[str, tuple[Any, ...], dict[str, Any]]] = [] + self.sessions = _FakeAdminSessions(self) + + async def _bifrost(self, *a: Any, **k: Any) -> Any: + self.calls.append(("bifrost", a, k)) + if self._error is not None: + raise self._error + return self._result + + def stream_events(self, *a: Any, **k: Any) -> Any: + self.calls.append(("stream_events", a, k)) + return self._astream() + + async def _astream(self) -> Any: + for ev in self._events: + yield ev + if self._stream_error is not None: + raise self._stream_error + + +class _FakeAdminClient: + def __init__(self, admin: _FakeAdmin) -> None: + self.admin = admin + + +def _wtad(admin: _FakeAdmin) -> WorldtreeClient: + """Cast the admin-surface fake (`.admin.sessions.bifrost` + `.admin.stream_events`) + to the nominal client type the slice-6 route functions are typed against.""" + return cast(WorldtreeClient, _FakeAdminClient(admin)) + + +class TestGetSessionBifrostWt: + """slice-6: get_session_bifrost → SDK admin.sessions.bifrost(id); open-world verbatim.""" + + async def test_happy_returns_dict_verbatim(self) -> None: + binding = {"endpoint_url": "https://b/mcp", "connected": True, "tools": []} + fake = _FakeAdmin(result=binding) + out = await get_session_bifrost(_wtad(fake), "s1") + assert out is binding + assert fake.calls[-1] == ("bifrost", ("s1",), {}) + + async def test_empty_id_asserts_no_call(self) -> None: + fake = _FakeAdmin(result={}) + with pytest.raises(AssertionError): + await get_session_bifrost(_wtad(fake), "") + assert fake.calls == [] + + async def test_403_maps_to_session_api_failed(self) -> None: + fake = _FakeAdmin(error=ApiError("auth_scope_denied", "no", status=403)) + with pytest.raises(SessionApiFailed) as ei: + await get_session_bifrost(_wtad(fake), "s1") + assert ei.value.status == 403 + + async def test_404_not_bound_maps_to_session_api_failed(self) -> None: + fake = _FakeAdmin(error=ApiError("session_not_bifrost_bound", "no", status=404)) + with pytest.raises(SessionApiFailed) as ei: + await get_session_bifrost(_wtad(fake), "s1") + assert ei.value.status == 404 + + +class TestStreamAdminEventsWt: + """slice-6: stream_admin_events → SDK admin.stream_events; re-wrap SDK AdminEvent → + ratatoskr AdminEvent (nan/None degraded), terminal errors → Sse* types.""" + + async def test_rewraps_events_to_ratatoskr_shape(self) -> None: + sdk_evs = [ + _SdkAdminEvent(5, "session.created", "t0", {"session_id": "s1"}), + _SdkAdminEvent(6, "turn.completed", "t1", {"session_id": "s1", "turn_id": 2}), + ] + fake = _FakeAdmin(events=sdk_evs) + out = await _drain(stream_admin_events(_wtad(fake))) + assert all(isinstance(e, AdminEvent) for e in out) + assert (out[0].id, out[0].type, out[0].timestamp) == (5, "session.created", "t0") + assert out[0].data == {"session_id": "s1"} + assert out[1].id == 6 + + async def test_nan_admin_id_degrades_to_zero(self) -> None: + fake = _FakeAdmin(events=[_SdkAdminEvent(float("nan"), "system.heartbeat", None, None)]) + out = await _drain(stream_admin_events(_wtad(fake))) + assert out[0].id == 0 # id-less envelope → 0, not nan + + async def test_none_type_and_data_degrade(self) -> None: + # A partial wire: type=None (would crash `.startswith` in the web filter) and + # data=None (would crash `.get`) → "" and {} at the adapter boundary. + fake = _FakeAdmin(events=[_SdkAdminEvent(1, None, None, None)]) + out = await _drain(stream_admin_events(_wtad(fake))) + assert out[0].type == "" + assert out[0].data == {} + + async def test_passes_last_event_id(self) -> None: + fake = _FakeAdmin(events=[]) + await _drain(stream_admin_events(_wtad(fake), last_event_id=42)) + assert fake.calls[-1] == ("stream_events", (), {"last_event_id": 42}) + + async def test_non_200_apierror_maps_to_sse_connect_failed(self) -> None: + # The SDK admin stream raises ApiError("admin_stream_failed", status=…) on a + # non-200 open (NOT ConnectFailed) — mapped → SseConnectFailed for the web. + fake = _FakeAdmin(stream_error=ApiError("admin_stream_failed", "no", status=502)) + with pytest.raises(SseConnectFailed) as ei: + await _drain(stream_admin_events(_wtad(fake))) + assert ei.value.status == 502 + + async def test_connection_dropped_maps_and_carries_cursor(self) -> None: + # Both a connect-time failure (cursor None) and a mid-stream drop / resumable + # EOF (cursor set) surface as ConnectionDropped → SseConnectionDropped. + fake = _FakeAdmin(stream_error=wtsdk.ConnectionDropped("42")) + with pytest.raises(SseConnectionDropped) as ei: + await _drain(stream_admin_events(_wtad(fake))) + assert ei.value.last_seen_sse_id == "42" diff --git a/uv.lock b/uv.lock index c993a30..4a2f070 100644 --- a/uv.lock +++ b/uv.lock @@ -472,7 +472,7 @@ wheels = [ [[package]] name = "ratatoskr" -version = "0.21.18" +version = "0.21.19" source = { editable = "." } dependencies = [ { name = "httpx" },