Task b1ed9c45: SSE subscription re-fires the four HA bus events (#168)

custom_components/nestquest/__init__.py

@@ -31,6 +31,7 @@ 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
@@ -496,14 +499,37 @@ 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=coordinator_client_from_entry(hass, entry),
+            api_client=api_client,
             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,
@@ -511,6 +537,7 @@ 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
@@ -543,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,
@@ -611,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

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