Task 92f75680: Admin transition SSE stream (#202)

api/routes_admin.py

@@ -290,6 +290,16 @@ an omitted field is never passed to
 :func:`nestquest_core.children.edit_child` and stays unchanged — while
 an explicit ``null`` IS passed and is rejected by the core rather than
 silently ignored.
+
+The EVENTS route (``GET /events``) is the admin plane's live
+transition stream for the Admin PWA: the SAME four transitions
+(quest-completed, quest-uncompleted, quest-missed, child-day-complete)
+off the SAME in-process publisher as the panel plane, framed by the
+shared :func:`api.sse.transition_event_stream` helper.  It inherits the
+router's ONE :func:`~api.auth.require_admin` check (the Authorization
+header — the PWA reads it with a fetch-based reader because
+``EventSource`` cannot set headers), so the panel service token is
+refused here exactly as on every other admin route.
 """
 from __future__ import annotations
 
@@ -297,6 +307,7 @@ import datetime
 from typing import Annotated, Literal, NoReturn
 
 from fastapi import APIRouter, Depends, HTTPException, Request, Response
+from fastapi.responses import StreamingResponse
 from pydantic import (
     BaseModel,
     StrictBool,
@@ -325,6 +336,7 @@ from api.nestquest_core import (
     core_snapshot,
     core_sweep,
 )
+from api.sse import transition_event_stream
 
 #: All admin-plane routes share this router; the ONE admin dependency
 #: (service-token refusal + JWT verification + nestquest-admins group
@@ -2205,4 +2217,25 @@ async def admin_snapshot(request: Request) -> AdminSnapshotResponse:
     )
 
 
+
+@router.get(
+    "/events",
+    summary="Stream admin transition events (SSE)",
+)
+async def admin_events(request: Request) -> StreamingResponse:
+    """Stream the app's transition events to an admin client
+    (Server-Sent Events).
+
+    Requires a valid admin JWT (the router's shared
+    :func:`~api.auth.require_admin` dependency — the ONE check; this
+    handler performs NO auth of its own), so an unauthenticated or
+    panel-token request is refused before any stream is established.
+
+    Returns the shared :func:`api.sse.transition_event_stream` response
+    off the app's single in-process publisher (``app.state.publisher``)
+    — the same publisher, framing and teardown as the panel stream.
+    """
+    return transition_event_stream(request)
+
+
 __all__ = ["router"]

api/routes_panel.py

@@ -53,9 +53,6 @@ from __future__ import annotations
 
 import datetime
 import importlib
-import json
-
-from collections.abc import AsyncIterator
 
 from fastapi import APIRouter, Depends, HTTPException, Request
 from fastapi.responses import StreamingResponse
@@ -70,6 +67,7 @@ from api.nestquest_core import (
     core_snapshot,
 )
 from api import transitions as api_transitions
+from api.sse import transition_event_stream
 
 #: The instance DAO, imported on the app's core copy for the
 #: child-mismatch guard on the complete route.
@@ -328,23 +326,6 @@ async def panel_complete_instance(
     return {"status": str(result)}
 
 
-async def _sse_stream(request: Request) -> AsyncIterator[str]:
-    """Yield the SSE frames for every published transition event.
-
-    One subscriber of the app's single publisher (resolved off
-    ``app.state.publisher``); each event becomes one
-    ``event: <type>\\ndata: <json>\\n\\n`` frame.  Breaking out of the
-    publisher's subscription (on client disconnect) tears the
-    subscription down cleanly — no queue is leaked.
-    """
-    publisher = request.app.state.publisher
-    async for event_type, payload in publisher.subscribe():
-        yield (
-            f"event: {event_type}\n"
-            f"data: {json.dumps(payload)}\n\n"
-        )
-
-
 @router.get(
     "/events",
     summary="Stream panel transition events (SSE)",
@@ -357,17 +338,13 @@ async def panel_events(request: Request) -> StreamingResponse:
     :func:`~api.dependencies.require_panel_token` dependency — the ONE
     token check; this handler performs NO auth of its own).
 
-    Returns a ``text/event-stream`` response whose body is produced by
+    Returns the shared :func:`api.sse.transition_event_stream` response
-    :func:`_sse_stream` off the app's single in-process publisher
+    off the app's single in-process publisher (``app.state.publisher``,
-    (``app.state.publisher``, installed in the lifespan).  The stream
+    installed in the lifespan).  The stream ends when the client
-    ends when the client disconnects, at which point the subscription
+    disconnects, at which point the subscription is torn down (its
-    is torn down (its queue removed) with nothing leaked.
+    queue removed) with nothing leaked.
     """
