SSE endpoint publishing quest-completed transitions (task 4d7bae00)

The panel needs live transitions, so add ONE in-process publisher
(api/events.py, installed on app.state.publisher in the lifespan) and
a GET /api/v1/panel/events SSE route on the panel router — it inherits
the single require_panel_token check. The complete route now builds
the event through the core builder and publishes it only when the
core's .appended flag says an actual transition happened; a no-op
re-complete publishes nothing. Tests stream from a real uvicorn
instance on an ephemeral port because httpx's ASGITransport buffers
endless SSE bodies.

api/app.py

@@ -39,6 +39,7 @@ from fastapi import FastAPI
 
 from api.config import ApiConfig
 from api.database import DatabaseState, make_database
+from api.events import TransitionPublisher
 from api.routes_panel import router as panel_router
 
 
@@ -47,12 +48,16 @@ def _build_lifespan(db_path: str):
 
     Captured as a closure so the app factory can hand the configured
     path in without threading it through FastAPI's lifespan signature.
+    The lifespan also installs the app's ONE in-process transition
+    publisher on ``app.state.publisher`` (see :mod:`api.events`) so
+    every route and SSE subscriber shares that single instance.
     """
 
     @asynccontextmanager
     async def lifespan(app: FastAPI) -> AsyncIterator[None]:
         state = make_database(db_path)
         app.state.db = state
+        app.state.publisher = TransitionPublisher()
         try:
             await state.open_and_migrate()
             yield

api/events.py

@@ -0,0 +1,78 @@
+"""The ONE in-process transition-event publisher for the API service.
+
+Route handlers call :meth:`TransitionPublisher.publish` synchronously
+(no ``await``, callable from any async handler); SSE subscribers stream
+those transitions through an async generator.  The scope of "ONE" is
+deliberate: a single :class:`TransitionPublisher` instance is created
+in the app lifespan and stored on ``app.state.publisher`` — per-app,
+so tests build their own hermetic app without module-level state, and
+both the panel complete route and the SSE endpoint resolve the SAME
+instance off ``request.app.state`` (exactly one publisher exists per
+running app; nothing else may create one).
+
+Design (small and dependency-free, asyncio only):
+
+- Each subscriber gets its own ``asyncio.Queue(maxsize=64)``.
+- ``publish`` is synchronous and non-blocking: it puts the event on
+  every live queue with ``put_nowait``.  A slow subscriber that lets
+  its queue fill gets the event DROPPED (oldest overflow) — never a
+  blocked publisher, never unbounded growth.
+- A disconnected subscriber's queue is removed in the generator's
+  ``finally`` block, so teardown is clean and a closed consumer cannot
+  leak its queue.
+
+Fan-out is fire-and-forget by design: the SSE plane mirrors transition
+events for the panel UI; a dropped event costs the panel one UI
+refresh, and the documented recovery path is the next snapshot poll.
+"""
+from __future__ import annotations
+
+import asyncio
+from collections.abc import AsyncIterator
+
+#: Per-subscriber queue capacity.  Bounded so a stalled consumer can
+#: never make the publisher or another subscriber wait; beyond this
+#: the slow subscriber drops the event (see module docstring).
+_QUEUE_MAXSIZE = 64
+
+__all__ = ["TransitionPublisher"]
+
+
+class TransitionPublisher:
+    """A single in-process fan-out point for transition events."""
+
+    def __init__(self) -> None:
+        """Create the publisher with no subscribers."""
+        self._subscribers: set[asyncio.Queue] = set()
+
+    def publish(self, event_type: str, payload: dict) -> None:
+        """Fan one transition event out to every current subscriber.
+
+        Synchronous and non-blocking: each subscriber's bounded queue
+        is filled with ``put_nowait``, and a FULL queue silently drops
+        the event for that (slow) subscriber rather than blocking the
+        route handler or growing without bound.
+        """
+        for queue in tuple(self._subscribers):
+            try:
+                queue.put_nowait((event_type, payload))
+            except asyncio.QueueFull:
+                # The subscriber is not keeping up; drop the event for
+                # it and move on (the panel recovers by polling).
+                pass
+
+    async def subscribe(self) -> AsyncIterator[tuple[str, dict]]:
+        """Yield ``(event_type, payload)`` tuples until the consumer breaks.
+
+        Registers ONE bounded queue for the duration of the iteration
+        and removes it in the ``finally`` block, so a client disconnect
+        (or any break out of the loop) tears the subscription down
+        cleanly with no leaked queue.
+        """
+        queue: asyncio.Queue = asyncio.Queue(maxsize=_QUEUE_MAXSIZE)
+        self._subscribers.add(queue)
+        try:
+            while True:
+                yield await queue.get()
+        finally:
+            self._subscribers.discard(queue)

api/nestquest_core.py

@@ -116,6 +116,10 @@ core_snapshot = importlib.import_module(f"{_CORE_MODULE_NAME}.snapshot")
 #: ``uncomplete_instance``) the panel's write routes call.
 core_completion = importlib.import_module(f"{_CORE_MODULE_NAME}.completion")
 
+#: The transition-event builders (``core.events.build_quest_completed_event``
+#: et al.) the panel's write routes build their SSE events with.
+core_events = importlib.import_module(f"{_CORE_MODULE_NAME}.events")
+
 #: The explicit settings object (``core.settings.NestQuestSettings``)
 #: the snapshot builder's ``settings`` parameter takes.
 core_settings = importlib.import_module(f"{_CORE_MODULE_NAME}.settings")
@@ -124,6 +128,7 @@ __all__ = [
     "core",
     "core_completion",
     "core_db",
+    "core_events",
     "core_migrations",
     "core_snapshot",
     "core_settings",

api/routes_panel.py

@@ -53,14 +53,19 @@ 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
 from pydantic import BaseModel, field_validator
 
 from api.database import DatabaseState
 from api.dependencies import require_panel_token
 from api.nestquest_core import (
     core_completion,
+    core_events,
     core_settings,
     core_snapshot,
 )
@@ -299,7 +304,57 @@ async def panel_complete_instance(
     # The derived state is "done" in both the appended and no-op cases;
     # ``result.appended`` distinguishes them for callers that care —
     # the panel route returns the same success either way.
+    if result.appended:
+        # An ACTUAL transition: publish the SSE transition event.  The
+        # payload is built by the core builder (never hand-built here)
+        # and fan-out is fire-and-forget (see api/events.py).
+        event_type, payload = await core_events.build_quest_completed_event(
+            database, instance_id, was_on_time=result.was_on_time
+        )
+        request.app.state.publisher.publish(event_type, payload)
     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)",
+)
+async def panel_events(request: Request) -> StreamingResponse:
+    """Stream the app's transition events to the panel (Server-Sent
+    Events).
+
+    Requires the panel service token (checked by the router's shared
+    :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
+    :func:`_sse_stream` off the app's single in-process publisher
+    (``app.state.publisher``, installed in the lifespan).  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"},
+    )
+
+
 __all__ = ["router"]

