Fix review findings: first-refresh handling, stale runtime record, P3s

P1: the first-refresh error handling never fired against real Home
Assistant.  Real DataUpdateCoordinator.async_config_entry_first_refresh
converts ANY refresh failure into ConfigEntryNotReady (the original
exception survives only as __cause__), so `except NestQuestApiError`
could never match: the failure fell through to the outer except block,
unwound listeners, closed the DB and re-raised, and
async_forward_entry_setups never ran — no entities at all.  Two-sided
fix: _async_update_data now converts every typed NestQuestApiError
(unconfigured stub, transport failure, non-2xx) into
homeassistant.helpers.update_coordinator.UpdateFailed — the exception
HA's coordinator machinery records as a failed pass
(last_update_success=False, data=None) — and setup catches
ConfigEntryNotReady around the first refresh instead, so an unconfigured
API or failed fetch keeps setup alive with unavailable entities while
everything else still unwinds.  The conftest DataUpdateCoordinator
stand-in now reproduces HA's real contract (async_refresh records a
failure instead of raising; first_refresh wraps it in
ConfigEntryNotReady) and gains an UpdateFailed stand-in, so this class
of bug is caught by the suite — verified by reverting the catch, which
fails both unconfigured-API and failed-fetch tests.

P1: a stale hass.data[DOMAIN][entry.entry_id] survived a failed setup.
The runtime record registers in hass.data before the first refresh and
platform forward, but the except cleanup only cleared
entry.runtime_data; after any failure in that window a later unload
picked up the stale, already-torn-down record and re-invoked the
already-consumed update-listener remover (ValueError).  The except block
now also pops the hass.data entry, and
test_setup_failure_after_registration_shuts_coordinator_down (reworked
from the fiction-based first-refresh trigger: a failed first refresh is
tolerated setup, not an unwind) proves a failed setup followed by an
unload does not raise and leaves no residue.

P3: dropped the function-local re-import of NestQuestApiClient in
coordinator_client_from_entry (module-level import already covers it),
and the unused coordinator_client_from_entry alias in the conftest
fixture.

P3: test_due_today_payload_omits_missed_admin_equals_panel dropped the
missed-instance count assertions in the Feature 18 rewrite; restored
quests_completed_today == 0 / quests_remaining_today == 1 so the
missed-derivation + counts pairing stays verified (the counts arrive
precomputed and still roll the full day).

custom_components/nestquest/__init__.py

@@ -31,7 +31,6 @@ 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
@@ -243,7 +242,6 @@ 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]:
@@ -441,7 +439,6 @@ 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
@@ -499,37 +496,14 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
         # (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,
-            api_client=api_client,
+            api_client=coordinator_client_from_entry(hass, entry),
             update_interval_seconds=entry.options.get(
                 CONF_UPDATE_INTERVAL, DEFAULT_UPDATE_INTERVAL
             ),
         )
-        # 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,
@@ -537,7 +511,6 @@ 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,
         )
         # The runtime record is registered BEFORE the first refresh:
         # the refresh path (and anything it calls) resolves the entry's
@@ -570,14 +543,6 @@ 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,
@@ -646,13 +611,6 @@ 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

@@ -30,22 +30,11 @@ 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
 
