SSE: publish uncompleted, child-day-complete and missed transitions

The panel stream so far carried only quest-completed. This adds the
three remaining documented transitions (§3) so the panel UI can react
to reversals, full days, and nightly misses through the same ONE
publisher:

- core/events.py gains build_quest_missed_event, mirroring the
  sweep's hand-built payload exactly (shared _instance_payload shape),
  so the API plane publishes the same shape the integration sweep fires.
- api/transitions.py adds the reusable publish helpers the later
  uncomplete route (da0226b3) and the API-side missed sweep will call;
  payloads are always built by the core builders, never hand-built.
- The panel complete route now evaluates the day-complete rule on an
  actual transition, off the completed event's payload and the route's
  single clock read; the core builder still decides, so a zero-quest
  day and a non-today instance never fire it and a no-op re-complete
  publishes nothing.
- Tests drive all three transitions through the real SSE stream
  (server-loop scheduling for the not-yet-routed helpers) and assert
  every documented payload field.

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

@@ -112,13 +112,23 @@ core_migrations = importlib.import_module(f"{_CORE_MODULE_NAME}.migrations")
 #: assembly the panel routes call.
 core_snapshot = importlib.import_module(f"{_CORE_MODULE_NAME}.snapshot")
 
+#: The completion service (``core.completion.complete_instance`` /
+#: ``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")
 
 __all__ = [
     "core",
+    "core_completion",
     "core_db",
+    "core_events",
     "core_migrations",
     "core_snapshot",
     "core_settings",

api/routes_panel.py

@@ -52,13 +52,28 @@ The same :class:`PanelSnapshotResponse` Pydantic model is the route's
 from __future__ import annotations
 
 import datetime
+import importlib
+import json
 
-from fastapi import APIRouter, Depends, Request
-from pydantic import BaseModel
+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_snapshot, core_settings
+from api.nestquest_core import (
+    core_completion,
+    core_events,
+    core_settings,
+    core_snapshot,
+)
+from api import transitions as api_transitions
+
+#: The instance DAO, imported on the app's core copy for the
+#: child-mismatch guard on the complete route.
+_dao_instances = importlib.import_module("nestquest_core.dao_instances")
 
 #: All panel-plane routes share this router; the service-token check is
 #: attached HERE so every panel route (current and later) requires it.
@@ -122,6 +137,37 @@ class PanelSnapshotResponse(BaseModel):
     children: list[PanelChildPayload]
 
 
+class PanelCompleteRequest(BaseModel):
+    """The body of ``POST /api/v1/panel/instances/{id}/complete``.
+
+    Matches the panel's ``nestquest.complete_quest`` payload (§2 of
+    design/ENTITIES-AND-SERVICES.md): ``actor_child_id`` is the tapped
+    profile.  A positive integer is required — the core completion
+    layer rejects non-int (and bool) ids anyway, but rejecting them
+    here with 422 keeps the contract visible in the OpenAPI document
+    and fails before any database work.
+    """
+
+    actor_child_id: int
+
+    model_config = {
+        "json_schema_extra": {
+            "description": (
+                "The tapped child profile the completion is "
+                "attributed to (decision 9)."
+            )
+        }
+    }
+
+    @field_validator("actor_child_id")
+    @classmethod
+    def _positive(cls, value: int) -> int:
+        """Reject zero and negative ids with a validation error."""
+        if value < 1:
+            raise ValueError("actor_child_id must be a positive integer")
+        return value
+
+
 def _local_now() -> datetime.datetime:
     """Return the API host's local time as ONE timezone-aware clock read.
 
@@ -180,4 +226,148 @@ async def panel_snapshot(request: Request) -> PanelSnapshotResponse:
     )
 
 
+@router.post(
+    "/instances/{instance_id}/complete",
+    summary="Complete one quest instance",
+    status_code=200,
+)
+async def panel_complete_instance(
+    instance_id: int, body: PanelCompleteRequest, request: Request
+) -> dict[str, str]:
+    """Complete one quest instance as the tapped child profile.
+
+    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).
+
+    Thin adapter only: resolves the live database off ``app.state.db``,
+    reads ONE timezone-aware clock (:func:`_local_now`), and delegates
+    to :func:`nestquest_core.completion.complete_instance` with
+    ``actor_source="panel"`` and the tapped ``actor_child_id`` — the
+    append/idempotence logic, the lock, and the actor-shape validation
+    all live in the core layer, never here.
+
+    Response and idempotence: ``complete_instance`` returns a
+    :class:`~nestquest_core.completion.CompletionResult` whose
+    ``appended`` flag reports whether THIS call wrote an event; a
+    repeat call for an already-completed instance is a no-op inside
+    the core layer (no duplicate event) and this route still answers
+    ``{"status": "done"}`` either way, so a retried panel tap succeeds.
+
+    Error mapping (deliberately narrow): the ONLY ``ValueError`` mapped
+    to 404 is the unknown-``instance_id`` one, identified by the
+    ``instance_id:`` prefix ``complete_instance`` names the failing
+    field with.  Every other ``ValueError`` (an invalid actor shape,
+    which the body model and the panel actor contract already make
+    unreachable) re-raises to the framework's 500 — nothing else is
+    turned into a 404 or a 500 of our own making.
+
+    Child mismatch: ``complete_instance`` verifies the tapped
+    ``actor_child_id`` exists but does NOT check it is the child the
+    instance belongs to.  The route performs that check here (one
+    ``QuestInstancesDao.get_by_id`` read) and answers 404 when the
+    instance's ``child_id`` differs — completing another child's quest
+    from the wrong profile must not succeed, and 404 (not 403) keeps
+    the instance's existence from leaking to a wrong-profile tap.
+    """
+    state: DatabaseState = request.app.state.db
+    database = state.database
+
+    # Child-mismatch guard (see docstring): the instance must belong to
+    # the tapped profile before the completion layer is entered.
+    instance = await _dao_instances.QuestInstancesDao(database).get_by_id(
+        instance_id
+    )
+    if instance is not None and instance.child_id != body.actor_child_id:
+        raise HTTPException(
+            status_code=404,
+            detail="Quest instance not found for this child",
+        )
+
+    now = _local_now()
+    try:
+        result = await core_completion.complete_instance(
+            database,
+            instance_id,
+            actor_source="panel",
+            actor_child_id=body.actor_child_id,
+            now=now,
+        )
+    except ValueError as error:
+        message = str(error)
+        if message.startswith("instance_id:"):
+            # complete_instance names the field in its ValueError:
+            # "instance_id: quest instance N does not exist".
+            raise HTTPException(
+                status_code=404,
+                detail="Quest instance not found",
+            ) from error
+        raise
+    # 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)
+        # THEN evaluate the day-complete rule off the completed event's
+        # payload (the core builder decides when the child's whole day
+        # is cleared — a zero-quest day and a non-today instance never
+        # fire it).  ``today`` comes from the route's single clock read
+        # so the calendar date can never disagree with the completion's.
+        await api_transitions.publish_child_day_complete(
+            database,
+            payload,
+            now.date(),
+            request.app.state.publisher,
+        )
+    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"]

api/transitions.py

@@ -0,0 +1,83 @@
+"""Reusable SSE transition publishers for the API service (task 44c93cfd).
+
+Each helper wraps the ONE publisher on ``app.state.publisher`` with a
+single documented transition: it builds the payload through the core
+event builders (``nestquest_core.events``) — never hand-built here —
+and fans it out with :meth:`~api.events.TransitionPublisher.publish`.
+
+The uncomplete route (follow-up task da0226b3) and the API-side missed
+sweep do not exist yet; these helpers are the functions they will call,
+and the SSE tests exercise them directly.  Fan-out stays fire-and-forget
+(see api/events.py): a slow subscriber drops the event and recovers by
+polling the snapshot.
+"""
+from __future__ import annotations
+
+import datetime
+
+from api.nestquest_core import core_events
+
+
+async def publish_quest_uncompleted(
+    database: object, instance_id: int, publisher: object
+) -> None:
+    """Publish ONE ``nestquest_quest_uncompleted`` event for an actual
+    reversal of ``instance_id``.
+
+    The payload is built by the core builder
+    (:func:`nestquest_core.events.build_quest_uncompleted_events`);
+    this helper only picks the single event out of the builder's list
+    and publishes it.
+    """
+    events = await core_events.build_quest_uncompleted_events(
+        database, instance_id
+    )
+    for event_type, payload in events:
+        publisher.publish(event_type, payload)
+
+
+async def publish_quest_missed(
+    database: object, instance_id: int, publisher: object
+) -> None:
+    """Publish ONE ``nestquest_quest_missed`` event for ``instance_id``.
+
+    The payload is built by the core builder
+    (:func:`nestquest_core.events.build_quest_missed_event`), which
+    mirrors the documented sweep payload exactly (§3 of
+    design/ENTITIES-AND-SERVICES.md).
+    """
+    event_type, payload = await core_events.build_quest_missed_event(
+        database, instance_id
+    )
+    publisher.publish(event_type, payload)
+
+
+async def publish_child_day_complete(
+    database: object,
+    completed_payload: dict,
+    today: datetime.date,
+    publisher: object,
+) -> bool:
+    """Publish ``nestquest_child_day_complete`` when a completion
+    cleared the child's whole day; return whether one was published.
+
+    The day-complete rule lives entirely in the core builder
+    (:func:`nestquest_core.events.build_child_day_complete_event`):
+    it returns ``None`` when the day did not clear (a zero-quest day
+    or a non-today instance) and this helper publishes nothing then.
+    """
+    event = await core_events.build_child_day_complete_event(
+        database, completed_payload, today=today
+    )
+    if event is None:
+        return False
+    event_type, payload = event
+    publisher.publish(event_type, payload)
+    return True
+
+
+__all__ = [
+    "publish_child_day_complete",
+    "publish_quest_missed",
+    "publish_quest_uncompleted",
+]

custom_components/nestquest/core/events.py

@@ -14,7 +14,9 @@ fires nothing), and ``nestquest_child_day_complete`` fires when a
 completion clears the child's whole day (quests were owed and none
 remain — a zero-quest day can never fire it, matching the all-done
 binary sensor's rule).  ``nestquest_quest_missed`` is the documented
-contract Feature 11's nightly sweep fires; nothing here builds it.
+contract Feature 11's nightly sweep fires; :func:`build_quest_missed_event`
+builds it with exactly the payload the sweep hand-builds (§3), so the
+API plane can publish the same shape without re-deriving it.
 
 The completed-event and day-complete-event builders are split so the
 shim can fire ``nestquest_quest_completed`` BEFORE it evaluates the
@@ -50,6 +52,7 @@ from .completion import derive_state
 from .const import (
     EVENT_CHILD_DAY_COMPLETE,
     EVENT_QUEST_COMPLETED,
+    EVENT_QUEST_MISSED,
     EVENT_QUEST_UNCOMPLETED,
 )
 from .dao_children import ChildrenDao
@@ -188,3 +191,20 @@ async def build_quest_uncompleted_events(
     """
     payload = await _instance_payload(database, instance_id, _now_stamp())
     return [(EVENT_QUEST_UNCOMPLETED, payload)]
