Serve the panel snapshot behind a reusable service-token dependency (task ae7e90ae)

The panel plane (D-012) needs an authenticated read surface the HA
integration polls: GET /api/v1/panel/snapshot. The token check lives in
ONE reusable dependency (api/dependencies.require_panel_token) — missing,
malformed, wrong, and empty-configured credentials all fail closed with
the same 401 (constant-time compare, uniform detail) — and the panel
router attaches it once so every later panel route inherits the check.

ApiConfig gains the required NESTQUEST_PANEL_TOKEN (same fail-fast
discipline as NESTQUEST_DB_PATH: unset/empty is a ConfigError, never a
silent default), create_app stores the resolved config on app.state so
the dependency reads it without module state, and the snapshot handler
stays a thin adapter: it calls the pure core build_snapshot with the
live database and one timezone-aware clock read, shapes each child's
instances through core.instance_payload(include_missed=False) so missed
instances are omitted from the panel payload (D-009), and returns the
shape documented by the PanelSnapshotResponse models in OpenAPI.

Tests (tests/test_api_panel.py) seed the household through the SAME
nestquest_core.* copy the app uses, pin the route's single clock read,
and assert 401 for missing/wrong/malformed tokens, 200 with the full
documented payload (children in sort order, counts, presence,
cycle_day), and that a past-due open (missed) instance never reaches
the payload.

api/app.py

@@ -39,7 +39,6 @@ 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
 
 
@@ -48,16 +47,12 @@ 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

@@ -1,78 +0,0 @@
-"""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,23 +112,13 @@ 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,28 +52,13 @@ The same :class:`PanelSnapshotResponse` Pydantic model is the route's
 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 fastapi import APIRouter, Depends, Request
+from pydantic import BaseModel
 
 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,
-)
-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")
+from api.nestquest_core import core_snapshot, core_settings
 
 #: All panel-plane routes share this router; the service-token check is
 #: attached HERE so every panel route (current and later) requires it.
@@ -137,37 +122,6 @@ 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.
 
@@ -226,148 +180,4 @@ 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

@@ -1,83 +0,0 @@
-"""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,9 +14,7 @@ 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; :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.
+contract Feature 11's nightly sweep fires; nothing here builds it.
 
 The completed-event and day-complete-event builders are split so the
 shim can fire ``nestquest_quest_completed`` BEFORE it evaluates the
@@ -52,7 +50,6 @@ 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
@@ -191,20 +188,3 @@ 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,30 +67,6 @@ _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,
@@ -239,15 +215,14 @@ class _PanelRunner:
             ApiConfig(db_path=db_path, panel_token=panel_token)
         )
 
-    async def __aenter__(self) -> tuple["_PanelRunner", httpx.AsyncClient]:
+    async def __aenter__(self) -> 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",
         )
-        await self._client.__aenter__()
-        return self, self._client
+        return await self._client.__aenter__()
 
     async def __aexit__(self, *exc) -> None:
         try:
@@ -265,8 +240,7 @@ 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 runner_and_client:
-        _, client = runner_and_client
+    async with _PanelRunner(temp_db_path) as client:
         yield client
 
 
@@ -277,9 +251,6 @@ 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")
@@ -288,11 +259,9 @@ 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 runner_and_client:
-        runner, client = runner_and_client
+    async with _PanelRunner(temp_db_path) as client:
         yield SimpleNamespace(
             client=client,
-            database=runner.app.state.db.database,
             seed=seed,
             today=today,
             pinned_now=pinned_now,
@@ -552,137 +521,3 @@ 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

@@ -1,456 +0,0 @@
-"""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,31 +791,3 @@ 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"]