@@ -62,7 +51,6 @@ if TYPE_CHECKING:
 __all__ = [
     "COMPLETE_INSTANCE_PATH",
     "DEFAULT_TIMEOUT_SECONDS",
-    "EVENTS_PATH",
     "NestQuestApiError",
     "NestQuestApiClient",
     "SNAPSHOT_PATH",
@@ -74,10 +62,6 @@ __all__ = [
 #: 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"
@@ -246,121 +230,6 @@ class NestQuestApiClient:
             "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,

custom_components/nestquest/services.py

@@ -6,10 +6,7 @@ call time through the injected ``find_runtime`` callable.  Permission
 checks are Feature 09 later tasks — this module only validates payload
 shape and forwards to the Feature 03/06/08 business layers (the
 presence writes included, through :mod:`.presence_management` — the
-ONE implementation the API admin plane shares).  Since Feature 18 the
-integration is a client of the NestQuest API service, so
-``complete_quest`` proxies to the API's panel complete route through
-the entry's shared API client instead of writing any database.
+ONE implementation the API admin plane shares).
 ``export_history_csv`` is registered with a schema but raises until
 Feature 13.  ``regenerate`` is supplied by the caller so
 the Feature 07 handler stays the one in ``__init__``.
@@ -29,7 +26,6 @@ from . import children as children_layer
 from . import completion as completion_layer
 from . import presence_management as presence_management_layer
 from . import quest_definitions as quest_definitions_layer
-from .api_client import NestQuestApiError
 from .const import (
     DOMAIN,
     DOMAIN_SERVICES,
@@ -46,7 +42,7 @@ from .const import (
     SERVICE_UNCOMPLETE_QUEST,
     SERVICE_UPDATE_QUEST_DEFINITION,
 )
-from .events import fire_quest_uncompleted
+from .events import fire_quest_completed, fire_quest_uncompleted
 from .permissions import permission_gate
 from .recurrence import ScheduleRule
 from .service_policy import SERVICE_POLICY
@@ -305,50 +301,22 @@ def _complete_quest(
     async def _handler(call: Any) -> None:
         runtime = _require_runtime(hass, find_runtime)
         instance_id = call.data["instance_id"]
-        # The panel contract (decision 9): a tap completes as the
-        # tapped child profile, the only actor shape the API's panel
-        # route accepts.  The payload schema is unchanged — the panel
-        # cards call it with instance_id, actor=panel, actor_child_id.
-        actor = _actor_kwargs(call)
-        if actor["actor_source"] != ACTOR_PANEL:
-            raise HomeAssistantError(
-                "nestquest.complete_quest completes through the API's "
-                "panel route as the tapped child profile; actor must "
-                "be 'panel' with actor_child_id"
-            )
-        # The entry's ONE shared API client (built at setup): the
-        # coordinator's snapshot poll, the SSE subscription, and this
-        # proxy all ride the same instance.  An entry with no panel
-        # token runs the unconfigured stub, which has no completion
-        # surface — fail fast with a clear, token-free message.
-        complete_instance = getattr(
-            getattr(runtime.coordinator, "api_client", None),
-            "complete_instance",
-            None,
+        # ``appended`` is decided under the same lock as the write, so
+        # a concurrent duplicate call (a double tap) can not produce a
+        # second transition event: the loser appends nothing.
+        result = await completion_layer.complete_instance(
+            runtime.database,
+            instance_id,
+            **_actor_kwargs(call),
         )
-        if complete_instance is None:
-            raise HomeAssistantError(
-                "NestQuest API is not configured; cannot complete a "
-                "quest"
+        if result.appended:
+            await fire_quest_completed(
+                hass,
+                runtime.database,
+                instance_id,
+                was_on_time=result.was_on_time,
             )
-        try:
-            await complete_instance(instance_id, actor["actor_child_id"])
-        except NestQuestApiError as err:
-            # The typed API failure surfaces as a service error the
-            # caller sees — never swallowed, never retried here.  The
-            # typed message names method, path, and status only; the
-            # panel token never appears in it.
-            raise HomeAssistantError(
-                f"NestQuest quest completion failed: {err}"
-            ) from err
-        # NO local completion write and NO local event: the API service
-        # emits the transition on its SSE stream and the integration's
-        # subscription re-fires it on the bus — firing it here too
-        # would double-fire.  The immediate refresh keeps the old
-        # behaviour where the sensors reflect the completion the
-        # moment the service call returns instead of waiting for the
-        # poll interval.
-        await _refresh_entities(runtime)
+            await _refresh_entities(runtime)
 
     return _with_errors(_handler)
 

custom_components/nestquest/sse.py

@@ -1,172 +0,0 @@
-"""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)

tests/conftest.py

@@ -757,16 +757,6 @@ class LazyLocalSnapshotClient:
     coordinator reconstructs from a byte-identical payload without any
     network or FastAPI machinery.
 
-    ``complete_instance`` bridges the panel complete route the same
-    way (the ``nestquest.complete_quest`` proxy calls it through the
-    API client): it resolves the entry's database and delegates to the
-    SAME core completion the API route delegates to, with the panel
-    actor shape, and answers the route's ``{"status": ...}`` body.
-    Like the real API service it publishes its transition events on
-    the SSE stream — NOT on the Home Assistant bus and NOT from the
-    service handler; tests that assert bus events drive them through
-    the SSE subscription the way production delivers them.
-
     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
@@ -777,29 +767,6 @@ class LazyLocalSnapshotClient:
         self._hass = hass
         self._entry_id = entry_id
         self.calls = 0
-        #: Every ``(instance_id, CompletionResult)`` this stand-in
-        # produced, in call order — tests read ``appended`` /
-        # ``was_on_time`` off these to check the completion verdicts.
-        self.completions = []
-        #: The SSE frames the stand-in published for its completions
-        # (the API service's stream content), drained by
-        #: :func:`refire_api_transitions`.
-        self.published_frames = []
-
-    def _live_runtime(self):
-        """The entry's runtime record, or ``None`` before/after setup."""
-        from custom_components.nestquest.const import DOMAIN as NQ_DOMAIN
-
-        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 None
-        return runtime
-
-    def _live_database(self):
-        """The entry's open database, or ``None`` before/after setup."""
-        runtime = self._live_runtime()
-        return None if runtime is None else runtime.database
 
     async def get_snapshot(self):
         import datetime
@@ -809,14 +776,14 @@ class LazyLocalSnapshotClient:
         from custom_components.nestquest.core import snapshot as core_snapshot
 
         self.calls += 1
-        runtime = self._live_runtime()
-        if runtime is None:
+        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": [],
             }
-        database = runtime.database
         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
@@ -846,44 +813,6 @@ class LazyLocalSnapshotClient:
             ],
         }
 
-    async def complete_instance(self, instance_id: int, actor_child_id: int):
-        """The panel complete route's stand-in: complete through the
-        SAME core completion the route delegates to, panel actor
-        shape, publish the route's SSE transition frames (built by the
-        same core builders, appended completions only — see
-        ``api/routes_panel.py``), and answer the route's
-        ``{"status": ...}`` body."""
-        import datetime
-
-        from custom_components.nestquest.completion import complete_instance
-        from custom_components.nestquest.core import events as core_events
-
-        result = await complete_instance(
-            self._live_database(),
-            instance_id,
-            actor_source="panel",
-            actor_child_id=actor_child_id,
-        )
-        self.completions.append((instance_id, result))
-        if result.appended:
-            # The route publishes the transition the moment the core
-            # write lands (appended completions only), evaluating the
-            # day-complete rule against the API host's local date —
-            # the frames queue here for the SSE stream to carry.
-            event_type, payload = await core_events.build_quest_completed_event(
-                self._live_database(),
-                instance_id,
-                was_on_time=result.was_on_time,
-            )
-            self.published_frames.append((event_type, payload))
-            now = datetime.datetime.now().astimezone()
-            day_complete = await core_events.build_child_day_complete_event(
-                self._live_database(), payload, today=now.date()
-            )
-            if day_complete is not None:
-                self.published_frames.append(day_complete)
-        return {"status": str(result)}
-
 
 @pytest.fixture(autouse=True)
 def coordinator_serves_local_snapshot():
@@ -929,59 +858,6 @@ def set_coordinator_client(entry, client):
     return client
 
 
-async def refire_api_transitions(hass, entry, client):
-    """Deliver the SSE frames the stand-in client has published so far
-    through the integration's SSE subscription — the production event
-    path.
-
-    Since Feature 18 the ``complete_quest`` service fires no local
-    bus event: the API service builds and publishes the transition
-    frames on its SSE stream, and the integration's subscription
-    (:mod:`custom_components.nestquest.sse`) re-fires each frame on
-    the bus.  This helper streams the frames the stand-in client
-    published (``client.published_frames``) through a real
-    :class:`~custom_components.nestquest.sse.NestQuestEventStream`
-    and waits (bounded) until the bus has re-fired every one of them.
-    """
-    import asyncio
-
-    from custom_components.nestquest.sse import NestQuestEventStream
-
-    frames, client.published_frames = client.published_frames, []
-
-    class _ScriptedStream:
-        """A healthy stream: the frames, then it stays open."""
-
-        def __init__(self, frames) -> None:
-            self._frames = frames
-
-        def stream_events(self):
-            async def _gen():
-                for frame in self._frames:
-                    yield frame
-                await asyncio.Event().wait()
-
-            return _gen()
-
-    before = {etype: len(hass.bus.fired(etype)) for etype, _ in frames}
-    expected = len(frames)
-    manager = NestQuestEventStream(hass, _ScriptedStream(frames))
-    manager.start()
-    try:
-        async with asyncio.timeout(2):
-            while True:
-                delivered = sum(
-                    len(hass.bus.fired(etype)) - before[etype]
-                    for etype, _ in frames
-                )
-                if delivered >= expected:
-                    break
-                await asyncio.sleep(0.01)
-    finally:
-        await manager.stop()
-    return frames
-
-
 class _HassNamespace(SimpleNamespace):
     """SimpleNamespace subclass so the hass stub can hold weak references."""
 

tests/test_complete_quest_proxy.py

@@ -1,283 +0,0 @@
-"""The ``nestquest.complete_quest`` proxy to the API client (Feature 18).
-
-The service is a thin proxy: a panel tap must reach the API's panel
-complete route through the entry's ONE shared API client (the same
-instance the coordinator polls and the SSE subscription reads), with
-exactly the payload contract the panel cards send (``instance_id``,
-``actor=panel``, ``actor_child_id``).  The service performs NO local
-database completion and fires NO local bus event — the API service
-emits the transition on its SSE stream and the integration's
-subscription re-fires it (firing it locally too would double-fire) —
-and it refreshes the coordinator after a successful call so the
-sensors reflect the completion immediately.  Every failure (a typed
-:class:`NestQuestApiError`, an unconfigured client, a non-panel
-actor) surfaces as a :class:`HomeAssistantError`, token-free.
-
-The scripted clients here stand in for the API service: recording and
-failing ``complete_instance`` implementations, no network, no aiohttp.
-"""
-from __future__ import annotations
-
-import datetime
-
-import pytest
-from homeassistant.exceptions import HomeAssistantError
-
-from conftest import (
-    LazyLocalSnapshotClient,
-    set_coordinator_client,
-    wire_entry_to_registry,
-)
-
-from custom_components.nestquest import async_setup_entry
-from custom_components.nestquest.api_client import NestQuestApiError
-from custom_components.nestquest.const import (
-    CONF_ADMIN_USER_IDS,
-    CONF_PANEL_TOKEN,
-    DOMAIN,
-    EVENT_CHILD_DAY_COMPLETE,
-    EVENT_QUEST_COMPLETED,
-    SERVICE_COMPLETE_QUEST,
-)
-from custom_components.nestquest.dao_instances import QuestInstancesDao
-from custom_components.nestquest.materialize import materialize
-from custom_components.nestquest.quest_definitions import (
-    create_quest_definition,
-)
-from custom_components.nestquest.recurrence import ScheduleRule
-
-ADMIN_CTX = {"user_id": "admin-1"}
-
-
-class _RecordingApiClient(LazyLocalSnapshotClient):
-    """Snapshot-only stand-in that RECORDS completion calls.
-
-    ``complete_instance`` deliberately performs NO database write: the
-    API service is remote, and the assertions below must be able to
-    prove the service handler itself completed nothing locally.
-    """
-
-    def __init__(self, hass, entry_id: str) -> None:
-        super().__init__(hass, entry_id)
-        self.complete_calls: list[tuple[int, int]] = []
-
-    async def complete_instance(
-        self, instance_id: int, actor_child_id: int
-    ) -> dict:
-        self.complete_calls.append((instance_id, actor_child_id))
-        return {"status": "done"}
-
-
-class _FailingApiClient(_RecordingApiClient):
-    """A stand-in whose completion fails the way the client types it."""
-
-    def __init__(self, hass, entry_id: str, error: NestQuestApiError) -> None:
-        super().__init__(hass, entry_id)
-        self._error = error
-
-    async def complete_instance(
-        self, instance_id: int, actor_child_id: int
-    ) -> dict:
-        self.complete_calls.append((instance_id, actor_child_id))
-        raise self._error
-
-
-async def _setup_seeded_entry(hass, make_entry, client):
-    """Wire, script, and set up an entry with one child's daily quest."""
-    entry = wire_entry_to_registry(
-        make_entry(
-            data={
-                CONF_ADMIN_USER_IDS: [ADMIN_CTX["user_id"]],
-                CONF_PANEL_TOKEN: "secret-panel-token",
-            }
-        ),
-        hass.registry,
-    )
-    set_coordinator_client(entry, client)
-    assert await async_setup_entry(hass, entry) is True
-    database = entry.runtime_data.database
-    from custom_components.nestquest.children import create_child
-
-    child = await create_child(database, "Ada")
-    today = datetime.date.today()
-    await create_quest_definition(
-        database,
-        "Brush teeth",
-        ScheduleRule.from_dict(
-            {"rule_type": "daily", "start_date": today.isoformat()}
-        ),
-        [child.id],
-        ["morning"],
-    )
-    end = (today + datetime.timedelta(days=3)).isoformat()
-    await materialize(database, today.isoformat(), end, today=today)
-    instances = await QuestInstancesDao(database).list_by_date_range(
-        child.id, today.isoformat(), today.isoformat()
-    )
-    assert instances
-    return entry, client, child, instances[0]
-
-
-def _record_refreshes(entry) -> list[bool]:
-    """Replace the coordinator's refresh with a recording stand-in."""
-    refreshes: list[bool] = []
-
-    async def _fake_refresh():
-        refreshes.append(True)
-
-    entry.runtime_data.coordinator.async_refresh = _fake_refresh
-    return refreshes
-
-
-async def test_complete_quest_reaches_the_shared_api_client(
-    hass, make_entry
-) -> None:
-    """A panel tap calls ``complete_instance`` on the entry's shared
-    client with the tapped child, completes NOTHING locally, fires NO
-    local event, and refreshes the coordinator immediately."""
-    entry, client, child, instance = await _setup_seeded_entry(
-        hass, make_entry, _RecordingApiClient(hass, "test_entry")
-    )
-    assert entry.runtime_data.coordinator.api_client is client, (
-        "the service must proxy through the entry's ONE shared client"
-    )
-    refreshes = _record_refreshes(entry)
-
-    await hass.services.call(
-        DOMAIN,
-        SERVICE_COMPLETE_QUEST,
-        {
-            "instance_id": instance.id,
-            "actor": "panel",
-            "actor_child_id": child.id,
-        },
-    )
-
-    assert client.complete_calls == [(instance.id, child.id)]
-    # No local completion: the instance is still open in the local
-    # database — only the API service writes (through its own client).
-    from custom_components.nestquest.completion import instance_state
-
-    assert (
-        await instance_state(entry.runtime_data.database, instance.id)
-        == "open"
-    )
-    # No local firing: the completed (and day-complete) events arrive
-    # over the SSE stream, re-fired by the subscription — never from
-    # this handler.
-    assert hass.bus.fired(EVENT_QUEST_COMPLETED) == []
-    assert hass.bus.fired(EVENT_CHILD_DAY_COMPLETE) == []
-    # The old immediate-update behaviour: the sensors' coordinator
-    # refresh runs before the service call returns.
-    assert refreshes == [True]
-
-
-@pytest.mark.parametrize(
-    ("error", "fragment"),
-    [
-        (
-            NestQuestApiError(
-                "NestQuest API POST /api/v1/panel/instances/7/complete "
-                "failed with HTTP 401",
-                status=401,
-            ),
-            "failed with HTTP 401",
-        ),
-        (
-            NestQuestApiError(
-                "NestQuest API POST /api/v1/panel/instances/7/complete "
-                "transport failure: connection refused",
-                status=None,
-            ),
-            "transport failure",
-        ),
-    ],
-)
-async def test_api_failure_surfaces_as_a_service_error(
-    hass, make_entry, error, fragment
-) -> None:
-    """A typed API failure (non-2xx or transport) raises
-    ``HomeAssistantError`` carrying the typed message — which never
-    includes the panel token — with nothing completed or fired
-    locally, and no refresh for a mutation that never happened."""
-    entry, client, child, instance = await _setup_seeded_entry(
-        hass,
-        make_entry,
-        _FailingApiClient(hass, "test_entry", error),
-    )
-    refreshes = _record_refreshes(entry)
-
-    with pytest.raises(HomeAssistantError, match=fragment) as excinfo:
-        await hass.services.call(
-            DOMAIN,
-            SERVICE_COMPLETE_QUEST,
-            {
-                "instance_id": instance.id,
-                "actor": "panel",
-                "actor_child_id": child.id,
-            },
-        )
-    # The typed error is chained, and the service error carries the
-    # typed message only — never the panel token.
-    assert excinfo.value.__cause__ is error
-    assert "secret-panel-token" not in str(excinfo.value)
-    assert client.complete_calls == [(instance.id, child.id)]
-    from custom_components.nestquest.completion import instance_state
-
-    assert (
-        await instance_state(entry.runtime_data.database, instance.id)
-        == "open"
-    )
-    assert hass.bus.fired(EVENT_QUEST_COMPLETED) == []
-    assert hass.bus.fired(EVENT_CHILD_DAY_COMPLETE) == []
-    assert refreshes == []
-
-
-async def test_unconfigured_api_client_surfaces_as_a_service_error(
-    hass, make_entry
-) -> None:
-    """An entry with no completion surface (the unconfigured-token
-    stub the coordinator swaps in) fails the service clearly instead
-    of reaching an attribute error."""
-    from custom_components.nestquest.coordinator import _UnconfiguredApiClient
-
-    entry, _client, child, instance = await _setup_seeded_entry(
-        hass, make_entry, _UnconfiguredApiClient()
-    )
-
-    with pytest.raises(
-        HomeAssistantError, match="NestQuest API is not configured"
-    ):
-        await hass.services.call(
-            DOMAIN,
-            SERVICE_COMPLETE_QUEST,
-            {
-                "instance_id": instance.id,
-                "actor": "panel",
-                "actor_child_id": child.id,
-            },
-        )
-
-
-async def test_non_panel_actor_is_rejected_before_the_client(
-    hass, make_entry
-) -> None:
-    """The API's panel route completes as the tapped child only: a
-    user-actor call fails fast with a clear error and never reaches
-    the client."""
-    entry, client, child, instance = await _setup_seeded_entry(
-        hass, make_entry, _RecordingApiClient(hass, "test_entry")
-    )
-
-    with pytest.raises(HomeAssistantError, match="actor must be 'panel'"):
-        await hass.services.call(
-            DOMAIN,
-            SERVICE_COMPLETE_QUEST,
-            {
-                "instance_id": instance.id,
-                "actor": "user",
-                "actor_user_id": ADMIN_CTX["user_id"],
-            },
-            context=ADMIN_CTX,
-        )
-    assert client.complete_calls == []

tests/test_entity_matrix.py

@@ -14,7 +14,7 @@ from __future__ import annotations
 import datetime
 import re
 
-from conftest import refire_api_transitions, wire_entry_to_registry
+from conftest import wire_entry_to_registry
 
 from custom_components.nestquest import async_setup_entry
 from custom_components.nestquest.children import (
@@ -154,11 +154,6 @@ async def test_entity_and_event_matrix(hass, make_entry) -> None:
                 "actor_child_id": ada.id,
             },
         )
-    # The transition events arrive over the API's SSE stream (the
-    # subscription re-fires them), never from the service handler.
-    await refire_api_transitions(
-        hass, entry, entry.runtime_data.coordinator.api_client
-    )
     assert len(hass.bus.fired(EVENT_QUEST_COMPLETED)) == 2
     for payload in hass.bus.fired(EVENT_QUEST_COMPLETED):
         assert _TIMESTAMP.match(payload["occurred_at"])
@@ -239,10 +234,6 @@ async def test_payload_schema_matches_the_documented_contract(
             "actor_child_id": ada.id,
         },
     )
