Re-fire the API's four panel SSE events on the HA bus

WHY: transitions that happen through the API service's panel route
(e.g. a tap on the PWA) published on the API's SSE stream but never
reached Home Assistant's event bus, so the four blueprints only fired
for completions made through the integration's own services. The
integration now subscribes: NestQuestApiClient.stream_events parses
the GET /api/v1/panel/events frames (typed NestQuestApiError on
non-2xx/transport failure, clean stop on cancellation), a new
NestQuestEventStream consumes it in a background task — firing each
documented event type with its documented payload on hass.bus,
reconnecting forever with a 5s-to-60s bounded backoff, never letting
an exception escape — and setup starts it / unload stops it, so no
task leaks.

custom_components/nestquest/__init__.py

@@ -26,11 +26,12 @@ from .const import (
 )
 from .core.settings import NestQuestSettings
 from .frontend import async_register_frontend
-from .coordinator import NestQuestCoordinator
+from .coordinator import NestQuestCoordinator, coordinator_client_from_entry
 from .db import NestQuestDatabase, make_database
 from .materialize import materialize as _materialize_run
 from .migrations import apply_migrations
 from .services import async_deregister_services, async_register_services
+from .sse import NestQuestEventStream
 from .store import async_get_db_path
 from .admin_allowlist import seed_setup_admin
 from .sweep import run_missed_sweep
@@ -242,6 +243,7 @@ class NestQuestRuntimeData:
     remove_update_listener: Callable[[], Any]
     remove_time_change_listener: Callable[[], Any]
     coordinator: Any = None
+    event_stream: NestQuestEventStream | None = None
 
 
 async def _async_owner_user_ids(hass: HomeAssistant) -> list[str]:
@@ -439,6 +441,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
     remove_update_listener: Callable[[], Any] | None = None
     remove_time_change_listener: Callable[[], Any] | None = None
     coordinator: Any = None
+    event_stream: NestQuestEventStream | None = None
     try:
         # Seed the admin allowlist on first setup so the owner is never
         # locked out: an empty database takes the persisted admin copy
@@ -490,18 +493,43 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
         _register_services(hass)
         # The shared Feature 10 coordinator: one refresh cycle every
         # entity reads from (CONF_UPDATE_INTERVAL seconds, default
-        # five minutes).  Created AFTER services so a failure below
+        # five minutes).  Since Feature 18 it polls the API service's
-        # unwinds them through the except path's listener handling.
+        # panel snapshot instead of reading the local database; an
+        # entry with no panel token yet gets the typed-error stub
+        # (unavailable entities, never a setup crash).  Created AFTER
+        # services so a failure below unwinds them through the except
+        # path's listener handling.
+        # ONE API client for the entry, shared by the coordinator's
+        # snapshot polling and the SSE subscription below (the client
+        # is stateless — one session, one connection pool).
+        api_client = coordinator_client_from_entry(hass, entry)
+        # The shared Feature 10 coordinator: one refresh cycle every
+        # entity reads from (CONF_UPDATE_INTERVAL seconds, default
+        # five minutes).  Since Feature 18 it polls the API service's
+        # panel snapshot instead of reading the local database; an
+        # entry with no panel token yet gets the typed-error stub
+        # (unavailable entities, never a setup crash).  Created AFTER
+        # services so a failure below unwinds them through the except
+        # path's listener handling.
         coordinator = NestQuestCoordinator(
             hass,
             entry_id=entry.entry_id,
-            database=database,
+            api_client=api_client,
-            settings=settings,
             update_interval_seconds=entry.options.get(
                 CONF_UPDATE_INTERVAL, DEFAULT_UPDATE_INTERVAL
             ),
         )
-        await coordinator.async_config_entry_first_refresh()
+        # The panel event subscription (the SSE task): a background
+        # task consumes the API service's transition-event stream and
+        # re-fires the four documented HA bus events.  It reuses the
+        # coordinator's client, so it only starts when that client
+        # actually offers a stream — an unconfigured-token stub (or a
+        # test snapshot client without ``stream_events``) leaves it
+        # unstarted, and the typed-error path already covers
+        # availability for those entries.
+        if callable(getattr(api_client, "stream_events", None)):
+            event_stream = NestQuestEventStream(hass, api_client)
+            event_stream.start()
         runtime_data = NestQuestRuntimeData(
             entry_id=entry.entry_id,
             settings=settings,
@@ -509,12 +537,32 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
             remove_update_listener=remove_update_listener,
             remove_time_change_listener=remove_time_change_listener,
             coordinator=coordinator,
+            event_stream=event_stream,
         )
-        # runtime_data is assigned BEFORE the platforms forward so each
+        # The runtime record is registered BEFORE the first refresh:
-        # platform's async_setup_entry reads its coordinator off the
+        # the refresh path (and anything it calls) resolves the entry's
-        # entry (the HA runtime-data contract); a forward failure below
+        # runtime state through hass.data, and it must find the live
+        # database there — never a half-set-up entry.  The platforms
+        # forward below reads the coordinator off the entry
+        # (the HA runtime-data contract); a forward failure below
         # unwinds it through the except path.
+        hass.data[DOMAIN][entry.entry_id] = runtime_data
         entry.runtime_data = runtime_data
+        # The first refresh polls the API snapshot; an API failure
+        # (unconfigured token, API service down) does NOT crash setup:
+        # the coordinator reports last_update_success=False, entities
+        # come up unavailable, and HA's standard retry cycle applies
+        # once the API is reachable.  HA's real
+        # ``async_config_entry_first_refresh`` wraps ANY failed update
+        # pass in ConfigEntryNotReady — the original error survives
+        # only as ``__cause__``, never as the raised exception — so the
+        # catch keys on ConfigEntryNotReady, NOT on the typed API
+        # error.  Everything else (a platform-forward failure,
+        # cancellation) still unwinds through the except block below.
+        try:
+            await coordinator.async_config_entry_first_refresh()
+        except ConfigEntryNotReady:
+            pass
         await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
     except BaseException:
         # A failure anywhere after the database is open must not leak the
@@ -522,6 +570,14 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
         # unwind EVERY remover we acquired before the failure, keeping the
         # original error as the one raised.  Removal is best-effort — a
         # failing remover must not mask the setup error that triggered it.