-    return StreamingResponse(
+    return transition_event_stream(request)
-        _sse_stream(request),
-        media_type="text/event-stream",
-        headers={"Cache-Control": "no-cache"},
-    )
 
 
 __all__ = ["router"]

api/sse.py

@@ -0,0 +1,50 @@
+"""The ONE Server-Sent Events response helper for the API service.
+
+Both SSE planes — the panel plane (``/api/v1/panel/events``, panel
+service token, LAN-only) and the admin plane (``/api/v1/admin/events``,
+Authentik admin JWT) — stream the SAME transitions off the SAME
+in-process publisher (``app.state.publisher``, see :mod:`api.events`)
+with the SAME framing and teardown, so the stream lives here once and
+each plane's route only supplies its own router-level auth.
+"""
+from __future__ import annotations
+
+import json
+from collections.abc import AsyncIterator
+
+from fastapi import Request
+from fastapi.responses import StreamingResponse
+
+__all__ = ["transition_event_stream"]
+
+
+async def _sse_stream(request: Request) -> AsyncIterator[str]:
+    """Yield the SSE frames for every published transition event.
+
+    One subscriber of the app's single publisher (resolved off
+    ``app.state.publisher``); each event becomes one
+    ``event: <type>\\ndata: <json>\\n\\n`` frame.  Breaking out of the
+    publisher's subscription (on client disconnect) tears the
+    subscription down cleanly — no queue is leaked.
+    """
+    publisher = request.app.state.publisher
+    async for event_type, payload in publisher.subscribe():
+        yield (
+            f"event: {event_type}\n"
+            f"data: {json.dumps(payload)}\n\n"
+        )
+
+
+def transition_event_stream(request: Request) -> StreamingResponse:
+    """A ``text/event-stream`` response over the app's transition events.
+
+    Performs NO auth of its own: the calling route's router-level
+    dependency is the ONE check for its plane.  The stream ends when
+    the client disconnects, at which point the subscription is torn
+    down (its queue removed) with nothing leaked.
+    """
+    return StreamingResponse(
+        _sse_stream(request),
+        media_type="text/event-stream",
+        headers={"Cache-Control": "no-cache"},
+    )

tests/test_api_admin_sse.py