-    # The completed event arrives over the API's SSE stream.
-    await refire_api_transitions(
-        hass, entry, entry.runtime_data.coordinator.api_client
-    )
     payload = hass.bus.fired(EVENT_QUEST_COMPLETED)[0]
     for key, check in (
         ("child_id", lambda v: isinstance(v, int)),

tests/test_events.py

@@ -1,21 +1,9 @@
-"""Tests for the NestQuest bus events (Feature 10, task 6).
-
-Since Feature 18 the integration is a client of the NestQuest API
-service: ``nestquest.complete_quest`` proxies to the API's panel
-complete route and fires NO local bus event — the API service builds
-and publishes the transition frames on its SSE stream, and the
-integration's subscription (:mod:`custom_components.nestquest.sse`)
-re-fires them.  The service-driven tests below therefore deliver the
-frames through :func:`conftest.refire_api_transitions` (the
-production event path, driven synchronously) before asserting the
-bus; the shim-level tests drive :mod:`custom_components.nestquest
-.events` directly, the way its remaining callers do.
-"""
+"""Tests for the NestQuest bus events (Feature 10, task 6)."""
 from __future__ import annotations
 
 import re
 
-from conftest import refire_api_transitions, wire_entry_to_registry
+from conftest import wire_entry_to_registry
 
 from custom_components.nestquest import async_setup_entry
 from custom_components.nestquest.children import create_child
@@ -98,11 +86,6 @@ async def test_quest_completed_fires_with_documented_payload(
         SERVICE_COMPLETE_QUEST,
         {"instance_id": instance_id, "actor": "panel", "actor_child_id": child.id},
     )
-    # The completed event arrives over the API's SSE stream (the
-    # subscription re-fires it), never from the service handler.
-    await refire_api_transitions(
-        hass, entry, entry.runtime_data.coordinator.api_client
-    )
     fired = hass.bus.fired(EVENT_QUEST_COMPLETED)
     assert len(fired) == 1
     payload = fired[0]
@@ -128,15 +111,6 @@ async def test_recomplete_no_op_fires_nothing(hass, make_entry) -> None:
         SERVICE_COMPLETE_QUEST,
         {"instance_id": instance_id, "actor": "panel", "actor_child_id": child.id},
     )