+        # The SSE subscription (started just before the runtime record)
+        # is stopped first: like the coordinator it must never outlive
+        # the database being closed below.
+        if event_stream is not None:
+            try:
+                await event_stream.stop()
+            except BaseException:
+                pass
         for remove_listener in (
             remove_time_change_listener,
             remove_update_listener,
@@ -548,10 +604,19 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
         # closes the already-closed database through the stale record.
         if getattr(entry, "runtime_data", None) is not None:
             entry.runtime_data = None
+        # The runtime record was ALSO registered in hass.data before
+        # the first refresh: remove it here on ANY failure, or a later
+        # unload picks up the stale, already-torn-down record
+        # (hass.data wins over entry.runtime_data there) and
+        # re-invokes the already-consumed update-listener remover.
+        hass.data.get(DOMAIN, {}).pop(entry.entry_id, None)
         await database.close()
         raise
     assert remove_update_listener is not None
     assert remove_time_change_listener is not None
+    # The runtime record was already registered before the first
+    # refresh; re-assert the (unchanged) record so the return path
+    # stays readable as one registration site.
     hass.data[DOMAIN][entry.entry_id] = runtime_data
     return True
 
@@ -581,6 +646,13 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
         return False
     unload_error: BaseException | None = None
     if runtime_data is not None:
+        # The SSE subscription is stopped FIRST: its task must be gone
+        # (no frame can fire) before anything else is torn down.
+        event_stream = getattr(runtime_data, "event_stream", None)
+        if event_stream is not None:
+            stop_stream = getattr(event_stream, "stop", None)
+            if stop_stream is not None:
+                await stop_stream()
         remove_update_listener = getattr(runtime_data, "remove_update_listener", None)
         remove_time_change_listener = getattr(
             runtime_data, "remove_time_change_listener", None

custom_components/nestquest/api_client.py

@@ -0,0 +1,414 @@
+"""Panel-plane API client for the NestQuest integration (Feature 18).
+
+The integration talks to the NestQuest API service (Feature 16) over
+HTTP instead of reading the database: this module is the ONE place that
+speaks the API's panel plane — ``GET /api/v1/panel/snapshot`` (today's
+household snapshot) and ``POST /api/v1/panel/instances/{id}/complete``
+(completion as the tapped child profile) — authenticated with the
+static panel service token as a Bearer credential on every request.
+
+Transport-injected by construction: the constructor takes the base URL,
+the panel token, and the HTTP session to use.  The session speaks the
+aiohttp ``ClientSession`` surface (``request(method, url, headers=...,
+json=...)`` returning an async context manager whose response carries
+``status`` and ``await text()``), so tests stub the transport with NO
+network and NO aiohttp installed; production builds the client through
+:func:`client_from_entry`, which hands in Home Assistant's shared
+aiohttp session (``homeassistant.helpers.aiohttp_client
+.async_get_clientsession``).  The module itself imports NEITHER
+homeassistant NOR aiohttp at runtime (aiohttp appears only behind
+``TYPE_CHECKING`` for annotations), so the client is usable anywhere —
+the HA coupling lives in the one production factory.
+
+Every failure is typed: a transport failure (connection refused, DNS
+failure, timeout — the exchange never produced a response) and every
+non-2xx HTTP response raise :class:`NestQuestApiError`.  The error
+carries the HTTP status where there is one (so 401 and 404 are
+distinguishable at the call site) and ``None`` when there is not; the
+original transport error is chained as ``__cause__`` — nothing is
+swallowed.  There are NO retries and the client logs nothing: refresh
+cadence is the coordinator's concern (the next task), and the panel
+token is a secret that must never reach a log line or an error
+message — the error texts name the method, path, and status only.
+
+The panel plane's SSE event stream (``GET /api/v1/panel/events``) is
+exposed by :meth:`NestQuestApiClient.stream_events` — an async
+generator yielding ``(event_type, payload)`` per Server-Sent Events
+frame (``event: <type>`` / ``data: <json>`` lines separated by a blank
+line).  Opening the stream gets the same one-timeout, typed-error
+contract as every other call; reading it has NO body-wide timeout (a
+healthy stream is long-lived — the subscription manager owns
+reconnection and backoff).
+"""
+from __future__ import annotations
+
+import asyncio
+import inspect
+import json
+from collections.abc import AsyncIterator
+from typing import TYPE_CHECKING, Any
+from urllib.parse import urlparse
+
+from .const import (
+    CONF_API_BASE_URL,
+    CONF_PANEL_TOKEN,
+    DEFAULT_API_BASE_URL,
+    DEFAULT_PANEL_TOKEN,
+)
+
+if TYPE_CHECKING:
+    import aiohttp
+
+__all__ = [
+    "COMPLETE_INSTANCE_PATH",
+    "DEFAULT_TIMEOUT_SECONDS",
+    "EVENTS_PATH",
+    "NestQuestApiError",
+    "NestQuestApiClient",
+    "SNAPSHOT_PATH",
+    "client_from_entry",
+    "is_valid_base_url",
+    "resolve_api_config",
+]
+
+#: The panel plane's snapshot route (Feature 16).
+SNAPSHOT_PATH = "/api/v1/panel/snapshot"
+
+#: The panel plane's SSE event-stream route (Feature 16): each
+#: transition arrives as an ``event: <type>`` / ``data: <json>`` frame.
+EVENTS_PATH = "/api/v1/panel/events"
+
+#: The panel plane's completion route template (Feature 16); the
+#: ``{instance_id}`` placeholder is filled per call.
+COMPLETE_INSTANCE_PATH = "/api/v1/panel/instances/{instance_id}/complete"
+
+#: Request timeout, in seconds.  One attempt, one timeout: a hung API
+#: call surfaces as a typed error for the coordinator to report, never
+#: a retry loop.
+DEFAULT_TIMEOUT_SECONDS = 10.0
+
+
+class NestQuestApiError(Exception):
+    """A failed NestQuest API exchange, raised for every failure mode.
+
+    ``status`` carries the HTTP status when a response arrived (a
+    non-2xx response — 401 for a bad token, 404 for an unknown
+    instance) and is ``None`` when the exchange failed at the transport
+    level (connection refused, DNS failure, timeout) — so callers
+    distinguish the two with ``is_transport_error``.  The original
+    transport error, where there is one, is chained as ``__cause__``.
+    Messages carry the method, path, and status — never the request
+    headers, so the panel token can never leak through an error.
+    """
+
+    def __init__(self, message: str, *, status: int | None = None) -> None:
+        super().__init__(message)
+        self.status = status
+
+    @property
+    def is_transport_error(self) -> bool:
+        """True when the exchange failed before any response arrived."""
+        return self.status is None
+
+
+def is_valid_base_url(value: object) -> bool:
+    """Return True when ``value`` is a usable API base URL.
+
+    A full ``http://`` or ``https://`` URL with a host — the exact
+    shape the options flow validates (form ergonomics) and the client
+    constructor enforces (fail fast, never build requests against a
+    scheme-less or host-less string).
+    """
+    if not isinstance(value, str):
+        return False
+    candidate = value.strip()
+    if not candidate:
+        return False
+    parsed = urlparse(candidate)
+    return parsed.scheme in ("http", "https") and bool(parsed.netloc)
+
+
+def resolve_api_config(entry: Any) -> tuple[str, str]:
+    """Resolve ``(base_url, panel_token)`` off a config entry.
+
+    Precedence mirrors the options flow's stored-values resolution:
+    ``entry.options`` first (the live options-flow copy), then
+    ``entry.data`` (the config-flow-era copy), then the const defaults.
+    The token is a secret: callers must never log the returned pair.
+    """
+    options = getattr(entry, "options", None) or {}
+    data = getattr(entry, "data", None) or {}
+
+    def _pick(key: str, default: str) -> str:
+        if key in options:
+            return options[key]
+        if key in data:
+            return data[key]
+        return default
+
+    return (
+        _pick(CONF_API_BASE_URL, DEFAULT_API_BASE_URL),
+        _pick(CONF_PANEL_TOKEN, DEFAULT_PANEL_TOKEN),
+    )
+
+
+def client_from_entry(hass: Any, entry: Any) -> NestQuestApiClient:
+    """Build the production client for a config entry.
+
+    Reads the configured base URL and panel token off the entry
+    (:func:`resolve_api_config`) and hands the client Home Assistant's
+    SHARED aiohttp session (``homeassistant.helpers.aiohttp_client
+    .async_get_clientsession``) so panel calls ride HA's one connection
+    pool.  The helper import is deferred into this function on purpose:
+    the test harness is mock-only and the module must import (and the
+    client must be stub-transport testable) without homeassistant or
+    aiohttp installed.
+    """
+    from homeassistant.helpers.aiohttp_client import async_get_clientsession
+
+    base_url, panel_token = resolve_api_config(entry)
+    return NestQuestApiClient(
+        base_url, panel_token, async_get_clientsession(hass)
+    )
+
+
+class NestQuestApiClient:
+    """Async client for the NestQuest API service's panel plane.
+
+    ONE instance per config entry (the coordinator task wires it into
+    the refresh cycle).  The client is deliberately thin: it adds the
+    Bearer credential and the base URL, enforces one timeout per
+    request, and turns every failure into :class:`NestQuestApiError` —
+    no business logic, no retries, no caching.
+    """
+
+    def __init__(
+        self,
+        base_url: str,
+        panel_token: str,
+        session: aiohttp.ClientSession,
+        *,
+        timeout_seconds: float = DEFAULT_TIMEOUT_SECONDS,
+    ) -> None:
+        """Store the connection settings; the session is injected.
+
+        ``session`` is an aiohttp ``ClientSession`` in production and a
+        stub transport in tests.  ``base_url`` must be a full http(s)
+        URL and ``panel_token`` must be non-empty — a client built
+        without either can only fail later at request time with a
+        less-explicable error, so construction fails fast instead.
+        """
+        if not is_valid_base_url(base_url):
+            raise ValueError(
+                f"base_url must be a full http(s) URL, got {base_url!r}"
+            )
+        if not isinstance(panel_token, str) or not panel_token.strip():
+            raise ValueError("panel_token must be a non-empty string")
+        if (
+            not isinstance(timeout_seconds, (int, float))
+            or isinstance(timeout_seconds, bool)
+            or timeout_seconds <= 0
+        ):
+            raise ValueError(
+                f"timeout_seconds must be a positive number, got {timeout_seconds!r}"
+            )
+        self._base_url = base_url.strip().rstrip("/")
+        self._panel_token = panel_token
+        self._session = session
+        self._timeout_seconds = float(timeout_seconds)
+
+    async def get_snapshot(self) -> dict[str, Any]:
+        """Fetch and parse today's household snapshot.
+
+        Returns the JSON the snapshot route returns (``today_iso``,
+        ``cycle_day``, and the per-child list — see
+        ``api/routes_panel.py`` for the documented shape).  Raises
+        :class:`NestQuestApiError` on transport failure, timeout, or a
+        non-2xx response.
+        """
+        return await self._request_json("GET", SNAPSHOT_PATH)
+
+    async def complete_instance(
+        self, instance_id: int, actor_child_id: int
+    ) -> dict[str, Any]:
+        """Complete one quest instance as the tapped child profile.
+
+        POSTs ``{"actor_child_id": actor_child_id}`` to the complete
+        route and returns the parsed response body (the route answers
+        ``{"status": ...}`` both for a first completion and for an
+        idempotent repeat, so a retried panel tap succeeds).  Raises
+        :class:`NestQuestApiError` on transport failure, timeout, or a
+        non-2xx response — 401 for a wrong token, 404 for an unknown
+        instance.
+        """
+        path = COMPLETE_INSTANCE_PATH.format(instance_id=instance_id)
+        return await self._request_json(
+            "POST", path, json_body={"actor_child_id": actor_child_id}
+        )
+
+    async def stream_events(
+        self,
+    ) -> AsyncIterator[tuple[str, dict[str, Any]]]:
+        """Yield ``(event_type, payload)`` for every panel SSE frame.
+
+        Opens ``GET /api/v1/panel/events`` with the Bearer credential
+        and parses the Server-Sent Events framing the API service
+        emits: ``event: <type>`` and ``data: <json>`` lines closed by a
+        blank line (one frame per transition event; a frame's ``data``
+        may span several ``data:`` lines, joined with newlines — the
+        standard SSE rule).  Line splits across chunk boundaries are
+        buffered, so frames are parsed correctly whatever chunking the
+        transport produces.
+
+        Opening the stream carries the SAME one-timeout, typed-error
+        contract as every other call: a non-2xx response raises
+        :class:`NestQuestApiError` with its status and a transport
+        failure (connection refused, DNS, the open timing out) raises
+        one with ``status=None`` and the original error chained.  The
+        reading loop itself has NO timeout — a healthy stream is
+        long-lived — but any failure raised from the transport
+        mid-stream raises typed as well, so the consumer can reconnect.
+        A frame whose ``data`` is not valid JSON is a protocol failure
+        and raises typed, carrying the response status it arrived
+        with.  Cancellation is never converted: closing the generator
+        (``aclose``, i.e. the consumer's task being cancelled) stops
+        cleanly and closes the underlying response in every path.
+        """
+        url = f"{self._base_url}{EVENTS_PATH}"
+        headers = {"Authorization": f"Bearer {self._panel_token}"}
+        # Opening the response gets ONE request timeout, exactly like
+        # ``_request_json``; the read loop below runs WITHOUT one — a
+        # body-wide timeout would kill a healthy long-lived stream.
+        # The response context manager is entered manually (not via
+        # ``async with``) because the block must span the yields below;
+        # the ``finally`` closes it on EVERY exit path.
+        try:
+            async with asyncio.timeout(self._timeout_seconds):
+                stream_cm = self._session.request(
+                    "GET", url, headers=headers, json=None
+                )
+                response = await stream_cm.__aenter__()
+        except asyncio.CancelledError:
+            # Cancellation is never converted (see ``_request_json``).
+            raise
+        except NestQuestApiError:
+            raise
+        except Exception as err:
+            raise NestQuestApiError(
+                f"NestQuest API GET {EVENTS_PATH} transport failure: {err}",
+                status=None,
+            ) from err
+
+        status = response.status
+        if not 200 <= status < 300:
+            await self._close_stream(stream_cm)
+            raise NestQuestApiError(
+                f"NestQuest API GET {EVENTS_PATH} failed with HTTP {status}",
+                status=status,
+            )
+
+        try:
+            event_type: str | None = None
+            data_lines: list[str] = []
+            buffer = ""
+            async for chunk in response:
+                if isinstance(chunk, (bytes, bytearray)):
+                    chunk = chunk.decode("utf-8")
+                buffer += chunk
+                while "\n" in buffer:
+                    line, buffer = buffer.split("\n", 1)
+                    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:
+                        # A blank line ends the frame.  A trailing
+                        # incomplete frame (stream ends without its
+                        # closing blank line) is discarded, per the
+                        # SSE spec's treatment of incomplete events.
+                        try:
+                            payload = json.loads("\n".join(data_lines))
+                        except ValueError as err:
+                            raise NestQuestApiError(
+                                f"NestQuest API GET {EVENTS_PATH} "
+                                f"delivered a malformed SSE frame: {err}",
+                                status=status,
+                            ) from err
+                        yield event_type, payload
+                        event_type = None
+                        data_lines = []
+        except asyncio.CancelledError:
+            raise
+        except NestQuestApiError:
+            raise
+        except Exception as err:
+            raise NestQuestApiError(
+                f"NestQuest API GET {EVENTS_PATH} transport failure: {err}",
+                status=None,
+            ) from err
+        finally:
+            await self._close_stream(stream_cm)
+
+    @staticmethod
+    async def _close_stream(stream_cm: Any) -> None:
+        """Close an opened streaming response context manager.
+
+        Works for aiohttp's response context manager (whose
+        ``__aexit__`` is a coroutine) and for test stubs that return
+        either a coroutine or a plain value.
+        """
+        result = stream_cm.__aexit__(None, None, None)
+        if inspect.isawaitable(result):
+            await result
+
+    async def _request_json(
+        self,
+        method: str,
+        path: str,
+        *,
+        json_body: dict[str, Any] | None = None,
+    ) -> Any:
+        """Run ONE request and return the parsed JSON body.
+
+        The Bearer token rides the Authorization header only.  A
+        non-2xx response raises :class:`NestQuestApiError` with its
+        status; anything the transport raises (connection, timeout) is
+        wrapped as a transport-level :class:`NestQuestApiError` with
+        the original error chained — never swallowed, never retried.
+        A 2xx body that is not valid JSON is a protocol failure and
+        raises typed too, carrying the status it arrived with.
+        """
+        url = f"{self._base_url}{path}"
+        headers = {"Authorization": f"Bearer {self._panel_token}"}
+        try:
+            async with asyncio.timeout(self._timeout_seconds):
+                async with self._session.request(
+                    method, url, headers=headers, json=json_body
+                ) as response:
+                    status = response.status
+                    if not 200 <= status < 300:
+                        raise NestQuestApiError(
+                            f"NestQuest API {method} {path} failed with "
+                            f"HTTP {status}",
+                            status=status,
+                        )
+                    body_text = await response.text()
+        except asyncio.CancelledError:
+            # Cancellation is never converted: a shutting-down event
+            # loop must stop, not read a synthesized API error.
+            raise
+        except NestQuestApiError:
+            raise
+        except Exception as err:
+            raise NestQuestApiError(
+                f"NestQuest API {method} {path} transport failure: {err}",
+                status=None,
+            ) from err
+        try:
+            return json.loads(body_text)
+        except ValueError as err:
+            raise NestQuestApiError(
+                f"NestQuest API {method} {path} returned malformed "
+                f"JSON: {err}",
+                status=status,
+            ) from err

custom_components/nestquest/coordinator.py

@@ -1,4 +1,4 @@
-"""The shared NestQuest DataUpdateCoordinator (Feature 10).
+"""The shared NestQuest DataUpdateCoordinator (Feature 10, Feature 18).
 
 ONE refresh pass builds the whole entity snapshot: active children,
 today's instances with their derived states (open / done / missed via
@@ -7,40 +7,56 @@ per-child rollup counts.  Every Feature 10 entity reads
 ``coordinator.data`` — no entity queries the database itself, so one
 refresh updates the whole board coherently.
 
-"Today" and "now" are HA-local (``hass.config.time_zone``), never the
+Since Feature 18 the snapshot no longer comes from the local database:
-host clock and never UTC, matching the materialization and completion
+the coordinator polls the NestQuest API service's panel plane —
-layers.
+``GET /api/v1/panel/snapshot`` through
+:class:`~.api_client.NestQuestApiClient` on its update interval — and
+RECONSTRUCTS the same core dataclasses the local builder produced
+(:class:`~.core.snapshot.NestQuestSnapshot`,
+:class:`~.core.snapshot.ChildDaySnapshot`,
+:class:`~.core.snapshot.QuestInstanceView`).  ``sensor.py`` and
+``binary_sensor.py`` read ``coordinator.data`` and
+:func:`~.core.snapshot.instance_payload` exactly as before, so every
+entity_id and attribute payload shape is unchanged: the panel cards and
+the four blueprints keep working untouched.
 
-The snapshot assembly itself lives in the Home-Assistant-free
+One loss is deliberate and documented (D-009 revisited): the panel
-:mod:`custom_components.nestquest.core.snapshot` module
+payload OMITS missed instances, so a snapshot rebuilt from the API can
-(:func:`~core.snapshot.build_snapshot`); this coordinator is now a thin
+no longer carry them — ``admin_instances`` (``include_missed=True``)
-HA wrapper that resolves the local timezone and reads the settings handed
+now equals ``instances`` (they can hold only open and completed rows).
-into its constructor, then delegates to that builder, passing the SAME
+The admin surface for missed quests is the PWA, which reads the API
-HA-local ``now`` it resolved in one clock read.  The API service's panel
+service directly; the integration's entities never needed missed rows
-snapshot route shares the same builder, so both consumers render ONE
+for their states or counts (the counts arrive precomputed).
-snapshot shape.
 
-Presence is resolved through the SAME engine the materializer uses
+Failure handling: constructing the client fails fast (ValueError) when
-(:class:`~.presence.PresenceEngine`): schedules and overrides are read
+the entry has no panel token configured, and a fetch failure raises the
-with the public DAO methods and the engine's anchor-date arithmetic
+typed :class:`~.api_client.NestQuestApiError`.  Neither crashes
-answers ``is this child at this house today`` (D-004 — never ISO week
+integration setup: the constructor logs a clear message and hands back
-parity).
+a client-shaped stub whose refresh raises the same typed error, and
-
+``_async_update_data`` converts every typed error into
-``cycle_day`` reports where today falls in the custody cycle of the
+:class:`~homeassistant.helpers.update_coordinator.UpdateFailed` — the
-first active child (display order) that HAS a schedule — 1-based day
+exception a HA ``DataUpdateCoordinator`` records as a failed pass
-``N`` of ``cycle_length_weeks * 7``, computed from that schedule's
+(``last_update_success = False``, ``data = None``), so the entities
-anchor date.  ``0`` means no active child has a schedule at all, so
+come up unavailable.  The typed error is never swallowed silently: it
-the admin header degrades to a harmless zero instead of guessing.
+is logged with its message and chained into the UpdateFailed.  (The
+conversion matters at setup: HA's real
+``async_config_entry_first_refresh`` wraps any failed pass in
+``ConfigEntryNotReady``, so a setup that keys on the original
+exception type would never match it.)
 """
 from __future__ import annotations
 
 import asyncio
 import datetime
-from zoneinfo import ZoneInfo
+from typing import Any
 
 from homeassistant.core import HomeAssistant
-from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
+from homeassistant.helpers.update_coordinator import (
+    DataUpdateCoordinator,
+    UpdateFailed,
+)
 
+from .api_client import NestQuestApiClient, NestQuestApiError, resolve_api_config
 from .const import (
     CONF_UPDATE_INTERVAL,
     DEFAULT_UPDATE_INTERVAL,
@@ -48,13 +64,10 @@ from .const import (
     LOGGER,
     MIN_UPDATE_INTERVAL,
 )
-from .core.db import NestQuestDatabase
-from .core.settings import NestQuestSettings
 from .core.snapshot import (
     ChildDaySnapshot,
     NestQuestSnapshot,
     QuestInstanceView,
-    build_snapshot,
 )
 
 __all__ = [
@@ -65,9 +78,120 @@ __all__ = [
 ]
 
 
+def snapshot_from_api_payload(payload: dict[str, Any]) -> NestQuestSnapshot:
+    """Reconstruct the core snapshot dataclasses from the API's JSON.
+
+    The panel snapshot route (``api/routes_panel.py``) returns today's
+    household as plain JSON — ``today_iso``, ``cycle_day``, and the
+    per-child list with precomputed counts and the instances in the
+    documented payload shape.  This function maps that shape BACK onto
+    the dataclasses ``sensor.py`` / ``binary_sensor.py`` read, field by
+    field:
+
+    - the API's ``state`` ``"completed"`` maps back to the core view's
+      ``"done"`` (so :func:`~.core.snapshot.instance_payload` renders
+      ``"completed"`` again — the round-trip is identity for open and
+      completed rows);
+    - ``overdue``, ``on_time`` (→ ``was_on_time``), ``completed_at``,
+      ``window``, ``due_time``, ``icon`` and the per-child counts pass
+      through verbatim;
+    - ``present`` and ``next_present`` pass through verbatim (the API
+      resolved them through the same presence engine server-side);
+    - the instance list is rebuilt IN ORDER, so the panel payload the
+      sensors emit matches the route's order byte for byte.
+
+    Because the route omits missed instances (D-009), a snapshot built
+    here cannot contain any ``missed`` view — see the module docstring.
+    """
+    children: list[ChildDaySnapshot] = []
+    for child in payload.get("children", []):
+        views = tuple(
+            QuestInstanceView(
+                instance_id=instance["id"],
+                definition_id=instance["definition_id"],
+                child_id=instance["child_id"],
+                title=instance["title"],
+                icon=instance.get("icon"),
+                window=instance["window"],
+                # The API payload does not carry due_date: every row
+                # is TODAY's instance (the route queries
+                # [today, today]), so the view's due_date is the
+                # snapshot's today_iso — the same value the local
+                # builder stamped on every today row.
+                due_date=payload["today_iso"],
+                due_time=instance.get("due_time"),
+                state=(
+                    "done" if instance["state"] == "completed" else
+                    instance["state"]
+                ),
+                overdue=instance["overdue"],
+                completed_at=instance.get("completed_at"),
+                was_on_time=instance.get("on_time"),
+            )
+            for instance in child.get("instances", [])
+        )
+        children.append(
+            ChildDaySnapshot(
+                child_id=child["child_id"],
+                child_name=child["child_name"],
+                instances=views,
+                present=child["present"],
+                due_today=child["due_today"],
+                completed_today=child["completed_today"],
+                remaining_today=child["remaining_today"],
+                completion_pct=child["completion_pct"],
+                next_present=child.get("next_present"),
+            )
+        )
+    return NestQuestSnapshot(
+        today_iso=payload["today_iso"],
+        children=tuple(children),
+        cycle_day=payload["cycle_day"],
+    )
+
+
+class _UnconfiguredApiClient:
+    """Client-shaped stand-in used when the entry has no panel token.
+
+    Constructing a real :class:`~.api_client.NestQuestApiClient` with an
+    empty token raises ValueError by design (fail fast on a client that
+    could only ever 401).  An entry that simply has not been configured
+    yet must not crash integration setup, so the coordinator swaps in
+    this stub: every refresh raises the SAME typed
+    :class:`~.api_client.NestQuestApiError` a transport failure would,
+    carrying a clear "API not configured" message;
+    :meth:`NestQuestCoordinator._async_update_data` converts it to
+    :class:`~homeassistant.helpers.update_coordinator.UpdateFailed`, so
+    the coordinator marks the entities unavailable and the HA standard
+    retry cycle applies.  The error is logged, never silent.
+    """
+
+    __slots__ = ()
+
+    async def get_snapshot(self) -> dict[str, Any]:
+        """Always raise: there is no configured API to fetch from."""
+        raise NestQuestApiError(
+            "NestQuest API not configured: set the panel token in the "
+            "integration options"
+        )
+
+
 class NestQuestCoordinator(DataUpdateCoordinator):
     """Shared refresh cycle for every NestQuest entity.
 
+    One refresh pass = one ``GET /api/v1/panel/snapshot`` through the
+    API client, reconstructed into the core snapshot dataclasses every
+    entity reads.  A failed pass (unconfigured API, transport failure,
+    non-2xx response) raises the typed error out of the client;
+    ``_async_update_data`` converts it to
+    :class:`~homeassistant.helpers.update_coordinator.UpdateFailed`, the
+    exception HA's coordinator machinery records as a failed pass: the
+    coordinator's ``data`` keeps its last good value while
+    ``last_update_success`` goes False — HA marks the entities
+    unavailable until a later pass succeeds.  An unexpected
+    non-typed error still propagates out of ``_async_update_data``
+    and is logged (with its traceback) by HA's machinery, the same way.
+
     Forced refreshes (the service paths pushing an immediate update
     after a completion) are SERIALIZED per entry: overlapping
     complete/uncomplete calls must not interleave their snapshot
@@ -82,8 +206,7 @@ class NestQuestCoordinator(DataUpdateCoordinator):
         hass: HomeAssistant,
         *,
         entry_id: str,
-        database: NestQuestDatabase,
+        api_client: NestQuestApiClient,
-        settings: NestQuestSettings,
         update_interval_seconds: int | None = None,
     ) -> None:
         interval = update_interval_seconds
@@ -100,8 +223,7 @@ class NestQuestCoordinator(DataUpdateCoordinator):
             update_interval=datetime.timedelta(seconds=interval),
         )
         self.entry_id = entry_id
-        self.database = database
+        self.api_client = api_client
-        self.settings = settings
         self._refresh_lock = asyncio.Lock()
 
     async def async_refresh(self) -> None:
@@ -111,12 +233,67 @@ class NestQuestCoordinator(DataUpdateCoordinator):
         async with self._refresh_lock:
             await super().async_refresh()
 
-    def _local_now(self) -> datetime.datetime:
-        time_zone = ZoneInfo(self.hass.config.time_zone)
-        return datetime.datetime.now(time_zone)
-
     async def _async_update_data(self) -> NestQuestSnapshot:
-        now = self._local_now()
+        """One refresh pass: fetch the panel snapshot and rebuild it.
-        return await build_snapshot(
+
-            self.database, self.settings, now
+        A typed API failure (unconfigured token stub, transport
+        failure, non-2xx response) is converted to
+        :class:`~homeassistant.helpers.update_coordinator.UpdateFailed`
+        — the exception HA's coordinator machinery records as a failed
+        pass (``last_update_success = False``, ``data`` unchanged) so
+        entities go unavailable and the setup-time first refresh
+        arrives as ``ConfigEntryNotReady`` — with the typed error
+        logged and chained.  Anything else still propagates.
+        """
+        try:
+            payload = await self.api_client.get_snapshot()
+        except NestQuestApiError as err:
+            LOGGER.error(
+                "NestQuest coordinator refresh failed: %s", err
+            )
+            raise UpdateFailed(str(err)) from err
+        return snapshot_from_api_payload(payload)
+
+
+def coordinator_client_from_entry(
+    hass: HomeAssistant, entry: Any, *, session_factory: Any = None
+) -> Any:
+    """Build the refresh client for a config entry, tolerating no token.
+
+    Reads ``(base_url, panel_token)`` off the entry and constructs the
+    real client.  An entry with no panel token configured (the shipped
+    default is the empty string) makes the constructor raise ValueError
+    by design; integration setup must not crash on an unconfigured
+    entry, so this factory swaps in the typed-error stub instead and
+    logs ONE clear message.  The ValueError itself is never re-raised
+    past this point — it is converted, with the reason logged.
+
+    ``session_factory`` overrides where the HTTP session comes from
+    (production: HA's shared aiohttp session via
+    :func:`~.api_client.client_from_entry`); it exists so tests can
+    inject a stub transport without importing aiohttp — the mock-only
+    test harness has no ``homeassistant.helpers.aiohttp_client``.
+    """
+    base_url, panel_token = resolve_api_config(entry)
+    if not isinstance(panel_token, str) or not panel_token.strip():
+        LOGGER.warning(
+            "NestQuest API panel token is not configured for entry %s: "
+            "entities will stay unavailable until a token is set in the "
+            "integration options",
+            entry.entry_id,
         )
