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

@@ -8,6 +8,10 @@ exposes the two endpoints every later task depends on:
   app has imported core, opened the database, and applied migrations;
   a cheap ``SELECT 1`` confirms the connection is live without leaking
   the filesystem path on an unauthenticated route.
+- ``GET /api/v1/panel/snapshot`` — the panel plane's today snapshot
+  (service-token protected; see :mod:`api.routes_panel`, whose router
+  carries the shared :func:`api.dependencies.require_panel_token`
+  check).
 - ``GET /docs`` — FastAPI's auto-served OpenAPI document (Swagger UI);
   ``/openapi.json`` is the raw schema.  No extra wiring is needed;
   FastAPI serves both by default.
@@ -35,6 +39,7 @@ from fastapi import FastAPI
 
 from api.config import ApiConfig
 from api.database import DatabaseState, make_database
+from api.routes_panel import router as panel_router
 
 
 def _build_lifespan(db_path: str):
@@ -79,6 +84,12 @@ def create_app(config: ApiConfig | None = None) -> FastAPI:
         lifespan=_build_lifespan(cfg.db_path),
     )
 
+    # The resolved config rides on app.state so the reusable route
+    # dependencies (the panel service-token check) read the configured
+    # values from the app rather than module state.
+    app.state.config = cfg
+    app.include_router(panel_router)
+
     @app.get("/health", summary="Service health")
     async def health() -> dict[str, str]:
         """Return service health.

api/config.py

@@ -1,16 +1,22 @@
 """Configuration for the NestQuest API service.
 