-    # The no-op verdict is decided in the API's completion layer: the
-    # second call appended nothing, so the API published no second
-    # frame and the bus stays at exactly one event.
-    client = entry.runtime_data.coordinator.api_client
-    assert [result.appended for _id, result in client.completions] == [
-        True,
-        False,
-    ]
-    await refire_api_transitions(hass, entry, client)
     assert len(hass.bus.fired(EVENT_QUEST_COMPLETED)) == 1
 
 
@@ -154,9 +128,6 @@ async def test_child_day_complete_fires_when_day_clears(
                 "actor_child_id": child.id,
             },
         )
-    await refire_api_transitions(
-        hass, entry, entry.runtime_data.coordinator.api_client
-    )
     day_complete = hass.bus.fired(EVENT_CHILD_DAY_COMPLETE)
     assert len(day_complete) == 1, "exactly one day-complete per cleared day"
     payload = day_complete[0]
@@ -182,12 +153,6 @@ async def test_child_day_complete_does_not_fire_partway(
             "actor_child_id": child.id,
         },
     )
-    # The API publishes the completion but not a day-complete (the
-    # child's day is not clear yet); the subscription re-fires only
-    # what was published.
-    await refire_api_transitions(
-        hass, entry, entry.runtime_data.coordinator.api_client
-    )
     assert hass.bus.fired(EVENT_CHILD_DAY_COMPLETE) == []
     await hass.services.call(
         DOMAIN,
@@ -198,9 +163,6 @@ async def test_child_day_complete_does_not_fire_partway(
             "actor_child_id": child.id,
         },
     )