+        return _UnconfiguredApiClient()
+    if session_factory is not None:
+        return NestQuestApiClient(
+            base_url, panel_token, session_factory(hass)
+        )
+    from custom_components.nestquest.api_client import client_from_entry
+
+    return client_from_entry(hass, entry)
+
+
+#: Test seam: entry_id → client, consulted by the conftest's client
+#: factory BEFORE the production path.  Always empty in production
+#: (nothing in the integration writes to it); tests register scripted
+#: clients through it and clear it between tests.
+_COORDINATOR_CLIENT_OVERRIDES: dict[str, Any] = {}

custom_components/nestquest/core/const.py

@@ -101,6 +101,32 @@ CONF_END_OF_DAY_REPORT_ENABLED = "end_of_day_report_enabled"
 CONF_CELEBRATION_ENABLED = "celebration_enabled"
 DEFAULT_AUTOMATION_ENABLED = True
 
+# Panel-plane API client (Feature 18): the integration reaches the
+# NestQuest API service (Feature 16) over HTTP instead of the database.
+# The base URL points at the API service host serving ``/api/v1/panel/*``
+# (uvicorn's default host/port is the sensible starting point); the panel
+# service token is the static Bearer credential the panel plane requires
+# (``NESTQUEST_PANEL_TOKEN`` on the API side).  Both are stored on the
+# entry like every other setting — options first, then the entry-data
+# copy, then defaults.  The TOKEN IS A SECRET: it is never logged, never
+# echoed into error messages, and never committed.  The shipped default
+# token IS the empty string (DEFAULT_PANEL_TOKEN) — an entry with no
+# configured token is therefore NOT a working client: the options flow
+# accepts an empty token as "not configured yet" and only rejects
+# malformed (padded/whitespace-only/non-string) values — the actual
+# non-emptiness guard is enforced at client construction, where
+# NestQuestApiClient fails fast on an empty/whitespace token
+# (api_client.py).  So a caller building the client from a bare
+# entry must either ensure a token is configured first or handle the
+# construction ValueError and surface an "API not configured" state —
+# an unconfigured entry never yields a client that can only 401.
+# This mirrors the API service's fail-closed stance (api/config.py: no
+# default token exists).
+CONF_API_BASE_URL = "api_base_url"
+DEFAULT_API_BASE_URL = "http://127.0.0.1:8000"
+CONF_PANEL_TOKEN = "panel_token"
+DEFAULT_PANEL_TOKEN = ""
+
 # Quest windows (D-008): a definition may declare several day windows;
 # each produces its own instance per day and groups the panel's Quest
 # Log into three columns.  The clock ranges are stored here so they stay

custom_components/nestquest/options_flow.py

@@ -12,6 +12,7 @@ from .const import (
     CONF_ADMIN_USER_IDS,
     CONF_AFTERNOON_REMINDER_ENABLED,
     CONF_AFTERNOON_REMINDER_TIME,
+    CONF_API_BASE_URL,
     CONF_CELEBRATION_ENABLED,
     CONF_DAY_ROLLOVER_TIME,
     CONF_END_OF_DAY_REPORT_ENABLED,
@@ -21,7 +22,9 @@ from .const import (
     CONF_MORNING_SUMMARY_TIME,
     CONF_NOTIFY_TARGET,
     CONF_PANEL_IDLE_TIMEOUT,
+    CONF_PANEL_TOKEN,
     CONF_UPDATE_INTERVAL,
+    DEFAULT_API_BASE_URL,
     DEFAULT_AFTERNOON_REMINDER_TIME,
     DEFAULT_AUTOMATION_ENABLED,
     DEFAULT_DAY_ROLLOVER_TIME,
@@ -29,12 +32,14 @@ from .const import (
     DEFAULT_HORIZON_DAYS,
     DEFAULT_MORNING_SUMMARY_TIME,
     DEFAULT_PANEL_IDLE_TIMEOUT,
+    DEFAULT_PANEL_TOKEN,
     DEFAULT_UPDATE_INTERVAL,
     DOMAIN,
     MIN_UPDATE_INTERVAL,
     TIME_PATTERN,
 )
 from .admin_allowlist import list_admin_ids, set_admin_ids
+from .api_client import is_valid_base_url
 
 MIN_HORIZON_DAYS = 1
 MIN_PANEL_IDLE_TIMEOUT = 30
@@ -116,6 +121,25 @@ def _async_validate(user_input: dict[str, Any]) -> dict[str, str]:
         if toggle_value is not None and not isinstance(toggle_value, bool):
             errors[toggle_key] = "invalid"
 
+    # The panel-plane API connection (Feature 18) is OPTIONAL in raw
+    # submits the same way: a missing key keeps the current value.  The
+    # base URL must be a full http(s) URL with a host — the same check
+    # the client's constructor enforces, so a malformed URL fails at
+    # the form, not at the first request.  The token may be EMPTY (the
+    # API connection is simply not configured yet) but a non-empty one
+    # must be pasted unpadded: a padded or whitespace-only value is a
+    # clipboard mistake that would fail the token check at request
+    # time.
+    api_base_url = user_input.get(CONF_API_BASE_URL)
+    if api_base_url is not None and not is_valid_base_url(api_base_url):
+        errors[CONF_API_BASE_URL] = "invalid_url"
+    panel_token = user_input.get(CONF_PANEL_TOKEN)
+    if panel_token is not None and (
+        not isinstance(panel_token, str)
+        or panel_token != panel_token.strip()
+    ):
+        errors[CONF_PANEL_TOKEN] = "invalid"
+
     return errors
 
 
@@ -186,6 +210,18 @@ def _build_schema(
             CONF_CELEBRATION_ENABLED,
             default=current[CONF_CELEBRATION_ENABLED],
         ): _strict_bool,
+        # Panel-plane API connection (Feature 18): the base URL the API
+        # client talks to and the panel service token it presents as
+        # the Bearer credential.  Schema-validated submits always carry
+        # both (the defaults pre-fill from the stored values); raw
+        # callers omitting them keep their stored values (see
+        # _async_validate).
+        vol.Required(
+            CONF_API_BASE_URL, default=current[CONF_API_BASE_URL]
+        ): str,
+        vol.Required(
+            CONF_PANEL_TOKEN, default=current[CONF_PANEL_TOKEN]
+        ): str,
     }
     if admin_choices is not None:
         # HA's supported multi-select validator: the frontend can
@@ -336,6 +372,11 @@ class NestQuestOptionsFlow(config_entries.OptionsFlowWithConfigEntry):
             data[notification_key] = user_input.get(
                 notification_key, current[notification_key]
             )
+        # The API connection section (Feature 18) stores the base URL
+        # and panel token the same fall-back-to-current way, so raw
+        # submits that predate the section keep their stored values.
+        for api_key in (CONF_API_BASE_URL, CONF_PANEL_TOKEN):
+            data[api_key] = user_input.get(api_key, current[api_key])
         if admin_ids_present:
             data[CONF_ADMIN_USER_IDS] = admin_ids
         elif admin_default:
@@ -365,6 +406,8 @@ class NestQuestOptionsFlow(config_entries.OptionsFlowWithConfigEntry):
             (CONF_AFTERNOON_REMINDER_ENABLED, DEFAULT_AUTOMATION_ENABLED),
             (CONF_END_OF_DAY_REPORT_ENABLED, DEFAULT_AUTOMATION_ENABLED),
             (CONF_CELEBRATION_ENABLED, DEFAULT_AUTOMATION_ENABLED),
+            (CONF_API_BASE_URL, DEFAULT_API_BASE_URL),
+            (CONF_PANEL_TOKEN, DEFAULT_PANEL_TOKEN),
         ):
             if key in options:
                 values[key] = options[key]

custom_components/nestquest/sse.py

@@ -0,0 +1,172 @@
+"""HA-bus subscription over the API service's panel SSE stream.
+
+The NestQuest API service (Feature 16) publishes every quest
+transition on ``GET /api/v1/panel/events`` as Server-Sent Events
+frames.  This module is the integration-side subscriber: ONE
+background task consumes :meth:`~.api_client.NestQuestApiClient
+.stream_events` and re-fires each frame on ``hass.bus`` with the SAME
+event type and payload fields the local transition path
+(:mod:`~.events`) emits — ``nestquest_quest_completed``,
+``nestquest_quest_uncompleted``, ``nestquest_quest_missed`` and
+``nestquest_child_day_complete`` — so the four blueprints fire
+identically whether a completion happens through the integration's own
+service or through the API service's panel route.
+
+Failure handling: a failed stream (API down, network blip, non-2xx, a
+dropped connection) is retried FOREVER with a bounded exponential
+backoff — :data:`RECONNECT_MIN_SECONDS` after the first failure,
+doubling up to :data:`RECONNECT_MAX_SECONDS` — so a long API outage
+neither pins the event loop nor spams the log, and a frame received on
+a healthy stream resets the backoff.  Nothing escapes the task: every
+failure is logged with its reason and the loop reconnects; a
+:class:`asyncio.CancelledError` (from :meth:`stop` or integration
+unload) is the only thing that ends it, and it stops cleanly — no
+frame fires after :meth:`stop` returns and the response is closed.
+"""
+from __future__ import annotations
+
+import asyncio
+from typing import Any
+
+from .const import LOGGER
+
+__all__ = [
+    "NestQuestEventStream",
+    "RECONNECT_MAX_SECONDS",
+    "RECONNECT_MIN_SECONDS",
+]
+
+#: Delay before the FIRST reconnect attempt after a stream failure.
+#: One failed frame must not turn into a hot reconnect loop.
+RECONNECT_MIN_SECONDS = 5.0
+
+#: Ceiling for the exponential reconnect backoff: a long API outage
+#: retries at a calm fixed cadence instead of growing without bound.
+RECONNECT_MAX_SECONDS = 60.0
+
+
+class NestQuestEventStream:
+    """Consume the panel SSE stream in the background, fire the HA bus.
+
+    ONE instance per config entry, built in integration setup with the
+    SAME API client the coordinator polls (the client is stateless —
+    one shared session, one connection pool).  ``start()`` schedules
+    the background task; ``stop()`` cancels and awaits it, so unload
+    never leaks the task and no frame fires after teardown.  The task
+    itself is failure-proof: any exception from the stream (typed
+    :class:`~.api_client.NestQuestApiError` or anything else) is
+    logged and retried after the current backoff, never allowed to
+    escape the task.
+    """
+
+    def __init__(
+        self,
+        hass: Any,
+        client: Any,
+        *,
+        reconnect_min_seconds: float = RECONNECT_MIN_SECONDS,
+        reconnect_max_seconds: float = RECONNECT_MAX_SECONDS,
+    ) -> None:
+        """Store the hass bus target, the API client, and the backoff.
+
+        The client is anything exposing ``stream_events()`` (the real
+        :class:`~.api_client.NestQuestApiClient` in production, a stub
+        in tests).  The backoff bounds must be positive and the minimum
+        must not exceed the maximum — a misconfiguration should fail
+        loudly at construction, not quietly reconnect at delay 0.
+        """
+        if (
+            not isinstance(reconnect_min_seconds, (int, float))
+            or isinstance(reconnect_min_seconds, bool)
+            or reconnect_min_seconds <= 0
+            or reconnect_min_seconds > reconnect_max_seconds
+        ):
+            raise ValueError(
+                "reconnect_min_seconds must be a positive number no "
+                f"greater than reconnect_max_seconds, got "
+                f"{reconnect_min_seconds!r}"
+            )
+        if (
+            not isinstance(reconnect_max_seconds, (int, float))
+            or isinstance(reconnect_max_seconds, bool)
+            or reconnect_max_seconds <= 0
+        ):
+            raise ValueError(
+                f"reconnect_max_seconds must be a positive number, got "
+                f"{reconnect_max_seconds!r}"
+            )
+        self._hass = hass
+        self._client = client
+        self._min_backoff = float(reconnect_min_seconds)
+        self._max_backoff = float(reconnect_max_seconds)
+        self._backoff = float(reconnect_min_seconds)
+        self._task: asyncio.Task | None = None
+
+    @property
+    def task(self) -> asyncio.Task | None:
+        """The background task, or ``None`` when never started/stopped."""
+        return self._task
+
+    def start(self) -> None:
+        """Schedule the subscription task (idempotent while it runs)."""
+        if self._task is not None and not self._task.done():
+            return
+        self._backoff = self._min_backoff
+        self._task = asyncio.create_task(
+            self._run(), name="nestquest-sse-subscription"
+        )
+
+    async def stop(self) -> None:
+        """Cancel and await the subscription task.
+
+        Returns only after the task has actually finished, so unload
+        can rely on: no further frame is fired after ``stop()``, and
+        the underlying stream response has been closed.  Safe to call
+        when the task was never started or already stopped.
+        """
+        task, self._task = self._task, None
+        if task is None:
+            return
+        task.cancel()
+        try:
+            await task
+        except asyncio.CancelledError:
+            # The task's own cancellation surfaces here — expected and
+            # swallowed; a cancelling the CALLER would re-raise out of
+            # the surrounding await on its own.
+            pass
+
+    async def _run(self) -> None:
+        """Consume stream frames forever, reconnecting on failure."""
+        while True:
+            try:
+                async for event_type, payload in self._client.stream_events():
+                    # A frame arrived on a healthy stream: the next
+                    # failure reconnects quickly again.
+                    self._backoff = self._min_backoff
+                    try:
+                        self._hass.bus.async_fire(event_type, dict(payload))
+                    except Exception as err:
+                        # Firing one frame must never tear down a
+                        # healthy connection: log and keep consuming.
+                        LOGGER.error(
+                            "NestQuest panel event %s could not be fired "
+                            "on the Home Assistant event bus: %s",
+                            event_type,
+                            err,
+                        )
+            except asyncio.CancelledError:
+                # Unload / shutdown: stop cleanly, never converted.
+                raise
+            except Exception as err:
+                delay = self._backoff
+                self._backoff = min(
+                    self._backoff * 2, self._max_backoff
+                )
+                LOGGER.warning(
+                    "NestQuest panel event stream failed (%s); "
+                    "reconnecting in %.1f s",
+                    err,
+                    delay,
+                )
+                await asyncio.sleep(delay)

custom_components/nestquest/strings.json