+
+
+async def build_quest_missed_event(
+    database: NestQuestDatabase, instance_id: int
+) -> tuple[str, dict]:
+    """Build the ``nestquest_quest_missed`` event for one instance.
+
+    The payload mirrors the nightly sweep's hand-built payload exactly
+    (the shared :func:`_instance_payload` shape: ``child_id``,
+    ``child_name``, ``instance_id``, ``quest_title``, ``window``,
+    ``due_date``, ``due_time``, ``occurred_at``), so the API plane's
+    missed path publishes the same event shape the integration's sweep
+    fires.  Like the sweep, this builder never mutates the instance or
+    writes a completion event — it reads only.
+    """
+    payload = await _instance_payload(database, instance_id, _now_stamp())
+    return (EVENT_QUEST_MISSED, payload)

tests/test_api_panel.py

@@ -67,6 +67,30 @@ _dao_instances = importlib.import_module("nestquest_core.dao_instances")
 _completion = importlib.import_module("nestquest_core.completion")
 
 
+async def _events_for_instance(database: object, instance_id: int) -> list:
+    """The instance's completion events, on the app's core copy.
+
+    Uses the SAME ``nestquest_core.*`` copy the app uses (the two-copy
+    caveat: never mix copies) against the app's live connection.
+    """
+    dao = _dao_instances.CompletionEventsDao(database)
+    return await dao.list_by_instance(instance_id)
+
+
+async def _derived_state(
+    database: object, instance: object, today: datetime.date
+) -> str:
+    """The instance's derived state on the app's core copy.
+
+    ``derive_state(instance, latest, today)`` from the app's core copy;
+    the latest event is fetched through the same copy's DAO against the
+    app's live connection.
+    """
+    dao = _dao_instances.CompletionEventsDao(database)
+    latest = await dao.get_latest_for_instance(instance.id)
+    return _completion.derive_state(instance, latest, today)
+
+
 async def _seed_household(
     db_path: str,
     today: datetime.date,
@@ -215,14 +239,15 @@ class _PanelRunner:
             ApiConfig(db_path=db_path, panel_token=panel_token)
         )
 