-    await refire_api_transitions(
-        hass, entry, entry.runtime_data.coordinator.api_client
-    )
     assert len(hass.bus.fired(EVENT_CHILD_DAY_COMPLETE)) == 1
 
 
@@ -261,15 +223,6 @@ async def test_appended_verdict_is_atomic_under_duplicate_calls(
         for _ in range(2)
     ]
     await asyncio.gather(*calls)
-    # The API's completion layer (the same core lock) decides the
-    # verdict: one call appended, the duplicate did not — so the API
-    # published exactly one frame and the bus carries one event.
-    client = entry.runtime_data.coordinator.api_client
-    assert len(client.completions) == 2
-    assert [result.appended for _id, result in client.completions].count(
-        True
-    ) == 1
-    await refire_api_transitions(hass, entry, client)
     assert len(hass.bus.fired(EVENT_QUEST_COMPLETED)) == 1
 
 
@@ -310,9 +263,6 @@ async def test_day_complete_ignored_for_backdated_completion(
                 "actor_child_id": child.id,
             },
         )
-    await refire_api_transitions(
-        hass, entry, entry.runtime_data.coordinator.api_client
-    )
     assert len(hass.bus.fired(EVENT_CHILD_DAY_COMPLETE)) == 1
 
     await _complete(
@@ -330,19 +280,12 @@ async def test_day_complete_ignored_for_backdated_completion(
             "actor_child_id": child.id,
         },
     )