@@ -0,0 +1,283 @@
+"""Admin-plane SSE tests: the admin transition stream (task 92f75680).
+
+These tests exercise ``GET /api/v1/admin/events`` — the admin plane's
+Server-Sent Events stream for the Admin PWA — against a REAL server:
+
+- The stream inherits the admin router's ONE ``require_admin``
+  dependency: an absent credential is 401, the panel service token and
+  a non-admin JWT are 403, all before any stream is established.
+- An admin JWT subscribes; each of the four documented transitions
+  (quest-completed, quest-uncompleted, quest-missed, child-day-complete)
+  arrives as an ``event: <type>`` / ``data: <json>`` frame carrying its
+  documented payload fields (design/ENTITIES-AND-SERVICES.md §3).
+- The admin and panel streams are fed by the SAME in-process publisher:
+  one transition arrives, identically, on both.
+
+Streaming approach: :mod:`tests.test_api_sse`'s live-uvicorn reader
+(httpx's ASGITransport buffers SSE bodies), on the admin-enabled
+:class:`tests.test_api_admin_uncomplete._AdminServer` whose JWKS cache
+is stubbed, so a real admin JWT verifies with NO network access.
+"""
+from __future__ import annotations
+
+import datetime
+from types import SimpleNamespace
+from zoneinfo import ZoneInfo
+
+import httpx
+import pytest
+
+from api import routes_admin, routes_panel
+from tests.admin_jwt_harness import make_token
+from tests.test_api_admin_uncomplete import _AdminServer
+from tests.test_api_panel import PANEL_TOKEN, _seed_household
+from tests.test_api_sse import (
+    EVENT_CHILD_DAY_COMPLETE,
+    EVENT_QUEST_COMPLETED,
+    EVENT_QUEST_MISSED,
+    EVENT_QUEST_UNCOMPLETED,
+    READ_TIMEOUT,
+    _stream,
+)
+
+EVENTS_PATH = "/api/v1/admin/events"
+
+
+@pytest.fixture
+def temp_db_path(tmp_path) -> str:
+    """A fresh per-test SQLite path under pytest's tmp_path."""
+    return str(tmp_path / "nestquest.db")
+
+
+@pytest.fixture
+async def admin_live(temp_db_path: str) -> SimpleNamespace:
+    """A seeded household served by a live admin-enabled uvicorn app.
+
+    Both planes' clocks are pinned to 12:00 UTC (past Ada's 10:00 due,
+    before Bo's 17:00) BEFORE the server starts — the app is created
+    inside uvicorn's factory thread.
+    """
+    today = datetime.datetime.now().astimezone().date()
+    pinned_now = datetime.datetime(
+        today.year, today.month, today.day, 12, 0, 0, tzinfo=ZoneInfo("UTC")
+    )
+    seed = await _seed_household(temp_db_path, today, pinned_now)
+    original_panel = routes_panel._local_now
+    original_admin = routes_admin._local_now
+    routes_panel._local_now = lambda: pinned_now
+    routes_admin._local_now = lambda: pinned_now
+    try:
+        with _AdminServer(temp_db_path) as server:
+            client = httpx.AsyncClient(
+                base_url=server.url(""), timeout=READ_TIMEOUT
+            )
+            async with client:
+                yield SimpleNamespace(client=client, seed=seed, server=server)
+    finally:
+        routes_panel._local_now = original_panel
+        routes_admin._local_now = original_admin
+
+
+async def _panel_complete(live: SimpleNamespace, instance_id: int, child_id: int):
+    """Complete ``instance_id`` through the panel route (a real transition)."""
+    response = await live.client.post(
+        f"/api/v1/panel/instances/{instance_id}/complete",
+        headers={"Authorization": f"Bearer {PANEL_TOKEN}"},
+        json={"actor_child_id": child_id},
+    )
+    assert response.status_code == 200
+
+
+# --- auth: the ONE require_admin dependency, no stream on refusal ----------
+
+
+async def test_admin_events_refuses_absent_panel_and_non_admin(
+    admin_live: SimpleNamespace,
+) -> None:
+    """No credential is 401; the panel token and a non-admin JWT are 403.
+
+    Each refusal is a complete (non-streaming) JSON error response, so
+    no stream is established for any of them.
+    """
+    client = admin_live.client
+
+    absent = await client.get(EVENTS_PATH)
+    assert absent.status_code == 401
+    assert not absent.headers["content-type"].startswith("text/event-stream")
+
+    panel = await client.get(
+        EVENTS_PATH, headers={"Authorization": f"Bearer {PANEL_TOKEN}"}
+    )
+    assert panel.status_code == 403
+    assert panel.json()["detail"] == (
+        "This credential type is not accepted for admin access"
+    )
+
+    non_admin = await client.get(
+        EVENTS_PATH,
+        headers={
+            "Authorization": f"Bearer {make_token(groups=('other-group',))}"
+        },
+    )
+    assert non_admin.status_code == 403
+    assert non_admin.json()["detail"] == (
+        "Admin access requires membership in the nestquest-admins group"
+    )
+
+
+async def test_admin_jwt_is_not_accepted_on_panel_events(
+    admin_live: SimpleNamespace,
+) -> None:
+    """The panel plane is unchanged: an admin JWT does not open it."""
+    response = await admin_live.client.get(
+        "/api/v1/panel/events",
+        headers={"Authorization": f"Bearer {make_token()}"},
+    )
+    assert response.status_code == 401
+
+
+async def test_admin_events_sets_no_cache(admin_live: SimpleNamespace) -> None:
+    """The admin stream is ``text/event-stream`` with ``no-cache``."""
+    async with admin_live.client.stream(
+        "GET",
+        EVENTS_PATH,
+        headers={"Authorization": f"Bearer {make_token()}"},
+    ) as response:
+        assert response.status_code == 200
+        assert response.headers["content-type"].startswith("text/event-stream")
+        assert response.headers["cache-control"] == "no-cache"
+
+
+# --- the four documented transitions on the admin stream -------------------
+
+
+async def test_admin_stream_carries_completed_and_day_complete(
+    admin_live: SimpleNamespace,
+) -> None:
+    """A panel completion streams quest-completed then child-day-complete.
+
+    Ada's only quest today is Pack bag, so completing it also clears her
+    day; both frames arrive on the ADMIN stream with their documented
+    payload fields.
+    """
+    seed = admin_live.seed
+    pack_bag = seed.pack_bag_instance
+    stream = _stream(admin_live.server.url(EVENTS_PATH), make_token())
+    async with stream:
+        await _panel_complete(admin_live, pack_bag.id, seed.ada.id)
+
+        event_type, payload = await stream.next_event()
+        assert event_type == EVENT_QUEST_COMPLETED == "nestquest_quest_completed"
+        assert payload["child_id"] == seed.ada.id
+        assert payload["child_name"] == "Ada"
+        assert payload["instance_id"] == pack_bag.id
+        assert payload["quest_title"] == "Pack bag"
+        assert payload["window"] == "morning"
+        assert payload["due_date"] == pack_bag.due_date
+        assert payload["due_time"] == "10:00"
+        assert payload["occurred_at"]
+        assert payload["was_on_time"] is False  # 10:00 due, 12:00 now
+        completed_at = payload["occurred_at"]
+
+        event_type, payload = await stream.next_event()
+        assert event_type == EVENT_CHILD_DAY_COMPLETE
+        assert event_type == "nestquest_child_day_complete"
+        assert payload["child_id"] == seed.ada.id
+        assert payload["child_name"] == "Ada"
+        assert payload["quests_due"] == 1
+        assert payload["quests_completed"] == 1
+        assert payload["occurred_at"] == completed_at
+
+
+async def test_admin_stream_carries_uncompleted(
+    admin_live: SimpleNamespace,
+) -> None:
+    """The admin uncomplete route's reversal streams quest-uncompleted."""
+    seed = admin_live.seed
+    brush = seed.brush_teeth_instance
+    admin_headers = {"Authorization": f"Bearer {make_token()}"}
+    stream = _stream(admin_live.server.url(EVENTS_PATH), make_token())
+    async with stream:
+        response = await admin_live.client.post(
+            f"/api/v1/admin/instances/{brush.id}/uncomplete",
+            headers=admin_headers,
+        )
+        assert response.status_code == 200
+        assert response.json()["appended"] is True
+
+        event_type, payload = await stream.next_event()
+    assert event_type == EVENT_QUEST_UNCOMPLETED
+    assert event_type == "nestquest_quest_uncompleted"
+    assert payload["child_id"] == seed.bo.id
+    assert payload["child_name"] == "Bo"
+    assert payload["instance_id"] == brush.id
+    assert payload["quest_title"] == "Brush teeth"
+    assert payload["window"] == "morning"
+    assert payload["due_date"] == brush.due_date
+    assert payload["due_time"] == "17:00"
+    assert payload["occurred_at"]
+    assert "was_on_time" not in payload
+
+
+async def test_admin_stream_carries_missed(admin_live: SimpleNamespace) -> None:
+    """The missed-sweep publisher's transition streams to the admin client.
+
+    Drives the API-side helper the sweep calls, on the SERVER's loop,
+    against the app's own database and its ONE publisher.
+    """
+    from api import transitions
+
+    seed = admin_live.seed
+    stale = seed.stale_instance
+    app = admin_live.server.app
+    stream = _stream(admin_live.server.url(EVENTS_PATH), make_token())
+    async with stream:
+        admin_live.server.run_on_loop(
+            transitions.publish_quest_missed(
+                app.state.db.database, stale.id, app.state.publisher
+            )
+        )
+        event_type, payload = await stream.next_event()
+    assert event_type == EVENT_QUEST_MISSED == "nestquest_quest_missed"
+    assert payload["child_id"] == seed.ada.id
+    assert payload["child_name"] == "Ada"
+    assert payload["instance_id"] == stale.id
+    assert payload["quest_title"] == "Stale chore"
+    assert payload["window"] == "morning"
+    assert payload["due_date"] == stale.due_date
+    assert payload["due_time"] == "09:00"
+    assert payload["occurred_at"]
+
+
+# --- one publisher feeds both planes ---------------------------------------
+
+
+async def test_admin_and_panel_streams_share_one_publisher(
+    admin_live: SimpleNamespace,
+) -> None:
+    """One transition arrives, identically, on the admin AND panel streams.
+
+    The route publishes once to ``app.state.publisher``; both planes'
+    subscribers receive the same frames (same ``occurred_at``), which a
+    second, admin-only publisher could never produce.
+    """
+    seed = admin_live.seed
+    pack_bag = seed.pack_bag_instance
+    admin_stream = _stream(admin_live.server.url(EVENTS_PATH), make_token())
+    panel_stream = _stream(admin_live.server.url("/api/v1/panel/events"))
+    async with admin_stream, panel_stream:
+        await _panel_complete(admin_live, pack_bag.id, seed.ada.id)
+
+        admin_events = [
+            await admin_stream.next_event(),
+            await admin_stream.next_event(),
+        ]
+        panel_events = [
+            await panel_stream.next_event(),
+            await panel_stream.next_event(),
+        ]
+    assert [event_type for event_type, _ in admin_events] == [
+        EVENT_QUEST_COMPLETED,
+        EVENT_CHILD_DAY_COMPLETE,
+    ]
+    assert admin_events == panel_events

