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)

tests/test_sse_subscription.py

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