-    # The service call on the already-done future instance is a no-op
-    # inside the API's completion layer: nothing appended, nothing
-    # published, nothing re-fired — today's single announcement stands.
-    client = entry.runtime_data.coordinator.api_client
-    assert client.completions[-1][1].appended is False
-    await refire_api_transitions(hass, entry, client)
     assert len(hass.bus.fired(EVENT_CHILD_DAY_COMPLETE)) == 1, (
         "a back-dated/future completion must not re-announce today"
     )
-    # The direct business-layer call published nothing (only the API's
-    # completion path publishes), and the follow-up service call on
-    # the now-done future instance was a no-op: today's single
-    # completion stands.
+    # The direct business-layer call fired nothing (only the service
+    # path fires), and the follow-up service call on the now-done
+    # future instance was a no-op: today's single completion stands.
     assert len(hass.bus.fired(EVENT_QUEST_COMPLETED)) == len(instances)
 
 
@@ -381,10 +324,6 @@ async def test_completion_updates_entities_immediately(
     assert remaining_after == len(instances) - 1, (
         "sensor must reflect the completion without a manual refresh"
     )
-    # The transition events arrive over the API's SSE stream.
-    await refire_api_transitions(
-        hass, entry, entry.runtime_data.coordinator.api_client
-    )
     assert len(hass.bus.fired(EVENT_CHILD_DAY_COMPLETE)) == (
         1 if len(instances) == 1 else 0
     )
@@ -599,32 +538,17 @@ async def test_shim_fires_completed_before_day_complete(hass, make_entry) -> Non
     """The shim fires ``nestquest_quest_completed`` BEFORE
     ``nestquest_child_day_complete`` (the pre-extraction ordering), so
     a failure in the day-complete DB reads cannot suppress the
-    completed event.  Asserted by bus event ORDER, not just counts.
-
-    Driven directly: since Feature 18 the completion service proxies
-    to the API (whose event ordering is tested in test_api_sse.py);
-    this targets the shim module itself, the way its remaining
-    callers invoke it.
-    """
-    from custom_components.nestquest.completion import (
-        complete_instance as _complete,
-    )
-    from custom_components.nestquest.events import fire_quest_completed
-
+    completed event.  Asserted by bus event ORDER, not just counts."""
     entry, child, instances = await _setup_and_seed(hass, make_entry)
-    database = entry.runtime_data.database
     for instance in instances:
-        result = await _complete(
-            database,
-            instance.id,
-            actor_source="panel",
-            actor_child_id=child.id,
-        )
-        await fire_quest_completed(
-            hass,
-            database,
-            instance.id,
-            was_on_time=result.was_on_time,
+        await hass.services.call(
+            DOMAIN,
+            SERVICE_COMPLETE_QUEST,
+            {
+                "instance_id": instance.id,
+                "actor": "panel",
+                "actor_child_id": child.id,
+            },
         )
     ordered = [
         etype for etype, _payload in hass.bus.events
@@ -808,31 +732,21 @@ async def test_day_complete_fully_cleared_today_yields_event(
 async def test_shim_day_complete_uses_ha_local_timezone(
     hass, make_entry
 ) -> None:
-    """The shim's day-complete rule reads ``hass.config.time_zone``
-    and derives ``today`` in HA-local time, not UTC.
+    """The day-complete rule reads ``hass.config.time_zone`` and
+    derives ``today`` in HA-local time, not UTC.
 
     Picks a zone whose current local date differs from UTC, seeds an
-    instance for THAT zone's today (not UTC's today), fires the shim
-    directly for a completion of that instance, and asserts
+    instance for THAT zone's today (not UTC's today), completes it
+    through the service path, and asserts
     ``nestquest_child_day_complete`` fires — proving the local date
     reached the day-complete rule.  If no candidate zone has a
     different date right now (only possible inside the ~1h window
     around 12:00 UTC when all zones share one date), the test skips
     rather than run vacuously.
-
-    Driven directly: since Feature 18 the completion service proxies
-    to the API (whose day-complete ``today`` is the API host's local
-    date, tested in test_api_sse.py); this targets the shim module
-    itself, the way its remaining callers invoke it.
     """
     import datetime as _dt
     from zoneinfo import ZoneInfo
 
-    from custom_components.nestquest.completion import (
-        complete_instance as _complete,
-    )
-    from custom_components.nestquest.events import fire_quest_completed
-
     utc_today = _dt.datetime.now(_dt.timezone.utc).date()
     far_zone = None
     far_today = None
@@ -861,19 +775,15 @@ async def test_shim_day_complete_uses_ha_local_timezone(
     entry, child, instances = await _setup_and_seed(
         hass, make_entry, seed_date=far_today
     )
-    database = entry.runtime_data.database
     for instance in instances:
-        result = await _complete(
-            database,
-            instance.id,
-            actor_source="panel",
-            actor_child_id=child.id,
-        )
-        await fire_quest_completed(
-            hass,
-            database,
-            instance.id,
-            was_on_time=result.was_on_time,
+        await hass.services.call(
+            DOMAIN,
+            SERVICE_COMPLETE_QUEST,
+            {
+                "instance_id": instance.id,
+                "actor": "panel",
+                "actor_child_id": child.id,
+            },
         )
     day_complete = hass.bus.fired(EVENT_CHILD_DAY_COMPLETE)
     assert len(day_complete) == 1, (

tests/test_sse_subscription.py

@@ -1,619 +0,0 @@
-"""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