-    async def __aenter__(self) -> httpx.AsyncClient:
+    async def __aenter__(self) -> tuple["_PanelRunner", httpx.AsyncClient]:
         self._lifespan = self.app.router.lifespan_context(self.app)
         await self._lifespan.__aenter__()
         self._client = httpx.AsyncClient(
             transport=httpx.ASGITransport(app=self.app),
             base_url="http://testserver",
         )
-        return await self._client.__aenter__()
+        await self._client.__aenter__()
+        return self, self._client
 
     async def __aexit__(self, *exc) -> None:
         try:
@@ -240,7 +265,8 @@ def temp_db_path(tmp_path: Path) -> str:
 @pytest.fixture
 async def panel_client(temp_db_path: str) -> httpx.AsyncClient:
     """An app client with an empty (migrated) household database."""
-    async with _PanelRunner(temp_db_path) as client:
+    async with _PanelRunner(temp_db_path) as runner_and_client:
+        _, client = runner_and_client
         yield client
 
 
@@ -251,6 +277,9 @@ async def seeded_panel(temp_db_path: str, monkeypatch) -> SimpleNamespace:
     One host clock read anchors ``today``; the pinned instant is
     12:00 UTC that day (past Ada's 10:00 due, before Bo's 17:00).  The
     route's ``_local_now`` is patched to return the pinned instant.