@@ -18,11 +18,13 @@
     "step": {
       "init": {
         "title": "NestQuest Settings",
-        "description": "Adjust how NestQuest plans, rolls over days, and idles its panel.",
+        "description": "Adjust how NestQuest plans, rolls over days, idles its panel, and reaches the NestQuest API service.",
         "data": {
           "horizon_days": "Generation horizon (days)",
           "day_rollover_time": "Day rollover time (HH:MM)",
           "panel_idle_timeout": "Panel idle timeout (seconds)",
+          "api_base_url": "NestQuest API base URL (http://host:port)",
+          "panel_token": "Panel service token",
           "admin_user_ids": "Admin users (parents)"
         }
       }
@@ -30,6 +32,7 @@
     "errors": {
       "invalid": "Invalid value.",
       "invalid_time": "Time must use 24-hour HH:MM format.",
+      "invalid_url": "Enter the API base URL as a full http:// or https:// URL with a host.",
       "no_admins": "Select at least one admin user; an empty allowlist would lock everyone out.",
       "invalid_admin": "Invalid admin selection."
     }

custom_components/nestquest/translations/en.json

@@ -18,11 +18,13 @@
     "step": {
       "init": {
         "title": "NestQuest Settings",
-        "description": "Adjust how NestQuest plans, rolls over days, and idles its panel.",
+        "description": "Adjust how NestQuest plans, rolls over days, idles its panel, and reaches the NestQuest API service.",
         "data": {
           "horizon_days": "Generation horizon (days)",
           "day_rollover_time": "Day rollover time (HH:MM)",
           "panel_idle_timeout": "Panel idle timeout (seconds)",
+          "api_base_url": "NestQuest API base URL (http://host:port)",
+          "panel_token": "Panel service token",
           "admin_user_ids": "Admin users (parents)"
         }
       }
@@ -30,6 +32,7 @@
     "errors": {
       "invalid": "Invalid value.",
       "invalid_time": "Time must use 24-hour HH:MM format.",
+      "invalid_url": "Enter the API base URL as a full http:// or https:// URL with a host.",
       "no_admins": "Select at least one admin user; an empty allowlist would lock everyone out.",
       "invalid_admin": "Invalid admin selection."
     }

tests/conftest.py

@@ -52,6 +52,7 @@ _HA_MODULES = (
     "homeassistant.helpers.entity",
     "homeassistant.helpers.event",
     "homeassistant.helpers.update_coordinator",
+    "homeassistant.helpers.aiohttp_client",
     "homeassistant.components",
     "homeassistant.components.sensor",
     "homeassistant.components.binary_sensor",
@@ -99,6 +100,7 @@ for _parent, _child in (
     ("homeassistant.helpers", "entity"),
     ("homeassistant.helpers", "event"),
     ("homeassistant.helpers", "update_coordinator"),
+    ("homeassistant.helpers", "aiohttp_client"),
     ("homeassistant", "components"),
     ("homeassistant.components", "sensor"),
     ("homeassistant.components", "binary_sensor"),
@@ -365,6 +367,15 @@ class BinarySensorEntity(Entity):
         return self._attr_device_class
 
 
+class UpdateFailed(HomeAssistantError):
+    """Stand-in mirroring homeassistant.helpers.update_coordinator.UpdateFailed.
+
+    The exception an update pass raises to report a failed fetch: the
+    coordinator records ``last_update_success = False`` instead of
+    propagating it out of the refresh.
+    """
+
+
 class DataUpdateCoordinator:
     """Functional stand-in for HA's DataUpdateCoordinator.
 
@@ -373,8 +384,20 @@ class DataUpdateCoordinator:
     ``async_add_listener``/``async_update_listeners`` for entity
     subscription, ``async_refresh``/``async_config_entry_first_refresh``
     driving the subclass's ``_async_update_data``, ``last_update_success``,
-    and ``data``.  Interval scheduling is NOT simulated (no real clock
+    ``last_exception``, and ``data``.  Interval scheduling is NOT
-    ticks in tests) — tests drive refreshes directly.
+    simulated (no real clock ticks in tests) — tests drive refreshes
+    directly.
+
+    The failure contract mirrors real HA exactly (it is the trap this
+    harness once hid): an update failure NEVER propagates out of
+    ``async_refresh`` — the coordinator records
+    ``last_update_success = False``, keeps its last good ``data``, and
+    still notifies listeners so entities re-render unavailable — and
+    ``async_config_entry_first_refresh`` wraps ANY failed pass in
+    :class:`ConfigEntryNotReady`, with the original error only
+    surviving as ``__cause__``.  A stand-in that re-raised the raw
+    update error would let setup code key on the original exception
+    type — a clause real HA can never reach.
     """
 
     def __init__(self, hass, logger, *, name=None, update_interval=None):
@@ -384,6 +407,7 @@ class DataUpdateCoordinator:
         self.update_interval = update_interval
         self.data = None
         self.last_update_success = False
+        self.last_exception = None
         self._listeners = []
 
     def async_add_listener(self, update_callback, context=None):
@@ -403,18 +427,35 @@ class DataUpdateCoordinator:
             callback()
 
     async def async_refresh(self):
-        """Run one update pass and notify listeners."""
+        """Run one update pass and notify listeners.
+
+        Mirrors HA: a failed update pass is RECORDED, never raised —
+        ``last_update_success`` goes False, the exception lands on
+        ``last_exception``, and listeners still fire; only cancellation
+        propagates.
+        """
         try:
             self.data = await self._async_update_data()
             self.last_update_success = True
-        except BaseException:
+        except Exception as err:
             self.last_update_success = False
-            raise
+            self.last_exception = err
         self.async_update_listeners()
 
     async def async_config_entry_first_refresh(self):
-        """First refresh at config-entry setup (alias of refresh)."""
+        """First refresh at config-entry setup, per HA's real contract.
+
+        HA's real ``async_config_entry_first_refresh`` wraps ANY failed
+        update pass in ``ConfigEntryNotReady`` — the original error
+        survives only as ``__cause__`` — so setup code that must
+        tolerate a failed first refresh keys on ConfigEntryNotReady,
+        never on the original exception type.
+        """
         await self.async_refresh()
+        if not self.last_update_success:
+            raise ConfigEntryNotReady(
+                f"error fetching {self.name} data: {self.last_exception}"
+            ) from self.last_exception
 
     async def async_shutdown(self):
         """Release listeners on teardown."""
@@ -424,6 +465,7 @@ class DataUpdateCoordinator:
 _update_coordinator_mock = _ha_mock("homeassistant.helpers.update_coordinator")
 _update_coordinator_mock.DataUpdateCoordinator = DataUpdateCoordinator
 _update_coordinator_mock.CoordinatorEntity = CoordinatorEntity
+_update_coordinator_mock.UpdateFailed = UpdateFailed
 
 _entity_mock = _ha_mock("homeassistant.helpers.entity")
 _entity_mock.DeviceInfo = DeviceInfo
@@ -700,6 +742,122 @@ def wire_entry_to_registry(entry, registry: ListenerRegistry):
     return entry
 
 
+class LazyLocalSnapshotClient:
+    """A test client that serves the snapshot built LOCALLY.
+
+    Feature 18 moved the coordinator's refresh onto the API service's
+    panel snapshot, but most integration tests still seed the local
+    database directly (the DB removal is a later task).  This stub
+    bridges the two: ``get_snapshot`` resolves the entry's runtime
+    record (its database and settings) and runs the SAME pure core
+    builder the API route runs
+    (:func:`~custom_components.nestquest.core.snapshot.build_snapshot`),
+    shaping the payload exactly like ``api/routes_panel.py`` —
+    instances via ``instance_payload(include_missed=False)`` — so the
+    coordinator reconstructs from a byte-identical payload without any
+    network or FastAPI machinery.
+
+    The resolution is LAZY on purpose: the coordinator's factory runs
+    mid-setup, before the runtime record exists, so the first refresh
+    (and any refresh while the entry is torn down) serves an EMPTY but
+    valid snapshot; every later refresh reads the seeded database.
+    """
+
+    def __init__(self, hass, entry_id: str) -> None:
+        self._hass = hass
+        self._entry_id = entry_id
+        self.calls = 0
+
+    async def get_snapshot(self):
+        import datetime
+        import zoneinfo
+
+        from custom_components.nestquest.const import DOMAIN as NQ_DOMAIN
+        from custom_components.nestquest.core import snapshot as core_snapshot
+
+        self.calls += 1
+        runtime = self._hass.data.get(NQ_DOMAIN, {}).get(self._entry_id)
+        database = getattr(runtime, "database", None)
+        if database is None or not getattr(database, "connected", True):
+            return {
+                "today_iso": datetime.date.today().isoformat(),
+                "cycle_day": 0,
+                "children": [],
+            }
+        settings = getattr(runtime, "settings", None)
+        # The HA-LOCAL now, the same single clock read the production
+        # coordinator used to hand the builder: the hass stub's
+        # configured time zone, never the host clock.
+        now = datetime.datetime.now(
+            zoneinfo.ZoneInfo(self._hass.config.time_zone)
+        )
+        snapshot = await core_snapshot.build_snapshot(database, settings, now)
+        return {
+            "today_iso": snapshot.today_iso,
+            "cycle_day": snapshot.cycle_day,
+            "children": [
+                {
+                    "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": core_snapshot.instance_payload(
+                        child.instances, include_missed=False
+                    ),
+                }
+                for child in snapshot.children
+            ],
+        }
+
+
+@pytest.fixture(autouse=True)
+def coordinator_serves_local_snapshot():
+    """Default every test entry's coordinator to the local-snapshot client.
+
+    The integration's setup builds its API client through
+    ``coordinator_client_from_entry``; the fixture swaps that factory
+    for a test one handing out :class:`LazyLocalSnapshotClient` (so
+    DB-seeded tests exercise the coordinator's API path against local
+    data).  Tests that script their own client register it with
+    :func:`set_coordinator_client`, which the factory honours first.
+    """
+    import custom_components.nestquest as nq_mod
+    import custom_components.nestquest.coordinator as coordinator_module
+
+    _COORDINATOR_CLIENT_OVERRIDES = (
+        coordinator_module._COORDINATOR_CLIENT_OVERRIDES
+    )
+
+    def _factory(hass, entry, **kwargs):
+        override = _COORDINATOR_CLIENT_OVERRIDES.get(entry.entry_id)
+        if override is not None:
+            return override
+        return LazyLocalSnapshotClient(hass, entry.entry_id)
+
+    import unittest.mock as mock
+
+    _COORDINATOR_CLIENT_OVERRIDES.clear()
+    patcher = mock.patch.object(
+        nq_mod, "coordinator_client_from_entry", _factory
+    )
+    patcher.start()
+    yield
+    patcher.stop()
+    _COORDINATOR_CLIENT_OVERRIDES.clear()
+
+
+def set_coordinator_client(entry, client):
+    """Register a scripted client for ``entry``'s coordinator factory."""
+    import custom_components.nestquest.coordinator as coordinator_module
+
+    coordinator_module._COORDINATOR_CLIENT_OVERRIDES[entry.entry_id] = client
+    return client
+
+
 class _HassNamespace(SimpleNamespace):
     """SimpleNamespace subclass so the hass stub can hold weak references."""
 

tests/test_api_client.py

@@ -0,0 +1,450 @@
+"""Tests for the panel-plane API client (Feature 18).
+
+The client is exercised against a STUB transport only — no network, no
+aiohttp, no homeassistant: the injected session is a fake that speaks
+the same ``request(...)`` surface aiohttp's ``ClientSession`` exposes
+and records every call, so the tests assert exactly what the client
+puts on the wire (method, URL, Bearer header, JSON body) and how every
+failure mode (non-2xx, connection failure, timeout, malformed JSON)
+surfaces as the typed :class:`NestQuestApiError`.
+"""
+from __future__ import annotations
+
+import asyncio
+import json
+import sys
+import types
+
+import pytest
+
+from conftest import make_config_entry
+
+from custom_components.nestquest import api_client
+from custom_components.nestquest.api_client import (
+    NestQuestApiClient,
+    NestQuestApiError,
+    is_valid_base_url,
+    resolve_api_config,
+)
+from custom_components.nestquest.const import (
+    CONF_API_BASE_URL,
+    CONF_PANEL_TOKEN,
+    DEFAULT_API_BASE_URL,
+    DEFAULT_PANEL_TOKEN,
+)
+
+BASE_URL = "http://api.test:8000"
+TOKEN = "test-panel-token"
+
+#: A realistic panel snapshot payload (the documented shape of the
+#: Feature 16 snapshot route).
+SNAPSHOT_PAYLOAD = {
+    "today_iso": "2026-09-23",
+    "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": None,
+                    "window": "morning",
+                    "due_time": "10:00",
+                    "state": "open",
+                    "overdue": True,
+                    "completed_at": None,
+                    "on_time": None,
+                }
+            ],
+        }
+    ],
+}
+
+
+class StubResponse:
+    """A fake aiohttp response: ``status`` plus ``await text()``."""
+
+    def __init__(self, status: int, body: str) -> None:
+        self.status = status
+        self._body = body
+
+    async def text(self) -> str:
+        return self._body
+
+
+class _OutcomeContext:
+    """Async context manager yielding one scripted outcome.
+
+    Mirrors aiohttp's ``session.request(...)`` return value: entering
+    the context performs the request, so a scripted transport error
+    raises THERE — before any response object exists, exactly where
+    aiohttp's connector raises on a refused connection.
+    """
+
+    def __init__(self, response, error) -> None:
+        self._response = response
+        self._error = error
+
+    async def __aenter__(self):
+        if self._error is not None:
+            raise self._error
+        return self._response
+
+    async def __aexit__(self, exc_type, exc, tb) -> bool:
+        return False
+
+
+class StubTransport:
+    """A no-network stand-in for aiohttp's ``ClientSession``.
+
+    Records every ``request(...)`` call as a dict (method, url, headers,
+    json body) and answers from a scripted queue of outcomes — a
+    :class:`StubResponse` or an exception per call, consumed in order —
+    so each test drives exactly one exchange.
+    """
+
+    def __init__(self, *outcomes) -> None:
+        self.calls: list[dict] = []
+        self._outcomes = list(outcomes)
+
+    def request(self, method: str, url: str, **kwargs):
+        self.calls.append(
+            {
+                "method": method,
+                "url": url,
+                "headers": kwargs.get("headers"),
+                "json": kwargs.get("json"),
+            }
+        )
+        if not self._outcomes:
+            raise AssertionError("StubTransport ran out of scripted outcomes")
+        outcome = self._outcomes.pop(0)
+        if isinstance(outcome, BaseException):
+            return _OutcomeContext(None, outcome)
+        return _OutcomeContext(outcome, None)
+
+
+class HangingTransport:
+    """A transport whose request never answers (for the timeout test)."""
+
+    def __init__(self) -> None:
+        self.calls: list[dict] = []
+
+    def request(self, method: str, url: str, **kwargs):
+        self.calls.append({"method": method, "url": url})
+        return _HangingContext()
+
+
+class _HangingContext:
+    async def __aenter__(self):
+        await asyncio.sleep(5)
+        raise AssertionError("hanging transport was never cancelled")
+
+    async def __aexit__(self, exc_type, exc, tb) -> bool:
+        return False
+
+
+def make_client(transport, *, timeout_seconds: float = 5.0) -> NestQuestApiClient:
+    """Build a client over the stub transport with the test credentials."""
+    return NestQuestApiClient(
+        BASE_URL, TOKEN, transport, timeout_seconds=timeout_seconds
+    )
+
+
+# ---------------------------------------------------------------------------
+# Success paths
+# ---------------------------------------------------------------------------
+
+
+async def test_get_snapshot_parses_payload_and_sends_bearer() -> None:
+    transport = StubTransport(
+        StubResponse(200, json.dumps(SNAPSHOT_PAYLOAD))
+    )
+    client = make_client(transport)
+
+    snapshot = await client.get_snapshot()
+
+    assert snapshot == SNAPSHOT_PAYLOAD
+    call = transport.calls[0]
+    assert call["method"] == "GET"
+    assert call["url"] == f"{BASE_URL}/api/v1/panel/snapshot"
+    assert call["headers"] == {"Authorization": f"Bearer {TOKEN}"}
+    assert call["json"] is None
+
+
+async def test_complete_instance_posts_body_and_headers() -> None:
+    transport = StubTransport(StubResponse(200, '{"status": "done"}'))
+    client = make_client(transport)
+
+    result = await client.complete_instance(42, 3)
+
+    assert result == {"status": "done"}
+    call = transport.calls[0]
+    assert call["method"] == "POST"
+    assert call["url"] == f"{BASE_URL}/api/v1/panel/instances/42/complete"
+    assert call["json"] == {"actor_child_id": 3}
+    assert call["headers"]["Authorization"] == f"Bearer {TOKEN}"
+
+
+async def test_base_url_trailing_slash_is_normalized() -> None:
+    """A base URL with a trailing slash must not double the path slash."""
+    transport = StubTransport(StubResponse(200, "{}"))
+    client = NestQuestApiClient(
+        f"{BASE_URL}/", TOKEN, transport, timeout_seconds=5.0
+    )
+    await client.get_snapshot()
+    assert transport.calls[0]["url"] == f"{BASE_URL}/api/v1/panel/snapshot"
+
+
+# ---------------------------------------------------------------------------
+# HTTP failures: the typed error carries the status
+# ---------------------------------------------------------------------------
+
+
+async def test_snapshot_401_raises_typed_error() -> None:
+    """A wrong/missing token answers 401 and surfaces with that status."""
+    transport = StubTransport(
+        StubResponse(401, '{"detail": "Missing or invalid panel service token"}')
+    )
+    client = make_client(transport)
+
+    with pytest.raises(NestQuestApiError) as excinfo:
+        await client.get_snapshot()
+
+    error = excinfo.value
+    assert error.status == 401
+    assert not error.is_transport_error
+    assert error.__cause__ is None
+    # The token is a secret: it never reaches the raised message.
+    assert TOKEN not in str(error)
+
+
+async def test_complete_unknown_instance_404_raises_typed_error() -> None:
+    transport = StubTransport(
+        StubResponse(404, '{"detail": "Quest instance not found"}')
+    )
+    client = make_client(transport)
+
+    with pytest.raises(NestQuestApiError) as excinfo:
+        await client.complete_instance(999999, 1)
+
+    assert excinfo.value.status == 404
+    assert not excinfo.value.is_transport_error
+
+
+async def test_snapshot_500_raises_typed_error() -> None:
+    """Any non-2xx response is typed, whatever the status."""
+    transport = StubTransport(StubResponse(500, "{}"))
+    client = make_client(transport)
+
+    with pytest.raises(NestQuestApiError) as excinfo:
+        await client.get_snapshot()
+
+    assert excinfo.value.status == 500
+
+
+async def test_malformed_2xx_body_raises_typed_with_status() -> None:
+    """A 2xx body that is not JSON is a protocol failure, carrying its status."""
+    transport = StubTransport(StubResponse(200, "<html>not json</html>"))
+    client = make_client(transport)
+
+    with pytest.raises(NestQuestApiError) as excinfo:
+        await client.get_snapshot()
+
+    assert excinfo.value.status == 200
+
+
+# ---------------------------------------------------------------------------
+# Transport failures: the typed error carries no status and chains the cause
+# ---------------------------------------------------------------------------
+
+
+async def test_connection_failure_raises_transport_error() -> None:
+    transport = StubTransport(OSError("connection refused"))
+    client = make_client(transport)
+
+    with pytest.raises(NestQuestApiError) as excinfo:
+        await client.get_snapshot()
+
+    error = excinfo.value
+    assert error.status is None
+    assert error.is_transport_error
+    assert isinstance(error.__cause__, OSError)
+
+
+async def test_connection_failure_on_complete_is_typed() -> None:
+    transport = StubTransport(OSError("connection refused"))
+    client = make_client(transport)
+
+    with pytest.raises(NestQuestApiError) as excinfo:
+        await client.complete_instance(7, 1)
+
+    assert excinfo.value.status is None
+    assert excinfo.value.is_transport_error
+
+
+async def test_request_timeout_raises_transport_error() -> None:
+    """A hung API call surfaces as a typed transport error, once, with
+    no retry: the one request is cancelled by the client's timeout."""
+    transport = HangingTransport()
+    client = make_client(transport, timeout_seconds=0.05)
+
+    with pytest.raises(NestQuestApiError) as excinfo:
+        await client.get_snapshot()
+
+    assert excinfo.value.status is None
+    assert excinfo.value.is_transport_error
+    assert len(transport.calls) == 1
+
+
+# ---------------------------------------------------------------------------
+# Construction and config resolution
+# ---------------------------------------------------------------------------
+
+
+def test_constructor_requires_full_base_url() -> None:
+    for bad_url in (None, "", "   ", "api.lan:8000", "http://", "ftp://x"):
+        with pytest.raises(ValueError):
+            NestQuestApiClient(bad_url, TOKEN, StubTransport())
+
+
+def test_constructor_requires_non_empty_token() -> None:
+    for bad_token in (None, "", "   "):
+        with pytest.raises(ValueError):
+            NestQuestApiClient(BASE_URL, bad_token, StubTransport())
+
+
+def test_constructor_requires_positive_timeout() -> None:
+    for bad_timeout in (0, -1, "5", True, None):
+        with pytest.raises(ValueError):
+            NestQuestApiClient(
+                BASE_URL, TOKEN, StubTransport(), timeout_seconds=bad_timeout
+            )
+
+
+def test_is_valid_base_url_accepts_full_http_urls() -> None:
+    assert is_valid_base_url("http://127.0.0.1:8000")
+    assert is_valid_base_url("https://nestquest.lan")
+    assert is_valid_base_url(" http://127.0.0.1:8000 ")
+    assert not is_valid_base_url("api.lan:8000")
+    assert not is_valid_base_url("http://")
+    assert not is_valid_base_url("")
+    assert not is_valid_base_url(None)
+    assert not is_valid_base_url(42)
+
+
+def test_resolve_api_config_prefers_options_then_data_then_defaults() -> None:
+    options_only = make_config_entry(
+        options={CONF_API_BASE_URL: "http://from-options:1"}
+    )
+    assert resolve_api_config(options_only) == (
+        "http://from-options:1",
+        DEFAULT_PANEL_TOKEN,
+    )
+
+    data_only = make_config_entry(
+        data={
+            CONF_API_BASE_URL: "http://from-data:2",
+            CONF_PANEL_TOKEN: "from-data",
+        }
+    )
+    assert resolve_api_config(data_only) == ("http://from-data:2", "from-data")
+
+    both = make_config_entry(
+        options={CONF_API_BASE_URL: "http://from-options:3"},
+        data={
+            CONF_API_BASE_URL: "http://from-data:4",
+            CONF_PANEL_TOKEN: "from-data",
+        },
+    )
+    assert resolve_api_config(both) == (
+        "http://from-options:3",
+        "from-data",
+    )
+
+    bare = make_config_entry()
+    assert resolve_api_config(bare) == (DEFAULT_API_BASE_URL, DEFAULT_PANEL_TOKEN)
+
+
+def test_client_from_entry_reads_config_and_shared_session(monkeypatch) -> None:
+    """The production factory resolves the entry config and hands in HA's
+    shared aiohttp session (stubbed module — no real HA, no network)."""
+    fake_module = types.ModuleType("homeassistant.helpers.aiohttp_client")
+    shared_session = object()
+    fake_module.async_get_clientsession = lambda hass: shared_session
+    monkeypatch.setitem(
+        sys.modules, "homeassistant.helpers.aiohttp_client", fake_module
+    )
+    entry = make_config_entry(
+        options={
+            CONF_API_BASE_URL: "http://panel.lan:8443",
+            CONF_PANEL_TOKEN: "entry-token",
+        }
+    )
+
+    client = api_client.client_from_entry(object(), entry)
+
+    assert client._session is shared_session
+    assert client._base_url == "http://panel.lan:8443"
+    assert client._panel_token == "entry-token"
+
+
+def test_client_from_entry_unconfigured_token_raises_at_construction(
+    monkeypatch,
+) -> None:
+    """An entry with NO configured token (the shipped default) raises
+    ValueError AT CONSTRUCTION — the client's fail-fast guard — rather
+    than silently building a client that can only 401 (the documented
+    contract: the options flow's non-empty validation is the user-facing
+    guard; the coordinator must ensure a token is configured first or
+    handle the construction error as "API not configured")."""
+    fake_module = types.ModuleType("homeassistant.helpers.aiohttp_client")
+    fake_module.async_get_clientsession = lambda hass: object()
+    monkeypatch.setitem(
+        sys.modules, "homeassistant.helpers.aiohttp_client", fake_module
+    )
+    entry = make_config_entry(
+        options={CONF_API_BASE_URL: DEFAULT_API_BASE_URL},
+    )
+    assert resolve_api_config(entry)[1] == DEFAULT_PANEL_TOKEN == ""
+
+    with pytest.raises(ValueError, match="panel_token must be a non-empty"):
+        api_client.client_from_entry(object(), entry)
+
+    # Same for a whitespace-only stored token: whitespace is not a token.
+    entry_ws = make_config_entry(options={CONF_PANEL_TOKEN: "   "})
+    with pytest.raises(ValueError, match="panel_token must be a non-empty"):
+        api_client.client_from_entry(object(), entry_ws)
+
+
+async def test_client_logs_nothing() -> None:
+    """The client emits no log records at all (the token must never
+    reach a log line), success or failure."""
+    import logging
+
+    records: list[logging.LogRecord] = []
+
+    class _Capture(logging.Handler):
+        def emit(self, record: logging.LogRecord) -> None:
+            records.append(record)
+
+    handler = _Capture()
+    logging.getLogger().addHandler(handler)
+    try:
+        transport = StubTransport(StubResponse(401, "{}"))
+        client = make_client(transport)
+        with pytest.raises(NestQuestApiError):
+            await client.get_snapshot()
+    finally:
+        logging.getLogger().removeHandler(handler)
+    assert records == []