tests/test_api_sse.py

@@ -0,0 +1,295 @@
+"""SSE endpoint tests: the panel event stream (task 4d7bae00).
+
+These tests exercise ``GET /api/v1/panel/events`` — the panel plane's
+Server-Sent Events stream — against a REAL server:
+
+- The stream requires the panel service token (D-012): no/wrong token
+  returns 401 (the router-level shared dependency, no second check).
+- The response is ``text/event-stream`` and each transition arrives as
+  an ``event: <type>`` / ``data: <json>`` frame.
+- Completing an instance through the panel complete route publishes
+  the ``nestquest_quest_completed`` transition carrying the documented
+  payload fields (design/ENTITIES-AND-SERVICES.md §3).
+- A no-op re-complete publishes nothing (no second completed event).
+
+Streaming approach (documented): httpx's ASGITransport BUFFERS the
+whole response body before returning, so an endless SSE stream hangs
+it — exactly the awkward case this task anticipated.  These tests
+therefore run the app with uvicorn on an ephemeral port (port 0, so
+the OS picks a free one) in a daemon thread, and httpx streams from
+``http://127.0.0.1:<port>``.  Every read is bounded by a timeout so a
+wedged stream fails an assertion instead of hanging the suite; the
+server is shut down in the fixture's finally block.
+"""
+from __future__ import annotations
+
+import asyncio
+import datetime
+import importlib
+import json
+import socket
+import threading
+import time
+from pathlib import Path
+from types import SimpleNamespace
+from zoneinfo import ZoneInfo
+
+import httpx
+import pytest
+import uvicorn
+
+from api import routes_panel
+from api.app import create_app
+from api.config import ApiConfig
+from tests.test_api_panel import PANEL_TOKEN, _seed_household
+
+#: The documented quest-completed event name (core const, §3).
+_core_const = importlib.import_module("nestquest_core.const")
+EVENT_QUEST_COMPLETED = _core_const.EVENT_QUEST_COMPLETED
+
+#: Upper bound for any single stream read; bounds the whole test so a
+#: wedged stream fails the assertion instead of hanging the suite.
+READ_TIMEOUT = 5.0
+
+
+def _free_port() -> int:
+    """Ask the OS for one free TCP port (bound then released)."""
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
+        sock.bind(("127.0.0.1", 0))
+        return sock.getsockname()[1]
+
+
+class _Server:
+    """Run the app with uvicorn on an ephemeral port, in a thread.
+
+    ``create_app`` runs INSIDE the server thread (uvicorn's factory
+    mode), so the lifespan — including ``app.state.publisher`` — is
+    set up per test on the server's own event loop.  Daemon thread +
+    explicit ``should_exit`` shutdown in the fixture's finally block.
+    """
+
+    def __init__(self, db_path: str) -> None:
+        self.port = _free_port()
+        self._server = uvicorn.Server(
+            uvicorn.Config(
+                # A closure factory: reads nothing at import time.
+                lambda: create_app(
+                    ApiConfig(db_path=db_path, panel_token=PANEL_TOKEN)
+                ),
+                host="127.0.0.1",
+                port=self.port,
+                log_level="warning",
+            )
+        )
+        self._thread = threading.Thread(
+            target=self._server.run, daemon=True
+        )
+
+    def __enter__(self) -> "_Server":
+        self._thread.start()
+        deadline = time.monotonic() + READ_TIMEOUT
+        while not self._server.started:
+            if time.monotonic() > deadline or not self._thread.is_alive():
+                raise RuntimeError("uvicorn did not start")
+            time.sleep(0.01)
+        return self
+
+    def __exit__(self, *exc) -> None:
+        self._server.should_exit = True
+        self._thread.join(timeout=READ_TIMEOUT)
+
+    def url(self, path: str) -> str:
+        """The full URL for ``path`` on this server."""
+        return f"http://127.0.0.1:{self.port}{path}"
+
+
+class _EventStream:
+    """A subscribed SSE reader over a live httpx connection.
+
+    Parses frames (``event:``/``data:`` pairs) into
+    ``(event_type, payload)`` tuples; ``next_event`` awaits the next
+    parsed frame with a timeout.
+    """
+
+    def __init__(self) -> None:
+        self.queue: asyncio.Queue = asyncio.Queue()
+        self._task: asyncio.Task | None = None
+        self._client: httpx.AsyncClient | None = None
+        self._stream_cm = None
+        self._response: httpx.Response | None = None
+
+    async def __aenter__(self) -> "_EventStream":
+        self._client = httpx.AsyncClient(timeout=READ_TIMEOUT)
+        self._stream_cm = self._client.stream(
+            "GET",
+            self.url,
+            headers={"Authorization": f"Bearer {PANEL_TOKEN}"},
+        )
+        self._response = await self._stream_cm.__aenter__()
+        assert self._response.status_code == 200
+        assert self._response.headers["content-type"].startswith(
+            "text/event-stream"
+        )
+        self._task = asyncio.create_task(self._pump())
+        return self
+
+    def _set_url(self, url: str) -> "_EventStream":
+        """Record the stream URL (called before ``__aenter__``)."""
+        self.url = url
+        return self
+
+    async def _pump(self) -> None:
+        """Read the stream and push parsed frames onto the queue."""
+        event_type: str | None = None
+        data_lines: list[str] = []
+        try:
+            async for chunk in self._response.aiter_text():
+                for line in chunk.splitlines():
+                    if line.startswith("event:"):
+                        event_type = line[len("event:"):].strip()
+                    elif line.startswith("data:"):
+                        data_lines.append(line[len("data:"):].strip())
+                    elif not line and event_type is not None:
+                        payload = json.loads("\n".join(data_lines))
+                        await self.queue.put((event_type, payload))
+                        event_type = None
+                        data_lines = []
+        except Exception as error:  # pragma: no cover - teardown path
+            await self.queue.put(error)
+
+    async def next_event(self) -> tuple[str, dict]:
+        """The next parsed ``(event_type, payload)``, with a timeout."""
+        item = await asyncio.wait_for(
+            self.queue.get(), timeout=READ_TIMEOUT
+        )
+        if isinstance(item, Exception):
+            raise item
+        return item
+
+    async def __aexit__(self, *exc) -> None:
+        if self._task is not None:
+            self._task.cancel()
+            try:
+                await self._task
+            except asyncio.CancelledError:
+                pass
+        await self._stream_cm.__aexit__(*exc)
+        await self._client.__aexit__(*exc)
+
+
+@pytest.fixture
+def temp_db_path(tmp_path: Path) -> str:
+    """A fresh per-test SQLite path under pytest's tmp_path."""
+    return str(tmp_path / "nestquest.db")
+
+
+@pytest.fixture
+async def seeded_panel(temp_db_path: str) -> SimpleNamespace:
+    """A seeded household served by a live uvicorn app, clock pinned.
+
+    Same pinning discipline as the panel tests (12:00 UTC, past Ada's
+    10:00 due, before Bo's 17:00): the route's ``_local_now`` is
+    patched so the completion's ``was_on_time`` is deterministic.  The
+    app is created inside uvicorn's factory, so the patch must happen
+    BEFORE the server starts.
+    """
+    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)
+    # The app is created inside uvicorn's factory (in the server
+    # thread), so the patch must be applied BEFORE the server starts.
+    original = routes_panel._local_now
+    routes_panel._local_now = lambda: pinned_now
+    try:
+        with _Server(temp_db_path) as server:
+            client = httpx.AsyncClient(base_url=server.url(""))
+            async with client:
+                yield SimpleNamespace(
+                    client=client,
+                    seed=seed,
+                    server=server,
+                    url=server.url,
+                )
+    finally:
+        routes_panel._local_now = original
+
+
+def _stream(url: str) -> _EventStream:
+    """A subscribed SSE reader for ``url``."""
+    return _EventStream()._set_url(url)
+
+
+# --- auth and content type ----------------------------------------------
+
+
+async def test_events_requires_panel_token(seeded_panel) -> None:
+    """The SSE route inherits the router's token check (no second one)."""
+    response = await seeded_panel.client.get("/api/v1/panel/events")
+    assert response.status_code == 401
+
+
+# --- the quest-completed transition --------------------------------------
+
+
+async def test_complete_publishes_quest_completed_event(
+    seeded_panel,
+) -> None:
+    """A panel completion emits the documented completed transition.
+
+    The stream is subscribed FIRST, the instance is completed through
+    the panel route, and the emitted frame's type and every documented
+    payload field (child_id, child_name, instance_id, quest_title,
+    window, due_date, due_time, occurred_at, was_on_time) are asserted.
+    """
+    seed = seeded_panel.seed
+    pack_bag = seed.pack_bag_instance
+    stream = _stream(seeded_panel.url("/api/v1/panel/events"))
+    async with stream:
+        response = await seeded_panel.client.post(
+            f"/api/v1/panel/instances/{pack_bag.id}/complete",
+            headers={"Authorization": f"Bearer {PANEL_TOKEN}"},
+            json={"actor_child_id": seed.ada.id},
+        )
+        assert response.status_code == 200
+
+        event_type, payload = await stream.next_event()
+        assert event_type == EVENT_QUEST_COMPLETED
+        assert event_type == "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"]  # strict UTC ISO-8601 stamp
+        assert payload["was_on_time"] is False  # 10:00 due, 12:00 now
+
+
+async def test_no_op_recomplete_publishes_nothing(seeded_panel) -> None:
+    """A second complete call emits NO second completed event.
+
+    After the first (real) transition, a re-complete is a no-op in the
+    core layer; the stream must stay silent.
+    """
+    seed = seeded_panel.seed
+    url = f"/api/v1/panel/instances/{seed.pack_bag_instance.id}/complete"
+    body = {"actor_child_id": seed.ada.id}
+    headers = {"Authorization": f"Bearer {PANEL_TOKEN}"}
+    stream = _stream(seeded_panel.url("/api/v1/panel/events"))
+    async with stream:
+        first = await seeded_panel.client.post(url, headers=headers, json=body)
+        assert first.status_code == 200
+        event_type, _ = await stream.next_event()
+        assert event_type == EVENT_QUEST_COMPLETED
+
+        second = await seeded_panel.client.post(
+            url, headers=headers, json=body
+        )
+        assert second.status_code == 200
+        with pytest.raises(asyncio.TimeoutError):
+            # Nothing arrives within the window: the no-op re-complete
+            # must publish nothing.
+            await asyncio.wait_for(stream.queue.get(), timeout=0.3)