-The API reads its single configuration knob — the SQLite database path —
-from the ``NESTQUEST_DB_PATH`` environment variable.  Unlike a dev
-convenience default, a missing or empty value is a configuration ERROR,
-not a silent fallback to a predictable world-writable temp path: the API
-owns the database (it is the writer of record per Feature 16's
-contract), and silently dropping data into ``$TMPDIR/nestquest.db``
-would lose it at reboot.  ``ApiConfig.from_env`` therefore RAISES when
-the variable is unset or empty.  This mirrors the integration's
-"explicit settings object" discipline (Feature 15's
-``core.settings``): the API never reads HA config; callers pass the
-resolved path into the database wiring.
+The API reads its configuration knobs from environment variables:
+
+- ``NESTQUEST_DB_PATH`` — the SQLite database file.  The API opens this
+  file on startup (creating it if absent) and applies migrations.
+- ``NESTQUEST_PANEL_TOKEN`` — the static panel-plane service token
+  (D-012): the credential ``/api/v1/panel/*`` routes require via
+  :func:`api.dependencies.require_panel_token`.  It is stored outside
+  the repo and handed to the service through the environment.
+
+Unlike a dev convenience default, a missing or empty value is a
+configuration ERROR, not a silent fallback: a default database path
+would silently drop data at reboot, and a default (or empty) service
+token would either authenticate everyone or lock the panel plane out
+for good.  ``ApiConfig.from_env`` therefore RAISES when either variable
+is unset or empty.  This mirrors the integration's "explicit settings
+object" discipline (Feature 15's ``core.settings``): the API never
+reads HA config; callers pass the resolved values into the wiring.
 
 Serving: the app is built with the factory ``api.app:create_app`` — run
 it with ``uvicorn --factory api.app:create_app`` so the factory reads
@@ -27,6 +33,14 @@ from dataclasses import dataclass
 #: :meth:`ApiConfig.from_env`.
 DB_PATH_ENV = "NESTQUEST_DB_PATH"
 
+#: Environment variable holding the static panel-plane service token
+#: (D-012), checked by :func:`api.dependencies.require_panel_token`.
+#: REQUIRED — unset or empty raises :class:`ConfigError` from
+#: :meth:`ApiConfig.from_env`; there is no silent default, because an
+#: empty configured token would either authenticate every caller or
+#: lock the panel plane out permanently.
+PANEL_TOKEN_ENV = "NESTQUEST_PANEL_TOKEN"
+
 
 class ConfigError(RuntimeError):
     """Raised when required API configuration is missing or invalid."""
@@ -36,12 +50,18 @@ class ConfigError(RuntimeError):
 class ApiConfig:
     """Resolved API configuration.
 
-    Currently a single field — the SQLite path — but kept as a dataclass
-    so later tasks (auth, SSE, settings persistence) can extend it
-    without touching the call sites that read ``db_path``.
+    Both fields are required (no defaults): ``from_env`` rejects an
+    unset or empty value for either, and a caller building the dataclass
+    directly must pass an explicit token — the token check itself fails
+    closed on an empty configured token, but expecting a caller to
+    discover that at request time is a wiring mistake, so the field is
+    mandatory.  Kept as a dataclass so later tasks (admin plane, SSE,
+    settings persistence) can extend it without touching the call sites
+    that read ``db_path``.
     """
 
     db_path: str
+    panel_token: str
 
     @classmethod
     def from_env(cls, env: dict[str, str] | None = None) -> "ApiConfig":
@@ -52,17 +72,27 @@ class ApiConfig:
 
         Raises :class:`ConfigError` when ``NESTQUEST_DB_PATH`` is unset
         or empty/whitespace-only — there is no silent default, because a
-        default in the temp directory would silently lose data at reboot.
+        default in the temp directory would silently lose data at reboot
+        — and likewise when ``NESTQUEST_PANEL_TOKEN`` is unset or empty
+        (a default token is never invented: the panel plane is closed
+        until a real one is configured).
         """
         source = env if env is not None else os.environ
-        raw = source.get(DB_PATH_ENV, "").strip()
-        if not raw:
+        raw_db_path = source.get(DB_PATH_ENV, "").strip()
+        if not raw_db_path:
             raise ConfigError(
                 f"{DB_PATH_ENV} must be set to a writable SQLite file path "
                 f"(it is unset or empty — the API will not fall back to a "
                 f"temp directory and silently lose data)"
             )
-        return cls(db_path=raw)
+        raw_panel_token = source.get(PANEL_TOKEN_ENV, "").strip()
+        if not raw_panel_token:
+            raise ConfigError(
+                f"{PANEL_TOKEN_ENV} must be set to the panel service token "
+                f"(it is unset or empty — the API will not start without "
+                f"one, and no default token is invented)"
+            )
+        return cls(db_path=raw_db_path, panel_token=raw_panel_token)
 
 
-__all__ = ["ApiConfig", "ConfigError", "DB_PATH_ENV"]
+__all__ = ["ApiConfig", "ConfigError", "DB_PATH_ENV", "PANEL_TOKEN_ENV"]

api/dependencies.py

@@ -0,0 +1,81 @@
+"""Reusable FastAPI dependencies for the NestQuest API routes.
+
+The panel plane (``/api/v1/panel/*``) is authenticated by the static
+service token configured through ``NESTQUEST_PANEL_TOKEN`` (D-012): the
+HA integration presents it as a Bearer credential on every panel
+request.  :func:`require_panel_token` is the ONE implementation of that
+check — every panel route (this task's snapshot route and each later
+panel route) declares it as a dependency rather than re-implementing
+token handling, so the credential, the failure behaviour, and the
+401 contract stay in exactly one place.
+
+The dependency FAILS CLOSED: a missing ``Authorization`` header, a
+malformed one (wrong scheme or empty credential), a token that does not
+match the configured value, and an empty configured token all yield
+HTTP 401.  The comparison is constant-time (:func:`secrets.compare_digest`)
+so the wall-clock difference between "wrong length" and "wrong value"
+cannot be used to probe the token byte by byte, and every rejection
+carries the same detail — nothing in the response distinguishes which
+part of the check failed.
+"""
+from __future__ import annotations
+
+import secrets
+from typing import Annotated
+
+from fastapi import Header, HTTPException, Request
+
+#: The authorization scheme the panel plane expects.  Matched
+#: case-insensitively, as RFC 9110 requires for scheme names.
+_BEARER_SCHEME = "bearer"
+
+#: The 401 detail shared by every rejection — deliberately uniform so
+#: the response never reveals which part of the check failed.
+_REJECTION_DETAIL = "Missing or invalid panel service token"
+
+#: Challenge header returned with a 401, per the Bearer auth scheme.
+_WWW_AUTHENTICATE = {"WWW-Authenticate": "Bearer"}
+
+
+def _reject() -> HTTPException:
+    """Build the single 401 the dependency raises for every failure."""
+    return HTTPException(
+        status_code=401,
+        detail=_REJECTION_DETAIL,
+        headers=_WWW_AUTHENTICATE,
+    )
+
+
+async def require_panel_token(
+    request: Request,
+    authorization: Annotated[str | None, Header()] = None,
+) -> None:
+    """Reject the request with 401 unless it carries the panel token.
+
+    Reads the configured token off ``app.state.config`` (set by
+    :func:`api.app.create_app`) so the dependency stays reusable across
+    routes and carries no module state of its own.  A missing or
+    malformed ``Authorization`` header, a non-matching credential, and
+    an EMPTY configured token (a wiring mistake) all raise the same
+    ``HTTPException(401)`` — the check is closed even when the service
+    is misconfigured.
+
+    Returns ``None`` on success; FastAPI discards a ``None`` return,
+    the route handler simply proceeds.
+    """
+    config = getattr(request.app.state, "config", None)
+    expected = config.panel_token if config is not None else ""
+    if not expected or authorization is None:
+        raise _reject()
+    scheme, _, credential = authorization.partition(" ")
+    if scheme.lower() != _BEARER_SCHEME or not credential.strip():
+        raise _reject()
+    # compare_digest on bytes so a non-ASCII token raises nothing and a
+    # timing side channel on the str path is not available.
+    if not secrets.compare_digest(
+        credential.strip().encode("utf-8"), expected.encode("utf-8")
+    ):
+        raise _reject()
+
+
+__all__ = ["require_panel_token"]

api/nestquest_core.py

@@ -107,4 +107,19 @@ core: ModuleType = _load_core()
 core_db = importlib.import_module(f"{_CORE_MODULE_NAME}.db")
 core_migrations = importlib.import_module(f"{_CORE_MODULE_NAME}.migrations")
 
-__all__ = ["core", "core_db", "core_migrations"]
+#: The panel snapshot builder and its payload shaper
+#: (``core.snapshot.build_snapshot`` / ``instance_payload``) — the pure
+#: assembly the panel routes call.
+core_snapshot = importlib.import_module(f"{_CORE_MODULE_NAME}.snapshot")
+
+#: 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_db",
+    "core_migrations",
+    "core_snapshot",
+    "core_settings",
+]

api/routes_panel.py

@@ -0,0 +1,183 @@
+"""Panel-plane routes for the NestQuest API service.
+
+The panel plane (``/api/v1/panel/*``, D-012) serves the wall panel's
+reads.  Every route on :data:`router` requires the panel service token
+through the ONE reusable dependency :func:`api.dependencies.require_panel_token`
+(declared once at the router level, so each later panel route added to
+this router inherits the check).
+
+The snapshot route is a thin adapter over the bundled core: it performs
+NO business logic.  It reads the live database off ``app.state.db``,
+resolves ONE timezone-aware ``now`` (the API host's local time — the
+API is the household's "local" host now that the domain no longer lives
+inside Home Assistant), calls the pure
+:func:`nestquest_core.snapshot.build_snapshot`, shapes each child's
+instances through :func:`nestquest_core.snapshot.instance_payload` with
+``include_missed=False`` (D-009: missed instances are OMITTED from the
+panel payload), and returns the documented JSON shape:
+
+.. code-block:: json
+
+    {
+      "today_iso": "2026-09-22",
+      "cycle_day": 2,
+      "children": [
+        {
+          "child_id": 1,
+          "child_name": "Ada",
+          "present": false,
+          "next_present": "2026-09-24",
+          "due_today": 1,
+          "completed_today": 0,
+          "remaining_today": 1,
+          "completion_pct": 0,
+          "instances": [
+            {"id": 7, "definition_id": 3, "child_id": 1,
+             "title": "Pack bag", "icon": null, "window": "morning",
+             "due_time": "10:00", "state": "open", "overdue": true,
+             "completed_at": null, "on_time": null}
+          ]
+        }
+      ]
+    }
+
+``children`` follows the household's sort order (the core builder lists
+active children ``ORDER BY sort_order, id``); ``state`` is the
+documented panel spelling ``open`` | ``completed`` (§1 of
+design/ENTITIES-AND-SERVICES.md); ``cycle_day`` is today's 1-based day
+in the first scheduled child's custody cycle (0 when no child has one).
+The same :class:`PanelSnapshotResponse` Pydantic model is the route's
+``response_model``, so the OpenAPI document carries the shape too.
+"""
+from __future__ import annotations
+
+import datetime
+
+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_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.
+router = APIRouter(
+    prefix="/api/v1/panel",
+    dependencies=[Depends(require_panel_token)],
+)
+
+
+class PanelInstancePayload(BaseModel):
+    """One instance in the documented panel payload shape (§1).
+
+    Mirrors :func:`nestquest_core.snapshot.instance_payload`:
+    ``state`` is ``open`` | ``completed`` (missed instances never reach
+    the panel payload), ``on_time`` passes the completion event's flag
+    through verbatim (``None`` for an open instance), and
+    ``completed_at`` is the event's UTC timestamp when completed.
+    """
+
+    id: int
+    definition_id: int
+    child_id: int
+    title: str
+    icon: str | None
+    window: str
+    due_time: str | None
+    state: str
+    overdue: bool
+    completed_at: str | None
+    on_time: bool | None
+
+
+class PanelChildPayload(BaseModel):
+    """One child's day: presence, rollup counts, and today's instances."""
+
+    child_id: int
+    child_name: str
+    present: bool
+    #: ISO date of the child's next present day while they are away
+    #: (the away plate renders ``Returns <weekday>, <Mon D>`` from it);
+    #: ``None`` when the child is present today.
+    next_present: str | None
+    due_today: int
+    completed_today: int
+    remaining_today: int
+    completion_pct: int
+    instances: list[PanelInstancePayload]
+
+
+class PanelSnapshotResponse(BaseModel):
+    """The whole household snapshot the panel renders from.
+
+    ``today_iso`` anchors the payload's day; ``cycle_day`` is today's
+    1-based day in the first scheduled child's custody cycle (0 when no
+    child has a schedule); ``children`` is in the household's sort
+    order.
+    """
+
+    today_iso: str
+    cycle_day: int
+    children: list[PanelChildPayload]
+
+
+def _local_now() -> datetime.datetime:
+    """Return the API host's local time as ONE timezone-aware clock read.
+
+    ``now().astimezone()`` attaches the host's local timezone to a
+    single clock read, so the calendar date and wall-clock time the
+    builder derives can never disagree.  Kept as a module function (not
+    inlined) so tests pin the instant the same way the coordinator's
+    ``_local_now`` is pinned.
+    """
+    return datetime.datetime.now().astimezone()
+
+
+@router.get(
+    "/snapshot",
+    summary="Today's household snapshot",
+    response_model=PanelSnapshotResponse,
+)
+async def panel_snapshot(request: Request) -> PanelSnapshotResponse:
+    """Return today's full household snapshot for the panel.
+
+    Requires the panel service token (checked by the router's shared
+    :func:`~api.dependencies.require_panel_token` dependency).  Builds
+    the snapshot through the pure core builder against the live
+    database and shapes each child's instances with
+    ``instance_payload(include_missed=False)`` — missed instances are
+    OMITTED from the panel payload (D-009) — then returns the shape
+    documented in this module's docstring and
+    :class:`PanelSnapshotResponse`.
+    """
+    state: DatabaseState = request.app.state.db
+    snapshot = await core_snapshot.build_snapshot(
+        state.database, core_settings.NestQuestSettings(), _local_now()
+    )
+    return PanelSnapshotResponse(
+        today_iso=snapshot.today_iso,
+        cycle_day=snapshot.cycle_day,
+        children=[
+            PanelChildPayload(
+                child_id=child.child_id,
+                child_name=child.child_name,
+                present=child.present,
+                next_present=child.next_present,
+                due_today=child.due_today,
+                completed_today=child.completed_today,
+                remaining_today=child.remaining_today,
+                completion_pct=child.completion_pct,
+                instances=[
+                    PanelInstancePayload.model_validate(instance)
+                    for instance in core_snapshot.instance_payload(
+                        child.instances, include_missed=False
+                    )
+                ],
+            )
+            for child in snapshot.children
+        ],
+    )
+
+
+__all__ = ["router"]

custom_components/nestquest/__init__.py

@@ -496,6 +496,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
             hass,
             entry_id=entry.entry_id,
             database=database,
+            settings=settings,
             update_interval_seconds=entry.options.get(
                 CONF_UPDATE_INTERVAL, DEFAULT_UPDATE_INTERVAL
             ),

custom_components/nestquest/coordinator.py

@@ -11,6 +11,15 @@ refresh updates the whole board coherently.
 host clock and never UTC, matching the materialization and completion
 layers.
 
+The snapshot assembly itself lives in the Home-Assistant-free
+:mod:`custom_components.nestquest.core.snapshot` module
+(:func:`~core.snapshot.build_snapshot`); this coordinator is now a thin
+HA wrapper that resolves the local timezone and reads the settings handed
+into its constructor, then delegates to that builder, passing the SAME
+HA-local ``now`` it resolved in one clock read.  The API service's panel
+snapshot route shares the same builder, so both consumers render ONE
+snapshot shape.
+
 Presence is resolved through the SAME engine the materializer uses
 (:class:`~.presence.PresenceEngine`): schedules and overrides are read
 with the public DAO methods and the engine's anchor-date arithmetic
@@ -27,13 +36,11 @@ from __future__ import annotations
 
 import asyncio
 import datetime
-from dataclasses import dataclass
 from zoneinfo import ZoneInfo
 
 from homeassistant.core import HomeAssistant
 from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
 
-from .completion import derive_state
 from .const import (
     CONF_UPDATE_INTERVAL,
     DEFAULT_UPDATE_INTERVAL,
@@ -41,66 +48,21 @@ from .const import (
     LOGGER,
     MIN_UPDATE_INTERVAL,
 )
-from .dao_children import ChildrenDao
-from .dao_instances import (
-    CompletionEventsDao,
-    QuestInstancesDao,
-)
-from .dao_rules import QuestDefinitionsDao
-from .dao_presence import PresenceOverridesDao, PresenceSchedulesDao
-from .db import NestQuestDatabase
-from .presence import (
-    MAX_PREVIEW_SCAN_DAYS,
-    PresenceEngine,
-    PresenceOverride,
-    PresenceSchedule,
+from .core.db import NestQuestDatabase
+from .core.settings import NestQuestSettings
+from .core.snapshot import (
+    ChildDaySnapshot,
+    NestQuestSnapshot,
+    QuestInstanceView,
+    build_snapshot,
 )
 
-
-@dataclass(frozen=True)
-class QuestInstanceView:
-    """One instance plus everything the entities render from it."""
-
-    instance_id: int
-    definition_id: int
-    child_id: int
-    title: str
-    icon: str | None
-    window: str
-    due_date: str
-    due_time: str | None
-    state: str  # open | done | missed
-    overdue: bool
-    completed_at: str | None
-    was_on_time: bool | None
-
-
-@dataclass(frozen=True)
-class ChildDaySnapshot:
-    """One active child's day: instances, presence, and rollup counts."""
-
-    child_id: int
-    child_name: str
-    instances: tuple[QuestInstanceView, ...]
-    present: bool
-    due_today: int
-    completed_today: int
-    remaining_today: int
-    completion_pct: int
-    #: For a child away today: the ISO date of their next present day
-    #: (the panel's away plate renders ``Returns <weekday>, <Mon D>``
-    #: from it).  ``None`` when the child is present today or no
-    #: present day exists within the presence engine's scan cap.
-    next_present: str | None = None
-
-
-@dataclass(frozen=True)
-class NestQuestSnapshot:
-    """The coordinator's whole-entity data payload."""
-
-    today_iso: str
-    children: tuple[ChildDaySnapshot, ...]
-    cycle_day: int
+__all__ = [
+    "ChildDaySnapshot",
+    "NestQuestCoordinator",
+    "NestQuestSnapshot",
+    "QuestInstanceView",
+]
 
 
 class NestQuestCoordinator(DataUpdateCoordinator):
@@ -121,6 +83,7 @@ class NestQuestCoordinator(DataUpdateCoordinator):
         *,
         entry_id: str,
         database: NestQuestDatabase,
+        settings: NestQuestSettings,
         update_interval_seconds: int | None = None,
     ) -> None:
         interval = update_interval_seconds
@@ -138,6 +101,7 @@ class NestQuestCoordinator(DataUpdateCoordinator):
         )
         self.entry_id = entry_id
         self.database = database
+        self.settings = settings
         self._refresh_lock = asyncio.Lock()
 
     async def async_refresh(self) -> None:
@@ -153,140 +117,6 @@ class NestQuestCoordinator(DataUpdateCoordinator):
 
     async def _async_update_data(self) -> NestQuestSnapshot:
         now = self._local_now()
-        today = now.date()
-        today_iso = today.isoformat()
-
-        children = await ChildrenDao(self.database).list_active()
-
-        schedules: dict[int, PresenceSchedule] = {}
-        overrides: dict[int, list[PresenceOverride]] = {}
-        for child in children:
-            schedule = await PresenceSchedulesDao(self.database).get_by_child(
-                child.id
-            )
-            if schedule is not None:
-                schedules[child.id] = PresenceSchedule.decode(
-                    child.id, schedule.anchor_date, schedule.pattern
-                )
-            window_end = today + datetime.timedelta(
-                days=MAX_PREVIEW_SCAN_DAYS
-            )
-            window_end_iso = window_end.isoformat()
-            ranged = await PresenceOverridesDao(
-                self.database
-            ).list_by_child_and_range(child.id, today_iso, window_end_iso)
-            if ranged:
-                # The engine validates MODEL objects, not DAO records;
-                # each record converts losslessly (the schema guarantees
-                # end >= start and a real is_present).
-                overrides[child.id] = [
-                    PresenceOverride(
-                        record.child_id,
-                        record.start_date,
-                        record.end_date,
-                        record.is_present,
-                        record.note,
-                    )
-                    for record in ranged
-                ]
-        engine = PresenceEngine(schedules, overrides)
-
-        instances_dao = QuestInstancesDao(self.database)
-        events_dao = CompletionEventsDao(self.database)
-        definitions_dao = QuestDefinitionsDao(self.database)
-
-        snapshots: list[ChildDaySnapshot] = []
-        for child in children:
-            instances = await instances_dao.list_by_date_range(
-                child.id, today_iso, today_iso
-            )
-            views: list[QuestInstanceView] = []
-            for instance in instances:
-                latest = await events_dao.get_latest_for_instance(instance.id)
-                definition = await definitions_dao.get(instance.definition_id)
-                state = derive_state(instance, latest, today)
-                overdue = False
-                if state == "open" and instance.due_date == today_iso:
-                    if instance.due_time is not None:
-                        due_moment = datetime.datetime.strptime(
-                            instance.due_time, "%H:%M"
-                        ).time()
-                        overdue = now.time() > due_moment
-                views.append(
-                    QuestInstanceView(
-                        instance_id=instance.id,
-                        definition_id=instance.definition_id,
-                        child_id=instance.child_id,
-                        title=definition.title if definition else "Quest",
-                        icon=definition.icon if definition else None,
-                        window=instance.window,
-                        due_date=instance.due_date,
-                        due_time=instance.due_time,
-                        state=state,
-                        overdue=overdue,
-                        completed_at=(
-                            latest.occurred_at
-                            if latest is not None
-                            and latest.event_type == "completed"
-                            else None
-                        ),
-                        was_on_time=(
-                            latest.was_on_time
-                            if latest is not None
-                            and latest.event_type == "completed"
-                            else None
-                        ),
-                    )
-                )
-            due = len(views)
-            completed = sum(1 for view in views if view.state == "done")
-            remaining = due - completed
-            child_present = engine.is_present(child.id, today)
-            next_present: str | None = None
-            if not child_present:
-                try:
-                    next_present = engine.next_present_dates(
-                        child.id, today, 1
-                    )[0].isoformat()
-                except ValueError:
-                    next_present = None
-            snapshots.append(
-                ChildDaySnapshot(
-                    child_id=child.id,
-                    child_name=child.display_name,
-                    instances=tuple(views),
-                    present=child_present,
-                    due_today=due,
-                    completed_today=completed,
-                    remaining_today=remaining,
-                    completion_pct=(
-                        round(completed * 100 / due) if due else 100
-                    ),
-                    next_present=next_present,
-                )
-            )
-
-        return NestQuestSnapshot(
-            today_iso=today_iso,
-            children=tuple(snapshots),
-            cycle_day=_cycle_day(schedules, children, today),
+        return await build_snapshot(
+            self.database, self.settings, now
         )
-
-
-def _cycle_day(
-    schedules: dict[int, PresenceSchedule], children, today: datetime.date
-) -> int:
-    """Return today's 1-based day in the first scheduled child's cycle.
-
-    Anchor-date arithmetic ONLY (D-004): the day offset is computed from
-    the schedule's own anchor, never ISO week numbers or parity, so a
-    53-week year cannot silently invert the cycle.
-    """
-    for child in children:
-        schedule = schedules.get(child.id)
-        if schedule is None:
-            continue
-        days = (today - schedule.anchor_date).days
-        cycle_length = schedule.cycle_length_weeks * 7
-        return (days % cycle_length) + 1
-    return 0

custom_components/nestquest/core/snapshot.py

@@ -0,0 +1,306 @@
+"""Pure, Home-Assistant-free panel snapshot builder.
+
+The household snapshot the NestQuest panel renders — active children in
+sort order, each child's today-instances in the documented payload shape
+(missed instances OMITTED from the entity-facing attribute payload, but
+retained on the snapshot for the count rollups), per-child counts and
+presence, and the custody cycle day — used to be assembled inline inside
+:mod:`custom_components.nestquest.coordinator`'s
+``_async_update_data``.  That assembly is HA-coupled only because its
+host is a :class:`~homeassistant.helpers.update_coordinator.DataUpdateCoordinator`;
+the assembly itself touches no HA surface.
+
+This module holds the pure pieces — the snapshot dataclasses
+(:class:`QuestInstanceView`, :class:`ChildDaySnapshot`,
+:class:`NestQuestSnapshot`), the cycle-day helper (:func:`_cycle_day`),
+the panel payload shaper (:func:`instance_payload`), and the async
+:func:`build_snapshot` builder — so the HA coordinator and the API
+service's panel snapshot route share ONE implementation of both the
+snapshot assembly and the entity-facing attribute payload shape.
+Nothing here imports :mod:`homeassistant`; the builder takes the
+database, the settings object, and the HA-local ``now`` (a single
+clock read the caller already resolved, from which it derives
+``today``) and returns the snapshot, mirroring exactly what the
+coordinator used to do inline.
+
+``settings`` is accepted because the API route hands in the entry's
+:class:`~.settings.NestQuestSettings`; it is CURRENTLY UNUSED — the
+assembly reads nothing from it — and is retained so the builder's
+contract stays stable if a future panel feature derives scan caps or
+rollup behaviour from configuration.  No speculative read is wired in.
+"""
+from __future__ import annotations
+
+import datetime
+from collections.abc import Iterable
+from dataclasses import dataclass
+from typing import Any
+
+from .completion import derive_state
+from .dao_children import ChildrenDao
+from .dao_instances import CompletionEventsDao, QuestInstancesDao
+from .dao_presence import PresenceOverridesDao, PresenceSchedulesDao
+from .dao_rules import QuestDefinitionsDao
+from .db import NestQuestDatabase
+from .presence import (
+    MAX_PREVIEW_SCAN_DAYS,
+    PresenceEngine,
+    PresenceOverride,
+    PresenceSchedule,
+)
+from .settings import NestQuestSettings
+
+
+@dataclass(frozen=True)
+class QuestInstanceView:
+    """One instance plus everything the entities render from it."""
+
+    instance_id: int
+    definition_id: int
+    child_id: int
+    title: str
+    icon: str | None
+    window: str
+    due_date: str
+    due_time: str | None
+    state: str  # open | done | missed
+    overdue: bool
+    completed_at: str | None
+    was_on_time: bool | None
+
+
+@dataclass(frozen=True)
+class ChildDaySnapshot:
+    """One active child's day: instances, presence, and rollup counts."""
+
+    child_id: int
+    child_name: str
+    instances: tuple[QuestInstanceView, ...]
+    present: bool
+    due_today: int
+    completed_today: int
+    remaining_today: int
+    completion_pct: int
+    #: For a child away today: the ISO date of their next present day
+    #: (the panel's away plate renders ``Returns <weekday>, <Mon D>``
+    #: from it).  ``None`` when the child is present today or no
+    #: present day exists within the presence engine's scan cap.
+    next_present: str | None = None
+
+
+@dataclass(frozen=True)
+class NestQuestSnapshot:
+    """The coordinator's whole-entity data payload."""
+
+    today_iso: str
+    children: tuple[ChildDaySnapshot, ...]
+    cycle_day: int
+
+
+def _cycle_day(
+    schedules: dict[int, PresenceSchedule], children, today: datetime.date
+) -> int:
+    """Return today's 1-based day in the first scheduled child's cycle.
+
+    Anchor-date arithmetic ONLY (D-004): the day offset is computed from
+    the schedule's own anchor, never ISO week numbers or parity, so a
+    53-week year cannot silently invert the cycle.
+    """
+    for child in children:
+        schedule = schedules.get(child.id)
+        if schedule is None:
+            continue
+        days = (today - schedule.anchor_date).days
+        cycle_length = schedule.cycle_length_weeks * 7
+        return (days % cycle_length) + 1
+    return 0
+
+
+def instance_payload(
+    instances: Iterable[QuestInstanceView], *, include_missed: bool
+) -> list[dict[str, Any]]:
+    """Shape instance dicts per ENTITIES-AND-SERVICES.md §1.
+
+    The panel payload ``state`` is the documented ``open`` | ``completed``
+    spelling, so the builder's derived ``done`` maps to ``completed`` here;
+    ``open`` passes through.  ``missed`` instances are OMITTED from the
+    entity-facing attribute payload (decision 5) unless ``include_missed``
+    is set — the admin payload (D-009) keeps them with their ``missed``
+    state.  ``on_time`` passes the builder's ``was_on_time`` through
+    verbatim (``True`` / ``False`` for a completed instance, ``None`` for
+    an open one).  This is the SINGLE shared implementation both the HA
+    sensor layer and the API panel route read; it touches no Home
+    Assistant surface.
+    """
+    payload: list[dict[str, Any]] = []
+    for view in instances:
+        if view.state == "missed" and not include_missed:
+            continue
+        payload.append(
+            {
+                "id": view.instance_id,
+                "definition_id": view.definition_id,
+                "child_id": view.child_id,
+                "title": view.title,
+                "icon": view.icon,
+                "window": view.window,
+                "due_time": view.due_time,
+                "state": (
+                    "completed" if view.state == "done" else view.state
+                ),
+                "overdue": view.overdue,
+                "completed_at": view.completed_at,
+                "on_time": view.was_on_time,
+            }
+        )
+    return payload
+
+
+async def build_snapshot(
+    database: NestQuestDatabase,
+    settings: NestQuestSettings,
+    now: datetime.datetime,
+) -> NestQuestSnapshot:
+    """Build the household panel snapshot for ``today`` at the ``now`` instant.
+
+    Mirrors the assembly that lived in the coordinator's
+    ``_async_update_data``: active children in sort order, each child's
+    today-instances with derived state (open / done / missed via
+    :func:`~.completion.derive_state`) and the overdue flag, per-child
+    presence (resolved through :class:`~.presence.PresenceEngine`), the
+    per-child rollup counts, and the custody cycle day of the first
+    scheduled active child.  Missed instances are retained on the
+    snapshot (the rollup counts and the panel's missed plate need them)
+    but the entity-facing attribute payload omits them at the
+    :func:`instance_payload` layer — this builder produces the shared
+    shape both consumers read.
+
+    ``now`` is the HA-local timezone-aware datetime the caller already
+    resolved (a SINGLE clock read); the overdue check compares
+    ``now.time()`` against the instance's ``due_time`` for today's open
+    instances.  The builder performs NO internal clock read — it never
+    calls :func:`datetime.datetime.now` — so the instant is pinned in
+    tests.  ``today`` is derived INSIDE the builder as ``now.date()``,
+    so the calendar date that anchors the due-date and cycle-day
+    arithmetic is always the SAME clock read as the wall-clock time;
+    the two can never disagree.  ``settings`` is accepted but currently
+    unused (see the module docstring).
+    """
+    today = now.date()
+    today_iso = today.isoformat()
+
+    children = await ChildrenDao(database).list_active()
+
+    schedules: dict[int, PresenceSchedule] = {}
+    overrides: dict[int, list[PresenceOverride]] = {}
+    for child in children:
+        schedule = await PresenceSchedulesDao(database).get_by_child(
+            child.id
+        )
+        if schedule is not None:
+            schedules[child.id] = PresenceSchedule.decode(
+                child.id, schedule.anchor_date, schedule.pattern
+            )
+        window_end = today + datetime.timedelta(
+            days=MAX_PREVIEW_SCAN_DAYS
+        )
+        window_end_iso = window_end.isoformat()
+        ranged = await PresenceOverridesDao(
+            database
+        ).list_by_child_and_range(child.id, today_iso, window_end_iso)
+        if ranged:
+            # The engine validates MODEL objects, not DAO records;
+            # each record converts losslessly (the schema guarantees
+            # end >= start and a real is_present).
+            overrides[child.id] = [
+                PresenceOverride(
+                    record.child_id,
+                    record.start_date,
+                    record.end_date,
+                    record.is_present,
+                    record.note,
+                )
+                for record in ranged
+            ]
+    engine = PresenceEngine(schedules, overrides)
+
+    instances_dao = QuestInstancesDao(database)
+    events_dao = CompletionEventsDao(database)
+    definitions_dao = QuestDefinitionsDao(database)
+
+    snapshots: list[ChildDaySnapshot] = []
+    for child in children:
+        instances = await instances_dao.list_by_date_range(
+            child.id, today_iso, today_iso
+        )
+        views: list[QuestInstanceView] = []
+        for instance in instances:
+            latest = await events_dao.get_latest_for_instance(instance.id)
+            definition = await definitions_dao.get(instance.definition_id)
+            state = derive_state(instance, latest, today)
+            overdue = False
+            if state == "open" and instance.due_date == today_iso:
+                if instance.due_time is not None:
+                    due_moment = datetime.datetime.strptime(
+                        instance.due_time, "%H:%M"
+                    ).time()
+                    overdue = now.time() > due_moment
+            views.append(
+                QuestInstanceView(
+                    instance_id=instance.id,
+                    definition_id=instance.definition_id,
+                    child_id=instance.child_id,
+                    title=definition.title if definition else "Quest",
+                    icon=definition.icon if definition else None,
+                    window=instance.window,
+                    due_date=instance.due_date,
+                    due_time=instance.due_time,
+                    state=state,
+                    overdue=overdue,
+                    completed_at=(
+                        latest.occurred_at
+                        if latest is not None
+                        and latest.event_type == "completed"
+                        else None
+                    ),
+                    was_on_time=(
+                        latest.was_on_time
+                        if latest is not None
+                        and latest.event_type == "completed"
+                        else None
+                    ),
+                )
+            )
+        due = len(views)
+        completed = sum(1 for view in views if view.state == "done")
+        remaining = due - completed
+        child_present = engine.is_present(child.id, today)
+        next_present: str | None = None
+        if not child_present:
+            try:
+                next_present = engine.next_present_dates(
+                    child.id, today, 1
+                )[0].isoformat()
+            except ValueError:
+                next_present = None
+        snapshots.append(
+            ChildDaySnapshot(
+                child_id=child.id,
+                child_name=child.display_name,
+                instances=tuple(views),
+                present=child_present,
+                due_today=due,
+                completed_today=completed,
+                remaining_today=remaining,
+                completion_pct=(
+                    round(completed * 100 / due) if due else 100
+                ),
+                next_present=next_present,
+            )
+        )
+
+    return NestQuestSnapshot(
+        today_iso=today_iso,
+        children=tuple(snapshots),
+        cycle_day=_cycle_day(schedules, children, today),
+    )

custom_components/nestquest/sensor.py

@@ -39,6 +39,7 @@ from .coordinator import (
     NestQuestCoordinator,
     QuestInstanceView,
 )
+from .core.snapshot import instance_payload
 
 #: Home Assistant's state machine refuses states longer than 255
 #: characters (its hard recorder/state limit).  These sensors' states
@@ -63,40 +64,6 @@ def _state_within_limit(state: Any) -> Any:
     return state
 
 
-def _instance_payload(
-    instances: Iterable[QuestInstanceView], *, include_missed: bool
-) -> list[dict[str, Any]]:
-    """Shape instance dicts per ENTITIES-AND-SERVICES.md §1.
-
-    ``state`` maps the coordinator's derived ``done`` to the panel's
-    ``completed`` spelling; ``missed`` instances are omitted unless
-    ``include_missed`` (the admin payload, D-009, keeps them with
-    their ``missed`` state).
-    """
-    payload: list[dict[str, Any]] = []
-    for view in instances:
-        if view.state == "missed" and not include_missed:
-            continue
-        payload.append(
-            {
-                "id": view.instance_id,
-                "definition_id": view.definition_id,
-                "child_id": view.child_id,
-                "title": view.title,
-                "icon": view.icon,
-                "window": view.window,
-                "due_time": view.due_time,
-                "state": (
-                    "completed" if view.state == "done" else view.state
-                ),
-                "overdue": view.overdue,
-                "completed_at": view.completed_at,
-                "on_time": view.was_on_time,
-            }
-        )
-    return payload
-
-
 #: State reported by the next-quest sensor when the child owes nothing
 #: else today.  A documented sentinel, never an empty string (an empty
 #: state reads as "sensor went unavailable" in HA).
@@ -215,10 +182,10 @@ class NestQuestQuestsDueTodaySensor(_NestQuestChildDaySensor):
         if child is None:
             return None
         return {
-            "instances": _instance_payload(
+            "instances": instance_payload(
                 child.instances, include_missed=False
             ),
-            "admin_instances": _instance_payload(
+            "admin_instances": instance_payload(
                 child.instances, include_missed=True
             ),
             "child_id": child.child_id,

tests/test_api_health.py

@@ -69,7 +69,12 @@ class _AppRunner:
     """
 
     def __init__(self, db_path: str) -> None:
-        self.app = create_app(ApiConfig(db_path=db_path))
+        # The token value is irrelevant to these tests (the health
+        # route is unauthenticated); it just satisfies the required
+        # field — the panel token's own tests live in test_api_panel.py.
+        self.app = create_app(
+            ApiConfig(db_path=db_path, panel_token="health-runner-token")
+        )
 
     async def __aenter__(self) -> httpx.AsyncClient:
         self._lifespan = self.app.router.lifespan_context(self.app)
@@ -97,9 +102,15 @@ async def app_client(temp_db_path: str) -> httpx.AsyncClient:
 
 
 def test_from_env_returns_set_value() -> None:
-    """A set, non-empty NESTQUEST_DB_PATH is returned verbatim."""
-    cfg = ApiConfig.from_env({"NESTQUEST_DB_PATH": "/data/nq.db"})
+    """Set, non-empty env values are returned verbatim."""
+    cfg = ApiConfig.from_env(
+        {
+            "NESTQUEST_DB_PATH": "/data/nq.db",
+            "NESTQUEST_PANEL_TOKEN": "panel-token",
+        }
+    )
     assert cfg.db_path == "/data/nq.db"
+    assert cfg.panel_token == "panel-token"
 
 
 def test_from_env_raises_on_empty_value() -> None:
@@ -283,10 +294,11 @@ _NO_HA_SCRIPT = textwrap.dedent(
 
     # Build the app (the lifespan is NOT run here, so the path is never
     # opened — it just needs to be a non-empty string to satisfy the
-    # fail-fast config check).  Use a temp path so nothing is created.
+    # fail-fast config check).  Use a temp path so nothing is created;
+    # the panel token value is irrelevant to this import-only check.
     app = create_app(ApiConfig(db_path=str(
         Path(tempfile.gettempdir()) / "nq-noha-buildonly.db"
-    )))
+    ), panel_token="noha-import-token"))
     assert app is not None
 
     print("API NO-HA IMPORT OK")

tests/test_api_panel.py

@@ -0,0 +1,523 @@
+"""Panel-plane route tests: service token + snapshot (task ae7e90ae).
+
+These tests exercise the NestQuest API's panel plane against the ASGI
+app with httpx:
+
+- ``GET /api/v1/panel/snapshot`` requires the panel service token
+  (D-012): a request with NO ``Authorization`` header, a MALFORMED
+  header, and a WRONG token all return 401; the correct Bearer token
+  returns 200.
+- The 200 body is the documented snapshot shape: ``today_iso``,
+  ``cycle_day``, and ``children`` in the household's sort order, each
+  carrying ``child_id``/``child_name``/``present``/``next_present``,
+  the per-day rollup counts, and its instances in the documented panel
+  payload shape (ENTITIES-AND-SERVICES.md §1) with missed instances
+  OMITTED (D-009).
+- ``ApiConfig.from_env`` requires ``NESTQUEST_PANEL_TOKEN`` with the
+  same fail-fast discipline as ``NESTQUEST_DB_PATH``.
+
+The household is seeded through the SAME core copy the app uses — the
+``nestquest_core.*`` package registered by ``api/nestquest_core.py`` —
+NOT the conftest-stubbed ``custom_components.nestquest.core`` copy (see
+the two-copy caveat in api/nestquest_core.py's docstring: never compare
+class or exception identity across the two copies).  The seed opens and
+migrates its own connection on the app's SQLite path and closes BEFORE
+the app's lifespan opens the file, so the two connections never
+overlap.
+
+The route's single clock read (:func:`api.routes_panel._local_now`) is
+pinned with monkeypatch the same way the coordinator's ``_local_now``
+is pinned in test_core_snapshot.py, so the seeded rows and the
+snapshot's ``today`` can never straddle midnight.
+"""
+from __future__ import annotations
+
+import datetime
+import importlib
+from pathlib import Path
+from types import SimpleNamespace
+from zoneinfo import ZoneInfo
+
+import httpx
+import pytest
+
+from api import routes_panel
+from api.app import create_app
+from api.config import ApiConfig, ConfigError
+from api.database import _executor
+from api.nestquest_core import (
+    core_db,
+    core_migrations,
+    core_snapshot,
+)
+
+#: The panel service token the test apps are configured with.
+PANEL_TOKEN = "panel-service-token-test"
+
+# The bundled core submodules, through the SAME copy the app uses
+# (nestquest_core.*, NOT the conftest-stubbed custom_components copy).
+_children = importlib.import_module("nestquest_core.children")
+_quest_definitions = importlib.import_module(
+    "nestquest_core.quest_definitions"
+)
+_materialize = importlib.import_module("nestquest_core.materialize")
+_recurrence = importlib.import_module("nestquest_core.recurrence")
+_dao_presence = importlib.import_module("nestquest_core.dao_presence")
+_dao_instances = importlib.import_module("nestquest_core.dao_instances")
+_completion = importlib.import_module("nestquest_core.completion")
+
+
+async def _seed_household(
+    db_path: str,
+    today: datetime.date,
+    pinned_now: datetime.datetime,
+) -> SimpleNamespace:
+    """Seed a two-child household on ``db_path`` and return its records.
+
+    Mirrors the test_core_snapshot.py household: an AWAY child (2-week
+    custody cycle anchored yesterday, empty week 0, an override returns
+    her today+2) with an open quest due 10:00 — overdue at the pinned
+    12:00 — and a present child whose 17:00 quest is completed at
+    12:00 (on time).  A third quest's instance is materialized due
+    YESTERDAY (via ``materialize`` with ``today`` pinned to yesterday,
+    since the no-past guard refuses past-dated writes): a past-due OPEN
+    instance, i.e. the "missed" view the panel payload must omit.
+
+    Opens, migrates, seeds, and CLOSES its own connection so the app's
+    lifespan connection never overlaps it.
+    """
+    database = core_db.NestQuestDatabase(_executor)
+    await database.open(db_path)
+    try:
+        await core_migrations.apply_migrations(database)
+        today_iso = today.isoformat()
+        yesterday = today - datetime.timedelta(days=1)
+        yesterday_iso = yesterday.isoformat()
+
+        ada = await _children.create_child(database, "Ada", sort_order=0)
+        bo = await _children.create_child(database, "Bo", sort_order=1)
+
+        # Every rule starts YESTERDAY so the yesterday materialization
+        # below (which pins ``today`` to yesterday) fires them.
+        await _quest_definitions.create_quest_definition(
+            database,
+            "Pack bag",
+            _recurrence.ScheduleRule.from_dict(
+                {"rule_type": "daily", "start_date": yesterday_iso}
+            ),
+            [ada.id],
+            [("morning", "10:00")],
+        )
+        await _quest_definitions.create_quest_definition(
+            database,
+            "Brush teeth",
+            _recurrence.ScheduleRule.from_dict(
+                {"rule_type": "daily", "start_date": yesterday_iso}
+            ),
+            [bo.id],
+            [("morning", "17:00")],
+        )
+        # Today's instances first, while Ada has no presence schedule
+        # (she is present every day, so her instance IS generated).
+        await _materialize.materialize(
+            database,
+            today_iso,
+            (today + datetime.timedelta(days=3)).isoformat(),
+            today=today,
+        )
+        # The stale quest is created AFTER the today walk, so its only
+        # instance is the yesterday one written below.
+        await _quest_definitions.create_quest_definition(
+            database,
+            "Stale chore",
+            _recurrence.ScheduleRule.from_dict(
+                {"rule_type": "daily", "start_date": yesterday_iso}
+            ),
+            [ada.id],
+            [("morning", "09:00")],
+        )
+        # Materialize yesterday with ``today`` pinned to yesterday: the
+        # no-past guard accepts the rows, and they land past-due (no
+        # completion events) — the missed view.
+        await _materialize.materialize(
+            database, yesterday_iso, yesterday_iso, today=yesterday
+        )
+        instances = _dao_instances.QuestInstancesDao(database)
+
+        ada_today = await instances.list_by_child_and_date(ada.id, today_iso)
+        assert len(ada_today) == 1
+        pack_bag_instance = ada_today[0]
+        bo_today = await instances.list_by_child_and_date(bo.id, today_iso)
+        assert len(bo_today) == 1
+        brush_teeth_instance = bo_today[0]
+        ada_yesterday = await instances.list_by_child_and_date(
+            ada.id, yesterday_iso
+        )
+        stale_instance = next(
+            record
+            for record in ada_yesterday
+            if record.due_time == "09:00"
+        )
+
+        # Ada's custody schedule, added AFTER materialization (the
+        # same order test_core_snapshot.py seeds): 2-week cycle
+        # anchored yesterday, empty week 0 covers today => absent
+        # today; an override returns her on today+2.
+        await _dao_presence.PresenceSchedulesDao(database).upsert_by_child(
+            ada.id,
+            cycle_length_weeks=2,
+            anchor_date=yesterday_iso,
+            pattern="|0,1,2,3,4,5,6",
+        )
+        returns_date = today + datetime.timedelta(days=2)
+        await _dao_presence.PresenceOverridesDao(database).create(
+            ada.id,
+            returns_date.isoformat(),
+            returns_date.isoformat(),
+            is_present=True,
+            note="visit",
+        )
+
+        # Complete Bo's today instance at the pinned 12:00 (17:00 due
+        # => on time).
+        await _completion.complete_instance(
+            database,
+            brush_teeth_instance.id,
+            actor_source="panel",
+            actor_child_id=bo.id,
+            now=pinned_now,
+            today=today,
+        )
+        return SimpleNamespace(
+            ada=ada,
+            bo=bo,
+            pack_bag_instance=pack_bag_instance,
+            brush_teeth_instance=brush_teeth_instance,
+            stale_instance=stale_instance,
+            returns_date=returns_date,
+        )
+    finally:
+        await database.close()
+
+
+class _PanelRunner:
+    """Run the app's lifespan around an httpx AsyncClient.
+
+    Same pattern as test_api_health.py's ``_AppRunner``: httpx's ASGI
+    transport does not run the lifespan, so it is driven explicitly and
+    shut down in ``__aexit__``'s finally block even on failure.
+    """
+
+    def __init__(
+        self, db_path: str, panel_token: str = PANEL_TOKEN
+    ) -> None:
+        self.app = create_app(
+            ApiConfig(db_path=db_path, panel_token=panel_token)
+        )
+
+    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",
+        )
+        return await self._client.__aenter__()
+
+    async def __aexit__(self, *exc) -> None:
+        try:
+            await self._client.__aexit__(*exc)
+        finally:
+            await self._lifespan.__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 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:
+        yield client
+
+
+@pytest.fixture
+async def seeded_panel(temp_db_path: str, monkeypatch) -> SimpleNamespace:
+    """A seeded household served by the app, with the clock pinned.
+
+    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.
+    """
+    today = datetime.datetime.now().astimezone().date()
+    time_zone = ZoneInfo("UTC")
+    pinned_now = datetime.datetime(
+        today.year, today.month, today.day, 12, 0, 0, tzinfo=time_zone
+    )
+    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:
+        yield SimpleNamespace(
+            client=client,
+            seed=seed,
+            today=today,
+            pinned_now=pinned_now,
+            pinned_utc_iso=pinned_now.astimezone(
+                datetime.timezone.utc
+            ).isoformat(),
+        )
+
+
+# --- ApiConfig.from_env: the panel token --------------------------------
+
+
+def test_from_env_reads_panel_token() -> None:
+    """A set, non-empty NESTQUEST_PANEL_TOKEN is returned verbatim."""
+    cfg = ApiConfig.from_env(
+        {
+            "NESTQUEST_DB_PATH": "/data/nq.db",
+            "NESTQUEST_PANEL_TOKEN": "panel-token",
+        }
+    )
+    assert cfg.panel_token == "panel-token"
+
+
+def test_from_env_requires_panel_token_when_missing() -> None:
+    """A missing panel token raises ConfigError (no silent default)."""
+    with pytest.raises(ConfigError):
+        ApiConfig.from_env({"NESTQUEST_DB_PATH": "/data/nq.db"})
+
+
+def test_from_env_requires_panel_token_when_empty() -> None:
+    """An empty/whitespace panel token raises ConfigError."""
+    with pytest.raises(ConfigError):
+        ApiConfig.from_env(
+            {
+                "NESTQUEST_DB_PATH": "/data/nq.db",
+                "NESTQUEST_PANEL_TOKEN": "   ",
+            }
+        )
+
+
+# --- service-token auth --------------------------------------------------
+
+
+async def test_missing_token_is_unauthorized(
+    panel_client: httpx.AsyncClient,
+) -> None:
+    """A request with no Authorization header returns 401."""
+    response = await panel_client.get("/api/v1/panel/snapshot")
+    assert response.status_code == 401
+    assert "www-authenticate" in {
+        key.lower() for key in response.headers
+    }
+
+
+async def test_wrong_token_is_unauthorized(
+    panel_client: httpx.AsyncClient,
+) -> None:
+    """A wrong Bearer token returns 401."""
+    response = await panel_client.get(
+        "/api/v1/panel/snapshot",
+        headers={"Authorization": "Bearer not-the-panel-token"},
+    )
+    assert response.status_code == 401
+
+
+async def test_malformed_authorization_header_is_unauthorized(
+    panel_client: httpx.AsyncClient,
+) -> None:
+    """A malformed Authorization header returns 401.
+
+    Wrong scheme, empty credential, and an empty header all fail the
+    same way — the response never reveals which part failed.
+    """
+    for header in (
+        "Basic dXNlcjpwYXNz",
+        "Bearer",
+        "Bearer   ",
+        "Token not-the-panel-token",
+        "",
+    ):
+        response = await panel_client.get(
+            "/api/v1/panel/snapshot",
+            headers={"Authorization": header},
+        )
+        assert response.status_code == 401, f"header {header!r} passed"
+
+
+# --- snapshot payload ----------------------------------------------------
+
+
+async def test_valid_token_returns_full_snapshot(
+    seeded_panel: SimpleNamespace,
+) -> None:
+    """The correct token returns 200 with the documented snapshot shape.
+
+    Children in sort order (Ada sort_order 0, Bo 1); per-child counts,
+    presence, next_present, and cycle day all asserted as explicit
+    literals; each child's instances are in the documented panel
+    payload shape (done -> completed spelling, on_time passthrough).
+    """
+    seed = seeded_panel.seed
+    response = await seeded_panel.client.get(
+        "/api/v1/panel/snapshot",
+        headers={"Authorization": f"Bearer {PANEL_TOKEN}"},
+    )
+    assert response.status_code == 200
+    body = response.json()
+
+    assert body["today_iso"] == seeded_panel.today.isoformat()
+    # Ada's 2-week cycle anchored yesterday => today is cycle day 2.
+    assert body["cycle_day"] == 2
+
+    children = body["children"]
+    assert [child["child_id"] for child in children] == [
+        seed.ada.id,
+        seed.bo.id,
+    ], "children appear in sort_order"
+
+    ada, bo = children
+
+    # Ada: away today, returns on the override date; her 10:00 quest
+    # is open and overdue at the pinned 12:00.
+    assert ada["child_name"] == "Ada"
+    assert ada["present"] is False
+    assert ada["next_present"] == seed.returns_date.isoformat()
+    assert ada["due_today"] == 1
+    assert ada["completed_today"] == 0
+    assert ada["remaining_today"] == 1
+    assert ada["completion_pct"] == 0
+    assert ada["instances"] == [
+        {
+            "id": seed.pack_bag_instance.id,
+            "definition_id": seed.pack_bag_instance.definition_id,
+            "child_id": seed.ada.id,
+            "title": "Pack bag",
+            "icon": None,
+            "window": "morning",
+            "due_time": "10:00",
+            "state": "open",
+            "overdue": True,
+            "completed_at": None,
+            "on_time": None,
+        }
+    ]
+
+    # Bo: present, completed his 17:00 quest at the pinned 12:00
+    # (on time), so the panel spelling is "completed".
+    assert bo["child_name"] == "Bo"
+    assert bo["present"] is True
+    assert bo["next_present"] is None
+    assert bo["due_today"] == 1
+    assert bo["completed_today"] == 1
+    assert bo["remaining_today"] == 0
+    assert bo["completion_pct"] == 100
+    assert bo["instances"] == [
+        {
+            "id": seed.brush_teeth_instance.id,
+            "definition_id": seed.brush_teeth_instance.definition_id,
+            "child_id": seed.bo.id,
+            "title": "Brush teeth",
+            "icon": None,
+            "window": "morning",
+            "due_time": "17:00",
+            "state": "completed",
+            "overdue": False,
+            "completed_at": seeded_panel.pinned_utc_iso,
+            "on_time": True,
+        }
+    ]
+
+
+async def test_missed_instances_are_omitted_from_panel_payload(
+    seeded_panel: SimpleNamespace,
+) -> None:
+    """A missed instance never reaches the panel payload (D-009).
+
+    The household carries a past-due OPEN instance due yesterday (the
+    state the panel reads as "missed"); it must not appear anywhere in
+    the response, and Ada's payload carries exactly her ONE today
+    instance — the past-due rows (the stale chore AND her pack bag due
+    yesterday) are both omitted rather than rendered.
+    """
+    seed = seeded_panel.seed
+    response = await seeded_panel.client.get(
+        "/api/v1/panel/snapshot",
+        headers={"Authorization": f"Bearer {PANEL_TOKEN}"},
+    )
+    assert response.status_code == 200
+    body = response.json()
+
+    payload_instances = [
+        instance
+        for child in body["children"]
+        for instance in child["instances"]
+    ]
+    assert "missed" not in [
+        instance["state"] for instance in payload_instances
+    ]
+    past_due_ids = {seed.stale_instance.id}
+    assert past_due_ids.isdisjoint(
+        instance["id"] for instance in payload_instances
+    ), "the past-due open instance leaked into the panel payload"
+
+    ada = next(
+        child
+        for child in body["children"]
+        if child["child_id"] == seed.ada.id
+    )
+    assert [instance["id"] for instance in ada["instances"]] == [
+        seed.pack_bag_instance.id
+    ]
+
+
+def test_panel_shaper_drops_missed_views() -> None:
+    """The route's shaper omits a missed view (D-009), on the app's copy.
+
+    ``instance_payload(include_missed=False)`` is what the snapshot
+    route shapes every child's instances with; exercised here on the
+    SAME core copy the app uses (the two-copy caveat makes this a
+    distinct object from the conftest-stubbed copy's identical source).
+    """
+    missed_view = core_snapshot.QuestInstanceView(
+        instance_id=999,
+        definition_id=1,
+        child_id=1,
+        title="Stale chore",
+        icon=None,
+        window="morning",
+        due_date="2026-09-21",
+        due_time="09:00",
+        state="missed",
+        overdue=False,
+        completed_at=None,
+        was_on_time=None,
+    )
+    assert core_snapshot.instance_payload(
+        [missed_view], include_missed=False
+    ) == []
+
+
+# --- route documentation -------------------------------------------------
+
+
+async def test_snapshot_route_is_documented(
+    panel_client: httpx.AsyncClient,
+) -> None:
+    """The snapshot route and its response shape are in the OpenAPI doc."""
+    response = await panel_client.get("/openapi.json")
+    assert response.status_code == 200
+    schema = response.json()
+    assert "/api/v1/panel/snapshot" in schema["paths"]
+    assert "get" in schema["paths"]["/api/v1/panel/snapshot"]
+    documented = {
+        model["title"]
+        for model in schema["components"]["schemas"].values()
+    }
+    assert "PanelSnapshotResponse" in documented
+    assert "PanelChildPayload" in documented
+    assert "PanelInstancePayload" in documented

tests/test_core_snapshot.py

@@ -0,0 +1,306 @@
+"""Test the pure panel snapshot builder (task 9047a97b).
+
+The snapshot assembly that lived inline in the coordinator's
+``_async_update_data`` was extracted into the Home-Assistant-free
+:func:`custom_components.nestquest.core.snapshot.build_snapshot`, and
+the panel payload shaper (:func:`instance_payload`) with it.  These
+tests seed a household and assert the pure builder's output against
+EXPLICIT literal expected values — not self-equality — so the AWAY
+child's ``next_present``, a due-timed quest's ``overdue`` under a
+pinned ``now``, the custody ``cycle_day`` (> 1), and the panel payload
+dict shape (``done`` -> ``completed``, ``on_time`` passthrough, and
+``missed`` omission) are all exercised as literals.
+"""
+from __future__ import annotations
+
+import datetime
+from zoneinfo import ZoneInfo
+
+from conftest import wire_entry_to_registry
+
+from custom_components.nestquest import async_setup_entry
+from custom_components.nestquest.children import create_child
+from custom_components.nestquest.completion import complete_instance
+from custom_components.nestquest.core.snapshot import (
+    ChildDaySnapshot,
+    NestQuestSnapshot,
+    QuestInstanceView,
+    build_snapshot,
+    instance_payload,
+)
+from custom_components.nestquest.dao_instances import QuestInstancesDao
+from custom_components.nestquest.dao_presence import (
+    PresenceOverridesDao,
+    PresenceSchedulesDao,
+)
+from custom_components.nestquest.materialize import materialize
+from custom_components.nestquest.quest_definitions import create_quest_definition
+from custom_components.nestquest.recurrence import ScheduleRule
+
+
+async def test_instance_payload_shape_and_missed_omission() -> None:
+    """``instance_payload`` is the documented panel shape (§1).
+
+    ``done`` maps to ``completed``; ``open`` passes through; ``on_time``
+    is the builder's ``was_on_time`` verbatim; ``missed`` instances are
+    omitted unless ``include_missed`` (the admin payload, D-009).
+    """
+    def view(state: str, *, overdue: bool = False, on_time: bool | None = None,
+             completed_at: str | None = None, instance_id: int = 1) -> QuestInstanceView:
+        return QuestInstanceView(
+            instance_id=instance_id,
+            definition_id=10,
+            child_id=3,
+            title="Fill bird feeder",
+            icon="bird",
+            window="afternoon",
+            due_date="2026-09-21",
+            due_time="17:00",
+            state=state,
+            overdue=overdue,
+            completed_at=completed_at,
+            was_on_time=on_time,
+        )
+
+    open_view = view("open", overdue=True)
+    done_view = view("done", on_time=True, completed_at="2026-09-21T12:00:00+00:00")
+    missed_view = view("missed", instance_id=2)
+
+    panel = instance_payload([open_view, done_view, missed_view], include_missed=False)
+    assert panel == [
+        {
+            "id": 1,
+            "definition_id": 10,
+            "child_id": 3,
+            "title": "Fill bird feeder",
+            "icon": "bird",
+            "window": "afternoon",
+            "due_time": "17:00",
+            "state": "open",
+            "overdue": True,
+            "completed_at": None,
+            "on_time": None,
+        },
+        {
+            "id": 1,
+            "definition_id": 10,
+            "child_id": 3,
+            "title": "Fill bird feeder",
+            "icon": "bird",
+            "window": "afternoon",
+            "due_time": "17:00",
+            "state": "completed",
+            "overdue": False,
+            "completed_at": "2026-09-21T12:00:00+00:00",
+            "on_time": True,
+        },
+    ], "panel payload omits missed and spells done as completed"
+
+    admin = instance_payload(
+        [open_view, done_view, missed_view], include_missed=True
+    )
+    assert len(admin) == 3
+    assert admin[2]["state"] == "missed", (
+        "admin payload keeps missed instances with their missed state"
+    )
+
+
+async def test_build_snapshot_explicit_values(hass, make_entry, monkeypatch) -> None:
+    """The builder produces the documented shape as literals, pinned in time.
+
+    Ada is AWAY today (a 2-week schedule whose absent week 0 covers today,
+    plus a presence override marking her present on today+2) so
+    ``present`` is False and ``next_present`` is the override date.  Her
+    open quest due at 10:00 is OVERDUE under the pinned 12:00 ``now``.
+    Bo's quest due at 17:00 is completed at 12:00 (on time) so its
+    payload ``state`` is ``completed`` and ``on_time`` is True.  The first
+    scheduled active child is Ada, whose 2-week cycle anchored one day
+    before today puts ``cycle_day`` at 2.
+    """
+    entry = wire_entry_to_registry(make_entry(), hass.registry)
+    assert await async_setup_entry(hass, entry) is True
+    coordinator = entry.runtime_data.coordinator
+    database = entry.runtime_data.database
+    settings = entry.runtime_data.settings
+
+    # Single clock read for the seed/snapshot anchor (no midnight skew).
+    today = coordinator._local_now().date()
+    time_zone = ZoneInfo(hass.config.time_zone)
+    # Pin the build instant at 12:00 local: past Ada's 10:00 due time
+    # (overdue) and before Bo's 17:00, and the completion moment for Bo.
+    pinned_now = datetime.datetime(
+        today.year, today.month, today.day, 12, 0, 0, tzinfo=time_zone
+    )
+    today_iso = today.isoformat()
+
+    ada = await create_child(database, "Ada")
+    bo = await create_child(database, "Bo")
+
+    await create_quest_definition(
+        database,
+        "Pack bag",
+        ScheduleRule.from_dict(
+            {"rule_type": "daily", "start_date": today_iso}
+        ),
+        [ada.id],
+        [("morning", "10:00")],
+    )
+    await create_quest_definition(
+        database,
+        "Brush teeth",
+        ScheduleRule.from_dict(
+            {"rule_type": "daily", "start_date": today_iso}
+        ),
+        [bo.id],
+        [("morning", "17:00")],
+    )
+    end = (today + datetime.timedelta(days=3)).isoformat()
+    await materialize(database, today_iso, end, today=today)
+
+    # Ada: 2-week cycle anchored yesterday. Week 0 (the empty segment)
+    # covers today, so she is absent today; an override marks her present
+    # on today+2, so next_present resolves to that date and cycle_day is
+    # ((today - (today-1)) % 14) + 1 == 2.
+    await PresenceSchedulesDao(database).upsert_by_child(
+        ada.id,
+        cycle_length_weeks=2,
+        anchor_date=(today - datetime.timedelta(days=1)).isoformat(),
+        pattern="|0,1,2,3,4,5,6",
+    )
+    returns_date = today + datetime.timedelta(days=2)
+    await PresenceOverridesDao(database).create(
+        ada.id,
+        returns_date.isoformat(),
+        returns_date.isoformat(),
+        is_present=True,
+        note="visit",
+    )
+
+    # Complete Bo's instance at the pinned 12:00 (17:00 due => on time).
+    bo_instances = await QuestInstancesDao(database).list_by_child_and_date(
+        bo.id, today_iso
+    )
+    await complete_instance(
+        database,
+        bo_instances[0].id,
+        actor_source="panel",
+        actor_child_id=bo.id,
+        now=pinned_now,
+        today=today,
+    )
+
+    built = await build_snapshot(database, settings, pinned_now)
+
+    assert isinstance(built, NestQuestSnapshot)
+    assert built.today_iso == today_iso
+    assert built.cycle_day == 2, (
+        "Ada's 2-week cycle anchored yesterday => day 2 today"
+    )
+
+    by_id = {child.child_id: child for child in built.children}
+    assert list(by_id) == [ada.id, bo.id], (
+        "active children appear in sort_order, id order"
+    )
+    ada_child = by_id[ada.id]
+    bo_child = by_id[bo.id]
+
+    # Ada: away today, returns on the override date, one open+overdue quest.
+    assert isinstance(ada_child, ChildDaySnapshot)
+    assert ada_child.present is False
+    assert ada_child.next_present == returns_date.isoformat()
+    assert ada_child.due_today == 1
+    assert ada_child.completed_today == 0
+    assert ada_child.remaining_today == 1
+    assert ada_child.completion_pct == 0
+    ada_view = ada_child.instances[0]
+    assert isinstance(ada_view, QuestInstanceView)
+    assert ada_view.state == "open"
+    assert ada_view.due_time == "10:00"
+    assert ada_view.overdue is True, "12:00 now is past the 10:00 due time"
+    assert ada_view.completed_at is None
+    assert ada_view.was_on_time is None
+
+    # Bo: present, one completed instance, on time, not overdue.
+    assert bo_child.present is True
+    assert bo_child.next_present is None
+    assert bo_child.due_today == 1
+    assert bo_child.completed_today == 1
+    assert bo_child.remaining_today == 0
+    assert bo_child.completion_pct == 100
+    bo_view = bo_child.instances[0]
+    assert bo_view.state == "done"
+    assert bo_view.due_time == "17:00"
+    assert bo_view.overdue is False, "overdue is only computed for open state"
+    assert bo_view.completed_at == pinned_now.astimezone(
+        datetime.timezone.utc
+    ).isoformat()
+    assert bo_view.was_on_time is True
+
+    # Panel payload shape: Ada open+overdue stays open; Bo done -> completed.
+    ada_payload = instance_payload(ada_child.instances, include_missed=False)
+    assert ada_payload == [
+        {
+            "id": ada_view.instance_id,
+            "definition_id": ada_view.definition_id,
+            "child_id": ada.id,
+            "title": "Pack bag",
+            "icon": None,
+            "window": "morning",
+            "due_time": "10:00",
+            "state": "open",
+            "overdue": True,
+            "completed_at": None,
+            "on_time": None,
+        }
+    ]
+    bo_payload = instance_payload(bo_child.instances, include_missed=False)
+    assert bo_payload == [
+        {
+            "id": bo_view.instance_id,
+            "definition_id": bo_view.definition_id,
+            "child_id": bo.id,
+            "title": "Brush teeth",
+            "icon": None,
+            "window": "morning",
+            "due_time": "17:00",
+            "state": "completed",
+            "overdue": False,
+            "completed_at": pinned_now.astimezone(
+                datetime.timezone.utc
+            ).isoformat(),
+            "on_time": True,
+        }
+    ]
+
+    # Missed omission is real: a missed view is dropped from the panel
+    # payload and kept in the admin payload with its missed state.
+    missed_view = QuestInstanceView(
+        instance_id=999,
+        definition_id=10,
+        child_id=ada.id,
+        title="Stale quest",
+        icon=None,
+        window="morning",
+        due_date=(today - datetime.timedelta(days=1)).isoformat(),
+        due_time="09:00",
+        state="missed",
+        overdue=False,
+        completed_at=None,
+        was_on_time=None,
+    )
+    assert instance_payload([missed_view], include_missed=False) == []
+    assert (
+        instance_payload([missed_view], include_missed=True)[0]["state"]
+        == "missed"
+    )
+
+    # REQUIRED wrapper tie-in: the coordinator's _async_update_data is a
+    # thin WRAPPER over build_snapshot, so pinning its clock to the SAME
+    # pinned_now the builder was called with must reproduce the builder's
+    # snapshot byte-for-byte.  The dataclasses are frozen, so == is a full
+    # deep compare — a wrapper that mis-passed today vs now, or stopped
+    # calling the builder, would diverge here even though the literal
+    # assertions above still pass.
+    monkeypatch.setattr(coordinator, "_local_now", lambda: pinned_now)
+    await coordinator.async_refresh()
+    assert coordinator.data == built

tests/test_entity_matrix.py

@@ -186,18 +186,18 @@ async def test_entity_and_event_matrix(hass, make_entry) -> None:
     original_refresh = coordinator._async_update_data
 
     async def _future_dated_refresh():
-        import custom_components.nestquest.coordinator as coordinator_module
+        import custom_components.nestquest.core.snapshot as snapshot_module
 
-        real_derive = coordinator_module.derive_state
+        real_derive = snapshot_module.derive_state
 
         def _derive(instance, latest, today):
             return real_derive(instance, latest, today + datetime.timedelta(days=1))
 
-        coordinator_module.derive_state = _derive
+        snapshot_module.derive_state = _derive
         try:
             return await original_refresh()
         finally:
-            coordinator_module.derive_state = real_derive
+            snapshot_module.derive_state = real_derive
 
     coordinator._async_update_data = _future_dated_refresh
     await coordinator.async_refresh()

tests/test_sensors.py

@@ -238,16 +238,16 @@ async def test_due_today_payload_omits_missed_for_panel_includes_for_admin(
     entry, coordinator, (ada, bo, cory) = await _setup_seeded_entry(
         hass, make_entry
     )
-    import custom_components.nestquest.coordinator as coordinator_module
+    import custom_components.nestquest.core.snapshot as snapshot_module
 
     tomorrow = _today() + datetime.timedelta(days=1)
-    real_derive_state = coordinator_module.derive_state
+    real_derive_state = snapshot_module.derive_state
 
     def _derive_against_pinned_future(instance, latest, today):
         return real_derive_state(instance, latest, tomorrow)
 
     monkeypatch.setattr(
-        coordinator_module, "derive_state", _derive_against_pinned_future
+        snapshot_module, "derive_state", _derive_against_pinned_future
     )
     await coordinator.async_refresh()