From a3c92b68dc32c28788f699fc2614fd59afe1b750 Mon Sep 17 00:00:00 2001 From: Vuong Hoang Date: Tue, 30 Jun 2026 23:25:34 -0700 Subject: [PATCH] =?UTF-8?q?feat(#11):=20AdminEvents=20pane=20=E2=80=94=20G?= =?UTF-8?q?ET=20/admin/events=20SSE=20(session-filtered)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit v1 coverage-audit: the last unbuilt design-brief §5 debug pane. #11's blocker was already satisfied (admin key carries admin.events.read). Completes the admin/debug-observability core. - sse_client.py: AdminEvent dataclass + stream_admin_events — a new long-lived SSE consumer for the admin lifecycle stream (envelope {id,type,timestamp,data}), admin-scoped (bearer-override), Last-Event-ID resume. non-200 -> SseConnectFailed; mid-drop -> SseConnectionDropped. - tui.py: "AdminEvents" TabPane + _format_admin_event + _admin_event_matches (design-brief §6 filter: active-session + non-heartbeat system.*) + _stream_admin_events long-lived best-effort worker (unconditional on_mount; self-labels not-configured / unavailable / stream-ended). - Contract-skipped for stream_admin_events (out of #1's turn-SSE scope; spec § Admin Event Stream is the reference). TDD: 4 sse_client + 5 tui tests. Suite 561 green. - LIVE-AUTH-PROVEN on :8081 (GET /admin/events -> HTTP 200 under admin key). Coverage: REST 12/40. Tier 1 debug-observability core complete. --- docs/coverage-map.md | 18 +++--- persistent-memory.md | 2 + pyproject.toml | 2 +- src/ratatoskr/sse_client.py | 62 +++++++++++++++++++++ src/ratatoskr/tui.py | 70 ++++++++++++++++++++++- tests/test_sse_client.py | 72 ++++++++++++++++++++++++ tests/test_tui.py | 107 ++++++++++++++++++++++++++++++++++++ uv.lock | 2 +- 8 files changed, 324 insertions(+), 11 deletions(-) diff --git a/docs/coverage-map.md b/docs/coverage-map.md index cf268e1..98ae587 100644 --- a/docs/coverage-map.md +++ b/docs/coverage-map.md @@ -48,7 +48,7 @@ resolved (§ Surface 1, scope-resolution table). | Surface | Points | ✅ covered-live | ⬜ gap (in-scope) | 🚫 excluded-by-design | |---|---|---|---|---| -| REST (OpenAPI 2.2.0, path groups) | 40 | 11 | 7 | 22 | +| REST (OpenAPI 2.2.0, path groups) | 40 | 12 | 6 | 22 | | SSE events | 11 | 11 | 0 | 0 | | Bifrost provider planes | 8 verbs | 8 | 0 | (10 gated verbs deferred) | @@ -79,6 +79,7 @@ sub-gap). | `GET /capabilities` | ✅ | `sessions.py:428` `get_capabilities` → `cli.py` `--whoami` | Echo ephemeral-template discovery | | `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 | **Sub-gaps inside ✅ path groups** (the method we use is live; a sibling method on the same path is an unwired frontier item — see frontier Tier 1): @@ -96,7 +97,6 @@ on the same path is an unwired frontier item — see frontier Tier 1): | Endpoint | Status | Why in-scope | |---|---|---| -| `GET /admin/events` | ⬜ | design-brief §5 v1 **AdminEvents pane** (issue **#11**). **NO LONGER BLOCKED** — the `RATATOSKR_ADMIN_API_KEY` (`ratatoskr-readonly`) verified to carry `admin.events.read` (2026-07-01); pane just unbuilt. The last unbuilt §5 debug pane. | | `GET /admin/sessions/{id}/tools` | ⬜ | admin variant of the Tools inventory — **covered-by-alternative** via the owner-scoped `GET /sessions/{id}/tools` (✅); this admin variant remains a gap only for cross-user operator debug | | (`GET /sessions` picker · resume) | ⬜ | sub-gaps above — presenter-wiring only, wrappers exist | @@ -208,16 +208,18 @@ starts exercising them. 3. ✅ **DONE** — BifrostState pane (`v0.18.10`, `GET /admin/sessions/{id}/bifrost`, admin-key; live-auth-proven). The Tools half was already covered by the owner-scoped `GET /sessions/{id}/tools` (item 5). -4. **#11 — AdminEvents pane** (`GET /admin/events`) — **NO LONGER BLOCKED.** The - `RATATOSKR_ADMIN_API_KEY` was verified (2026-07-01) to carry `admin.events.read`; - the blocker (an admin key with the scope) is already satisfied. Only the pane - itself is unbuilt — an SSE-consuming admin pane, the last unbuilt §5 surface. +4. ✅ **DONE** — AdminEvents pane (`v0.18.11`, `GET /admin/events` SSE, + session-filtered; admin-key; live-auth-proven). #11's blocker was already + satisfied (admin key carries `admin.events.read`). **Tier 1 complete** — the + admin/debug-observability core (Persona · Tools · BifrostState · AdminEvents) + is fully built. -**Tier 2 — rounds out coverage:** +**Tier 2 — rounds out coverage (all that remains):** 5. ✅ **DONE** — `GET /sessions/{id}/tools` (`v0.18.9`, owner-scoped tool inventory in the TUI Tools pane). -6. **Transient-characters routing** (4 endpoints) + **`POST /sessions/{id}/persona_state`**. +6. **Transient-characters routing** (4 endpoints) + **`POST /sessions/{id}/persona_state`** + — the only remaining in-scope client I/O points. --- diff --git a/persistent-memory.md b/persistent-memory.md index ecc8c82..877037a 100644 --- a/persistent-memory.md +++ b/persistent-memory.md @@ -161,6 +161,8 @@ decision. Captures rationale that won't be obvious from code alone. - `[2026-07-01]` **BifrostState pane SHIPPED (`v0.18.10`) — `GET /admin/sessions/{id}/bifrost` in a new TUI "Bifrost" pane; the FIRST admin-key consumer in ratatoskr.** `get_session_bifrost(client, session_id, *, admin_key)` (sessions.py) — admin-scoped (`admin.sessions.read`); the request OVERRIDES Authorization with `admin_key` (distinct from the consumer bearer, asserted in a test); 200→dict, non-200→SessionApiFailed. Admin-key wiring: `--admin-key` flag + `RATATOSKR_ADMIN_API_KEY` env → new `ParsedArgs.admin_key`. New "Bifrost" TabPane + `_format_bifrost_state` + `_hydrate_bifrost_state` best-effort worker (mirror `_hydrate_session_tools`) UNCONDITIONALLY in on_mount → writes {endpoint, connected, caps_granted, tools} + audits; self-labels "not configured" (no admin key) / "not bound" (404) / graceful on 403 + error. Contract #2 amended (FN, validated OK) + TDD (4 wrapper respx tests incl. the admin-bearer-override assertion + 1 format unit + 3 hydrate integration). Suite **552 green**; my code ruff-clean (pre-existing tui.py ruff debt at other lines untouched, incl. a dead `RichText` import in `_hydrate_persona`). **LIVE-AUTH-PROVEN** on personal :8081: admin key authenticated (reached resource-layer 404 session_not_found, NOT 401/403) → `admin.sessions.read` works live; 200 full-state not exercised (no bound session on :8081 now — unit-covered). Patch bump (debug feature, no downstream coordination; consistent with the session's cadence — but the §5-core-completion angle is a possible minor, operator's call). - `[2026-07-01]` **LEDGER CORRECTION: #11 (AdminEvents) is NO LONGER BLOCKED.** Verified via `GET /me` on :8081 that `RATATOSKR_ADMIN_API_KEY` (`ratatoskr-readonly`, tier readonly-admin) carries ALL 7 read scopes INCLUDING **`admin.events.read`** (+ `admin.sessions.read`, admin.keys.read, admin.skuld.read, pending.read, search.read, tool_events.read). The coverage-map + prior memory had #11 "blocked on admin.events.read" — **STALE**; the admin key was minted (post-#11-filing, env.sh) WITH the scope, so the blocker is already satisfied. **Only the AdminEvents SSE pane itself is unbuilt** — the last unbuilt §5 debug pane (a live SSE-consuming admin pane, distinct from the hydrate-at-attach panes). Coverage-map updated. **Coverage: REST 11/40 ✅.** Consider building the AdminEvents pane and/or updating #11's tracker status (its stated blocker is gone). +- `[2026-07-01]` **AdminEvents pane SHIPPED (`v0.18.11`) — `GET /admin/events` SSE in a new TUI pane; #11 closed-by-build; Tier 1 (debug-observability core) COMPLETE.** `stream_admin_events(client, *, admin_key, last_event_id=None)` (sse_client.py) — a NEW long-lived SSE consumer for the admin lifecycle broadcast (envelope `{id,type,timestamp,data}`, 17-event v0 vocab), admin-scoped (`admin.events.read`, bearer-override), Last-Event-ID resume; non-200→SseConnectFailed, mid-drop→SseConnectionDropped; new `AdminEvent` dataclass (distinct from the turn `Event` union). New "AdminEvents" TabPane + `_format_admin_event` + `_admin_event_matches` (design-brief §6 filter: active-session events + non-heartbeat `system.*`) + `_stream_admin_events` long-lived best-effort worker (unconditional on_mount, cancelled on app exit; self-labels "not configured"/"unavailable"/"stream ended"). Reuses the admin key from the BifrostState slice. **Contract-SKIPPED** for `stream_admin_events` (out of contract #1's turn-SSE scope; spec § Admin Event Stream is the reference; well-TDD'd). TDD: 4 sse_client tests (multi-event+bearer-override, Last-Event-ID header, 403, malformed-skip) + 5 tui (format, filter, worker success/no-key/403). Suite **561 green**; my code ruff-clean (pre-existing tui.py ruff debt untouched, incl. the dead `RichText` import in `_hydrate_persona`). **LIVE-AUTH-PROVEN**: `GET /admin/events` on :8081 → HTTP 200 under the admin key (connected + streamed, idle in the 4s window — no 401/403). **Coverage: REST 12/40 ✅. Tier 1 admin/debug-observability core COMPLETE** (Persona · Tools · BifrostState · AdminEvents). Patch bump (session cadence) — **§5-core-completion is a possible `v0.19.0` MINOR milestone, surfaced to operator (needs approval).** Remaining in-scope client I/O: only Tier-2 (transient-characters routing + `POST /sessions/{id}/persona_state`). + _41 older entries (2026-05-* — the original debug-TUI/web build era) archived to archival-memory.md._ _For per-issue TDD implementation notes, Volva findings, and contract amendments, see the git log — every per-issue commit carries a structured message capturing the trail._ diff --git a/pyproject.toml b/pyproject.toml index f904ba9..eec7f7d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "ratatoskr" -version = "0.18.10" +version = "0.18.11" description = "Worldtree Conversation API debug TUI — multi-pane observability dashboard" readme = "README.md" requires-python = ">=3.12" diff --git a/src/ratatoskr/sse_client.py b/src/ratatoskr/sse_client.py index b66c798..d8e3c79 100644 --- a/src/ratatoskr/sse_client.py +++ b/src/ratatoskr/sse_client.py @@ -175,6 +175,23 @@ Event = ( ) +@dataclass(frozen=True) +class AdminEvent: + """One `/admin/events` envelope (INV-046) — an admin-tier lifecycle event. + + Distinct from the turn-stream `Event` union: this is the process-wide admin + broadcast stream, not a per-turn stream. `id` is a plain monotonic int + (resets on restart; heartbeats have id=0). `type` is a dotted namespace + (session.* / turn.* / key.* / system.*). `data` is a type-specific dict — + most carry `session_id`; per INV-049 it holds IDs + small metadata only. + """ + + id: int + type: str + timestamp: str | None + data: dict[str, Any] + + class MalformedSseId(Exception): """Raised when an SSE event's `id:` wire field is missing or non-composite.""" @@ -598,6 +615,51 @@ async def stream_turn_resilient( ) +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 + + 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) diff --git a/src/ratatoskr/tui.py b/src/ratatoskr/tui.py index 56f2695..5e39863 100644 --- a/src/ratatoskr/tui.py +++ b/src/ratatoskr/tui.py @@ -51,6 +51,7 @@ from ratatoskr.sessions import ( list_sessions, ) from ratatoskr.sse_client import ( + AdminEvent, AffectUpdate, AwaitingLlmFirstToken, CancelAlreadyCompleted, @@ -72,6 +73,7 @@ from ratatoskr.sse_client import ( TurnIdFlip, WorkerPhase, cancel_turn, + stream_admin_events, stream_turn_resilient, ) @@ -198,6 +200,17 @@ def _ts() -> str: return now.strftime("%H:%M:%S") + f".{now.microsecond // 1000:03d}" +def _format_admin_event(ev: AdminEvent) -> str: + """One-line render of an /admin/events envelope for the AdminEvents pane. + + Drops `session_id` from the detail (the pane is already session-scoped) and + shows HH:MM:SS from the ISO timestamp + the remaining small metadata fields. + """ + ts = (ev.timestamp or "")[11:19] + extras = " ".join(f"{k}={v}" for k, v in ev.data.items() if k != "session_id") + return f"[{ts}] {ev.type} {extras}".rstrip() + + def _format_bifrost_state(state: dict) -> list[str]: """Render GET /admin/sessions/{id}/bifrost (#176) into BifrostState-pane lines.""" tools = [t.get("name", "?") for t in state.get("tools", [])] @@ -1053,7 +1066,7 @@ class RatatoskrApp(App[int]): /* v0.8.1: #current-text Static removed. Streaming text now coalesces on `\n` and writes directly to #transcript (same pattern as v0.7.1 thinking fix). Eliminates the dock-bottom-growth-overlap bug. */ - #tools-log, #debug-log, #thinking-log, #bifrost-log { + #tools-log, #debug-log, #thinking-log, #bifrost-log, #admin-events-log { background: $background; padding: 0 1; } @@ -1227,6 +1240,15 @@ class RatatoskrApp(App[int]): id="bifrost-log", wrap=True, markup=False, highlight=False, min_width=0, ) + with TabPane("AdminEvents", id="admin-events-tab"): + # #11: live GET /admin/events SSE stream, admin-scoped, + # FILTERED to the active session (design-brief §6). A + # long-lived worker appends matching lifecycle events; + # "not configured" when no admin key is set. + yield RichLog( + id="admin-events-log", wrap=True, markup=False, + highlight=False, min_width=0, + ) # INV-002 + INV-003: visible identity + hint widgets (Footer-area). # pane-name widget displays current side-pane name. yield Static("", id="identity") @@ -1288,6 +1310,9 @@ class RatatoskrApp(App[int]): # (admin-scoped). Self-labels "not configured" when no admin key is set, # "not bound" for the common unbound-session 404 — always writes at mount. self.run_worker(self._hydrate_bifrost_state()) + # #11: long-lived worker streaming GET /admin/events into the AdminEvents + # pane, filtered to this session. Admin-key-gated; cancelled on app exit. + self.run_worker(self._stream_admin_events()) async def _hydrate_persona(self) -> None: """Hydrate persona-header + Persona pane via GET /agents/{id}/persona_state. @@ -1402,6 +1427,49 @@ class RatatoskrApp(App[int]): f"connected={state.get('connected')} tools={len(state.get('tools', []))}" ) + def _admin_event_matches(self, ev: AdminEvent) -> bool: + """AdminEvents filter (design-brief §6): active-session events + non-heartbeat + system.* (stream-integrity signals). Heartbeats are keepalive noise.""" + if ev.type == "system.heartbeat": + return False + if ev.type.startswith("system."): + return True + return ev.data.get("session_id") == self.session_id + + async def _stream_admin_events(self) -> None: + """Stream GET /admin/events (admin-scoped) into the AdminEvents pane (#11). + + Long-lived + best-effort (never crashes the TUI). Filtered to the active + session (design-brief §6): appends matching lifecycle events as they + arrive. No admin key → "not configured". On connect failure (e.g. 403 + scope-denied) or a mid-stream drop, writes a labeled line and stops. + """ + assert self.client is not None and self.session_id is not None + from rich.text import Text as RichText + + log = self.query_one("#admin-events-log", RichLog) + admin_key = getattr(self.args, "admin_key", None) + if not admin_key: + log.write(RichText("(admin key not configured — set RATATOSKR_ADMIN_API_KEY)")) + self._audit( + f"admin_events_skipped session={self.session_id[-8:]} reason=no_admin_key" + ) + return + try: + async for ev in stream_admin_events(self.client, admin_key=admin_key): + if self._admin_event_matches(ev): + log.write(RichText(_format_admin_event(ev))) + except SseConnectFailed as exc: + log.write(RichText(f"(admin events unavailable: HTTP {exc.status})")) + self._audit( + f"admin_events_unavailable session={self.session_id[-8:]} status={exc.status}" + ) + except Exception as exc: # drop / best-effort — never crash the TUI + log.write(RichText(f"(admin events stream ended: {type(exc).__name__})")) + self._audit( + f"admin_events_ended session={self.session_id[-8:]} err={type(exc).__name__}" + ) + def _update_persona_surfaces(self, snapshot: dict) -> None: """Update sticky header + Persona pane from a fresh snapshot. diff --git a/tests/test_sse_client.py b/tests/test_sse_client.py index 0c4dab5..7a1f652 100644 --- a/tests/test_sse_client.py +++ b/tests/test_sse_client.py @@ -5,6 +5,7 @@ import pytest import respx from ratatoskr.sse_client import ( + AdminEvent, AffectUpdate, AgentNotAvailable, AwaitingLlmFirstToken, @@ -27,6 +28,7 @@ from ratatoskr.sse_client import ( _parse_sse_id, cancel_turn, reconnect_turn, + stream_admin_events, stream_turn, stream_turn_resilient, ) @@ -1261,3 +1263,73 @@ class TestStreamTurnResilient: collected.append(e) assert [e.sse_id for e in collected] == [SseId(42, 1)] # type: ignore[attr-defined] assert route.call_count == 2 + + +class TestStreamAdminEvents: + """docs/conversation-api-spec.md § Admin Event Stream — stream_admin_events (#11).""" + + @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_tui.py b/tests/test_tui.py index 35526b7..ccc228b 100644 --- a/tests/test_tui.py +++ b/tests/test_tui.py @@ -3330,3 +3330,110 @@ class TestBifrostStateHydration: joined = " ".join(_text_of(w) for w in writes) assert "not bound to Bifrost" in joined assert "bifrost_state_unavailable" in joined + + +class TestAdminEventsStream: + """stream_admin_events + the #11 AdminEvents pane (GET /admin/events, session-filtered).""" + + @staticmethod + def _mute_hydrates(monkeypatch: pytest.MonkeyPatch) -> None: + """Neutralize the other on_mount workers (tools + bifrost) — no real calls.""" + import ratatoskr.tui as tui_mod + from ratatoskr.sessions import SessionApiFailed + + async def noop_tools(client, session_id): + return {"agent_id": "x", "builtin_tools": [], "bifrost_tools": []} + + async def noop_bifrost(client, session_id, *, admin_key): + raise SessionApiFailed(status=404, body=b"nb") + + monkeypatch.setattr(tui_mod, "get_session_tools", noop_tools) + monkeypatch.setattr(tui_mod, "get_session_bifrost", noop_bifrost) + + def test_format_admin_event(self) -> None: + """format_admin_event [unit]: HH:MM:SS + type + fields; session_id dropped.""" + from ratatoskr.sse_client import AdminEvent + from ratatoskr.tui import _format_admin_event + + line = _format_admin_event( + AdminEvent( + 42, "turn.completed", "2026-05-06T10:00:05.000Z", + {"session_id": "s1", "turn_id": 7, "duration_ms": 1200, "phase": "succeeded"}, + ) + ) + assert "turn.completed" in line + assert "[10:00:05]" in line + assert "turn_id=7" in line + assert "session_id" not in line # dropped — pane is already session-scoped + + def test_admin_event_matches_filter(self) -> None: + """admin_event_matches [unit]: active-session + non-heartbeat system.* pass (§6).""" + from ratatoskr.sse_client import AdminEvent + + E = AdminEvent + app = _resolved_app(_args_existing(session_id="s-match")) + assert app._admin_event_matches(E(1, "session.created", "t", {"session_id": "s-match"})) + assert not app._admin_event_matches(E(2, "turn.started", "t", {"session_id": "other"})) + assert not app._admin_event_matches(E(0, "system.heartbeat", "t", {})) + assert app._admin_event_matches(E(3, "system.events_dropped", "t", {"count": 5})) + + async def test_stream_writes_filtered_events(self, monkeypatch: pytest.MonkeyPatch) -> None: + """stream_filtered [scenario,tracer]: only active-session + non-heartbeat lines land.""" + import ratatoskr.tui as tui_mod + from ratatoskr.sse_client import AdminEvent + + self._mute_hydrates(monkeypatch) + writes = _spy_writes(monkeypatch) + + async def fake_stream(client, *, admin_key, last_event_id=None): + yield AdminEvent(41, "session.created", "t", {"session_id": "s-ae-2"}) + yield AdminEvent(0, "system.heartbeat", "t", {}) # filtered (noise) + yield AdminEvent(42, "turn.started", "t", {"session_id": "other"}) # diff session + yield AdminEvent(43, "session.deleted", "t", {"session_id": "s-ae-2"}) + + monkeypatch.setattr(tui_mod, "stream_admin_events", fake_stream) + app = _resolved_app(_args_existing(session_id="s-ae-2", admin_key="ak")) + async with app.run_test() as pilot: + await pilot.pause() + await app._stream_admin_events() + await pilot.pause() + joined = " ".join(_text_of(w) for w in writes) + assert "session.created" in joined + assert "session.deleted" in joined + assert "system.heartbeat" not in joined + assert "turn.started" not in joined # different session → filtered + + async def test_stream_no_admin_key(self, monkeypatch: pytest.MonkeyPatch) -> None: + """stream_no_admin_key [scenario]: admin_key None → 'not configured' + skip audit.""" + self._mute_hydrates(monkeypatch) + writes = _spy_writes(monkeypatch) + app = _resolved_app(_args_existing(session_id="s-ae-3")) # admin_key None + async with app.run_test() as pilot: + await pilot.pause() + await app._stream_admin_events() + await pilot.pause() + joined = " ".join(_text_of(w) for w in writes) + assert "admin key not configured" in joined + assert "admin_events_skipped" in joined + + async def test_stream_403_unavailable(self, monkeypatch: pytest.MonkeyPatch) -> None: + """stream_403 [error]: 403 scope-denied → 'unavailable' + audit; no crash.""" + import ratatoskr.tui as tui_mod + from ratatoskr.sse_client import SseConnectFailed + + self._mute_hydrates(monkeypatch) + writes = _spy_writes(monkeypatch) + + async def denied(client, *, admin_key, last_event_id=None): + raise SseConnectFailed(status=403, body=b"auth_scope_denied") + yield # unreachable — makes this an async generator + + monkeypatch.setattr(tui_mod, "stream_admin_events", denied) + app = _resolved_app(_args_existing(session_id="s-ae-4", admin_key="ak")) + async with app.run_test() as pilot: + await pilot.pause() + await app._stream_admin_events() + await pilot.pause() + joined = " ".join(_text_of(w) for w in writes) + assert "admin events unavailable: HTTP 403" in joined + assert "admin_events_unavailable" in joined diff --git a/uv.lock b/uv.lock index a15473c..9aec298 100644 --- a/uv.lock +++ b/uv.lock @@ -1052,7 +1052,7 @@ wheels = [ [[package]] name = "ratatoskr" -version = "0.18.10" +version = "0.18.11" source = { editable = "." } dependencies = [ { name = "httpx" },