tests/test_coordinator.py

@@ -1,263 +1,512 @@
-"""Tests for the shared NestQuest coordinator (Feature 10, task 1)."""
+"""Tests for the shared NestQuest coordinator (Feature 10, Feature 18).
+
+The coordinator now polls the API service's panel snapshot through the
+API client (Feature 18) and reconstructs the core snapshot dataclasses
+every entity reads.  The tests stub the client — NO network, NO aiohttp
+— with a FIXTURE snapshot payload (the documented shape of the Feature
+16 snapshot route, ``api/routes_panel.py``) and assert the
+reconstruction: per-child counts and presence, the instance views
+(``completed`` mapped back to ``done``), the household rollups, the
+cycle day, and the unconfigured / failed-fetch behaviour.
+"""
 from __future__ import annotations
 
 import datetime
 
 import pytest
 
-from conftest import wire_entry_to_registry
+from conftest import set_coordinator_client, wire_entry_to_registry
 
 from custom_components.nestquest import async_setup_entry, async_unload_entry
-from custom_components.nestquest.children import (
+from custom_components.nestquest.api_client import (
-    create_child,
+    NestQuestApiClient,
-    list_children,
+    NestQuestApiError,
-    set_child_active,
 )
-from custom_components.nestquest.completion import complete_instance
 from custom_components.nestquest.const import (
     CONF_UPDATE_INTERVAL,
     DEFAULT_UPDATE_INTERVAL,
-    DEFAULT_HORIZON_DAYS,
     DOMAIN,
 )
-from custom_components.nestquest.coordinator import NestQuestCoordinator
+from custom_components.nestquest.coordinator import (
-from custom_components.nestquest.dao_presence import PresenceSchedulesDao
+    NestQuestCoordinator,
-from custom_components.nestquest.materialize import materialize
+    coordinator_client_from_entry,
-from custom_components.nestquest.quest_definitions import (
+    snapshot_from_api_payload,
-    create_quest_definition,
 )
-from custom_components.nestquest.recurrence import ScheduleRule
+from custom_components.nestquest.core.snapshot import instance_payload
-
+from homeassistant.helpers.update_coordinator import UpdateFailed
-ADMIN_CTX = {"user_id": "admin-1"}
 
 
 def _today_iso() -> str:
     return datetime.date.today().isoformat()
 
 
-async def _seed_day(database, *, completed: bool = False) -> None:
+class StubSnapshotClient:
-    """Two active children + one inactive, one daily definition each."""
+    """A no-network stand-in for the API client's snapshot surface.
-    await create_child(database, "Ada")
+
-    await create_child(database, "Bo")
+    Answers ``get_snapshot()`` from a scripted queue of outcomes — a
-    inactive = await create_child(database, "Ghost")
+    payload dict or an exception per call, consumed in order — and
-    await set_child_active(database, inactive.id, False)
+    records every call, so each test drives exactly one refresh.
-    today = _today_iso()
+    """
-    await create_quest_definition(
+
-        database,
+    def __init__(self, *outcomes) -> None:
-        "Brush teeth",
+        self.calls = 0
-        ScheduleRule.from_dict({"rule_type": "daily", "start_date": today}),
+        self._outcomes = list(outcomes)
-        [child.id for child in await list_children(database, active_only=True)],
+        self._last = outcomes[-1] if outcomes else None
-        ["morning"],
+
+    async def get_snapshot(self):
+        self.calls += 1
+        if not self._outcomes:
+            # Re-serve the last outcome: setup's first refresh and the
+            # test's explicit pass both poll the same scripted state.
+            outcome = self._last
+        else:
+            outcome = self._outcomes.pop(0)
+            self._last = outcome
+        if isinstance(outcome, BaseException):
+            raise outcome
+        return outcome
+
+
+#: A fixture panel snapshot (the documented Feature 16 route shape):
+#: two children — Ada owes one, owes none of it yet, is away today with
+#: a returns date; Bo completed his only quest and is present.
+FIXTURE_SNAPSHOT = {
+    "today_iso": _today_iso(),
+    "cycle_day": 3,
+    "children": [
+        {
+            "child_id": 1,
+            "child_name": "Ada",
+            "present": False,
+            "next_present": "2026-09-24",
+            "due_today": 2,
+            "completed_today": 1,
+            "remaining_today": 1,
+            "completion_pct": 50,
+            "instances": [
+                {
+                    "id": 7,
+                    "definition_id": 3,
+                    "child_id": 1,
+                    "title": "Brush teeth",
+                    "icon": "🦷",
+                    "window": "morning",
+                    "due_time": "08:00",
+                    "state": "completed",
+                    "overdue": False,
+                    "completed_at": f"{_today_iso()}T07:55:00+00:00",
+                    "on_time": True,
+                },
+                {
+                    "id": 8,
+                    "definition_id": 3,
+                    "child_id": 1,
+                    "title": "Pack bag",
+                    "icon": None,
+                    "window": "morning",
+                    "due_time": "10:00",
+                    "state": "open",
+                    "overdue": True,
+                    "completed_at": None,
+                    "on_time": None,
+                },
+            ],
+        },
+        {
+            "child_id": 2,
+            "child_name": "Bo",
+            "present": True,
+            "next_present": None,
+            "due_today": 1,
+            "completed_today": 1,
+            "remaining_today": 0,
+            "completion_pct": 100,
+            "instances": [
+                {
+                    "id": 9,
+                    "definition_id": 4,
+                    "child_id": 2,
+                    "title": "Make bed",
+                    "icon": "🛏",
+                    "window": "morning",
+                    "due_time": "07:30",
+                    "state": "completed",
+                    "overdue": False,
+                    "completed_at": f"{_today_iso()}T07:31:00+00:00",
+                    "on_time": True,
+                }
+            ],
+        },
+    ],
+}
+
+
+async def _wire_entry(hass, make_entry, client, options=None):
+    """Build an entry whose coordinator uses the scripted API client.
+
+    The client is registered with the conftest override BEFORE setup:
+    the first refresh at setup already polls the snapshot, so the
+    platform setup sees real data and creates the whole entity set.
+    """
+    from conftest import set_coordinator_client
+
+    entry = wire_entry_to_registry(
+        make_entry(
+            options=options,
+            data={"api_base_url": "http://api.test:8000", "panel_token": "tok"},
+        ),
+        hass.registry,
     )
-    end = (
+    set_coordinator_client(entry, client)
-        datetime.date.today() + datetime.timedelta(days=DEFAULT_HORIZON_DAYS)
-    ).isoformat()
-    await materialize(database, today, end, today=datetime.date.today())
-
-
-async def test_coordinator_snapshot_counts_and_states(hass, make_entry) -> None:
-    entry = wire_entry_to_registry(make_entry(), hass.registry)
     assert await async_setup_entry(hass, entry) is True
     coordinator = entry.runtime_data.coordinator
-    assert coordinator is not None
+    assert coordinator.api_client is client
-    database = entry.runtime_data.database
+    return entry, coordinator
-    await _seed_day(database)
 
+
+async def test_coordinator_builds_snapshot_from_fixture(hass, make_entry) -> None:
+    """One refresh = one snapshot GET; the fixture's values land on the
+    snapshot the entities read, per child and household-wide."""
+    client = StubSnapshotClient(FIXTURE_SNAPSHOT)
+    _entry, coordinator = await _wire_entry(hass, make_entry, client)
+    calls_after_setup = client.calls
     snapshot = await coordinator._async_update_data()
+
+    assert client.calls == calls_after_setup + 1
     assert snapshot.today_iso == _today_iso()
-    assert len(snapshot.children) == 2, "inactive child must be absent"
+    assert snapshot.cycle_day == 3
-    by_name = {child.child_name: child for child in snapshot.children}
+    assert len(snapshot.children) == 2
-    ada = by_name["Ada"]
+
-    assert ada.due_today == 1
+    ada, bo = snapshot.children
-    assert ada.completed_today == 0
+    assert (ada.child_id, ada.child_name) == (1, "Ada")
+    assert ada.due_today == 2
+    assert ada.completed_today == 1
     assert ada.remaining_today == 1
-    assert ada.completion_pct == 0
+    assert ada.completion_pct == 50
-    instance = ada.instances[0]
+    assert ada.present is False
-    assert instance.state == "open"
+    assert ada.next_present == "2026-09-24"
-    assert instance.overdue is False
+
-    assert instance.title == "Brush teeth"
+    done_view, open_view = ada.instances
-    assert instance.window == "morning"
+    assert done_view.instance_id == 7
-    assert ada.present is True, "no schedule means present every day"
+    assert done_view.title == "Brush teeth"
-    assert ada.next_present is None, "present child has no returns date"
+    assert done_view.icon == "🦷"
-    assert snapshot.cycle_day == 0, "no schedule anywhere"
+    assert done_view.window == "morning"
+    assert done_view.due_time == "08:00"
+    assert done_view.due_date == _today_iso()
+    # The API's "completed" maps back to the core view's "done".
+    assert done_view.state == "done"
+    assert done_view.overdue is False
+    assert done_view.completed_at == f"{_today_iso()}T07:55:00+00:00"
+    assert done_view.was_on_time is True
+
+    assert open_view.state == "open"
+    assert open_view.overdue is True
+    assert open_view.completed_at is None
+    assert open_view.was_on_time is None
+
+    assert (bo.child_id, bo.child_name) == (2, "Bo")
+    assert bo.due_today == 1
+    assert bo.completed_today == 1
+    assert bo.remaining_today == 0
+    assert bo.completion_pct == 100
+    assert bo.present is True
+    assert bo.next_present is None
+
+    # Household rollups: the sums the household sensors compute.
+    assert sum(child.due_today for child in snapshot.children) == 3
+    assert sum(child.completed_today for child in snapshot.children) == 2
+    assert sum(child.remaining_today for child in snapshot.children) == 1
 
 
-async def test_coordinator_next_present_for_away_child(
+async def test_coordinator_refresh_publishes_to_listeners(hass, make_entry) -> None:
-    hass, make_entry
+    """The refresh pass notifies listeners and reports success — the
-) -> None:
+    ONE refresh pass every entity reads from."""
-    """A child absent today carries the ISO date of their next present
+    client = StubSnapshotClient(FIXTURE_SNAPSHOT)
-    day; an always-absent pattern carries None instead."""
+    _entry, coordinator = await _wire_entry(hass, make_entry, client)
-    entry = wire_entry_to_registry(make_entry(), hass.registry)
+    called = []
-    assert await async_setup_entry(hass, entry) is True
+    coordinator.async_add_listener(lambda: called.append("entity"))
-    coordinator = entry.runtime_data.coordinator
+    await coordinator.async_refresh()
-    database = entry.runtime_data.database
+    assert called == ["entity"]
-    await create_child(database, "Ada")
+    assert coordinator.last_update_success is True
-    ada = (await list_children(database))[0]
+    assert coordinator.data is not None
-    today = datetime.date.today()
+    assert coordinator.data.children[0].child_name == "Ada"
-    pattern = ",".join(
+
-        str(day) for day in range(7) if day != today.weekday()
+
-    )
+async def test_instance_payload_shape_is_unchanged(hass, make_entry) -> None:
-    await PresenceSchedulesDao(database).upsert_by_child(
+    """The per-child instance attribute payload the due-today sensor
-        ada.id, 1, today.isoformat(), pattern
+    emits keeps its exact documented shape, and admin_instances now
-    )
+    equals instances (the API omits missed rows — D-009; the admin
-    await create_child(database, "Bo")
+    surface for missed quests is the PWA)."""
-    bo = (await list_children(database))[1]
+    client = StubSnapshotClient(FIXTURE_SNAPSHOT)
-    await PresenceSchedulesDao(database).upsert_by_child(
+    _entry, coordinator = await _wire_entry(hass, make_entry, client)
-        bo.id, 1, today.isoformat(), ""
-    )
     snapshot = await coordinator._async_update_data()
-    by_name = {child.child_name: child for child in snapshot.children}
+    ada = snapshot.children[0]
-    assert by_name["Ada"].present is False
+
-    assert by_name["Ada"].next_present == (
+    panel = instance_payload(ada.instances, include_missed=False)
-        today + datetime.timedelta(days=1)
+    assert panel == [
-    ).isoformat()
+        {
-    assert by_name["Bo"].present is False
+            "id": 7,
-    assert by_name["Bo"].next_present is None
+            "definition_id": 3,
+            "child_id": 1,
+            "title": "Brush teeth",
+            "icon": "🦷",
+            "window": "morning",
+            "due_time": "08:00",
+            "state": "completed",
+            "overdue": False,
+            "completed_at": f"{_today_iso()}T07:55:00+00:00",
+            "on_time": True,
+        },
+        {
+            "id": 8,
+            "definition_id": 3,
+            "child_id": 1,
+            "title": "Pack bag",
+            "icon": None,
+            "window": "morning",
+            "due_time": "10:00",
+            "state": "open",
+            "overdue": True,
+            "completed_at": None,
+            "on_time": None,
+        },
+    ]
+    # The API payload omits missed rows, so include_missed cannot
+    # re-add them: admin_instances == instances, byte for byte.
+    assert instance_payload(ada.instances, include_missed=True) == panel
 
 
-async def test_coordinator_reflects_completion_and_uncompletion(
+async def test_coordinator_entities_read_the_api_snapshot(hass, make_entry) -> None:
-    hass, make_entry
+    """The registered sensors and binary sensors read the rebuilt
-) -> None:
+    snapshot: entity values come from the fixture, with the SAME
-    entry = wire_entry_to_registry(make_entry(), hass.registry)
+    entity_ids as ever."""
-    assert await async_setup_entry(hass, entry) is True
+    client = StubSnapshotClient(FIXTURE_SNAPSHOT)
-    coordinator = entry.runtime_data.coordinator
+    entry, coordinator = await _wire_entry(hass, make_entry, client)
-    database = entry.runtime_data.database
+    await coordinator.async_refresh()
-    await _seed_day(database)
+
-    ada = (
+    ada_due = hass.entities[f"{DOMAIN}_child_1_quests_due_today"]
+    assert ada_due.native_value == 2
+    attributes = ada_due.extra_state_attributes
+    assert attributes["child_id"] == 1
+    assert attributes["child_name"] == "Ada"
+    assert attributes["present"] is False
+    assert [row["id"] for row in attributes["instances"]] == [7, 8]
+
+    assert (
+        hass.entities[f"{DOMAIN}_child_1_quests_completed_today"].native_value
+        == 1
+    )
+    assert (
+        hass.entities[f"{DOMAIN}_child_1_quests_remaining_today"].native_value
+        == 1
+    )
+    assert (
+        hass.entities[f"{DOMAIN}_child_1_completion_pct_today"].native_value
+        == 50
+    )
+    assert hass.entities[f"{DOMAIN}_child_1_next_quest"].native_value == (
+        "Pack bag"
+    )
+    # Ada owes one still: not all done; away today.
+    assert hass.entities[f"{DOMAIN}_child_1_all_done"].is_on is False
+    assert hass.entities[f"{DOMAIN}_child_1_present_today"].is_on is False
+
+    assert (
+        hass.entities[f"{DOMAIN}_child_2_completion_pct_today"].native_value
+        == 100
+    )
+    # Bo had a quest and owes none of it: all done, present today.
+    assert hass.entities[f"{DOMAIN}_child_2_all_done"].is_on is True
+    assert hass.entities[f"{DOMAIN}_child_2_present_today"].is_on is True
+    bo_attributes = hass.entities[
+        f"{DOMAIN}_child_2_present_today"
+    ].extra_state_attributes
+    assert bo_attributes["next_present"] is None
+
+    assert (
+        hass.entities[f"{DOMAIN}_household_quests_due_today"].native_value == 3
+    )
+    assert (
+        hass.entities[f"{DOMAIN}_household_quests_completed_today"].native_value
+        == 2
+    )
+    cycle = hass.entities[f"{DOMAIN}_cycle_day"]
+    assert cycle.native_value == 3
+    assert cycle.extra_state_attributes == {
+        "cycle_day": 3,
+        "today": _today_iso(),
+    }
+
+    await async_unload_entry(hass, entry)
+
+
+async def test_coordinator_failed_fetch_marks_unavailable(hass, make_entry) -> None:
+    """A failed fetch (transport failure or non-2xx — both typed) is a
+    FAILED refresh pass: the update pass converts the typed error to
+    UpdateFailed (the exception HA's coordinator machinery records as a
+    failed pass, chaining the typed error as ``__cause__``), the refresh
+    records the failure instead of raising, and data keeps its last
+    good value."""
+    error = NestQuestApiError(
+        "NestQuest API GET /api/v1/panel/snapshot failed with HTTP 503",
+        status=503,
+    )
+    # The stub serves the good snapshot for setup's first refresh, then
+    # the scripted failure.
+    client = StubSnapshotClient(FIXTURE_SNAPSHOT, error)
+    _entry, coordinator = await _wire_entry(hass, make_entry, client)
+    # First refresh (at setup) served the good snapshot.
+    assert coordinator.data is not None
+    assert coordinator.last_update_success is True
+
+    # The next pass hits the scripted failure: the update pass converts
+    # the typed error to UpdateFailed, with the typed error chained.
+    with pytest.raises(UpdateFailed, match="HTTP 503") as excinfo:
         await coordinator._async_update_data()
-    ).children[0]
+    assert isinstance(excinfo.value.__cause__, NestQuestApiError)
-    instance_id = ada.instances[0].instance_id
 
-    await complete_instance(
+    # The refresh pass records the failure the HA way — it does NOT
-        database,
+    # raise: last_update_success goes False, data keeps the last GOOD
-        instance_id,
+    # snapshot, and the failed pass lands on last_exception as the
-        actor_source="panel",
+    # converted UpdateFailed with the typed error chained.
-        actor_child_id=ada.child_id,
+    await coordinator.async_refresh()
-    )
+    assert coordinator.last_update_success is False
-    snapshot = await coordinator._async_update_data()
+    assert coordinator.data.children[0].child_name == "Ada"
-    child = snapshot.children[0]
+    assert isinstance(coordinator.last_exception, UpdateFailed)
-    assert child.completed_today == 1
+    assert isinstance(coordinator.last_exception.__cause__, NestQuestApiError)
-    assert child.remaining_today == 0
-    assert child.completion_pct == 100
-    view = child.instances[0]
-    assert view.state == "done"
-    assert view.completed_at is not None
-    assert view.was_on_time is not None
-
-    from custom_components.nestquest.completion import uncomplete_instance
-
-    await uncomplete_instance(
-        database, instance_id, actor_source="user", actor_user_id="admin-1"
-    )
-    snapshot = await coordinator._async_update_data()
-    child = snapshot.children[0]
-    assert child.completed_today == 0
-    assert child.remaining_today == 1
-    assert child.instances[0].state == "open"
 
 
-async def test_coordinator_zero_quests_reports_full_percentage(
+async def test_failed_first_refresh_does_not_crash_setup(
     hass, make_entry
 ) -> None:
+    """A failed fetch at the FIRST refresh (at setup) does not crash
+    setup: the coordinator reports last_update_success=False, the setup
+    catches the ConfigEntryNotReady that HA's real first refresh raises
+    for a failed pass, and the platforms still forward so the entities
+    come up unavailable."""
+    # The stub fails every snapshot fetch (HTTP 503), including setup's
+    # first refresh.
+    client = StubSnapshotClient(
+        NestQuestApiError(
+            "NestQuest API GET /api/v1/panel/snapshot failed with HTTP 503",
+            status=503,
+        )
+    )
+    entry = wire_entry_to_registry(
+        make_entry(
+            data={"api_base_url": "http://api.test:8000", "panel_token": "tok"}
+        ),
+        hass.registry,
+    )
+    set_coordinator_client(entry, client)
+    assert await async_setup_entry(hass, entry) is True
+    coordinator = entry.runtime_data.coordinator
+    assert coordinator.api_client is client
+    # The failed first pass: the entities' surface stays unavailable.
+    assert coordinator.last_update_success is False
+    assert coordinator.data is None
+    # Setup survived the failed first refresh — the platforms were
+    # forwarded (the regression the review caught: a first-refresh
+    # failure used to abort setup before any entity was created).
+    assert hass.config_entries.forwarded_platforms == [
+        "sensor",
+        "binary_sensor",
+    ]
+    # The unload after the failed-first-refresh setup stays clean.
+    assert await async_unload_entry(hass, entry) is True
+
+
+async def test_coordinator_unconfigured_api_does_not_crash_setup(
+    hass, make_entry, monkeypatch
+) -> None:
+    """An entry with no panel token set up fine: the coordinator gets
+    the typed-error stub, the first refresh fails cleanly (unavailable
+    entities), and setup does NOT crash."""
+    # Restore the production factory: the conftest default would serve
+    # the local snapshot, but this test exercises the token guard.
+    import custom_components.nestquest as nq
+
+    monkeypatch.setattr(
+        nq, "coordinator_client_from_entry", coordinator_client_from_entry
+    )
     entry = wire_entry_to_registry(make_entry(), hass.registry)
     assert await async_setup_entry(hass, entry) is True
     coordinator = entry.runtime_data.coordinator
-    await create_child(entry.runtime_data.database, "Ada")
+    from custom_components.nestquest.coordinator import _UnconfiguredApiClient
-    snapshot = await coordinator._async_update_data()
+
-    child = snapshot.children[0]
+    assert isinstance(coordinator.api_client, _UnconfiguredApiClient)
-    assert child.due_today == 0
+    # A real NestQuestApiClient was never built (it would ValueError).
-    assert child.completion_pct == 100, "zero quests is a perfect day"
+    assert not isinstance(coordinator.api_client, NestQuestApiClient)
-    assert child.instances == ()
+    # The first refresh failed the HA way: the update pass converted
+    # the stub's typed error to UpdateFailed, the first refresh raised
+    # ConfigEntryNotReady, and setup tolerated it — platforms forwarded
+    # so the entities come up unavailable.
+    assert coordinator.last_update_success is False
+    assert coordinator.data is None
+    with pytest.raises(UpdateFailed, match="not configured"):
+        await coordinator._async_update_data()
+    assert hass.config_entries.forwarded_platforms == [
+        "sensor",
+        "binary_sensor",
+    ]
+    assert await async_unload_entry(hass, entry) is True
 
 
-async def test_coordinator_cycle_day_from_first_scheduled_child(
+async def test_coordinator_configured_token_builds_real_client(
     hass, make_entry
 ) -> None:
-    entry = wire_entry_to_registry(make_entry(), hass.registry)
+    """An entry WITH a panel token builds the real client through the
+    production factory (a stub transport injected via session_factory,
+    the mock-only harness having no aiohttp module to hand in), and the
+    coordinator's refresh polls the snapshot route end to end."""
+    import json as json_module
+
+    class _OneShotSession:
+        """A session-shaped stub answering every request with the fixture."""
+
+        def __init__(self, payload) -> None:
+            self._body = json_module.dumps(payload)
+
+        def request(self, method, url, **kwargs):
+            body = self._body
+
+            class _Response:
+                def __init__(self) -> None:
+                    self.status = 200
+
+                async def text(self):
+                    return body
+
+            class _Context:
+                async def __aenter__(self):
+                    return _Response()
+
+                async def __aexit__(self, *exc):
+                    return False
+
+            return _Context()
+
+    def _session_factory(_hass):
+        return _OneShotSession(FIXTURE_SNAPSHOT)
+
+    entry = wire_entry_to_registry(
+        make_entry(
+            data={"api_base_url": "http://api.test:8000", "panel_token": "tok"}
+        ),
+        hass.registry,
+    )
+    real_client = coordinator_client_from_entry(
+        hass, entry, session_factory=_session_factory
+    )
+    assert isinstance(real_client, NestQuestApiClient)
+
+    set_coordinator_client(entry, real_client)
     assert await async_setup_entry(hass, entry) is True
     coordinator = entry.runtime_data.coordinator
-    database = entry.runtime_data.database
+    assert isinstance(coordinator.api_client, NestQuestApiClient)
-    await create_child(database, "Ada")
+    # The real client polls the stub transport; the snapshot lands.
-    ada = (await list_children(database))[0]
+    await coordinator.async_refresh()
-    today = datetime.date.today()
+    assert coordinator.data.children[0].child_name == "Ada"
-    # A 2-week cycle anchored 3 days ago: today is day 4.
-    anchor = today - datetime.timedelta(days=3)
-    await PresenceSchedulesDao(database).upsert_by_child(
-        ada.id, 2, anchor.isoformat(), "0,1,2,3,4,5,6|"
-    )
-    snapshot = await coordinator._async_update_data()
-    assert snapshot.cycle_day == 4
-
-
-async def test_coordinator_presence_resolves_schedule_and_override(
-    hass, make_entry
-) -> None:
-    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
-    await create_child(database, "Ada")
-    ada = (await list_children(database))[0]
-    today = datetime.date.today()
-    # Scheduled absent today (empty weekday set for this cycle week).
-    await PresenceSchedulesDao(database).upsert_by_child(
-        ada.id, 1, today.isoformat(), ""
-    )
-    snapshot = await coordinator._async_update_data()
-    assert snapshot.children[0].present is False
-
-    # An override beats the pattern.
-    from custom_components.nestquest.dao_presence import (
-        PresenceOverridesDao,
-    )
-
-    await PresenceOverridesDao(database).create(
-        ada.id, today.isoformat(), today.isoformat(), True
-    )
-    snapshot = await coordinator._async_update_data()
-    assert snapshot.children[0].present is True
-
-
-async def test_coordinator_next_present_honours_future_override(
-    hass, make_entry
-) -> None:
-    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
-    await create_child(database, "Ada")
-    ada = (await list_children(database))[0]
-    today = datetime.date.today()
-    tomorrow = today + datetime.timedelta(days=1)
-    # Scheduled present tomorrow (single-day cycle, weekday set set below
-    # covers every day), but overridden absent tomorrow.
-    await PresenceSchedulesDao(database).upsert_by_child(
-        ada.id, 1, today.isoformat(), "0,1,2,3,4,5,6"
-    )
-    from custom_components.nestquest.dao_presence import (
-        PresenceOverridesDao,
-    )
-
-    await PresenceOverridesDao(database).create(
-        ada.id, tomorrow.isoformat(), tomorrow.isoformat(), False
-    )
-    snapshot = await coordinator._async_update_data()
-    child = snapshot.children[0]
-    assert child.present is True
-    assert child.next_present != tomorrow.isoformat(), (
-        "a future absent override must remove tomorrow from the preview"
-    )
-
-
-async def test_coordinator_derives_missed_for_past_due_open(
-    hass, make_entry
-) -> None:
-    entry = wire_entry_to_registry(make_entry(), hass.registry)
-    assert await async_setup_entry(hass, entry) is True
-    coordinator = entry.runtime_data.coordinator
-    snapshot = await coordinator._async_update_data()
-    # No children: empty but valid snapshot.
-    assert snapshot.children == ()
 
 
 async def test_setup_creates_coordinator_with_configured_interval(
@@ -270,12 +519,9 @@ async def test_setup_creates_coordinator_with_configured_interval(
     coordinator = entry.runtime_data.coordinator
     assert isinstance(coordinator, NestQuestCoordinator)
     assert coordinator.update_interval == datetime.timedelta(seconds=60)
-    assert coordinator.data is not None, "first refresh ran at setup"
 
 
-async def test_setup_defaults_to_five_minute_interval(
+async def test_setup_defaults_to_five_minute_interval(hass, make_entry) -> None:
-    hass, make_entry
-) -> None:
     entry = wire_entry_to_registry(make_entry(), hass.registry)
     assert await async_setup_entry(hass, entry) is True
     coordinator = entry.runtime_data.coordinator
@@ -298,24 +544,23 @@ async def test_unload_shuts_coordinator_down(hass, make_entry) -> None:
     assert torn_down == [coordinator]
 
 
-async def test_coordinator_notifies_listeners_on_refresh(
+async def test_setup_failure_after_registration_shuts_coordinator_down(
-    hass, make_entry
-) -> None:
-    entry = wire_entry_to_registry(make_entry(), hass.registry)
-    assert await async_setup_entry(hass, entry) is True
-    coordinator = entry.runtime_data.coordinator
-    called = []
-    coordinator.async_add_listener(lambda: called.append("entity"))
-    await coordinator.async_refresh()
-    assert called == ["entity"]
-    assert coordinator.last_update_success is True
-
-
-async def test_failed_first_refresh_shuts_coordinator_down(
     hass, make_entry, monkeypatch
 ) -> None:
     """A setup failure after coordinator creation must not leak its
-    listeners against the database the except path closes."""
+    listeners or leave a stale runtime record behind.
+
+    The runtime record registers in ``hass.data`` BEFORE the platform
+    forward, so a forward failure reaches the unwind with the record
+    already registered.  The unwind must shut the coordinator down,
+    clear ``entry.runtime_data``, AND remove the stale ``hass.data``
+    entry — otherwise a later unload picks up the stale, already-torn-
+    down record and re-invokes the already-consumed update-listener
+    remover, which raises.  (A failed first refresh is NOT such a
+    failure: HA's real first refresh wraps it in ConfigEntryNotReady,
+    which setup deliberately tolerates so entities come up
+    unavailable.)
+    """
     from custom_components.nestquest import async_setup_entry
     import custom_components.nestquest.coordinator as coordinator_module
 
@@ -330,17 +575,37 @@ async def test_failed_first_refresh_shuts_coordinator_down(
         _record_shutdown,
     )
 
-    async def _boom(self):
+    async def _boom(entry, platforms):
-        raise RuntimeError("refresh failed")
+        raise RuntimeError("platform forward failed")
 
     monkeypatch.setattr(
-        coordinator_module.NestQuestCoordinator,
+        hass.config_entries,
-        "_async_update_data",
+        "async_forward_entry_setups",
         _boom,
     )
 
     entry = wire_entry_to_registry(make_entry(), hass.registry)
-    with pytest.raises(RuntimeError, match="refresh failed"):
+    with pytest.raises(RuntimeError, match="platform forward failed"):
         await async_setup_entry(hass, entry)
     assert shutdown_calls == [entry.entry_id]
     assert getattr(entry, "runtime_data", None) is None
+    # The stale hass.data registration was removed with the runtime
+    # record: the unload below finds nothing stale, must not raise,
+    # and must leave no residue behind.
+    assert await async_unload_entry(hass, entry) is True
+    assert DOMAIN not in hass.data
+    assert hass.data == {}
+    assert hass.registry.size == 0
+    assert hass.time_change.size == 0
+    assert entry.runtime_data is None
+
+
+async def test_snapshot_from_api_payload_empty_household() -> None:
+    """An empty household (no children) is a valid snapshot: empty
+    children tuple, cycle day passes through."""
+    snapshot = snapshot_from_api_payload(
+        {"today_iso": "2026-09-23", "cycle_day": 0, "children": []}
+    )
+    assert snapshot.children == ()
+    assert snapshot.cycle_day == 0
+    assert snapshot.today_iso == "2026-09-23"

tests/test_core_snapshot.py

@@ -123,9 +123,11 @@ async def test_build_snapshot_explicit_values(hass, make_entry, monkeypatch) ->
     database = entry.runtime_data.database
     settings = entry.runtime_data.settings
 
-    # Single clock read for the seed/snapshot anchor (no midnight skew).
+    # Single clock read for the seed/snapshot anchor (no midnight skew);
-    today = coordinator._local_now().date()
+    # the coordinator no longer owns a clock read (the API serves the
+    # snapshot), so the test resolves the HA-local zone directly.
     time_zone = ZoneInfo(hass.config.time_zone)
+    today = datetime.datetime.now(time_zone).date()
     # 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(
@@ -294,13 +296,38 @@ async def test_build_snapshot_explicit_values(hass, make_entry, monkeypatch) ->
         == "missed"
     )
 
-    # REQUIRED wrapper tie-in: the coordinator's _async_update_data is a
+    # REQUIRED wrapper tie-in: since Feature 18 the coordinator's
-    # thin WRAPPER over build_snapshot, so pinning its clock to the SAME
+    # refresh polls the snapshot through its API client, so pinning the
-    # pinned_now the builder was called with must reproduce the builder's
+    # client to serve the payload of the SAME snapshot built with
-    # snapshot byte-for-byte.  The dataclasses are frozen, so == is a full
+    # pinned_now must reproduce the builder's snapshot byte-for-byte.
-    # deep compare — a wrapper that mis-passed today vs now, or stopped
+    # The dataclasses are frozen, so == is a full deep compare — a
-    # calling the builder, would diverge here even though the literal
+    # reconstruction that dropped or reshaped a field would diverge
-    # assertions above still pass.
+    # here even though the literal assertions above still pass.
-    monkeypatch.setattr(coordinator, "_local_now", lambda: pinned_now)
+    from conftest import set_coordinator_client
+
+    class _PinnedPayloadClient:
+        async def get_snapshot(self):
+            return {
+                "today_iso": built.today_iso,
+                "cycle_day": built.cycle_day,
+                "children": [
+                    {
+                        "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": instance_payload(
+                            child.instances, include_missed=False
+                        ),
+                    }
+                    for child in built.children
+                ],
+            }
+
+    set_coordinator_client(entry, _PinnedPayloadClient())
     await coordinator.async_refresh()
     assert coordinator.data == built

tests/test_entity_matrix.py

@@ -180,9 +180,12 @@ async def test_entity_and_event_matrix(hass, make_entry) -> None:
     assert _sensor(hass, ada.id, "all_done").is_on is False
     assert _sensor(hass, ada.id, "quests_remaining_today").native_value == 1
 
-    # --- The instances payload omits missed instances while
+    # --- The instances payload omits missed instances (D-009) — and
-    # admin_instances includes them (D-009): pin the derive clock
+    # since Feature 18 the snapshot comes from the API panel route,
-    # forward in the TEST ONLY so today's open instances read missed.
+    # which omits missed rows ENTIRELY, so admin_instances cannot
+    # carry them either (the admin surface for missed quests is the
+    # PWA, reading the API directly): pin the derive clock forward in
+    # the TEST ONLY so today's open instances read missed.
     original_refresh = coordinator._async_update_data
 
     async def _future_dated_refresh():
@@ -203,9 +206,9 @@ async def test_entity_and_event_matrix(hass, make_entry) -> None:
     await coordinator.async_refresh()
     attributes = _sensor(hass, bo.id, "quests_due_today").extra_state_attributes
     assert attributes["instances"] == [], "panel payload omits missed"
-    assert len(attributes["admin_instances"]) == 2
+    assert attributes["admin_instances"] == [], (
-    assert all(
+        "the API payload omits missed rows, so the admin payload "
-        entry["state"] == "missed" for entry in attributes["admin_instances"]
+        "carries none either (the PWA is the admin surface now)"
     )
     coordinator._async_update_data = original_refresh
     await coordinator.async_refresh()

tests/test_options_flow.py

@@ -15,22 +15,26 @@ from custom_components.nestquest import config_flow
 from custom_components.nestquest import options_flow
 from custom_components.nestquest.const import (
     CONF_ADMIN_USER_IDS,
-    CONF_DAY_ROLLOVER_TIME,
-    CONF_HORIZON_DAYS,
-    CONF_PANEL_IDLE_TIMEOUT,
-    CONF_UPDATE_INTERVAL,
     CONF_AFTERNOON_REMINDER_ENABLED,
     CONF_AFTERNOON_REMINDER_TIME,
+    CONF_API_BASE_URL,
     CONF_CELEBRATION_ENABLED,
+    CONF_DAY_ROLLOVER_TIME,
     CONF_END_OF_DAY_REPORT_ENABLED,
     CONF_END_OF_DAY_REPORT_TIME,
+    CONF_HORIZON_DAYS,
     CONF_MORNING_SUMMARY_ENABLED,
     CONF_MORNING_SUMMARY_TIME,
     CONF_NOTIFY_TARGET,
+    CONF_PANEL_IDLE_TIMEOUT,
+    CONF_PANEL_TOKEN,
+    CONF_UPDATE_INTERVAL,
+    DEFAULT_API_BASE_URL,
     DEFAULT_AFTERNOON_REMINDER_TIME,
     DEFAULT_AUTOMATION_ENABLED,
     DEFAULT_END_OF_DAY_REPORT_TIME,
     DEFAULT_MORNING_SUMMARY_TIME,
+    DEFAULT_PANEL_TOKEN,
     DEFAULT_UPDATE_INTERVAL,
     DEFAULT_DAY_ROLLOVER_TIME,
     DEFAULT_HORIZON_DAYS,
@@ -49,8 +53,9 @@ VALID_INPUT = {
 }
 
 #: What a schema-validated submit produces from a bare VALID_INPUT on a
-#: fresh entry: the Feature 10 refresh interval and the Feature 11
+#: fresh entry: the Feature 10 refresh interval, the Feature 11
-#: notification section fill from their defaults.
+#: notification section, and the Feature 18 API connection fill from
+#: their defaults.
 FULL_INPUT = {
     **VALID_INPUT,
     CONF_UPDATE_INTERVAL: DEFAULT_UPDATE_INTERVAL,
@@ -62,6 +67,8 @@ FULL_INPUT = {
     CONF_AFTERNOON_REMINDER_ENABLED: DEFAULT_AUTOMATION_ENABLED,
     CONF_END_OF_DAY_REPORT_ENABLED: DEFAULT_AUTOMATION_ENABLED,
     CONF_CELEBRATION_ENABLED: DEFAULT_AUTOMATION_ENABLED,
+    CONF_API_BASE_URL: DEFAULT_API_BASE_URL,
+    CONF_PANEL_TOKEN: DEFAULT_PANEL_TOKEN,
 }
 
 
@@ -259,6 +266,8 @@ def test_options_flow_schema_applies_stored_defaults() -> None:
         CONF_AFTERNOON_REMINDER_ENABLED: True,
         CONF_END_OF_DAY_REPORT_ENABLED: True,
         CONF_CELEBRATION_ENABLED: True,
+        CONF_API_BASE_URL: DEFAULT_API_BASE_URL,
+        CONF_PANEL_TOKEN: DEFAULT_PANEL_TOKEN,
         CONF_ADMIN_USER_IDS: [],
     }
 
@@ -272,6 +281,9 @@ def test_options_flow_schema_coerces_field_types() -> None:
     assert validators[CONF_HORIZON_DAYS] is int
     assert validators[CONF_DAY_ROLLOVER_TIME] is str
     assert validators[CONF_PANEL_IDLE_TIMEOUT] is int
+    # The Feature 18 API connection fields are free-text strings.
+    assert validators[CONF_API_BASE_URL] is str
+    assert validators[CONF_PANEL_TOKEN] is str
 
 
 @pytest.mark.parametrize("bad_horizon", [0, -5])
@@ -326,8 +338,10 @@ def test_options_flow_rejects_invalid_idle_timeout(bad_timeout) -> None:
     assert result["errors"] == {CONF_PANEL_IDLE_TIMEOUT: "invalid"}
 
 
-def test_options_flow_creates_entry_with_four_keys() -> None:
+def test_options_flow_creates_entry_with_validated_keys() -> None:
-    """A valid submit creates an entry with exactly the four validated keys."""
+    """A valid submit creates an entry with exactly the validated keys
+    (the legacy settings plus every later section's keys, filled from
+    defaults where the raw submit omitted them)."""
     flow = _make_flow(_make_entry())
     result = _run(flow.async_step_init(dict(VALID_INPUT)))
     assert result["type"] == "create_entry"
@@ -799,3 +813,91 @@ def test_options_flow_no_hard_coded_personal_target_in_source() -> None:
     assert const.__dict__.get("DEFAULT_NOTIFY_TARGET", "") == ""
     form = _run(_make_flow(_make_entry()).async_step_init(None))
     assert form["data_schema"]({})[CONF_NOTIFY_TARGET] == ""
+
+
+# ---------------------------------------------------------------------------
+# Panel-plane API connection section (Feature 18)
+# ---------------------------------------------------------------------------
+
+
+def test_options_flow_api_section_defaults_saved() -> None:
+    """A bare submit saves the API connection defaults: the const base
+    URL and an EMPTY panel token (no token is ever invented)."""
+    flow = _make_flow(_make_entry())
+    result = _run(flow.async_step_init(dict(VALID_INPUT)))
+    assert result["type"] == "create_entry"
+    assert result["data"][CONF_API_BASE_URL] == DEFAULT_API_BASE_URL
+    assert result["data"][CONF_PANEL_TOKEN] == ""
+
+
+def test_options_flow_api_base_url_must_be_full_http_url() -> None:
+    """Scheme-less, host-less, empty, and non-string base URLs reject
+    with a field error — the same rule the client's constructor
+    enforces, so a malformed URL fails at the form, not at the first
+    request."""
+    flow = _make_flow(_make_entry())
+    for bad_url in ("api.lan:8000", "http://", "", "   ", 42):
+        result = _run(
+            flow.async_step_init(
+                {**VALID_INPUT, CONF_API_BASE_URL: bad_url}
+            )
+        )
+        assert result["type"] == "form"
+        assert result["errors"] == {CONF_API_BASE_URL: "invalid_url"}
+
+
+def test_options_flow_panel_token_validation() -> None:
+    """The EMPTY token is valid (the API connection is simply not
+    configured yet); padded and non-string values reject — a padded
+    paste would fail the token check at request time instead."""
+    flow = _make_flow(_make_entry())
+    result = _run(
+        flow.async_step_init({**VALID_INPUT, CONF_PANEL_TOKEN: ""})
+    )
+    assert result["type"] == "create_entry"
+    assert result["data"][CONF_PANEL_TOKEN] == ""
+
+    for bad_token in ("  secret  ", "   ", 42):
+        flow = _make_flow(_make_entry())
+        result = _run(
+            flow.async_step_init(
+                {**VALID_INPUT, CONF_PANEL_TOKEN: bad_token}
+            )
+        )
+        assert result["type"] == "form"
+        assert result["errors"] == {CONF_PANEL_TOKEN: "invalid"}
+
+
+def test_options_flow_api_section_stored_values_round_trip() -> None:
+    """Stored API settings are the form's defaults and a raw legacy
+    submit without them keeps the stored values."""
+    entry = _make_entry(
+        options={
+            **VALID_INPUT,
+            CONF_API_BASE_URL: "http://panel.lan:8443",
+            CONF_PANEL_TOKEN: "stored-token",
+        }
+    )
+    flow = _make_flow(entry)
+    form = _run(flow.async_step_init(None))
+    schema_defaults = form["data_schema"]({})
+    assert schema_defaults[CONF_API_BASE_URL] == "http://panel.lan:8443"
+    assert schema_defaults[CONF_PANEL_TOKEN] == "stored-token"
+
+    # A raw submit carrying only the legacy keys keeps the stored
+    # API connection settings.
+    result = _run(flow.async_step_init(dict(VALID_INPUT)))
+    assert result["type"] == "create_entry"
+    assert result["data"][CONF_API_BASE_URL] == "http://panel.lan:8443"
+    assert result["data"][CONF_PANEL_TOKEN] == "stored-token"
+
+
+def test_options_flow_strings_cover_api_section(
+    strings: dict, translations_en: dict
+) -> None:
+    """Both string files name the API fields and the URL error."""
+    data = strings["options"]["step"]["init"]["data"]
+    for key in (CONF_API_BASE_URL, CONF_PANEL_TOKEN):
+        assert data.get(key), f"Missing field name for '{key}'"
+    assert strings["options"]["errors"]["invalid_url"]
+    assert translations_en["options"] == strings["options"]

tests/test_sensors.py

@@ -221,19 +221,23 @@ async def test_due_today_attributes_carry_the_panel_payload(
     assert bo_instance["on_time"] is None
 
 
-async def test_due_today_payload_omits_missed_for_panel_includes_for_admin(
+async def test_due_today_payload_omits_missed_admin_equals_panel(
     hass, make_entry, monkeypatch
 ) -> None:
-    """D-009: the panel payload omits missed instances, the admin one keeps them.
+    """D-009: the panel payload omits missed instances.
 
-    A missed instance is derived (``due_date`` before the snapshot's
+    Since Feature 18 the snapshot comes from the API panel route,
-    "today"), never stored.  The coordinator's day listing is keyed on
+    which omits missed rows entirely (the admin surface for missed
-    ``_local_now``'s date, so pinning the clock forward would move the
+    quests is the PWA, reading the API directly).  The entity payload
-    listing window off today's rows entirely; the pinned FUTURE clock
+    therefore cannot carry missed rows any more: ``admin_instances``
-    is therefore applied at the ``derive_state`` boundary instead —
+    equals ``instances`` — open and completed rows only.  A missed
-    today's open instances then read as missed through the same
+    instance is still derived locally (``due_date`` before the
-    derivation rule, without back-dating instance rows past the DAO's
+    snapshot's "today"); the pinned-FUTURE clock at the
-    no-past guard or touching append-only completion events.
+    ``derive_state`` boundary makes today's open instances read
+    missed through the same derivation rule, and this test pins the
+    change: neither payload carries them, and the count rollups
+    (precomputed server-side from the full day) are unaffected here
+    because the API counts arrive per the route's own build.
     """
     entry, coordinator, (ada, bo, cory) = await _setup_seeded_entry(
         hass, make_entry
@@ -256,10 +260,13 @@ async def test_due_today_payload_omits_missed_for_panel_includes_for_admin(
     assert attributes["instances"] == [], (
         "missed instances are omitted from the panel payload"
     )
-    [missed] = attributes["admin_instances"]
+    assert attributes["admin_instances"] == [], (
-    assert missed["state"] == "missed"
+        "the API payload omits missed rows, so the admin payload "
-    assert missed["completed_at"] is None
+        "carries none either (the PWA is the admin surface now)"
-    # Counts still roll the missed instance up as not-completed.
+    )
+    # Counts still roll the missed instance up as not-completed: the
+    # counts arrive precomputed from the route's own build, which rolls
+    # the FULL day (the missed row included) into the numbers.
     assert _sensor(hass, bo.id, "quests_completed_today").native_value == 0
     assert _sensor(hass, bo.id, "quests_remaining_today").native_value == 1
 

tests/test_sse_subscription.py

@@ -0,0 +1,619 @@
+"""Tests for the panel SSE subscription (HA-bus re-firing task).
+
+Three layers are exercised, all STUB-driven — no network, no aiohttp,
+no real Home Assistant:
+
+- The API client's ``stream_events`` (transport stub): what goes on
+  the wire (GET, the events URL, the Bearer header), SSE frame
+  parsing (``event:``/``data:`` lines, blank-line frame ends, chunk
+  boundaries, multi-line data), and every failure mode (non-2xx,
+  connection failure, mid-stream drop, malformed frame) surfacing as
+  the typed :class:`NestQuestApiError`; cancellation closes the
+  stream cleanly.
+- The :class:`~custom_components.nestquest.sse.NestQuestEventStream`
+  subscription manager (duck-typed stream client): the four documented
+  transition events (design/ENTITIES-AND-SERVICES.md §3, exactly as
+  ``tests/test_api_sse.py`` observes them on the API side) fire on
+  ``hass.bus`` with the documented payload fields; ``stop()`` cancels
+  the task so no frame fires afterwards; a stream failure reconnects
+  with a bounded, reset-on-success backoff and never crashes the
+  task.
+- The integration wiring: setup starts the stream alongside the
+  runtime record, unload tears it down (no firing afterwards, task
+  cancelled), and an entry whose client has no stream surface (the
+  unconfigured-token stub, the test snapshot client) starts none.
+"""
+from __future__ import annotations
+
+import asyncio
+import json
+
+import pytest
+
+from conftest import set_coordinator_client, wire_entry_to_registry
+
+import custom_components.nestquest as nq_integration
+from custom_components.nestquest import (
+    async_setup_entry,
+    async_unload_entry,
+)
+from custom_components.nestquest import sse as sse_module
+from custom_components.nestquest.api_client import (
+    NestQuestApiClient,
+    NestQuestApiError,
+)
+from custom_components.nestquest.sse import NestQuestEventStream
+
+BASE_URL = "http://api.test:8000"
+TOKEN = "test-panel-token"
+
+# Captured BEFORE any test patches asyncio.sleep (the reconnect test
+# swaps in a recording fake; this reference keeps helpers sleeping for
+# real regardless).
+_REAL_SLEEP = asyncio.sleep
+
+#: The four documented transition events with their documented payload
+#: fields — the exact shapes the API side streams (test_api_sse.py).
+DOCUMENTED_FRAMES = (
+    (
+        "nestquest_quest_completed",
+        {
+            "child_id": 1,
+            "child_name": "Ada",
+            "instance_id": 7,
+            "quest_title": "Pack bag",
+            "window": "morning",
+            "due_date": "2026-09-23",
+            "due_time": "10:00",
+            "occurred_at": "2026-09-23T12:00:00+00:00",
+            "was_on_time": False,
+        },
+    ),
+    (
+        "nestquest_quest_uncompleted",
+        {
+            "child_id": 1,
+            "child_name": "Ada",
+            "instance_id": 7,
+            "quest_title": "Pack bag",
+            "window": "morning",
+            "due_date": "2026-09-23",
+            "due_time": "10:00",
+            "occurred_at": "2026-09-23T12:00:00+00:00",
+        },
+    ),
+    (
+        "nestquest_quest_missed",
+        {
+            "child_id": 1,
+            "child_name": "Ada",
+            "instance_id": 9,
+            "quest_title": "Stale chore",
+            "window": "morning",
+            "due_date": "2026-09-23",
+            "due_time": "09:00",
+            "occurred_at": "2026-09-23T23:30:00+00:00",
+        },
+    ),
+    (
+        "nestquest_child_day_complete",
+        {
+            "child_id": 1,
+            "child_name": "Ada",
+            "quests_due": 1,
+            "quests_completed": 1,
+            "occurred_at": "2026-09-23T12:00:00+00:00",
+        },
+    ),
+)
+
+#: The documented payload field sets per event type.
+DOCUMENTED_FIELDS = {
+    "nestquest_quest_completed": {
+        "child_id",
+        "child_name",
+        "instance_id",
+        "quest_title",
+        "window",
+        "due_date",
+        "due_time",
+        "occurred_at",
+        "was_on_time",
+    },
+    "nestquest_quest_uncompleted": {
+        "child_id",
+        "child_name",
+        "instance_id",
+        "quest_title",
+        "window",
+        "due_date",
+        "due_time",
+        "occurred_at",
+    },
+    "nestquest_quest_missed": {
+        "child_id",
+        "child_name",
+        "instance_id",
+        "quest_title",
+        "window",
+        "due_date",
+        "due_time",
+        "occurred_at",
+    },
+    "nestquest_child_day_complete": {
+        "child_id",
+        "child_name",
+        "quests_due",
+        "quests_completed",
+        "occurred_at",
+    },
+}
+
+#: A minimal panel snapshot payload for the full-setup wiring tests
+#: (the shape ``snapshot_from_api_payload`` consumes; no children).
+MINIMAL_SNAPSHOT = {"today_iso": "2026-09-23", "cycle_day": 1, "children": []}
+
+
+# ---------------------------------------------------------------------------
+# Stub transports and stream clients (no network anywhere)
+# ---------------------------------------------------------------------------
+
+
+class _StreamingResponse:
+    """A fake aiohttp streaming response: ``status`` + chunk iteration.
+
+    ``chunks`` are the body pieces the stream yields in order; a chunk
+    may be an exception, raised when the iteration reaches it (a
+    mid-stream transport drop).  ``hold_open`` keeps the stream open
+    forever after the chunks run out, for cancellation tests.
+    """
+
+    def __init__(self, status: int, chunks, *, hold_open: bool = False) -> None:
+        self.status = status
+        self._chunks = list(chunks)
+        self._hold_open = hold_open
+        self.closed = False
+
+    def __aiter__(self):
+        return self._iterate()
+
+    async def _iterate(self):
+        for chunk in self._chunks:
+            if isinstance(chunk, BaseException):
+                raise chunk
+            yield chunk
+        if self._hold_open:
+            await asyncio.Event().wait()  # cancelled, never ends
+
+
+class _RequestContext:
+    """Async context manager mirroring ``session.request(...)``'s return.
+
+    Entering performs the request (a scripted transport error raises
+    THERE, where aiohttp's connector raises on a refused connection);
+    exiting marks the response closed, so tests can assert the stream
+    was released.
+    """
+
+    def __init__(self, response, error) -> None:
+        self._response = response
+        self._error = error
+
+    async def __aenter__(self):
+        if self._error is not None:
+            raise self._error
+        return self._response
+
+    async def __aexit__(self, exc_type, exc, tb) -> bool:
+        if self._response is not None:
+            self._response.closed = True
+        return False
+
+
+class _StreamTransport:
+    """A no-network stand-in for aiohttp's ``ClientSession``.
+
+    Records every ``request(...)`` call and answers from a scripted
+    queue of outcomes (a :class:`_StreamingResponse` or an exception
+    per attempt, consumed in order).
+    """
+
+    def __init__(self, *outcomes) -> None:
+        self.calls: list[dict] = []
+        self._outcomes = list(outcomes)
+
+    def request(self, method: str, url: str, **kwargs):
+        self.calls.append(
+            {
+                "method": method,
+                "url": url,
+                "headers": kwargs.get("headers"),
+                "json": kwargs.get("json"),
+            }
+        )
+        outcome = self._outcomes.pop(0)
+        if isinstance(outcome, BaseException):
+            return _RequestContext(None, outcome)
+        return _RequestContext(outcome, None)
+
+
+def make_stream_client(transport) -> NestQuestApiClient:
+    """Build a client over the streaming stub with the test credentials."""
+    return NestQuestApiClient(BASE_URL, TOKEN, transport)
+
+
+def _frame_text(event_type: str, payload: dict) -> str:
+    """One documented SSE frame: event line, data line, blank line."""
+    return f"event: {event_type}\ndata: {json.dumps(payload)}\n\n"
+
+
+class _ScriptedStreamClient:
+    """Duck-typed API client: one scripted stream per connection.
+
+    ``script`` holds generator FACTORIES — ``stream_events()`` pops the
+    next one and returns its async generator, so a test scripts exactly
+    what each connection attempt yields (or how it fails).
+    """
+
+    def __init__(self, script) -> None:
+        self._script = list(script)
+        self.calls = 0
+
+    def stream_events(self):
+        self.calls += 1
+        return self._script.pop(0)()
+
+
+def _frames(*frames):
+    """A live stream factory: yields ``frames``, then stays open."""
+
+    def _factory():
+        async def _gen():
+            for frame in frames:
+                yield frame
+            await asyncio.Event().wait()  # cancelled, never ends
+
+        return _gen()
+
+    return _factory
+
+
+def _fail(error: BaseException):
+    """A stream factory that fails before yielding anything."""
+
+    def _factory():
+        async def _gen():
+            raise error
+            yield  # noqa: unreachable — makes _gen an async generator
+
+        return _gen()
+
+    return _factory
+
+
+def _drop_after(error: BaseException, *frames):
+    """A stream factory that yields ``frames`` then drops mid-stream."""
+
+    def _factory():
+        async def _gen():
+            for frame in frames:
+                yield frame
+            raise error
+
+        return _gen()
+
+    return _factory
+
+
+async def _wait_until(predicate, *, timeout: float = 2.0) -> None:
+    """Poll ``predicate`` until true or the deadline passes."""
+    loop = asyncio.get_running_loop()
+    deadline = loop.time() + timeout
+    while not predicate():
+        if loop.time() > deadline:
+            raise AssertionError("condition not met within the deadline")
+        await _REAL_SLEEP(0.01)
+
+
+def make_manager(
+    hass, client, *, min_backoff: float = 0.01, max_backoff: float = 0.05
+) -> NestQuestEventStream:
+    """A manager with fast (test-sized) backoff bounds."""
+    return NestQuestEventStream(
+        hass,
+        client,
+        reconnect_min_seconds=min_backoff,
+        reconnect_max_seconds=max_backoff,
+    )
+
+
+# ---------------------------------------------------------------------------
+# The client's stream_events: parsing and typed failures
+# ---------------------------------------------------------------------------
+
+
+async def test_stream_events_yields_documented_frames_and_sends_bearer() -> None:
+    """All four documented frames parse, with the Bearer GET on the wire."""
+    body = "".join(
+        _frame_text(event_type, payload)
+        for event_type, payload in DOCUMENTED_FRAMES
+    )
+    transport = _StreamTransport(_StreamingResponse(200, [body]))
+    client = make_stream_client(transport)
+
+    frames = [frame async for frame in client.stream_events()]
+
+    assert frames == list(DOCUMENTED_FRAMES)
+    call = transport.calls[0]
+    assert call["method"] == "GET"
+    assert call["url"] == f"{BASE_URL}/api/v1/panel/events"
+    assert call["headers"] == {"Authorization": f"Bearer {TOKEN}"}
+    assert call["json"] is None
+
+
+async def test_stream_events_buffers_lines_across_chunks_and_joins_data() -> None:
+    """Chunk-split lines parse, and multi-line data joins per SSE rules."""
+    first_type, first_payload = DOCUMENTED_FRAMES[0]
+    # The first frame is cut MID-LINE across two chunks; the second
+    # frame spreads its JSON over two ``data:`` lines.
+    chunks = [
+        f"event: {first_type}\nda",
+        f"ta: {json.dumps(first_payload)}\n\nevent: "
+        f"{DOCUMENTED_FRAMES[1][0]}\n",
+        f'data: {{"extra":\ndata: true}}\n\n',
+    ]
+    transport = _StreamTransport(_StreamingResponse(200, chunks))
+    client = make_stream_client(transport)
+
+    frames = [frame async for frame in client.stream_events()]
+
+    assert len(frames) == 2
+    assert frames[0] == DOCUMENTED_FRAMES[0]
+    # The split-out frame's data spread over two ``data:`` lines joins
+    # with a newline (the SSE standard) and parses as one JSON body.
+    assert frames[1] == (DOCUMENTED_FRAMES[1][0], {"extra": True})
+
+
+async def test_stream_events_non_2xx_raises_typed_and_closes() -> None:
+    """A wrong token (401) is typed with its status; no stream is read."""
+    response = _StreamingResponse(401, ["event: x\ndata: {}\n\n"])
+    transport = _StreamTransport(response)
+    client = make_stream_client(transport)
+
+    with pytest.raises(NestQuestApiError) as excinfo:
+        async for _ in client.stream_events():
+            pass
+
+    assert excinfo.value.status == 401
+    assert not excinfo.value.is_transport_error
+    assert TOKEN not in str(excinfo.value)
+    assert response.closed
+
+
+async def test_stream_events_connection_failure_raises_typed_transport_error() -> None:
+    """A failure before any response is a typed transport error."""
+    transport = _StreamTransport(OSError("connection refused"))
+    client = make_stream_client(transport)
+
+    with pytest.raises(NestQuestApiError) as excinfo:
+        async for _ in client.stream_events():
+            pass
+
+    assert excinfo.value.status is None
+    assert excinfo.value.is_transport_error
+    assert isinstance(excinfo.value.__cause__, OSError)
+
+
+async def test_stream_events_midstream_drop_raises_typed() -> None:
+    """A transport error mid-stream is typed, so the consumer reconnects."""
+    response = _StreamingResponse(
+        200,
+        [_frame_text(DOCUMENTED_FRAMES[0][0], DOCUMENTED_FRAMES[0][1]), OSError("reset")],
+    )
+    transport = _StreamTransport(response)
+    client = make_stream_client(transport)
+
+    with pytest.raises(NestQuestApiError) as excinfo:
+        async for _ in client.stream_events():
+            pass
+
+    assert excinfo.value.status is None
+    assert excinfo.value.is_transport_error
+    assert isinstance(excinfo.value.__cause__, OSError)
+    assert response.closed
+
+
+async def test_stream_events_stops_cleanly_on_cancellation() -> None:
+    """Closing the generator (the consumer task's cancellation) stops
+    cleanly: no error, the underlying response is closed, and nothing
+    further is yielded."""
+    response = _StreamingResponse(
+        200,
+        [_frame_text(DOCUMENTED_FRAMES[0][0], DOCUMENTED_FRAMES[0][1])],
+        hold_open=True,
+    )
+    transport = _StreamTransport(response)
+    client = make_stream_client(transport)
+
+    generator = client.stream_events()
+    first = await asyncio.wait_for(generator.__anext__(), timeout=2.0)
+    assert first == DOCUMENTED_FRAMES[0]
+
+    await asyncio.wait_for(generator.aclose(), timeout=2.0)
+    assert response.closed
+
+
+async def test_stream_events_malformed_frame_raises_typed() -> None:
+    """A frame whose data is not JSON is a typed protocol failure."""
+    response = _StreamingResponse(
+        200, ["event: nestquest_quest_completed\ndata: not-json\n\n"]
+    )
+    transport = _StreamTransport(response)
+    client = make_stream_client(transport)
+
+    with pytest.raises(NestQuestApiError) as excinfo:
+        async for _ in client.stream_events():
+            pass
+
+    assert excinfo.value.status == 200
+    assert response.closed
+
+
+# ---------------------------------------------------------------------------
+# The subscription manager: firing, teardown, reconnect
+# ---------------------------------------------------------------------------
+
+
+async def test_manager_fires_documented_frames_on_ha_bus(hass) -> None:
+    """Every documented transition fires on hass.bus with its payload."""
+    client = _ScriptedStreamClient([_frames(*DOCUMENTED_FRAMES)])
+    manager = make_manager(hass, client)
+    manager.start()
+    try:
+        await _wait_until(lambda: len(hass.bus.events) >= 4)
+        assert hass.bus.events == [tuple(f) for f in DOCUMENTED_FRAMES]
+        for event_type, payload in DOCUMENTED_FRAMES:
+            fired = hass.bus.fired(event_type)
+            assert fired, f"{event_type} never fired"
+            assert set(fired[-1]) == DOCUMENTED_FIELDS[event_type]
+    finally:
+        await manager.stop()
+
+
+async def test_manager_stop_cancels_task_and_blocks_later_frames(hass) -> None:
+    """stop() cancels and awaits the task: no frame fires afterwards,
+    even when the (dead) stream is later offered more data."""
+    stream_release = asyncio.Event()
+    first_type, first_payload = DOCUMENTED_FRAMES[0]
+    later_type, later_payload = DOCUMENTED_FRAMES[1]
+
+    def _factory():
+        async def _gen():
+            yield (first_type, dict(first_payload))
+            await stream_release.wait()
+            yield (later_type, dict(later_payload))
+
+        return _gen()
+
+    client = _ScriptedStreamClient([_factory])
+    manager = make_manager(hass, client)
+    manager.start()
+    await _wait_until(lambda: len(hass.bus.events) == 1)
+
+    task = manager.task
+    await manager.stop()
+    assert task.cancelled()
+    assert manager.task is None
+
+    # Offer the stream more data after teardown: nothing fires.
+    stream_release.set()
+    await _REAL_SLEEP(0.05)
+    assert hass.bus.events == [(first_type, dict(first_payload))]
+
+
+async def test_manager_reconnects_with_bounded_backoff(hass, monkeypatch) -> None:
+    """A failed stream reconnects with a bounded, reset-on-success
+    backoff and never crashes the task."""
+    delays: list[float] = []
+
+    async def fake_sleep(delay, *args, **kwargs):
+        delays.append(delay)
+        await _REAL_SLEEP(0)
+
+    monkeypatch.setattr(sse_module.asyncio, "sleep", fake_sleep)
+
+    stream_error = ConnectionError("stream reset")
+    client = _ScriptedStreamClient(
+        [
+            _fail(RuntimeError("api down")),
+            _fail(RuntimeError("api down")),
+            _fail(RuntimeError("api down")),
+            _fail(RuntimeError("api down")),
+            _drop_after(stream_error, DOCUMENTED_FRAMES[0]),
+            _frames(*DOCUMENTED_FRAMES),
+        ]
+    )
+    manager = make_manager(hass, client, min_backoff=0.05, max_backoff=0.2)
+    manager.start()
+    try:
+        await _wait_until(lambda: len(hass.bus.events) >= 5)
+        # 5s → 10s → 20s (cap) → 20s (cap), then a frame on a healthy
+        # stream reset it to the 5s minimum for the next failure.
+        assert delays == [0.05, 0.1, 0.2, 0.2, 0.05]
+        assert hass.bus.events == [
+            tuple(DOCUMENTED_FRAMES[0]),
+            *[tuple(f) for f in DOCUMENTED_FRAMES],
+        ]
+        # The task is still alive (subscribed and idle), not crashed.
+        assert manager.task is not None and not manager.task.done()
+    finally:
+        await manager.stop()
+
+
+def test_manager_rejects_invalid_backoff_bounds(hass) -> None:
+    """Non-positive or inverted backoff bounds fail at construction."""
+    with pytest.raises(ValueError):
+        NestQuestEventStream(hass, object(), reconnect_min_seconds=0)
+    with pytest.raises(ValueError):
+        NestQuestEventStream(hass, object(), reconnect_min_seconds=-1)
+    with pytest.raises(ValueError):
+        NestQuestEventStream(hass, object(), reconnect_max_seconds=0)
+    with pytest.raises(ValueError):
+        NestQuestEventStream(hass, object(), reconnect_min_seconds=10.0, reconnect_max_seconds=5.0)
+
+
+# ---------------------------------------------------------------------------
+# Integration wiring: setup starts the stream, unload stops it
+# ---------------------------------------------------------------------------
+
+
+class _SetupClient:
+    """Client for full-setup wiring tests: scripted snapshot + stream."""
+
+    def __init__(self, script) -> None:
+        self._script = list(script)
+        self.stream_calls = 0
+
+    async def get_snapshot(self) -> dict:
+        return dict(MINIMAL_SNAPSHOT)
+
+    def stream_events(self):
+        self.stream_calls += 1
+        return self._script.pop(0)()
+
+
+async def test_setup_starts_stream_and_unload_stops_it(hass, make_entry) -> None:
+    """Setup starts the subscription alongside the runtime record; unload
+    tears the task down so no frame fires afterwards."""
+    client = _SetupClient([_frames(*DOCUMENTED_FRAMES)])
+    set_coordinator_client(make_entry(), client)
+    entry = wire_entry_to_registry(make_entry(), hass.registry)
+    assert await nq_integration.async_setup_entry(hass, entry) is True
+
+    runtime = entry.runtime_data
+    assert runtime.event_stream is not None
+    await _wait_until(lambda: len(hass.bus.events) >= 4)
+    assert [event for event, _ in hass.bus.events] == [
+        frame[0] for frame in DOCUMENTED_FRAMES
+    ]
+
+    task = runtime.event_stream.task
+    assert await nq_integration.async_unload_entry(hass, entry) is True
+    assert task.cancelled()
+    assert runtime.event_stream.task is None
+
+    events_after_unload = len(hass.bus.events)
+    await _REAL_SLEEP(0.05)
+    assert len(hass.bus.events) == events_after_unload
+
+
+async def test_setup_without_stream_surface_starts_no_subscription(
+    hass, make_entry
+) -> None:
+    """An entry whose API client has no stream surface (the local-snapshot
+    stub the harness installs, like the unconfigured-token stub in
+    production) starts NO subscription — and setup still succeeds."""
+    entry = wire_entry_to_registry(make_entry(), hass.registry)
+    assert await nq_integration.async_setup_entry(hass, entry) is True
+    assert entry.runtime_data.event_stream is None
+    assert await nq_integration.async_unload_entry(hass, entry) is True