+
+    Yields the runner too so tests reach the app's live database (the
+    SAME ``nestquest_core`` connection) for assertions.
     """
     today = datetime.datetime.now().astimezone().date()
     time_zone = ZoneInfo("UTC")
@@ -259,9 +288,11 @@ async def seeded_panel(temp_db_path: str, monkeypatch) -> SimpleNamespace:
     )
     seed = await _seed_household(temp_db_path, today, pinned_now)
     monkeypatch.setattr(routes_panel, "_local_now", lambda: pinned_now)
-    async with _PanelRunner(temp_db_path) as client:
+    async with _PanelRunner(temp_db_path) as runner_and_client:
+        runner, client = runner_and_client
         yield SimpleNamespace(
             client=client,
+            database=runner.app.state.db.database,
             seed=seed,
             today=today,
             pinned_now=pinned_now,
@@ -521,3 +552,137 @@ async def test_snapshot_route_is_documented(
     assert "PanelSnapshotResponse" in documented
     assert "PanelChildPayload" in documented
     assert "PanelInstancePayload" in documented
+
+
+# --- complete route -------------------------------------------------------
+
+
+async def test_complete_writes_panel_completion(
+    seeded_panel: SimpleNamespace,
+) -> None:
+    """The first POST completes the instance as the tapped profile.
+
+    Returns success; the completion event recorded is
+    ``actor_source='panel'`` with the tapped ``actor_child_id`` and NO
+    ``actor_user_id`` (the panel actor contract), and the instance's
+    derived state becomes ``done``.
+    """
+    seed = seeded_panel.seed
+    response = await seeded_panel.client.post(
+        f"/api/v1/panel/instances/{seed.pack_bag_instance.id}/complete",
+        headers={"Authorization": f"Bearer {PANEL_TOKEN}"},
+        json={"actor_child_id": seed.ada.id},
+    )
+    assert response.status_code == 200
+    assert response.json() == {"status": "done"}
+
+    # Assert the recorded event and derived state through the SAME core
+    # copy the app uses (two-copy caveat: never mix copies), against the
+    # app's live connection.
+    events = await _events_for_instance(
+        seeded_panel.database, seed.pack_bag_instance.id
+    )
+    assert len(events) == 1
+    event = events[0]
+    assert event.event_type == "completed"
+    assert event.actor_source == "panel"
+    assert event.actor_child_id == seed.ada.id
+    assert event.actor_user_id is None
+
+    instance = await _dao_instances.QuestInstancesDao(
+        seeded_panel.database
+    ).get_by_id(seed.pack_bag_instance.id)
+    assert await _derived_state(
+        seeded_panel.database, instance, seeded_panel.today
+    ) == "done"
+
+
+async def test_complete_is_idempotent(
+    seeded_panel: SimpleNamespace,
+) -> None:
+    """A second call is a no-op that still returns success.
+
+    The core completion layer's ``appended`` flag decides: the repeat
+    call appends nothing, so exactly ONE completion event row exists
+    after two calls.
+    """
+    seed = seeded_panel.seed
+    headers = {"Authorization": f"Bearer {PANEL_TOKEN}"}
+    url = f"/api/v1/panel/instances/{seed.pack_bag_instance.id}/complete"
+    body = {"actor_child_id": seed.ada.id}
+    first = await seeded_panel.client.post(url, headers=headers, json=body)
+    assert first.status_code == 200
+    second = await seeded_panel.client.post(url, headers=headers, json=body)
+    assert second.status_code == 200
+    assert second.json() == {"status": "done"}
+
+    events = await _events_for_instance(
+        seeded_panel.database, seed.pack_bag_instance.id
+    )
+    assert len(events) == 1, (
+        "the idempotent call must not append a second event"
+    )
+
+
+async def test_complete_unknown_instance_is_not_found(
+    panel_client: httpx.AsyncClient,
+) -> None:
+    """Completing a non-existent instance returns 404."""
+    response = await panel_client.post(
+        "/api/v1/panel/instances/999999/complete",
+        headers={"Authorization": f"Bearer {PANEL_TOKEN}"},
+        json={"actor_child_id": 1},
+    )
+    assert response.status_code == 404
+
+
+async def test_complete_without_token_is_unauthorized(
+    panel_client: httpx.AsyncClient,
+) -> None:
+    """A complete request with no Authorization header returns 401."""
+    response = await panel_client.post(
+        "/api/v1/panel/instances/1/complete",
+        json={"actor_child_id": 1},
+    )
+    assert response.status_code == 401
+
+
+async def test_complete_wrong_child_is_not_found(
+    seeded_panel: SimpleNamespace,
+) -> None:
+    """Tapping the wrong child's profile answers 404, not success.
+
+    ``complete_instance`` does not itself check that the tapped
+    ``actor_child_id`` is the child the instance belongs to; the route
+    guards it so a wrong-profile tap can never complete another
+    child's quest.
+    """
+    seed = seeded_panel.seed
+    response = await seeded_panel.client.post(
+        f"/api/v1/panel/instances/{seed.pack_bag_instance.id}/complete",
+        headers={"Authorization": f"Bearer {PANEL_TOKEN}"},
+        json={"actor_child_id": seed.bo.id},
+    )
+    assert response.status_code == 404
+    # And nothing was completed.
+    events = await _events_for_instance(
+        seeded_panel.database, seed.pack_bag_instance.id
+    )
+    assert events == []
+
+
+async def test_complete_route_is_documented(
+    panel_client: httpx.AsyncClient,
+) -> None:
+    """The complete route and its request shape are in the OpenAPI doc."""
+    response = await panel_client.get("/openapi.json")
+    assert response.status_code == 200
+    schema = response.json()
+    path = "/api/v1/panel/instances/{instance_id}/complete"
+    assert path in schema["paths"]
+    assert "post" in schema["paths"][path]
+    documented = {
+        model["title"]
+        for model in schema["components"]["schemas"].values()
+    }
+    assert "PanelCompleteRequest" in documented

tests/test_api_sse.py

@@ -0,0 +1,456 @@
+"""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
+EVENT_QUEST_UNCOMPLETED = _core_const.EVENT_QUEST_UNCOMPLETED
+EVENT_QUEST_MISSED = _core_const.EVENT_QUEST_MISSED
+EVENT_CHILD_DAY_COMPLETE = _core_const.EVENT_CHILD_DAY_COMPLETE
+
+#: 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_loop: asyncio.AbstractEventLoop | None = None
+        self._server = uvicorn.Server(
+            uvicorn.Config(
+                # A closure factory that captures the app's event loop
+                # as it is created INSIDE the server thread: the
+                # lifespan's database and publisher belong to that
+                # thread's loop, so tests driving an API-side
+                # transition helper directly schedule it there.
+                self._make_app_factory(db_path),
+                host="127.0.0.1",
+                port=self.port,
+                log_level="warning",
+            )
+        )
+        self._thread = threading.Thread(
+            target=self._server.run, daemon=True
+        )
+
+    def _make_app_factory(self, db_path: str):
+        """The uvicorn app factory, closing over ``db_path``.
+
+        Records the running loop the first time uvicorn calls it (in
+        the server thread) so :meth:`run_on_loop` can target it.
+        """
+
+        def factory():
+            app = create_app(
+                ApiConfig(db_path=db_path, panel_token=PANEL_TOKEN)
+            )
+            self._server_loop = asyncio.get_running_loop()
+            self.app = app
+            return app
+
+        return factory
+
+    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}"
+
+    def run_on_loop(self, coro) -> None:
+        """Run ``coro`` on the SERVER's event loop and wait for it.
+
+        The app (its database and publisher) lives on the uvicorn
+        thread's loop, so a test that drives an API-side transition
+        helper directly must schedule it there: the helper awaits core
+        DAO calls and calls ``publisher.publish`` synchronously, and
+        both must run on the loop that owns them.  Bounded by
+        READ_TIMEOUT so a wedged schedule fails instead of hanging.
+        """
+        return asyncio.run_coroutine_threadsafe(
+            coro, self._server_loop
+        ).result(timeout=READ_TIMEOUT)
+
+
+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
+        # Ada's only quest today: the completion also clears her whole
+        # day, so the day-complete event follows (task 44c93cfd).
+        event_type, _ = await stream.next_event()
+        assert event_type == EVENT_CHILD_DAY_COMPLETE
+
+        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)
+
+
+# --- the remaining documented transitions (task 44c93cfd) -----------------
+
+
+async def test_publish_quest_uncompleted_streams_event(seeded_panel) -> None:
+    """The reusable uncompleted publisher streams the documented
+    uncompleted transition with its documented payload fields.
+
+    The admin uncomplete route does not exist yet (deferred to task
+    da0226b3); the test drives the API-side helper the route will call.
+    The helper runs on the SERVER's event loop (``run_coroutine_threadsafe``
+    on the uvicorn thread's loop) against the app's own live database,
+    exactly the way the later route handler will call it.
+    """
+    from api import transitions
+
+    seed = seeded_panel.seed
+    pack_bag = seed.pack_bag_instance
+    app = seeded_panel.server.app
+    database = app.state.db.database
+    stream = _stream(seeded_panel.url("/api/v1/panel/events"))
+    async with stream:
+        seeded_panel.server.run_on_loop(
+            transitions.publish_quest_uncompleted(
+                database, pack_bag.id, app.state.publisher
+            )
+        )
+        event_type, payload = await stream.next_event()
+    assert event_type == EVENT_QUEST_UNCOMPLETED
+    assert event_type == "nestquest_quest_uncompleted"
+    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 "was_on_time" not in payload
+
+
+async def test_last_quest_of_day_publishes_day_complete(seeded_panel) -> None:
+    """Completing the child's LAST open quest of the day emits the
+    completed event AND the child-day-complete event carrying
+    ``quests_due`` and ``quests_completed``.
+
+    Ada has exactly one instance today (Pack bag), so completing it
+    clears her whole day; a no-op re-complete afterwards publishes
+    nothing.
+    """
+    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
+        completed_occurred_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
+        # One transition, one shared timestamp (the core builder
+        # reuses the completed event's stamp).
+        assert payload["occurred_at"] == completed_occurred_at
+
+        # No-op re-complete: nothing further on the stream.
+        second = 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 second.status_code == 200
+        with pytest.raises(asyncio.TimeoutError):
+            await asyncio.wait_for(stream.queue.get(), timeout=0.3)
+
+
+async def test_publish_quest_missed_streams_event(seeded_panel) -> None:
+    """The reusable missed publisher streams the documented missed
+    transition with its documented payload fields (the sweep payload).
+
+    The API-side missed sweep does not exist yet at this queue
+    position; the test drives the API-side helper the sweep will call
+    against the seeded past-due open instance, on the server's own
+    event loop and database.
+    """
+    from api import transitions
+
+    seed = seeded_panel.seed
+    stale = seed.stale_instance
+    app = seeded_panel.server.app
+    database = app.state.db.database
+    stream = _stream(seeded_panel.url("/api/v1/panel/events"))
+    async with stream:
+        seeded_panel.server.run_on_loop(
+            transitions.publish_quest_missed(
+                database, stale.id, app.state.publisher
+            )
+        )
+        event_type, payload = await stream.next_event()
+    assert event_type == EVENT_QUEST_MISSED
+    assert event_type == "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"]

