feat(#20): admin (bifrost inspection + admin-events stream) onto the wt adapter (slice-6)

Slice-6 of the worldtree-sdk cutover: migrate the two admin routes off the
hand-rolled paths onto the `ratatoskr.wt` adapter over `client.admin.*`, and delete
the retired code. Both are web-only (the coverage-map's `tui.py` rows were stale —
corrected to `web/server.py`).

Adapter (`wt.py`): `get_session_bifrost` → `client.admin.sessions.bifrost` (open-world
dict verbatim, any error → SessionApiFailed default); `stream_admin_events` →
`client.admin.stream_events`, re-wrapping the SDK's `AdminEvent` → ratatoskr's at the
boundary.

Decisions (contract § slice-6 notes):
- Admin auth moves from a per-call `Authorization` header override to the client's
  `admin_auth` (`_wt_client(admin_key=…)`, extended this slice) — the SDK's admin.*
  routes use the provider, not a header.
- `AdminEvent` re-wrap (chosen over yield-through): the SDK's `admin_id`(nan)/None-able
  `type`/`data` diverge from ratatoskr's `id`/`type`/`data` that the web filter reads;
  re-wrapping (nan→0, None→""/{}) degrades the open-world None/nan ONCE at the adapter
  and keeps the web endpoint + `_admin_event_matches_web` + the `AdminEvent` domain type
  unchanged (preserves the web surface). Rejected: yield SDK events + rewire the web
  filter (heavier churn, scattered hardening).
- Admin-stream error map: a NON-200 open raises `ApiError("admin_stream_failed")`
  (NOT `ConnectFailed`) → SseConnectFailed; `ConnectionDropped` (connect-time OR
  mid-stream/resumable-EOF) → SseConnectionDropped. The web integration test caught the
  ApiError-not-ConnectFailed gotcha the unit fake couldn't.

Web (`web/server.py`): both admin endpoints build the wt client with admin_key and call
`wt.*`; the bifrost endpoint gains ConnectFailed→502 handling (cutover foot-gun); the
admin-events endpoint closes the injected transport (INV-CUT-1), never the wt client.

Deleted the hand-rolled `sessions.get_session_bifrost` + `sse_client.stream_admin_events`
(+ orphaned httpx/httpx_sse/json/AsyncIterator imports); the ratatoskr `AdminEvent`
dataclass stays in `sse_client.py` (re-wrap target, imported by wt + web) until slice-7.
Retired `test_sse_client.py` entirely (its last test was the admin stream) and the
`test_sessions.py` `TestGetSessionBifrost`; added the slice-6 adapter tests.

LIVE SMOKE (:8081, readonly-admin key) — INV-CUT-5 / DEC-4 cleared: the web bifrost
endpoint returned an admin-authed clean 404 envelope (auth + route + mapping proven);
a real `session.created` admin event (id=32) re-wrapped cleanly on live wire (driven by
a session-create, throwaway session cleaned up).

Suite 490 green; ruff clean; mypy net-improved on web/server.py (16→12 pre-existing, no
new). Patch bump 0.21.18 → 0.21.19 (the cutover MINOR is deferred to slice-7, DEC-6).
This commit is contained in:
vh
2026-07-19 12:48:04 -07:00
parent 5bc39a092e
commit de9a5baf45
11 changed files with 309 additions and 266 deletions
@@ -295,6 +295,46 @@ re-anchor its coverage-map rows.
web-server caller — only the `--whoami` and `--characters` CLI one-shot probes. The web-server caller — only the `--whoami` and `--characters` CLI one-shot probes. The
web surface is untouched this slice. 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 ## Out of scope
- Bifrost PROVIDER planes (memory/affect) — hand-rolled, ADR-0009, untouched. - Bifrost PROVIDER planes (memory/affect) — hand-rolled, ADR-0009, untouched.
+2 -2
View File
@@ -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 /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 /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 /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/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) | ✅ | `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/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` | | `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…` | | `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 | | `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 |
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
[project] [project]
name = "ratatoskr" name = "ratatoskr"
version = "0.21.18" version = "0.21.19"
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"
-28
View File
@@ -6,9 +6,6 @@ Implements docs/contracts/issues/2.contract.md.
from __future__ import annotations from __future__ import annotations
from dataclasses import dataclass from dataclasses import dataclass
from typing import Any
import httpx
@dataclass(frozen=True) @dataclass(frozen=True)
@@ -228,28 +225,3 @@ def endpoint_for_plane(plane: str, base_host: str) -> str:
if plane not in ports: if plane not in ports:
raise ValueError(f"unknown plane: {plane!r} (expected 'memory', 'affect', or 'combined')") raise ValueError(f"unknown plane: {plane!r} (expected 'memory', 'affect', or 'combined')")
return f"http://{base_host}:{ports[plane]}" 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)
-53
View File
@@ -5,14 +5,9 @@ Implements docs/contracts/issues/1.contract.md.
from __future__ import annotations from __future__ import annotations
import json
from collections.abc import AsyncIterator
from dataclasses import dataclass from dataclasses import dataclass
from typing import Any, NamedTuple from typing import Any, NamedTuple
import httpx
import httpx_sse
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."""
@@ -173,51 +168,3 @@ class CancelFailed(Exception):
super().__init__(f"cancel failed: status={status}, body={body[:128]!r}") super().__init__(f"cancel failed: status={status}, body={body[:128]!r}")
self.status = status self.status = status
self.body = body 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
+41 -17
View File
@@ -45,15 +45,14 @@ from ratatoskr.sessions import (
BifrostConsumerKeyMissing, BifrostConsumerKeyMissing,
BifrostHandshakeFailed, BifrostHandshakeFailed,
PersonaNotConfigured, PersonaNotConfigured,
SessionApiFailed,
endpoint_for_plane, endpoint_for_plane,
get_session_bifrost,
) )
# The turn path (create / stream / cancel / tools / messages) AND the agents / # The turn path (create / stream / cancel / tools / messages), the agents /
# persona-state reads are served by the worldtree-sdk adapter (`wt.*`), which raises # persona-state reads, AND the admin surface (bifrost inspection + admin-events stream)
# ratatoskr's caller-semantic exceptions (DEC-2). The remaining hand-rolled endpoint # are all served by the worldtree-sdk adapter (`wt.*`), which raises ratatoskr's
# (admin bifrost) stays on the `sessions` / `sse_client` wrappers until slice 6. # 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 ( from ratatoskr.sse_client import (
AdminEvent, AdminEvent,
CancelAlreadyCompleted, CancelAlreadyCompleted,
@@ -64,16 +63,21 @@ from ratatoskr.sse_client import (
SseConnectFailed, SseConnectFailed,
SseConnectionDropped, SseConnectionDropped,
TurnIdFlip, 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: """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 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 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 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" base_url = str(client.base_url) or "http://localhost"
header = client.headers.get("Authorization", "") header = client.headers.get("Authorization", "")
# Case-insensitive scheme + tolerant of extra whitespace, so a valid bearer is # 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) parts = header.split(None, 1)
api_key = parts[1].strip() if len(parts) == 2 and parts[0].lower() == "bearer" else "" api_key = parts[1].strip() if len(parts) == 2 and parts[0].lower() == "bearer" else ""
return wt.build_client( 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) return JSONResponse({"error_code": "admin_key_not_configured"}, status_code=400)
client_factory = request.app.state.client_factory client_factory = request.app.state.client_factory
try: try:
async with client_factory() as client: async with client_factory() as transport:
bstate = await get_session_bifrost(client, session_id, admin_key=admin_key) # slice-6: the SDK's admin.* routes use the client's admin_auth (built with
except SessionApiFailed as exc: # 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( return JSONResponse(
{"error_code": "bifrost_state_unavailable", "status": exc.status}, {"error_code": "bifrost_state_unavailable", "status": exc.status},
status_code=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: 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 client_factory = request.app.state.client_factory
async def gen() -> AsyncIterator[bytes]: 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: 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): if not _admin_event_matches_web(ev, session_id):
continue continue
# Fixed SSE event name so the browser renders EVERY admin type # 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: except asyncio.CancelledError:
raise # browser disconnect — let the generator unwind raise # browser disconnect — let the generator unwind
finally: 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") return StreamingResponse(gen(), media_type="text/event-stream")
+64
View File
@@ -60,6 +60,7 @@ from .sessions import (
Tier3UserIdUnsupported, Tier3UserIdUnsupported,
) )
from .sse_client import ( from .sse_client import (
AdminEvent,
AgentNotAvailable, AgentNotAvailable,
CancelAlreadyCompleted, CancelAlreadyCompleted,
CancelFailed, CancelFailed,
@@ -684,3 +685,66 @@ async def delete_character(
return await client.characters.delete(character_id) return await client.characters.delete(character_id)
except ApiError as exc: except ApiError as exc:
raise translate_error(exc) from 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
+7 -70
View File
@@ -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 pytest
import respx
from ratatoskr.sessions import ( from ratatoskr.sessions import endpoint_for_plane
SessionApiFailed,
endpoint_for_plane,
get_session_bifrost,
)
class TestEndpointForPlane: class TestEndpointForPlane:
@@ -30,65 +29,3 @@ class TestEndpointForPlane:
"""unknown_plane [adversarial]: any other plane → ValueError (PRE-001).""" """unknown_plane [adversarial]: any other plane → ValueError (PRE-001)."""
with pytest.raises(ValueError): with pytest.raises(ValueError):
endpoint_for_plane("persona", "10.100.10.50") 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
-94
View File
@@ -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"]
+153
View File
@@ -38,6 +38,7 @@ from ratatoskr.sessions import (
Tier3UserIdUnsupported, Tier3UserIdUnsupported,
) )
from ratatoskr.sse_client import ( from ratatoskr.sse_client import (
AdminEvent,
AgentNotAvailable, AgentNotAvailable,
CancelAlreadyCompleted, CancelAlreadyCompleted,
CancelFailed, CancelFailed,
@@ -62,6 +63,7 @@ from ratatoskr.wt import (
get_character_state, get_character_state,
get_me, get_me,
get_persona_state, get_persona_state,
get_session_bifrost,
get_session_messages, get_session_messages,
get_session_tools, get_session_tools,
list_agents, list_agents,
@@ -69,6 +71,7 @@ from ratatoskr.wt import (
list_sessions, list_sessions,
patch_agent, patch_agent,
set_persona_state, set_persona_state,
stream_admin_events,
stream_turn, stream_turn,
translate_error, translate_error,
write_authored_history, write_authored_history,
@@ -1066,3 +1069,153 @@ class TestDeleteCharacter:
with pytest.raises(SessionApiFailed) as ei: with pytest.raises(SessionApiFailed) as ei:
await delete_character(_wtm(fake), "char_x") await delete_character(_wtm(fake), "char_x")
assert ei.value.status == 500 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"
Generated
+1 -1
View File
@@ -472,7 +472,7 @@ wheels = [
[[package]] [[package]]
name = "ratatoskr" name = "ratatoskr"
version = "0.21.18" version = "0.21.19"
source = { editable = "." } source = { editable = "." }
dependencies = [ dependencies = [
{ name = "httpx" }, { name = "httpx" },