tests/test_api_sse.py

@@ -158,7 +158,8 @@ class _EventStream:
     parsed frame with a timeout.
     """
 
-    def __init__(self) -> None:
+    def __init__(self, token: str = PANEL_TOKEN) -> None:
+        self.token = token
         self.queue: asyncio.Queue = asyncio.Queue()
         self._task: asyncio.Task | None = None
         self._client: httpx.AsyncClient | None = None
@@ -170,7 +171,7 @@ class _EventStream:
         self._stream_cm = self._client.stream(
             "GET",
             self.url,
-            headers={"Authorization": f"Bearer {PANEL_TOKEN}"},
+            headers={"Authorization": f"Bearer {self.token}"},
         )
         self._response = await self._stream_cm.__aenter__()
         assert self._response.status_code == 200
@@ -263,9 +264,13 @@ async def seeded_panel(temp_db_path: str) -> SimpleNamespace:
         routes_panel._local_now = original
 
 
-def _stream(url: str) -> _EventStream:
+def _stream(url: str, token: str = PANEL_TOKEN) -> _EventStream:
-    """A subscribed SSE reader for ``url``."""
+    """A subscribed SSE reader for ``url``, sending ``token`` as Bearer.
-    return _EventStream()._set_url(url)
+
+    Defaults to the panel service token; the admin-plane stream tests
+    pass an admin JWT instead.
+    """
+    return _EventStream(token)._set_url(url)
 
 
 # --- auth and content type ----------------------------------------------