tests/test_events.py

@@ -791,3 +791,31 @@ async def test_shim_day_complete_uses_ha_local_timezone(
         f"{far_today.isoformat()}, UTC today {utc_today.isoformat()}); "
         f"got {len(day_complete)} events"
     )
+
+
+async def test_build_quest_missed_event_payload_key_set(
+    hass, make_entry
+) -> None:
+    """The missed-event payload carries exactly the documented keys —
+    the same shape the nightly sweep hand-builds (§3), with no
+    ``was_on_time`` (completion-only)."""
+    from custom_components.nestquest.core.events import (
+        build_quest_missed_event,
+    )
+
+    entry, child, instances = await _setup_and_seed(hass, make_entry)
+    database = entry.runtime_data.database
+    event_type, payload = await build_quest_missed_event(
+        database, instances[0].id
+    )
+    assert event_type == EVENT_QUEST_MISSED
+    assert event_type == "nestquest_quest_missed"
+    assert set(payload) == _UNCOMPLETED_KEYS, (
+        f"missed payload keys drift: got {set(payload)} "
+        f"expected {_UNCOMPLETED_KEYS}"
+    )
+    assert payload["child_id"] == child.id
+    assert payload["child_name"] == "Ada"
+    assert payload["instance_id"] == instances[0].id
+    assert payload["quest_title"] == "Brush teeth"
+    assert _TIMESTAMP.match(payload["occurred_at"]), payload["occurred_at"]