From f31cd08df6e1687a499a759a98d86cbbd0bcf0e2 Mon Sep 17 00:00:00 2001 From: rhammen <75572839+rhammen@users.noreply.github.com> Date: Tue, 28 Jul 2026 21:49:35 +0200 Subject: [PATCH 01/16] Add design spec for stream reconnect fix (#417) Co-Authored-By: Claude Sonnet 5 --- .../2026-07-28-stream-reconnect-design.md | 176 ++++++++++++++++++ 1 file changed, 176 insertions(+) create mode 100644 docs/superpowers/specs/2026-07-28-stream-reconnect-design.md diff --git a/docs/superpowers/specs/2026-07-28-stream-reconnect-design.md b/docs/superpowers/specs/2026-07-28-stream-reconnect-design.md new file mode 100644 index 00000000..596581f0 --- /dev/null +++ b/docs/superpowers/specs/2026-07-28-stream-reconnect-design.md @@ -0,0 +1,176 @@ +# Stream reconnect on transient connection failure — design + +**Issue:** [custom-components/zaptec#417](https://github.com/custom-components/zaptec/issues/417) + +## Problem + +`Installation.stream_main()` ([zaptec/api.py](../../../custom_components/zaptec/zaptec/api.py)) +opens a persistent Azure Service Bus (AMQP) connection for live push updates. +On any exception — including a transient connection failure such as +`azure.servicebus.exceptions.ServiceBusConnectionError` — it catches the +exception with a blanket `except Exception: _LOGGER.exception("Stream +failed")` and returns normally. + +`ZaptecManager.create_streams()` ([manager.py](../../../custom_components/zaptec/manager.py)) +starts `stream_main()` exactly once per installation, as a background task, +during `async_setup_entry`. Nothing supervises that task or restarts it if it +exits. So a one-off network blip (e.g. a home router reboot) permanently +kills the live stream until the user reloads the integration or restarts +Home Assistant — a silent, indefinite degradation to poll-interval-only +freshness. + +## Goals + +- A transient stream failure (connection error, AMQP error, etc.) is + retried automatically with backoff, indefinitely — the stream should + self-heal without user intervention. +- A permanent condition (no permission to the stream, HTTP 403) is *not* + retried — retrying forever would just be log/network noise for a + situation retrying can't fix. +- `cancel_streams()` (integration unload/reload) must still cleanly stop + the stream, including while a reconnect backoff is in progress. +- Reconnect activity is logged just enough to diagnose an outage, without + spamming a stack trace on every backoff cycle during a prolonged outage. + +## Non-goals + +- Changing polling-coordinator behavior (already handles its own + independent retry — see #393). +- Changing the fallback-poll cadence while the stream is down (tracked + separately, see the "stream reconciliation gap" note referenced from + issue #378 discussion — out of scope here). +- `Installation.stream()` (a second, currently-unused convenience entry + point that wraps `stream_main()` in its own bare `asyncio.create_task`) + is not otherwise used by production code (only `manager.create_streams()` + is). Its behavior changes as a side effect of `stream_main()` no longer + swallowing exceptions (see below), but no new supervision logic is added + there — it isn't called from anywhere in this integration today. + +## Design + +### `stream_main()` contract change + +Remove the outer `except Exception: _LOGGER.exception("Stream failed")` +that currently wraps the whole connect-and-consume body. After the change: + +- **Returns normally** → permanent stop. Currently this is only the + 403/Forbidden case when fetching stream connection details (already + handled today by logging a warning and `return`-ing early). No other + code path returns normally after this change — reaching the end of the + `async for msg in receiver:` loop only happens when the receiver itself + ends the iteration, which in practice means the connection is closing. +- **Raises an exception** → transient failure. The exception propagates to + whoever awaited `stream_main()`. +- **`asyncio.CancelledError`** → not caught by `except Exception` (it is a + `BaseException`, not `Exception`, on the Python versions this integration + targets), so it is unaffected by this change and continues to propagate + straight through, as it does today. + +The `finally` block (clearing `_stream_receiver`, `_stream_running`, and +logging "Servicebus stream stopped for %s") is unchanged — it still runs +on every exit path. + +### Supervising wrapper + +New coroutine `ZaptecManager._stream_supervisor(install: Installation)` in +`manager.py`, used as the task body in `create_streams()` in place of the +direct `install.stream_main(...)` call: + +```python +async def _stream_supervisor(self, install: Installation) -> None: + delay = STREAM_RECONNECT_INIT_DELAY + connected_at: float | None = None + warned = False + while True: + connected_at = time.monotonic() + try: + await install.stream_main(cb=self.stream_callback, ssl_context=get_default_context()) + return # permanent stop (e.g. 403) + except Exception: + if time.monotonic() - connected_at >= STREAM_RECONNECT_MAX_DELAY: + delay = STREAM_RECONNECT_INIT_DELAY # reset after a long-lived connection + if not warned: + _LOGGER.warning( + "Stream for %s disconnected, reconnecting", install.qual_id, exc_info=True + ) + warned = True + else: + _LOGGER.debug("Stream for %s still reconnecting", install.qual_id, exc_info=True) + await asyncio.sleep(delay) + delay = min(delay * STREAM_RECONNECT_FACTOR, STREAM_RECONNECT_MAX_DELAY) + delay = random.normalvariate(delay, delay * STREAM_RECONNECT_JITTER) +``` + +(Illustrative — final implementation may adjust variable names/structure to +match repo style, but the behavior above is the contract.) + +`asyncio.CancelledError` is not caught here either, so `cancel_streams()`'s +existing `task.cancel()` + `await task` continues to stop the supervisor +(and whatever `stream_main()` call is in flight, or the backoff sleep) +immediately, unchanged from today. + +`warned` resets to `False` implicitly each time the loop returns to the top +after a successful reconnect (a fresh `_stream_supervisor` iteration only +warns again if *this* connection attempt also fails) — i.e. only the first +failure of a given outage logs at `warning`; the rest of that outage's +retries log at `debug`. A brand new outage after a successful reconnect +warns again. + +### New constants + +In `zaptec/const.py`, alongside the existing `API_RETRY_*` constants: + +```python +STREAM_RECONNECT_INIT_DELAY = 1.0 +STREAM_RECONNECT_FACTOR = 2.0 +STREAM_RECONNECT_JITTER = 0.1 +STREAM_RECONNECT_MAX_DELAY = 300.0 # 5 minutes +``` + +Kept separate from `API_RETRY_*` since they govern a different thing (a +long-lived connection's reconnect cadence, not a single HTTP request's +retry count) even though the shape (exponential + jitter, capped) matches. + +### Reconnected signal + +No new logging plumbing needed: `stream_main()` already logs +`_LOGGER.info("Running service bus stream for %s", self.qual_id)` once it +successfully opens the receiver. That existing line now doubles as the +"reconnected" signal once a prior failure has warned — satisfying the +Home Assistant integration quality-scale guidance ("log a warning once +when unavailable, log once when reconnected") without adding a new log +statement. + +## Error handling summary + +| Condition | `stream_main()` behavior | Supervisor behavior | +|---|---|---| +| 403 fetching stream connection details | logs warning, returns | stops, no retry | +| `ServiceBusConnectionError` / other transient error | raises | logs once (warn), backs off, retries | +| Integration unload (`cancel_streams()`) | `CancelledError` propagates | `CancelledError` propagates, loop exits | +| Malformed individual stream message | already handled inside the `async for` loop (existing `except Exception: _LOGGER.exception("Couldn't process stream message")`, unchanged) — does not end the stream | n/a | + +## Testing + +- `stream_main()`: existing tests around `stream_update` routing are + unaffected. Add a test confirming a non-403 exception now propagates + out of `stream_main()` instead of being swallowed (behavior change). +- `_stream_supervisor()`: new tests — + - retries with backoff on a raised exception, calling `stream_main()` + again; + - stops (single call, no retry) when `stream_main()` returns normally; + - propagates `CancelledError` without retrying; + - resets backoff delay after a connection that stayed up past + `STREAM_RECONNECT_MAX_DELAY`; + - logs at `warning` only for the first failure of an outage, `debug` + for subsequent ones within the same outage. + +## Open questions / risks + +- `stream_main()` no longer catching its own exceptions means any *unexpected* + bug in message processing that somehow escapes the inner per-message + `try/except` would now also be treated as "transient, retry" by the + supervisor rather than silently logged once and left stopped. This is + considered acceptable — retrying is a reasonable default reaction to an + unexpected stream failure, and the per-message handler already isolates + normal message-processing errors from ending the stream at all. From bd9ea142a114a8ae1890d610f53184caf70b49fb Mon Sep 17 00:00:00 2001 From: rhammen <75572839+rhammen@users.noreply.github.com> Date: Tue, 28 Jul 2026 22:01:05 +0200 Subject: [PATCH 02/16] Add implementation plan for stream reconnect fix (#417) Co-Authored-By: Claude Sonnet 5 --- .../plans/2026-07-28-stream-reconnect.md | 520 ++++++++++++++++++ 1 file changed, 520 insertions(+) create mode 100644 docs/superpowers/plans/2026-07-28-stream-reconnect.md diff --git a/docs/superpowers/plans/2026-07-28-stream-reconnect.md b/docs/superpowers/plans/2026-07-28-stream-reconnect.md new file mode 100644 index 00000000..bb381ade --- /dev/null +++ b/docs/superpowers/plans/2026-07-28-stream-reconnect.md @@ -0,0 +1,520 @@ +# Stream Reconnect on Transient Failure Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Make the Zaptec live-update stream (`Installation.stream_main()`) automatically reconnect with exponential backoff after a transient connection failure (e.g. a home internet outage), instead of permanently dying until the user reloads Home Assistant. + +**Architecture:** `stream_main()` stops swallowing its own exceptions, so a transient failure now propagates to its caller instead of returning silently. A new module-level `_stream_supervisor()` coroutine in `manager.py` becomes the actual background-task body (replacing the direct `stream_main()` call): it calls `stream_main()` in a loop, retries with exponential backoff+jitter on any raised exception, and stops for good if `stream_main()` ever returns normally (its existing signal for "no permission to the stream", HTTP 403). + +**Tech Stack:** Python 3.14, Home Assistant custom integration, `asyncio`, `azure-servicebus` (vendored stream client), `pytest` + `pytest-asyncio` (plain pytest harness, no `pytest-homeassistant-custom-component` yet in this repo). + +## Global Constraints + +- Follow `ruff format` / `ruff check` (repo uses `select = ["ALL"]` in `.ruff.toml`) — run both before considering a task done. +- Tests run via: `SKIP_ZAPTEC_API_TEST=true "C:/Users/rhamm/anaconda3/envs/py314/python.exe" -m pytest -p no:homeassistant -q` — the `-p no:homeassistant` flag is required on this Windows dev setup (avoids an `fcntl`-dependent pytest plugin that doesn't exist on Windows); never add a compatibility shim instead, just pass the flag. +- Never commit without the user's explicit go-ahead (existing session convention — ask before every `git commit`). +- `asyncio.CancelledError` is a `BaseException`, not an `Exception`, on the Python version this repo targets — `except Exception` blocks must not (and currently do not) catch it. Don't add a bare `except:` or `except BaseException` anywhere in this work. + +--- + +### Task 1: Stop `stream_main()` swallowing its own exceptions + +**Files:** +- Modify: `custom_components/zaptec/zaptec/api.py:397` and `:474-475` (the `try:`/`except Exception:` wrapping `Installation.stream_main()`'s body) +- Test: `tests/zaptec/test_api.py` + +**Interfaces:** +- Consumes: nothing new — this task only changes existing `Installation.stream_main()` control flow. +- Produces: `Installation.stream_main()` now propagates any exception raised while fetching stream connection details or consuming the stream, instead of catching it and returning `None`. The existing 403/Forbidden case is unchanged — it still logs a warning and returns `None` (this is the "permanent stop, don't retry" signal Task 2's supervisor relies on). + +Current code (for reference — do not copy verbatim, this is what you're changing): + +```python + try: + self._stream_running = True + + # Get connection details + try: + conf = await self.live_stream_connection_details() + except RequestError as err: + if err.error_code != HTTPStatus.FORBIDDEN: + raise + _LOGGER.warning( + "Failed to get live stream info. " + "Check if user have access in the zaptec portal" + ) + return + + # ... (connection setup and the `async for msg in receiver:` loop, unchanged) ... + + except Exception: + # Do this in order to show the error in the log. + _LOGGER.exception("Stream failed") + finally: + self._stream_receiver = None + self._stream_running = False + _LOGGER.info("Servicebus stream stopped for %s", self.qual_id) +``` + +- [ ] **Step 1: Write the failing tests** + +Add to the bottom of `tests/zaptec/test_api.py` (after the existing `Installation.stream_update routing` section, i.e. after the `test_stream_update_zero_guid_is_ignored` test): + +```python +# --------------------------------------------------------------------------- +# Installation.stream_main error propagation +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_stream_main_propagates_non_forbidden_error() -> None: + """A non-403 error fetching stream connection details now propagates. + + Previously this was swallowed internally (logged, then stream_main() + returned None) so a caller had no way to distinguish "transient + failure, please retry" from "stream ended cleanly". See issue #417. + """ + inst = Installation({"Id": "inst-1"}, _fake_owner()) + inst.live_stream_connection_details = AsyncMock( # type: ignore[method-assign] + side_effect=RequestError("server error", HTTPStatus.BAD_GATEWAY) + ) + with pytest.raises(RequestError): + await inst.stream_main() + + +@pytest.mark.asyncio +async def test_stream_main_forbidden_returns_none() -> None: + """A 403 fetching stream connection details still returns cleanly. + + This remains the "permanent stop, don't retry" signal the stream + supervisor (manager.py) relies on. + """ + inst = Installation({"Id": "inst-1"}, _fake_owner()) + inst.live_stream_connection_details = AsyncMock( # type: ignore[method-assign] + side_effect=RequestError("no access", HTTPStatus.FORBIDDEN) + ) + result = await inst.stream_main() + assert result is None +``` + +- [ ] **Step 2: Run tests to verify the first one fails** + +Run: `SKIP_ZAPTEC_API_TEST=true "C:/Users/rhamm/anaconda3/envs/py314/python.exe" -m pytest tests/zaptec/test_api.py -k stream_main -p no:homeassistant -v` + +Expected: `test_stream_main_propagates_non_forbidden_error` FAILS (no exception raised — it's currently swallowed). `test_stream_main_forbidden_returns_none` PASSES already (existing behavior, unaffected by this task — it's here as a regression guard). + +- [ ] **Step 3: Remove the outer `except Exception` in `stream_main()`** + +In `custom_components/zaptec/zaptec/api.py`, change: + +```python + except Exception: + # Do this in order to show the error in the log. + _LOGGER.exception("Stream failed") + finally: +``` + +to: + +```python + finally: +``` + +(i.e. delete the `except Exception:` block entirely, keeping the `try:` / `finally:` around the same body unchanged.) + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `SKIP_ZAPTEC_API_TEST=true "C:/Users/rhamm/anaconda3/envs/py314/python.exe" -m pytest tests/zaptec/test_api.py -p no:homeassistant -v` + +Expected: PASS (full file, to confirm nothing else in `test_api.py` broke). + +- [ ] **Step 5: Lint** + +Run: +``` +"C:/Users/rhamm/anaconda3/envs/py314/python.exe" -m ruff format custom_components/zaptec/zaptec/api.py tests/zaptec/test_api.py --diff +"C:/Users/rhamm/anaconda3/envs/py314/python.exe" -m ruff check custom_components/zaptec/zaptec/api.py tests/zaptec/test_api.py +``` +Expected: no diff, no lint errors. If `ruff format --diff` shows a diff, run it without `--diff` to apply, then re-run `ruff check`. + +- [ ] **Step 6: Commit** + +Ask the user for explicit go-ahead first (per this repo's convention — never commit automatically). Once approved: + +```bash +git add custom_components/zaptec/zaptec/api.py tests/zaptec/test_api.py +git commit -m "$(cat <<'EOF' +Let stream_main() propagate transient failures instead of swallowing them + +Only the existing 403/Forbidden case still returns cleanly; any other +failure now raises so a caller can distinguish "retry me" from +"permanent stop". Prep for issue #417's reconnect supervisor. + +Co-Authored-By: Claude Sonnet 5 +EOF +)" +``` + +--- + +### Task 2: Add reconnect-with-backoff supervisor and wire it into `create_streams()` + +**Files:** +- Modify: `custom_components/zaptec/const.py` (add `STREAM_RECONNECT_*` constants) +- Modify: `custom_components/zaptec/manager.py` (add module-level `_stream_supervisor()`, update imports, update `create_streams()`) +- Test: `tests/test_manager.py` (new file) + +**Interfaces:** +- Consumes: `Installation.stream_main(cb, ssl_context) -> None` from Task 1 (raises on transient failure, returns `None` on permanent stop or normal completion). +- Produces: `_stream_supervisor(install: Installation, cb: Callable[[dict], Awaitable[None]], ssl_context: ssl.SSLContext | None) -> None` — a module-level coroutine in `manager.py` (not a `ZaptecManager` method, so it's testable without constructing a full manager/config-entry). `create_streams()` now schedules `_stream_supervisor(...)` as the background task instead of calling `install.stream_main(...)` directly. No other public interface changes — `cancel_streams()` is untouched and still works because `task.cancel()` interrupts whatever `_stream_supervisor` is currently awaiting (either inside `stream_main()`, or the backoff `asyncio.sleep`). + +- [ ] **Step 1: Add the new constants** + +In `custom_components/zaptec/const.py`, after the existing `ZAPTEC_POLL_INSTALLATION_TRIGGER_DELAYS` constant (and before `REQUEST_REFRESH_DELAY`), add: + +```python +STREAM_RECONNECT_INIT_DELAY = 1.0 +"""Initial delay in seconds before the first stream reconnect attempt.""" + +STREAM_RECONNECT_FACTOR = 2.0 +"""Exponential backoff multiplier applied between stream reconnect attempts.""" + +STREAM_RECONNECT_JITTER = 0.1 +"""Relative jitter applied to the stream reconnect backoff delay.""" + +STREAM_RECONNECT_MAX_DELAY = 300.0 +"""Maximum delay in seconds between stream reconnect attempts (5 minutes).""" +``` + +- [ ] **Step 2: Write the failing tests** + +Create `tests/test_manager.py`: + +```python +"""Tests for custom_components.zaptec.manager.""" + +import asyncio +import logging +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import pytest + +from custom_components.zaptec.manager import _stream_supervisor + + +def _fake_install() -> SimpleNamespace: + """Return a stand-in for Installation carrying only what _stream_supervisor uses.""" + return SimpleNamespace(qual_id="Installation[nst-1]", stream_main=AsyncMock()) + + +@pytest.mark.asyncio +async def test_stream_supervisor_stops_when_stream_main_returns_normally() -> None: + """stream_main() returning None (e.g. 403/Forbidden) is a permanent stop.""" + install = _fake_install() + install.stream_main.return_value = None + + await _stream_supervisor(install, cb=AsyncMock(), ssl_context=None) + + install.stream_main.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_stream_supervisor_retries_on_exception(monkeypatch: pytest.MonkeyPatch) -> None: + """A raised exception is retried, not left dead.""" + install = _fake_install() + install.stream_main.side_effect = [ConnectionError("boom"), None] + monkeypatch.setattr(asyncio, "sleep", AsyncMock()) + + await _stream_supervisor(install, cb=AsyncMock(), ssl_context=None) + + assert install.stream_main.await_count == 2 + + +@pytest.mark.asyncio +async def test_stream_supervisor_propagates_cancelled_error_without_retrying( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Task cancellation (integration unload/reload) is not treated as a retryable failure.""" + install = _fake_install() + install.stream_main.side_effect = asyncio.CancelledError() + monkeypatch.setattr(asyncio, "sleep", AsyncMock()) + + with pytest.raises(asyncio.CancelledError): + await _stream_supervisor(install, cb=AsyncMock(), ssl_context=None) + + install.stream_main.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_stream_supervisor_logs_warning_once_then_debug( + monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture +) -> None: + """Only the first failure of an outage logs at warning; the rest log at debug.""" + install = _fake_install() + install.stream_main.side_effect = [ConnectionError("1"), ConnectionError("2"), None] + monkeypatch.setattr(asyncio, "sleep", AsyncMock()) + + with caplog.at_level(logging.DEBUG, logger="custom_components.zaptec.manager"): + await _stream_supervisor(install, cb=AsyncMock(), ssl_context=None) + + warnings = [r for r in caplog.records if r.levelno == logging.WARNING] + debugs = [r for r in caplog.records if r.levelno == logging.DEBUG] + assert len(warnings) == 1 + assert len(debugs) == 1 + + +@pytest.mark.asyncio +async def test_stream_supervisor_resets_backoff_after_long_lived_connection( + monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture +) -> None: + """A connection that outlived the max backoff delay counts as a fresh outage. + + Verified indirectly through logging: each failure that's treated as a + *new* outage logs at WARNING (the "warned" flag resets alongside the + delay). If the reset didn't happen, the second failure would log at + DEBUG instead (see test_stream_supervisor_logs_warning_once_then_debug + for that same-outage case). + """ + from custom_components.zaptec import manager as manager_module + + install = _fake_install() + install.stream_main.side_effect = [ConnectionError("1"), ConnectionError("2"), None] + monkeypatch.setattr(asyncio, "sleep", AsyncMock()) + monkeypatch.setattr(manager_module, "STREAM_RECONNECT_MAX_DELAY", 100.0) + # 5 monotonic() calls total: connected_at + failure-check for each of the + # 2 failed attempts, plus connected_at for the 3rd (successful) attempt. + # The gap between attempt 2's connected_at (0.0) and its failure-check + # (200.0) exceeds MAX_DELAY (100.0), so that failure counts as a new + # outage rather than a continuation of the first. + clock = iter([0.0, 0.0, 0.0, 200.0, 500.0]) + monkeypatch.setattr(manager_module.time, "monotonic", lambda: next(clock)) + + with caplog.at_level(logging.DEBUG, logger="custom_components.zaptec.manager"): + await _stream_supervisor(install, cb=AsyncMock(), ssl_context=None) + + warnings = [r for r in caplog.records if r.levelno == logging.WARNING] + assert len(warnings) == 2 # both failures counted as separate outages +``` + +- [ ] **Step 3: Run tests to verify they fail** + +Run: `SKIP_ZAPTEC_API_TEST=true "C:/Users/rhamm/anaconda3/envs/py314/python.exe" -m pytest tests/test_manager.py -p no:homeassistant -v` + +Expected: FAIL with `ModuleNotFoundError`/`ImportError` (`_stream_supervisor` doesn't exist yet). + +- [ ] **Step 4: Update `manager.py` imports** + +Change the import block at the top of `custom_components/zaptec/manager.py` from: + +```python +import asyncio +from collections.abc import Iterable +import contextlib +from copy import copy +from dataclasses import dataclass +import logging +from typing import Any + +from homeassistant.config_entries import ConfigEntry +from homeassistant.core import HomeAssistant +from homeassistant.helpers.entity import DeviceInfo, EntityDescription +from homeassistant.util.ssl import get_default_context + +from .const import DOMAIN, KEYS_TO_SKIP_ENTITY_AVAILABILITY_CHECK, MANUFACTURER +from .coordinator import ZaptecUpdateCoordinator +from .entity import KeyUnavailableError, ZaptecBaseEntity +from .zaptec import Charger, Installation, Zaptec, ZaptecBase +``` + +to: + +```python +import asyncio +from collections.abc import Awaitable, Callable, Iterable +import contextlib +from copy import copy +from dataclasses import dataclass +import logging +import random +import ssl +import time +from typing import Any + +from homeassistant.config_entries import ConfigEntry +from homeassistant.core import HomeAssistant +from homeassistant.helpers.entity import DeviceInfo, EntityDescription +from homeassistant.util.ssl import get_default_context + +from .const import ( + DOMAIN, + KEYS_TO_SKIP_ENTITY_AVAILABILITY_CHECK, + MANUFACTURER, + STREAM_RECONNECT_FACTOR, + STREAM_RECONNECT_INIT_DELAY, + STREAM_RECONNECT_JITTER, + STREAM_RECONNECT_MAX_DELAY, +) +from .coordinator import ZaptecUpdateCoordinator +from .entity import KeyUnavailableError, ZaptecBaseEntity +from .zaptec import Charger, Installation, Zaptec, ZaptecBase +``` + +- [ ] **Step 5: Add `_stream_supervisor()`** + +In `custom_components/zaptec/manager.py`, add this module-level function right before `class ZaptecManager:`: + +```python +async def _stream_supervisor( + install: Installation, + cb: Callable[[dict], Awaitable[None]], + ssl_context: ssl.SSLContext | None, +) -> None: + """Run install.stream_main(), reconnecting after a transient failure. + + stream_main() returning normally means a permanent stop (e.g. no + permission to the live stream) -- this loop ends without retrying. It + raising means a transient failure to retry with exponential backoff. + asyncio.CancelledError is a BaseException, not an Exception, so it is + never caught here: cancelling the task (integration unload/reload) + still stops this immediately, whether currently inside stream_main() + or in the backoff sleep below. + """ + delay = STREAM_RECONNECT_INIT_DELAY + warned = False + while True: + connected_at = time.monotonic() + try: + await install.stream_main(cb=cb, ssl_context=ssl_context) + return + except Exception: + if time.monotonic() - connected_at >= STREAM_RECONNECT_MAX_DELAY: + # The previous connection lived long enough to count this + # as a fresh outage rather than a continuation of the last. + delay = STREAM_RECONNECT_INIT_DELAY + warned = False + if not warned: + _LOGGER.warning( + "Stream for %s disconnected, reconnecting", install.qual_id, exc_info=True + ) + warned = True + else: + _LOGGER.debug( + "Stream for %s still reconnecting", install.qual_id, exc_info=True + ) + await asyncio.sleep(delay) + delay = delay * STREAM_RECONNECT_FACTOR + delay = random.normalvariate(delay, delay * STREAM_RECONNECT_JITTER) + delay = min(delay, STREAM_RECONNECT_MAX_DELAY) +``` + +- [ ] **Step 6: Wire it into `create_streams()`** + +In `custom_components/zaptec/manager.py`, change: + +```python + def create_streams(self) -> None: + """Create the streams for all installations.""" + for install in self.zaptec.installations: + if install.id in self.zaptec: + task = self.config_entry.async_create_background_task( + self.hass, + install.stream_main( + cb=self.stream_callback, + ssl_context=get_default_context(), + ), + name=f"Zaptec Stream for {install.qual_id}", + ) + self.streams.append((task, install)) +``` + +to: + +```python + def create_streams(self) -> None: + """Create the streams for all installations.""" + for install in self.zaptec.installations: + if install.id in self.zaptec: + task = self.config_entry.async_create_background_task( + self.hass, + _stream_supervisor( + install, + cb=self.stream_callback, + ssl_context=get_default_context(), + ), + name=f"Zaptec Stream for {install.qual_id}", + ) + self.streams.append((task, install)) +``` + +(`cancel_streams()` below it is unchanged — no edit needed there.) + +- [ ] **Step 7: Run tests to verify they pass** + +Run: `SKIP_ZAPTEC_API_TEST=true "C:/Users/rhamm/anaconda3/envs/py314/python.exe" -m pytest tests/test_manager.py -p no:homeassistant -v` + +Expected: all 5 tests PASS. + +- [ ] **Step 8: Lint** + +Run: +``` +"C:/Users/rhamm/anaconda3/envs/py314/python.exe" -m ruff format custom_components/zaptec/const.py custom_components/zaptec/manager.py tests/test_manager.py --diff +"C:/Users/rhamm/anaconda3/envs/py314/python.exe" -m ruff check custom_components/zaptec/const.py custom_components/zaptec/manager.py tests/test_manager.py +``` +Expected: no diff, no lint errors. If `ruff format --diff` shows a diff, run it without `--diff` to apply, then re-run `ruff check`. + +- [ ] **Step 9: Commit** + +Ask the user for explicit go-ahead first. Once approved: + +```bash +git add custom_components/zaptec/const.py custom_components/zaptec/manager.py tests/test_manager.py +git commit -m "$(cat <<'EOF' +Add reconnect-with-backoff supervisor for the live update stream + +Wraps stream_main() in an exponential-backoff retry loop so a transient +connection failure (e.g. a brief home-internet outage) reconnects +automatically instead of permanently killing live updates until the +user reloads the integration. Fixes #417. + +Co-Authored-By: Claude Sonnet 5 +EOF +)" +``` + +--- + +### Task 3: Full verification + +**Files:** none (verification only) + +**Interfaces:** +- Consumes: everything from Tasks 1-2. +- Produces: nothing new — confirms the full suite and lint are green together, not just per-file. + +- [ ] **Step 1: Run the full test suite** + +Run: `SKIP_ZAPTEC_API_TEST=true "C:/Users/rhamm/anaconda3/envs/py314/python.exe" -m pytest tests -p no:homeassistant -q` + +Expected: all tests pass (aside from any pre-existing `SKIP_ZAPTEC_API_TEST`/live-network skips unrelated to this change). + +- [ ] **Step 2: Run full-repo lint** + +Run: +``` +"C:/Users/rhamm/anaconda3/envs/py314/python.exe" -m ruff format . --diff +"C:/Users/rhamm/anaconda3/envs/py314/python.exe" -m ruff check +``` +Expected: no diff, no errors. This matches what CI's `lint-ruff` job runs (see `CLAUDE.md`), scoped to the whole repo, not just the files touched here. + +- [ ] **Step 3: Report results to the user** + +Summarize pass/fail counts and any lint findings. Do not open a PR or push — per this repo's AI-policy convention, that's a separate, explicit ask. + +--- + +## Design doc + +Full rationale, alternatives considered, and error-handling table: +`docs/superpowers/specs/2026-07-28-stream-reconnect-design.md` From a24ca888f5574570bba88deb4626a1d9df2e664d Mon Sep 17 00:00:00 2001 From: rhammen <75572839+rhammen@users.noreply.github.com> Date: Tue, 28 Jul 2026 22:13:21 +0200 Subject: [PATCH 03/16] Ignore .superpowers/ SDD scratch workspace Co-Authored-By: Claude Sonnet 5 --- .gitignore | 1 + 1 file changed, 1 insertion(+) diff --git a/.gitignore b/.gitignore index 161bb6ba..6e5a4fa7 100644 --- a/.gitignore +++ b/.gitignore @@ -11,6 +11,7 @@ __pycache__ coverage.xml .ruff_cache htmlcov/ +.superpowers/ # Home Assistant configuration config/* From cd33db5be709e7461eb30522c199539031ee421d Mon Sep 17 00:00:00 2001 From: rhammen <75572839+rhammen@users.noreply.github.com> Date: Tue, 28 Jul 2026 22:15:58 +0200 Subject: [PATCH 04/16] Let stream_main() propagate transient failures instead of swallowing them Only the existing 403/Forbidden case still returns cleanly; any other failure now raises so a caller can distinguish "retry me" from "permanent stop". Prep for issue #417's reconnect supervisor. Co-Authored-By: Claude Sonnet 5 --- custom_components/zaptec/zaptec/api.py | 4 --- tests/zaptec/test_api.py | 36 ++++++++++++++++++++++++++ 2 files changed, 36 insertions(+), 4 deletions(-) diff --git a/custom_components/zaptec/zaptec/api.py b/custom_components/zaptec/zaptec/api.py index cd818dd4..be677b48 100644 --- a/custom_components/zaptec/zaptec/api.py +++ b/custom_components/zaptec/zaptec/api.py @@ -471,10 +471,6 @@ async def stream_main( # remove the msg from the "queue" await receiver.complete_message(msg) - except Exception: - # Do this in order to show the error in the log. - _LOGGER.exception("Stream failed") - finally: # Cleanup self._stream_receiver = None diff --git a/tests/zaptec/test_api.py b/tests/zaptec/test_api.py index 825b05d9..e82589d7 100644 --- a/tests/zaptec/test_api.py +++ b/tests/zaptec/test_api.py @@ -546,6 +546,42 @@ def test_stream_update_zero_guid_is_ignored() -> None: charger.set_attributes.assert_not_called() +# --------------------------------------------------------------------------- +# Installation.stream_main error propagation +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_stream_main_propagates_non_forbidden_error() -> None: + """A non-403 error fetching stream connection details now propagates. + + Previously this was swallowed internally (logged, then stream_main() + returned None) so a caller had no way to distinguish "transient + failure, please retry" from "stream ended cleanly". See issue #417. + """ + inst = Installation({"Id": "inst-1"}, _fake_owner()) + inst.live_stream_connection_details = AsyncMock( # type: ignore[method-assign] + side_effect=RequestError("server error", HTTPStatus.BAD_GATEWAY) + ) + with pytest.raises(RequestError): + await inst.stream_main() + + +@pytest.mark.asyncio +async def test_stream_main_forbidden_returns_none() -> None: + """A 403 fetching stream connection details still returns cleanly. + + This remains the "permanent stop, don't retry" signal the stream + supervisor (manager.py) relies on. + """ + inst = Installation({"Id": "inst-1"}, _fake_owner()) + inst.live_stream_connection_details = AsyncMock( # type: ignore[method-assign] + side_effect=RequestError("no access", HTTPStatus.FORBIDDEN) + ) + result = await inst.stream_main() + assert result is None + + # --------------------------------------------------------------------------- # Zaptec mapping / registry + poll dispatch # --------------------------------------------------------------------------- From 80f5e5a5e137b2c08a3421ef9037aa34300f27b9 Mon Sep 17 00:00:00 2001 From: rhammen <75572839+rhammen@users.noreply.github.com> Date: Tue, 28 Jul 2026 22:35:46 +0200 Subject: [PATCH 05/16] Add reconnect-with-backoff supervisor for the live update stream Wraps stream_main() in an exponential-backoff retry loop so a transient connection failure (e.g. a brief home-internet outage) reconnects automatically instead of permanently killing live updates until the user reloads the integration. Fixes #417. Co-Authored-By: Claude Sonnet 5 --- custom_components/zaptec/const.py | 12 +++ custom_components/zaptec/manager.py | 60 ++++++++++++++- tests/test_manager.py | 111 ++++++++++++++++++++++++++++ 3 files changed, 180 insertions(+), 3 deletions(-) create mode 100644 tests/test_manager.py diff --git a/custom_components/zaptec/const.py b/custom_components/zaptec/const.py index 739326f3..f939e33a 100644 --- a/custom_components/zaptec/const.py +++ b/custom_components/zaptec/const.py @@ -36,6 +36,18 @@ ZAPTEC_POLL_INSTALLATION_TRIGGER_DELAYS = [2, 7] """Delays in seconds for installation state updates after a change.""" +STREAM_RECONNECT_INIT_DELAY = 1.0 +"""Initial delay in seconds before the first stream reconnect attempt.""" + +STREAM_RECONNECT_FACTOR = 2.0 +"""Exponential backoff multiplier applied between stream reconnect attempts.""" + +STREAM_RECONNECT_JITTER = 0.1 +"""Relative jitter applied to the stream reconnect backoff delay.""" + +STREAM_RECONNECT_MAX_DELAY = 300.0 +"""Maximum delay in seconds between stream reconnect attempts (5 minutes).""" + # This sets the delay after doing actions and the poll of updated values. # It was 0.3 and evidently that is a bit too fast for Zaptec cloud to handle. REQUEST_REFRESH_DELAY = 1 diff --git a/custom_components/zaptec/manager.py b/custom_components/zaptec/manager.py index 79b9fc02..b31c212d 100644 --- a/custom_components/zaptec/manager.py +++ b/custom_components/zaptec/manager.py @@ -3,11 +3,14 @@ from __future__ import annotations import asyncio -from collections.abc import Iterable +from collections.abc import Awaitable, Callable, Iterable import contextlib from copy import copy from dataclasses import dataclass import logging +import random +import ssl +import time from typing import Any from homeassistant.config_entries import ConfigEntry @@ -15,7 +18,15 @@ from homeassistant.helpers.entity import DeviceInfo, EntityDescription from homeassistant.util.ssl import get_default_context -from .const import DOMAIN, KEYS_TO_SKIP_ENTITY_AVAILABILITY_CHECK, MANUFACTURER +from .const import ( + DOMAIN, + KEYS_TO_SKIP_ENTITY_AVAILABILITY_CHECK, + MANUFACTURER, + STREAM_RECONNECT_FACTOR, + STREAM_RECONNECT_INIT_DELAY, + STREAM_RECONNECT_JITTER, + STREAM_RECONNECT_MAX_DELAY, +) from .coordinator import ZaptecUpdateCoordinator from .entity import KeyUnavailableError, ZaptecBaseEntity from .zaptec import Charger, Installation, Zaptec, ZaptecBase @@ -32,6 +43,48 @@ class ZaptecEntityDescription(EntityDescription): cls: type[ZaptecBaseEntity[Any]] +async def _stream_supervisor( + install: Installation, + cb: Callable[[dict], Awaitable[None]], + ssl_context: ssl.SSLContext | None, +) -> None: + """Run install.stream_main(), reconnecting after a transient failure. + + stream_main() returning normally means a permanent stop (e.g. no + permission to the live stream) -- this loop ends without retrying. It + raising means a transient failure to retry with exponential backoff. + asyncio.CancelledError is a BaseException, not an Exception, so it is + never caught here: cancelling the task (integration unload/reload) + still stops this immediately, whether currently inside stream_main() + or in the backoff sleep below. + """ + delay = STREAM_RECONNECT_INIT_DELAY + warned = False + while True: + connected_at = time.monotonic() + try: + await install.stream_main(cb=cb, ssl_context=ssl_context) + except Exception: + if time.monotonic() - connected_at >= STREAM_RECONNECT_MAX_DELAY: + # The previous connection lived long enough to count this + # as a fresh outage rather than a continuation of the last. + delay = STREAM_RECONNECT_INIT_DELAY + warned = False + if not warned: + _LOGGER.warning( + "Stream for %s disconnected, reconnecting", install.qual_id, exc_info=True + ) + warned = True + else: + _LOGGER.debug("Stream for %s still reconnecting", install.qual_id, exc_info=True) + await asyncio.sleep(delay) + delay = delay * STREAM_RECONNECT_FACTOR + delay = random.normalvariate(delay, delay * STREAM_RECONNECT_JITTER) + delay = min(delay, STREAM_RECONNECT_MAX_DELAY) + else: + return + + class ZaptecManager: """Manager for Zaptec data.""" @@ -196,7 +249,8 @@ def create_streams(self) -> None: if install.id in self.zaptec: task = self.config_entry.async_create_background_task( self.hass, - install.stream_main( + _stream_supervisor( + install, cb=self.stream_callback, ssl_context=get_default_context(), ), diff --git a/tests/test_manager.py b/tests/test_manager.py new file mode 100644 index 00000000..ad0a57aa --- /dev/null +++ b/tests/test_manager.py @@ -0,0 +1,111 @@ +"""Tests for custom_components.zaptec.manager.""" + +import asyncio +import logging +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import pytest + +from custom_components.zaptec.manager import _stream_supervisor + + +def _fake_install() -> SimpleNamespace: + """Return a stand-in for Installation carrying only what _stream_supervisor uses.""" + return SimpleNamespace(qual_id="Installation[nst-1]", stream_main=AsyncMock()) + + +@pytest.mark.asyncio +async def test_stream_supervisor_stops_when_stream_main_returns_normally() -> None: + """stream_main() returning None (e.g. 403/Forbidden) is a permanent stop.""" + install = _fake_install() + install.stream_main.return_value = None + + await _stream_supervisor(install, cb=AsyncMock(), ssl_context=None) + + install.stream_main.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_stream_supervisor_retries_on_exception(monkeypatch: pytest.MonkeyPatch) -> None: + """A raised exception is retried, not left dead.""" + install = _fake_install() + install.stream_main.side_effect = [ConnectionError("boom"), None] + monkeypatch.setattr(asyncio, "sleep", AsyncMock()) + + await _stream_supervisor(install, cb=AsyncMock(), ssl_context=None) + + assert install.stream_main.await_count == 2 # noqa: PLR2004 + + +@pytest.mark.asyncio +async def test_stream_supervisor_propagates_cancelled_error_without_retrying( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Task cancellation (integration unload/reload) is not treated as a retryable failure.""" + install = _fake_install() + install.stream_main.side_effect = asyncio.CancelledError() + monkeypatch.setattr(asyncio, "sleep", AsyncMock()) + + with pytest.raises(asyncio.CancelledError): + await _stream_supervisor(install, cb=AsyncMock(), ssl_context=None) + + install.stream_main.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_stream_supervisor_logs_warning_once_then_debug( + monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture +) -> None: + """Only the first failure of an outage logs at warning; the rest log at debug.""" + install = _fake_install() + install.stream_main.side_effect = [ConnectionError("1"), ConnectionError("2"), None] + monkeypatch.setattr(asyncio, "sleep", AsyncMock()) + + with caplog.at_level(logging.DEBUG, logger="custom_components.zaptec.manager"): + await _stream_supervisor(install, cb=AsyncMock(), ssl_context=None) + + warnings = [r for r in caplog.records if r.levelno == logging.WARNING] + debugs = [r for r in caplog.records if r.levelno == logging.DEBUG] + assert len(warnings) == 1 + assert len(debugs) == 1 + + +@pytest.mark.asyncio +async def test_stream_supervisor_resets_backoff_after_long_lived_connection( + monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture +) -> None: + """A connection that outlived the max backoff delay counts as a fresh outage. + + Verified indirectly through logging: each failure that's treated as a + *new* outage logs at WARNING (the "warned" flag resets alongside the + delay). If the reset didn't happen, the second failure would log at + DEBUG instead (see test_stream_supervisor_logs_warning_once_then_debug + for that same-outage case). + """ + from custom_components.zaptec import manager as manager_module # noqa: PLC0415 + + install = _fake_install() + install.stream_main.side_effect = [ConnectionError("1"), ConnectionError("2"), None] + monkeypatch.setattr(asyncio, "sleep", AsyncMock()) + monkeypatch.setattr(manager_module, "STREAM_RECONNECT_MAX_DELAY", 100.0) + # 5 monotonic() calls total: connected_at + failure-check for each of the + # 2 failed attempts, plus connected_at for the 3rd (successful) attempt. + # The gap between attempt 2's connected_at (0.0) and its failure-check + # (200.0) exceeds MAX_DELAY (100.0), so that failure counts as a new + # outage rather than a continuation of the first. + # `next(clock, 500.0)` (not bare `next(clock)`): on Windows, asyncio's + # ProactorEventLoop calls time.monotonic() a few more times during event + # loop teardown, after _stream_supervisor has already returned. A bare + # next(clock) would raise StopIteration from inside asyncio's own + # teardown for those extra calls; the fallback keeps that harmless while + # the 5 calls _stream_supervisor actually makes still get the exact + # scripted sequence below. + clock = iter([0.0, 0.0, 0.0, 200.0, 500.0]) + monkeypatch.setattr(manager_module.time, "monotonic", lambda: next(clock, 500.0)) + + with caplog.at_level(logging.DEBUG, logger="custom_components.zaptec.manager"): + await _stream_supervisor(install, cb=AsyncMock(), ssl_context=None) + + warnings = [r for r in caplog.records if r.levelno == logging.WARNING] + assert len(warnings) == 2 # noqa: PLR2004 # both failures counted as separate outages From 21232936b8fb2f1c4bc16ab8bb395e0257f77d51 Mon Sep 17 00:00:00 2001 From: rhammen <75572839+rhammen@users.noreply.github.com> Date: Tue, 28 Jul 2026 23:23:40 +0200 Subject: [PATCH 06/16] Address final-review findings: backoff test coverage, jitter-at-cap fix, docstring - Add test coverage for the actual asyncio.sleep delay values (first-call value and max-delay cap enforcement across a sustained outage) - Clamp exponential growth to STREAM_RECONNECT_MAX_DELAY before applying jitter, so jitter isn't nullified once backoff saturates - Document _stream_supervisor()'s actual warn-once-per-outage semantics Co-Authored-By: Claude Sonnet 5 --- custom_components/zaptec/manager.py | 11 ++++++++- tests/test_manager.py | 37 +++++++++++++++++++++++++---- 2 files changed, 42 insertions(+), 6 deletions(-) diff --git a/custom_components/zaptec/manager.py b/custom_components/zaptec/manager.py index b31c212d..437f0478 100644 --- a/custom_components/zaptec/manager.py +++ b/custom_components/zaptec/manager.py @@ -57,6 +57,15 @@ async def _stream_supervisor( never caught here: cancelling the task (integration unload/reload) still stops this immediately, whether currently inside stream_main() or in the backoff sleep below. + + The first failure of an outage is logged at WARNING; subsequent + failures of the same outage are logged at DEBUG only. An outage is + considered "new" -- and thus re-warned -- only once the previous + connection stayed up for at least STREAM_RECONNECT_MAX_DELAY seconds + before failing again. A rapidly flapping stream (reconnecting and + failing faster than that) will therefore log its first failure as + WARNING and every subsequent failure in that flapping sequence as + DEBUG only -- it does not periodically re-warn while flapping. """ delay = STREAM_RECONNECT_INIT_DELAY warned = False @@ -78,7 +87,7 @@ async def _stream_supervisor( else: _LOGGER.debug("Stream for %s still reconnecting", install.qual_id, exc_info=True) await asyncio.sleep(delay) - delay = delay * STREAM_RECONNECT_FACTOR + delay = min(delay * STREAM_RECONNECT_FACTOR, STREAM_RECONNECT_MAX_DELAY) delay = random.normalvariate(delay, delay * STREAM_RECONNECT_JITTER) delay = min(delay, STREAM_RECONNECT_MAX_DELAY) else: diff --git a/tests/test_manager.py b/tests/test_manager.py index ad0a57aa..f6209e5a 100644 --- a/tests/test_manager.py +++ b/tests/test_manager.py @@ -7,7 +7,12 @@ import pytest -from custom_components.zaptec.manager import _stream_supervisor +from custom_components.zaptec import manager as manager_module +from custom_components.zaptec.manager import ( + STREAM_RECONNECT_INIT_DELAY, + STREAM_RECONNECT_MAX_DELAY, + _stream_supervisor, +) def _fake_install() -> SimpleNamespace: @@ -28,14 +33,38 @@ async def test_stream_supervisor_stops_when_stream_main_returns_normally() -> No @pytest.mark.asyncio async def test_stream_supervisor_retries_on_exception(monkeypatch: pytest.MonkeyPatch) -> None: - """A raised exception is retried, not left dead.""" + """A raised exception is retried, not left dead, using the initial backoff delay.""" install = _fake_install() install.stream_main.side_effect = [ConnectionError("boom"), None] - monkeypatch.setattr(asyncio, "sleep", AsyncMock()) + sleep_mock = AsyncMock() + monkeypatch.setattr(asyncio, "sleep", sleep_mock) + # Neutralize jitter so the first delay is deterministically the init delay. + monkeypatch.setattr(manager_module.random, "normalvariate", lambda mu, sigma: mu) # noqa: ARG005 await _stream_supervisor(install, cb=AsyncMock(), ssl_context=None) assert install.stream_main.await_count == 2 # noqa: PLR2004 + sleep_mock.assert_awaited_once() + (delay,), _ = sleep_mock.await_args + assert delay == pytest.approx(STREAM_RECONNECT_INIT_DELAY) + + +@pytest.mark.asyncio +async def test_stream_supervisor_backoff_never_exceeds_max_delay( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Across many consecutive failures, every sleep delay stays within the cap.""" + install = _fake_install() + install.stream_main.side_effect = [ConnectionError("boom")] * 10 + [None] + sleep_mock = AsyncMock() + monkeypatch.setattr(asyncio, "sleep", sleep_mock) + + await _stream_supervisor(install, cb=AsyncMock(), ssl_context=None) + + assert install.stream_main.await_count == 11 # noqa: PLR2004 + assert sleep_mock.await_count == 10 # noqa: PLR2004 + for (delay,), _ in sleep_mock.await_args_list: + assert delay <= STREAM_RECONNECT_MAX_DELAY @pytest.mark.asyncio @@ -83,8 +112,6 @@ async def test_stream_supervisor_resets_backoff_after_long_lived_connection( DEBUG instead (see test_stream_supervisor_logs_warning_once_then_debug for that same-outage case). """ - from custom_components.zaptec import manager as manager_module # noqa: PLC0415 - install = _fake_install() install.stream_main.side_effect = [ConnectionError("1"), ConnectionError("2"), None] monkeypatch.setattr(asyncio, "sleep", AsyncMock()) From 7bc348f378dafe5daf548d7c1a47c3955941c68a Mon Sep 17 00:00:00 2001 From: rhammen <75572839+rhammen@users.noreply.github.com> Date: Tue, 28 Jul 2026 23:39:20 +0200 Subject: [PATCH 07/16] Tighten _stream_supervisor() docstring wording Co-Authored-By: Claude Sonnet 5 --- custom_components/zaptec/manager.py | 12 ++++-------- 1 file changed, 4 insertions(+), 8 deletions(-) diff --git a/custom_components/zaptec/manager.py b/custom_components/zaptec/manager.py index 437f0478..5264d8ef 100644 --- a/custom_components/zaptec/manager.py +++ b/custom_components/zaptec/manager.py @@ -58,14 +58,10 @@ async def _stream_supervisor( still stops this immediately, whether currently inside stream_main() or in the backoff sleep below. - The first failure of an outage is logged at WARNING; subsequent - failures of the same outage are logged at DEBUG only. An outage is - considered "new" -- and thus re-warned -- only once the previous - connection stayed up for at least STREAM_RECONNECT_MAX_DELAY seconds - before failing again. A rapidly flapping stream (reconnecting and - failing faster than that) will therefore log its first failure as - WARNING and every subsequent failure in that flapping sequence as - DEBUG only -- it does not periodically re-warn while flapping. + The first failure of an outage logs at WARNING, later failures of the + same outage at DEBUG. An outage only counts as "new" once the prior + connection survived STREAM_RECONNECT_MAX_DELAY seconds -- a stream + flapping faster than that warns once, then stays at DEBUG. """ delay = STREAM_RECONNECT_INIT_DELAY warned = False From ef93474425721d180d16b549d8e808eab7be2c89 Mon Sep 17 00:00:00 2001 From: rhammen <75572839+rhammen@users.noreply.github.com> Date: Tue, 28 Jul 2026 23:49:21 +0200 Subject: [PATCH 08/16] Trim _stream_supervisor() docstring to the correctness-critical parts Per this repo's demonstrated review bar (PR #415, commit 2359337): maintainers cut explanatory comments even when accurate, not just stale/duplicative ones. The warn-once-per-outage paragraph explained a behavior detail already covered by the inline comment at the reset condition -- kept the return/raise contract and the CancelledError note since those are safety-critical to not breaking the function. Co-Authored-By: Claude Sonnet 5 --- custom_components/zaptec/manager.py | 16 ++++------------ 1 file changed, 4 insertions(+), 12 deletions(-) diff --git a/custom_components/zaptec/manager.py b/custom_components/zaptec/manager.py index 5264d8ef..0f35104c 100644 --- a/custom_components/zaptec/manager.py +++ b/custom_components/zaptec/manager.py @@ -50,18 +50,10 @@ async def _stream_supervisor( ) -> None: """Run install.stream_main(), reconnecting after a transient failure. - stream_main() returning normally means a permanent stop (e.g. no - permission to the live stream) -- this loop ends without retrying. It - raising means a transient failure to retry with exponential backoff. - asyncio.CancelledError is a BaseException, not an Exception, so it is - never caught here: cancelling the task (integration unload/reload) - still stops this immediately, whether currently inside stream_main() - or in the backoff sleep below. - - The first failure of an outage logs at WARNING, later failures of the - same outage at DEBUG. An outage only counts as "new" once the prior - connection survived STREAM_RECONNECT_MAX_DELAY seconds -- a stream - flapping faster than that warns once, then stays at DEBUG. + stream_main() returning normally means a permanent stop; raising means + a transient failure to retry with backoff. asyncio.CancelledError is a + BaseException, not an Exception, so it's never caught here -- task + cancellation still stops this immediately. """ delay = STREAM_RECONNECT_INIT_DELAY warned = False From baa2db8c9048f8a2dbd57544968fb041ebcc6518 Mon Sep 17 00:00:00 2001 From: rhammen <75572839+rhammen@users.noreply.github.com> Date: Tue, 28 Jul 2026 23:56:09 +0200 Subject: [PATCH 09/16] Trim backoff-reset test's comments to the load-bearing facts Same bar as the manager.py docstring trim: keep what's necessary to not accidentally break the test (the clock-sequence meaning, why the next(clock, 500.0) fallback exists), cut the restatement/elaboration. Co-Authored-By: Claude Sonnet 5 --- tests/test_manager.py | 25 ++++++++----------------- 1 file changed, 8 insertions(+), 17 deletions(-) diff --git a/tests/test_manager.py b/tests/test_manager.py index f6209e5a..6919ba81 100644 --- a/tests/test_manager.py +++ b/tests/test_manager.py @@ -106,28 +106,19 @@ async def test_stream_supervisor_resets_backoff_after_long_lived_connection( ) -> None: """A connection that outlived the max backoff delay counts as a fresh outage. - Verified indirectly through logging: each failure that's treated as a - *new* outage logs at WARNING (the "warned" flag resets alongside the - delay). If the reset didn't happen, the second failure would log at - DEBUG instead (see test_stream_supervisor_logs_warning_once_then_debug - for that same-outage case). + Verified via logging: a failure treated as a new outage warns again (see + test_stream_supervisor_logs_warning_once_then_debug for the same-outage + case, which stays at DEBUG). """ install = _fake_install() install.stream_main.side_effect = [ConnectionError("1"), ConnectionError("2"), None] monkeypatch.setattr(asyncio, "sleep", AsyncMock()) monkeypatch.setattr(manager_module, "STREAM_RECONNECT_MAX_DELAY", 100.0) - # 5 monotonic() calls total: connected_at + failure-check for each of the - # 2 failed attempts, plus connected_at for the 3rd (successful) attempt. - # The gap between attempt 2's connected_at (0.0) and its failure-check - # (200.0) exceeds MAX_DELAY (100.0), so that failure counts as a new - # outage rather than a continuation of the first. - # `next(clock, 500.0)` (not bare `next(clock)`): on Windows, asyncio's - # ProactorEventLoop calls time.monotonic() a few more times during event - # loop teardown, after _stream_supervisor has already returned. A bare - # next(clock) would raise StopIteration from inside asyncio's own - # teardown for those extra calls; the fallback keeps that harmless while - # the 5 calls _stream_supervisor actually makes still get the exact - # scripted sequence below. + # 5 monotonic() calls: connected_at + failure-check per failed attempt (x2), + # then connected_at for the successful 3rd. Attempt 2's gap (0.0 -> 200.0) + # exceeds MAX_DELAY (100.0), so it counts as a new outage. + # Fallback (not bare next(clock)): Windows' ProactorEventLoop calls + # monotonic() more times during teardown, after the coroutine has returned. clock = iter([0.0, 0.0, 0.0, 200.0, 500.0]) monkeypatch.setattr(manager_module.time, "monotonic", lambda: next(clock, 500.0)) From 1085a7dc71d0f7736b262a8b0b71610687b84dfa Mon Sep 17 00:00:00 2001 From: rhammen <75572839+rhammen@users.noreply.github.com> Date: Tue, 28 Jul 2026 23:57:28 +0200 Subject: [PATCH 10/16] Trim stream_main test docstrings to match file convention Every other test in this file uses a single-line docstring; these two stood out at 3 lines each. Kept the issue-#417 reference as a compact parenthetical, matching the existing precedent in test_init.py:25. Co-Authored-By: Claude Sonnet 5 --- tests/zaptec/test_api.py | 13 ++----------- 1 file changed, 2 insertions(+), 11 deletions(-) diff --git a/tests/zaptec/test_api.py b/tests/zaptec/test_api.py index e82589d7..1a43b779 100644 --- a/tests/zaptec/test_api.py +++ b/tests/zaptec/test_api.py @@ -553,12 +553,7 @@ def test_stream_update_zero_guid_is_ignored() -> None: @pytest.mark.asyncio async def test_stream_main_propagates_non_forbidden_error() -> None: - """A non-403 error fetching stream connection details now propagates. - - Previously this was swallowed internally (logged, then stream_main() - returned None) so a caller had no way to distinguish "transient - failure, please retry" from "stream ended cleanly". See issue #417. - """ + """A non-403 error fetching stream connection details now propagates (issue #417).""" inst = Installation({"Id": "inst-1"}, _fake_owner()) inst.live_stream_connection_details = AsyncMock( # type: ignore[method-assign] side_effect=RequestError("server error", HTTPStatus.BAD_GATEWAY) @@ -569,11 +564,7 @@ async def test_stream_main_propagates_non_forbidden_error() -> None: @pytest.mark.asyncio async def test_stream_main_forbidden_returns_none() -> None: - """A 403 fetching stream connection details still returns cleanly. - - This remains the "permanent stop, don't retry" signal the stream - supervisor (manager.py) relies on. - """ + """A 403 fetching stream connection details still returns cleanly.""" inst = Installation({"Id": "inst-1"}, _fake_owner()) inst.live_stream_connection_details = AsyncMock( # type: ignore[method-assign] side_effect=RequestError("no access", HTTPStatus.FORBIDDEN) From 4405b1e7960dfd46ce434b465a83ff92b16f7a4d Mon Sep 17 00:00:00 2001 From: rhammen <75572839+rhammen@users.noreply.github.com> Date: Wed, 29 Jul 2026 00:03:16 +0200 Subject: [PATCH 11/16] docs: remove planning docs, archived on docs/ai-planning-archive Co-Authored-By: Claude Sonnet 5 --- .../plans/2026-07-28-stream-reconnect.md | 520 ------------------ .../2026-07-28-stream-reconnect-design.md | 176 ------ 2 files changed, 696 deletions(-) delete mode 100644 docs/superpowers/plans/2026-07-28-stream-reconnect.md delete mode 100644 docs/superpowers/specs/2026-07-28-stream-reconnect-design.md diff --git a/docs/superpowers/plans/2026-07-28-stream-reconnect.md b/docs/superpowers/plans/2026-07-28-stream-reconnect.md deleted file mode 100644 index bb381ade..00000000 --- a/docs/superpowers/plans/2026-07-28-stream-reconnect.md +++ /dev/null @@ -1,520 +0,0 @@ -# Stream Reconnect on Transient Failure Implementation Plan - -> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. - -**Goal:** Make the Zaptec live-update stream (`Installation.stream_main()`) automatically reconnect with exponential backoff after a transient connection failure (e.g. a home internet outage), instead of permanently dying until the user reloads Home Assistant. - -**Architecture:** `stream_main()` stops swallowing its own exceptions, so a transient failure now propagates to its caller instead of returning silently. A new module-level `_stream_supervisor()` coroutine in `manager.py` becomes the actual background-task body (replacing the direct `stream_main()` call): it calls `stream_main()` in a loop, retries with exponential backoff+jitter on any raised exception, and stops for good if `stream_main()` ever returns normally (its existing signal for "no permission to the stream", HTTP 403). - -**Tech Stack:** Python 3.14, Home Assistant custom integration, `asyncio`, `azure-servicebus` (vendored stream client), `pytest` + `pytest-asyncio` (plain pytest harness, no `pytest-homeassistant-custom-component` yet in this repo). - -## Global Constraints - -- Follow `ruff format` / `ruff check` (repo uses `select = ["ALL"]` in `.ruff.toml`) — run both before considering a task done. -- Tests run via: `SKIP_ZAPTEC_API_TEST=true "C:/Users/rhamm/anaconda3/envs/py314/python.exe" -m pytest -p no:homeassistant -q` — the `-p no:homeassistant` flag is required on this Windows dev setup (avoids an `fcntl`-dependent pytest plugin that doesn't exist on Windows); never add a compatibility shim instead, just pass the flag. -- Never commit without the user's explicit go-ahead (existing session convention — ask before every `git commit`). -- `asyncio.CancelledError` is a `BaseException`, not an `Exception`, on the Python version this repo targets — `except Exception` blocks must not (and currently do not) catch it. Don't add a bare `except:` or `except BaseException` anywhere in this work. - ---- - -### Task 1: Stop `stream_main()` swallowing its own exceptions - -**Files:** -- Modify: `custom_components/zaptec/zaptec/api.py:397` and `:474-475` (the `try:`/`except Exception:` wrapping `Installation.stream_main()`'s body) -- Test: `tests/zaptec/test_api.py` - -**Interfaces:** -- Consumes: nothing new — this task only changes existing `Installation.stream_main()` control flow. -- Produces: `Installation.stream_main()` now propagates any exception raised while fetching stream connection details or consuming the stream, instead of catching it and returning `None`. The existing 403/Forbidden case is unchanged — it still logs a warning and returns `None` (this is the "permanent stop, don't retry" signal Task 2's supervisor relies on). - -Current code (for reference — do not copy verbatim, this is what you're changing): - -```python - try: - self._stream_running = True - - # Get connection details - try: - conf = await self.live_stream_connection_details() - except RequestError as err: - if err.error_code != HTTPStatus.FORBIDDEN: - raise - _LOGGER.warning( - "Failed to get live stream info. " - "Check if user have access in the zaptec portal" - ) - return - - # ... (connection setup and the `async for msg in receiver:` loop, unchanged) ... - - except Exception: - # Do this in order to show the error in the log. - _LOGGER.exception("Stream failed") - finally: - self._stream_receiver = None - self._stream_running = False - _LOGGER.info("Servicebus stream stopped for %s", self.qual_id) -``` - -- [ ] **Step 1: Write the failing tests** - -Add to the bottom of `tests/zaptec/test_api.py` (after the existing `Installation.stream_update routing` section, i.e. after the `test_stream_update_zero_guid_is_ignored` test): - -```python -# --------------------------------------------------------------------------- -# Installation.stream_main error propagation -# --------------------------------------------------------------------------- - - -@pytest.mark.asyncio -async def test_stream_main_propagates_non_forbidden_error() -> None: - """A non-403 error fetching stream connection details now propagates. - - Previously this was swallowed internally (logged, then stream_main() - returned None) so a caller had no way to distinguish "transient - failure, please retry" from "stream ended cleanly". See issue #417. - """ - inst = Installation({"Id": "inst-1"}, _fake_owner()) - inst.live_stream_connection_details = AsyncMock( # type: ignore[method-assign] - side_effect=RequestError("server error", HTTPStatus.BAD_GATEWAY) - ) - with pytest.raises(RequestError): - await inst.stream_main() - - -@pytest.mark.asyncio -async def test_stream_main_forbidden_returns_none() -> None: - """A 403 fetching stream connection details still returns cleanly. - - This remains the "permanent stop, don't retry" signal the stream - supervisor (manager.py) relies on. - """ - inst = Installation({"Id": "inst-1"}, _fake_owner()) - inst.live_stream_connection_details = AsyncMock( # type: ignore[method-assign] - side_effect=RequestError("no access", HTTPStatus.FORBIDDEN) - ) - result = await inst.stream_main() - assert result is None -``` - -- [ ] **Step 2: Run tests to verify the first one fails** - -Run: `SKIP_ZAPTEC_API_TEST=true "C:/Users/rhamm/anaconda3/envs/py314/python.exe" -m pytest tests/zaptec/test_api.py -k stream_main -p no:homeassistant -v` - -Expected: `test_stream_main_propagates_non_forbidden_error` FAILS (no exception raised — it's currently swallowed). `test_stream_main_forbidden_returns_none` PASSES already (existing behavior, unaffected by this task — it's here as a regression guard). - -- [ ] **Step 3: Remove the outer `except Exception` in `stream_main()`** - -In `custom_components/zaptec/zaptec/api.py`, change: - -```python - except Exception: - # Do this in order to show the error in the log. - _LOGGER.exception("Stream failed") - finally: -``` - -to: - -```python - finally: -``` - -(i.e. delete the `except Exception:` block entirely, keeping the `try:` / `finally:` around the same body unchanged.) - -- [ ] **Step 4: Run tests to verify they pass** - -Run: `SKIP_ZAPTEC_API_TEST=true "C:/Users/rhamm/anaconda3/envs/py314/python.exe" -m pytest tests/zaptec/test_api.py -p no:homeassistant -v` - -Expected: PASS (full file, to confirm nothing else in `test_api.py` broke). - -- [ ] **Step 5: Lint** - -Run: -``` -"C:/Users/rhamm/anaconda3/envs/py314/python.exe" -m ruff format custom_components/zaptec/zaptec/api.py tests/zaptec/test_api.py --diff -"C:/Users/rhamm/anaconda3/envs/py314/python.exe" -m ruff check custom_components/zaptec/zaptec/api.py tests/zaptec/test_api.py -``` -Expected: no diff, no lint errors. If `ruff format --diff` shows a diff, run it without `--diff` to apply, then re-run `ruff check`. - -- [ ] **Step 6: Commit** - -Ask the user for explicit go-ahead first (per this repo's convention — never commit automatically). Once approved: - -```bash -git add custom_components/zaptec/zaptec/api.py tests/zaptec/test_api.py -git commit -m "$(cat <<'EOF' -Let stream_main() propagate transient failures instead of swallowing them - -Only the existing 403/Forbidden case still returns cleanly; any other -failure now raises so a caller can distinguish "retry me" from -"permanent stop". Prep for issue #417's reconnect supervisor. - -Co-Authored-By: Claude Sonnet 5 -EOF -)" -``` - ---- - -### Task 2: Add reconnect-with-backoff supervisor and wire it into `create_streams()` - -**Files:** -- Modify: `custom_components/zaptec/const.py` (add `STREAM_RECONNECT_*` constants) -- Modify: `custom_components/zaptec/manager.py` (add module-level `_stream_supervisor()`, update imports, update `create_streams()`) -- Test: `tests/test_manager.py` (new file) - -**Interfaces:** -- Consumes: `Installation.stream_main(cb, ssl_context) -> None` from Task 1 (raises on transient failure, returns `None` on permanent stop or normal completion). -- Produces: `_stream_supervisor(install: Installation, cb: Callable[[dict], Awaitable[None]], ssl_context: ssl.SSLContext | None) -> None` — a module-level coroutine in `manager.py` (not a `ZaptecManager` method, so it's testable without constructing a full manager/config-entry). `create_streams()` now schedules `_stream_supervisor(...)` as the background task instead of calling `install.stream_main(...)` directly. No other public interface changes — `cancel_streams()` is untouched and still works because `task.cancel()` interrupts whatever `_stream_supervisor` is currently awaiting (either inside `stream_main()`, or the backoff `asyncio.sleep`). - -- [ ] **Step 1: Add the new constants** - -In `custom_components/zaptec/const.py`, after the existing `ZAPTEC_POLL_INSTALLATION_TRIGGER_DELAYS` constant (and before `REQUEST_REFRESH_DELAY`), add: - -```python -STREAM_RECONNECT_INIT_DELAY = 1.0 -"""Initial delay in seconds before the first stream reconnect attempt.""" - -STREAM_RECONNECT_FACTOR = 2.0 -"""Exponential backoff multiplier applied between stream reconnect attempts.""" - -STREAM_RECONNECT_JITTER = 0.1 -"""Relative jitter applied to the stream reconnect backoff delay.""" - -STREAM_RECONNECT_MAX_DELAY = 300.0 -"""Maximum delay in seconds between stream reconnect attempts (5 minutes).""" -``` - -- [ ] **Step 2: Write the failing tests** - -Create `tests/test_manager.py`: - -```python -"""Tests for custom_components.zaptec.manager.""" - -import asyncio -import logging -from types import SimpleNamespace -from unittest.mock import AsyncMock - -import pytest - -from custom_components.zaptec.manager import _stream_supervisor - - -def _fake_install() -> SimpleNamespace: - """Return a stand-in for Installation carrying only what _stream_supervisor uses.""" - return SimpleNamespace(qual_id="Installation[nst-1]", stream_main=AsyncMock()) - - -@pytest.mark.asyncio -async def test_stream_supervisor_stops_when_stream_main_returns_normally() -> None: - """stream_main() returning None (e.g. 403/Forbidden) is a permanent stop.""" - install = _fake_install() - install.stream_main.return_value = None - - await _stream_supervisor(install, cb=AsyncMock(), ssl_context=None) - - install.stream_main.assert_awaited_once() - - -@pytest.mark.asyncio -async def test_stream_supervisor_retries_on_exception(monkeypatch: pytest.MonkeyPatch) -> None: - """A raised exception is retried, not left dead.""" - install = _fake_install() - install.stream_main.side_effect = [ConnectionError("boom"), None] - monkeypatch.setattr(asyncio, "sleep", AsyncMock()) - - await _stream_supervisor(install, cb=AsyncMock(), ssl_context=None) - - assert install.stream_main.await_count == 2 - - -@pytest.mark.asyncio -async def test_stream_supervisor_propagates_cancelled_error_without_retrying( - monkeypatch: pytest.MonkeyPatch, -) -> None: - """Task cancellation (integration unload/reload) is not treated as a retryable failure.""" - install = _fake_install() - install.stream_main.side_effect = asyncio.CancelledError() - monkeypatch.setattr(asyncio, "sleep", AsyncMock()) - - with pytest.raises(asyncio.CancelledError): - await _stream_supervisor(install, cb=AsyncMock(), ssl_context=None) - - install.stream_main.assert_awaited_once() - - -@pytest.mark.asyncio -async def test_stream_supervisor_logs_warning_once_then_debug( - monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture -) -> None: - """Only the first failure of an outage logs at warning; the rest log at debug.""" - install = _fake_install() - install.stream_main.side_effect = [ConnectionError("1"), ConnectionError("2"), None] - monkeypatch.setattr(asyncio, "sleep", AsyncMock()) - - with caplog.at_level(logging.DEBUG, logger="custom_components.zaptec.manager"): - await _stream_supervisor(install, cb=AsyncMock(), ssl_context=None) - - warnings = [r for r in caplog.records if r.levelno == logging.WARNING] - debugs = [r for r in caplog.records if r.levelno == logging.DEBUG] - assert len(warnings) == 1 - assert len(debugs) == 1 - - -@pytest.mark.asyncio -async def test_stream_supervisor_resets_backoff_after_long_lived_connection( - monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture -) -> None: - """A connection that outlived the max backoff delay counts as a fresh outage. - - Verified indirectly through logging: each failure that's treated as a - *new* outage logs at WARNING (the "warned" flag resets alongside the - delay). If the reset didn't happen, the second failure would log at - DEBUG instead (see test_stream_supervisor_logs_warning_once_then_debug - for that same-outage case). - """ - from custom_components.zaptec import manager as manager_module - - install = _fake_install() - install.stream_main.side_effect = [ConnectionError("1"), ConnectionError("2"), None] - monkeypatch.setattr(asyncio, "sleep", AsyncMock()) - monkeypatch.setattr(manager_module, "STREAM_RECONNECT_MAX_DELAY", 100.0) - # 5 monotonic() calls total: connected_at + failure-check for each of the - # 2 failed attempts, plus connected_at for the 3rd (successful) attempt. - # The gap between attempt 2's connected_at (0.0) and its failure-check - # (200.0) exceeds MAX_DELAY (100.0), so that failure counts as a new - # outage rather than a continuation of the first. - clock = iter([0.0, 0.0, 0.0, 200.0, 500.0]) - monkeypatch.setattr(manager_module.time, "monotonic", lambda: next(clock)) - - with caplog.at_level(logging.DEBUG, logger="custom_components.zaptec.manager"): - await _stream_supervisor(install, cb=AsyncMock(), ssl_context=None) - - warnings = [r for r in caplog.records if r.levelno == logging.WARNING] - assert len(warnings) == 2 # both failures counted as separate outages -``` - -- [ ] **Step 3: Run tests to verify they fail** - -Run: `SKIP_ZAPTEC_API_TEST=true "C:/Users/rhamm/anaconda3/envs/py314/python.exe" -m pytest tests/test_manager.py -p no:homeassistant -v` - -Expected: FAIL with `ModuleNotFoundError`/`ImportError` (`_stream_supervisor` doesn't exist yet). - -- [ ] **Step 4: Update `manager.py` imports** - -Change the import block at the top of `custom_components/zaptec/manager.py` from: - -```python -import asyncio -from collections.abc import Iterable -import contextlib -from copy import copy -from dataclasses import dataclass -import logging -from typing import Any - -from homeassistant.config_entries import ConfigEntry -from homeassistant.core import HomeAssistant -from homeassistant.helpers.entity import DeviceInfo, EntityDescription -from homeassistant.util.ssl import get_default_context - -from .const import DOMAIN, KEYS_TO_SKIP_ENTITY_AVAILABILITY_CHECK, MANUFACTURER -from .coordinator import ZaptecUpdateCoordinator -from .entity import KeyUnavailableError, ZaptecBaseEntity -from .zaptec import Charger, Installation, Zaptec, ZaptecBase -``` - -to: - -```python -import asyncio -from collections.abc import Awaitable, Callable, Iterable -import contextlib -from copy import copy -from dataclasses import dataclass -import logging -import random -import ssl -import time -from typing import Any - -from homeassistant.config_entries import ConfigEntry -from homeassistant.core import HomeAssistant -from homeassistant.helpers.entity import DeviceInfo, EntityDescription -from homeassistant.util.ssl import get_default_context - -from .const import ( - DOMAIN, - KEYS_TO_SKIP_ENTITY_AVAILABILITY_CHECK, - MANUFACTURER, - STREAM_RECONNECT_FACTOR, - STREAM_RECONNECT_INIT_DELAY, - STREAM_RECONNECT_JITTER, - STREAM_RECONNECT_MAX_DELAY, -) -from .coordinator import ZaptecUpdateCoordinator -from .entity import KeyUnavailableError, ZaptecBaseEntity -from .zaptec import Charger, Installation, Zaptec, ZaptecBase -``` - -- [ ] **Step 5: Add `_stream_supervisor()`** - -In `custom_components/zaptec/manager.py`, add this module-level function right before `class ZaptecManager:`: - -```python -async def _stream_supervisor( - install: Installation, - cb: Callable[[dict], Awaitable[None]], - ssl_context: ssl.SSLContext | None, -) -> None: - """Run install.stream_main(), reconnecting after a transient failure. - - stream_main() returning normally means a permanent stop (e.g. no - permission to the live stream) -- this loop ends without retrying. It - raising means a transient failure to retry with exponential backoff. - asyncio.CancelledError is a BaseException, not an Exception, so it is - never caught here: cancelling the task (integration unload/reload) - still stops this immediately, whether currently inside stream_main() - or in the backoff sleep below. - """ - delay = STREAM_RECONNECT_INIT_DELAY - warned = False - while True: - connected_at = time.monotonic() - try: - await install.stream_main(cb=cb, ssl_context=ssl_context) - return - except Exception: - if time.monotonic() - connected_at >= STREAM_RECONNECT_MAX_DELAY: - # The previous connection lived long enough to count this - # as a fresh outage rather than a continuation of the last. - delay = STREAM_RECONNECT_INIT_DELAY - warned = False - if not warned: - _LOGGER.warning( - "Stream for %s disconnected, reconnecting", install.qual_id, exc_info=True - ) - warned = True - else: - _LOGGER.debug( - "Stream for %s still reconnecting", install.qual_id, exc_info=True - ) - await asyncio.sleep(delay) - delay = delay * STREAM_RECONNECT_FACTOR - delay = random.normalvariate(delay, delay * STREAM_RECONNECT_JITTER) - delay = min(delay, STREAM_RECONNECT_MAX_DELAY) -``` - -- [ ] **Step 6: Wire it into `create_streams()`** - -In `custom_components/zaptec/manager.py`, change: - -```python - def create_streams(self) -> None: - """Create the streams for all installations.""" - for install in self.zaptec.installations: - if install.id in self.zaptec: - task = self.config_entry.async_create_background_task( - self.hass, - install.stream_main( - cb=self.stream_callback, - ssl_context=get_default_context(), - ), - name=f"Zaptec Stream for {install.qual_id}", - ) - self.streams.append((task, install)) -``` - -to: - -```python - def create_streams(self) -> None: - """Create the streams for all installations.""" - for install in self.zaptec.installations: - if install.id in self.zaptec: - task = self.config_entry.async_create_background_task( - self.hass, - _stream_supervisor( - install, - cb=self.stream_callback, - ssl_context=get_default_context(), - ), - name=f"Zaptec Stream for {install.qual_id}", - ) - self.streams.append((task, install)) -``` - -(`cancel_streams()` below it is unchanged — no edit needed there.) - -- [ ] **Step 7: Run tests to verify they pass** - -Run: `SKIP_ZAPTEC_API_TEST=true "C:/Users/rhamm/anaconda3/envs/py314/python.exe" -m pytest tests/test_manager.py -p no:homeassistant -v` - -Expected: all 5 tests PASS. - -- [ ] **Step 8: Lint** - -Run: -``` -"C:/Users/rhamm/anaconda3/envs/py314/python.exe" -m ruff format custom_components/zaptec/const.py custom_components/zaptec/manager.py tests/test_manager.py --diff -"C:/Users/rhamm/anaconda3/envs/py314/python.exe" -m ruff check custom_components/zaptec/const.py custom_components/zaptec/manager.py tests/test_manager.py -``` -Expected: no diff, no lint errors. If `ruff format --diff` shows a diff, run it without `--diff` to apply, then re-run `ruff check`. - -- [ ] **Step 9: Commit** - -Ask the user for explicit go-ahead first. Once approved: - -```bash -git add custom_components/zaptec/const.py custom_components/zaptec/manager.py tests/test_manager.py -git commit -m "$(cat <<'EOF' -Add reconnect-with-backoff supervisor for the live update stream - -Wraps stream_main() in an exponential-backoff retry loop so a transient -connection failure (e.g. a brief home-internet outage) reconnects -automatically instead of permanently killing live updates until the -user reloads the integration. Fixes #417. - -Co-Authored-By: Claude Sonnet 5 -EOF -)" -``` - ---- - -### Task 3: Full verification - -**Files:** none (verification only) - -**Interfaces:** -- Consumes: everything from Tasks 1-2. -- Produces: nothing new — confirms the full suite and lint are green together, not just per-file. - -- [ ] **Step 1: Run the full test suite** - -Run: `SKIP_ZAPTEC_API_TEST=true "C:/Users/rhamm/anaconda3/envs/py314/python.exe" -m pytest tests -p no:homeassistant -q` - -Expected: all tests pass (aside from any pre-existing `SKIP_ZAPTEC_API_TEST`/live-network skips unrelated to this change). - -- [ ] **Step 2: Run full-repo lint** - -Run: -``` -"C:/Users/rhamm/anaconda3/envs/py314/python.exe" -m ruff format . --diff -"C:/Users/rhamm/anaconda3/envs/py314/python.exe" -m ruff check -``` -Expected: no diff, no errors. This matches what CI's `lint-ruff` job runs (see `CLAUDE.md`), scoped to the whole repo, not just the files touched here. - -- [ ] **Step 3: Report results to the user** - -Summarize pass/fail counts and any lint findings. Do not open a PR or push — per this repo's AI-policy convention, that's a separate, explicit ask. - ---- - -## Design doc - -Full rationale, alternatives considered, and error-handling table: -`docs/superpowers/specs/2026-07-28-stream-reconnect-design.md` diff --git a/docs/superpowers/specs/2026-07-28-stream-reconnect-design.md b/docs/superpowers/specs/2026-07-28-stream-reconnect-design.md deleted file mode 100644 index 596581f0..00000000 --- a/docs/superpowers/specs/2026-07-28-stream-reconnect-design.md +++ /dev/null @@ -1,176 +0,0 @@ -# Stream reconnect on transient connection failure — design - -**Issue:** [custom-components/zaptec#417](https://github.com/custom-components/zaptec/issues/417) - -## Problem - -`Installation.stream_main()` ([zaptec/api.py](../../../custom_components/zaptec/zaptec/api.py)) -opens a persistent Azure Service Bus (AMQP) connection for live push updates. -On any exception — including a transient connection failure such as -`azure.servicebus.exceptions.ServiceBusConnectionError` — it catches the -exception with a blanket `except Exception: _LOGGER.exception("Stream -failed")` and returns normally. - -`ZaptecManager.create_streams()` ([manager.py](../../../custom_components/zaptec/manager.py)) -starts `stream_main()` exactly once per installation, as a background task, -during `async_setup_entry`. Nothing supervises that task or restarts it if it -exits. So a one-off network blip (e.g. a home router reboot) permanently -kills the live stream until the user reloads the integration or restarts -Home Assistant — a silent, indefinite degradation to poll-interval-only -freshness. - -## Goals - -- A transient stream failure (connection error, AMQP error, etc.) is - retried automatically with backoff, indefinitely — the stream should - self-heal without user intervention. -- A permanent condition (no permission to the stream, HTTP 403) is *not* - retried — retrying forever would just be log/network noise for a - situation retrying can't fix. -- `cancel_streams()` (integration unload/reload) must still cleanly stop - the stream, including while a reconnect backoff is in progress. -- Reconnect activity is logged just enough to diagnose an outage, without - spamming a stack trace on every backoff cycle during a prolonged outage. - -## Non-goals - -- Changing polling-coordinator behavior (already handles its own - independent retry — see #393). -- Changing the fallback-poll cadence while the stream is down (tracked - separately, see the "stream reconciliation gap" note referenced from - issue #378 discussion — out of scope here). -- `Installation.stream()` (a second, currently-unused convenience entry - point that wraps `stream_main()` in its own bare `asyncio.create_task`) - is not otherwise used by production code (only `manager.create_streams()` - is). Its behavior changes as a side effect of `stream_main()` no longer - swallowing exceptions (see below), but no new supervision logic is added - there — it isn't called from anywhere in this integration today. - -## Design - -### `stream_main()` contract change - -Remove the outer `except Exception: _LOGGER.exception("Stream failed")` -that currently wraps the whole connect-and-consume body. After the change: - -- **Returns normally** → permanent stop. Currently this is only the - 403/Forbidden case when fetching stream connection details (already - handled today by logging a warning and `return`-ing early). No other - code path returns normally after this change — reaching the end of the - `async for msg in receiver:` loop only happens when the receiver itself - ends the iteration, which in practice means the connection is closing. -- **Raises an exception** → transient failure. The exception propagates to - whoever awaited `stream_main()`. -- **`asyncio.CancelledError`** → not caught by `except Exception` (it is a - `BaseException`, not `Exception`, on the Python versions this integration - targets), so it is unaffected by this change and continues to propagate - straight through, as it does today. - -The `finally` block (clearing `_stream_receiver`, `_stream_running`, and -logging "Servicebus stream stopped for %s") is unchanged — it still runs -on every exit path. - -### Supervising wrapper - -New coroutine `ZaptecManager._stream_supervisor(install: Installation)` in -`manager.py`, used as the task body in `create_streams()` in place of the -direct `install.stream_main(...)` call: - -```python -async def _stream_supervisor(self, install: Installation) -> None: - delay = STREAM_RECONNECT_INIT_DELAY - connected_at: float | None = None - warned = False - while True: - connected_at = time.monotonic() - try: - await install.stream_main(cb=self.stream_callback, ssl_context=get_default_context()) - return # permanent stop (e.g. 403) - except Exception: - if time.monotonic() - connected_at >= STREAM_RECONNECT_MAX_DELAY: - delay = STREAM_RECONNECT_INIT_DELAY # reset after a long-lived connection - if not warned: - _LOGGER.warning( - "Stream for %s disconnected, reconnecting", install.qual_id, exc_info=True - ) - warned = True - else: - _LOGGER.debug("Stream for %s still reconnecting", install.qual_id, exc_info=True) - await asyncio.sleep(delay) - delay = min(delay * STREAM_RECONNECT_FACTOR, STREAM_RECONNECT_MAX_DELAY) - delay = random.normalvariate(delay, delay * STREAM_RECONNECT_JITTER) -``` - -(Illustrative — final implementation may adjust variable names/structure to -match repo style, but the behavior above is the contract.) - -`asyncio.CancelledError` is not caught here either, so `cancel_streams()`'s -existing `task.cancel()` + `await task` continues to stop the supervisor -(and whatever `stream_main()` call is in flight, or the backoff sleep) -immediately, unchanged from today. - -`warned` resets to `False` implicitly each time the loop returns to the top -after a successful reconnect (a fresh `_stream_supervisor` iteration only -warns again if *this* connection attempt also fails) — i.e. only the first -failure of a given outage logs at `warning`; the rest of that outage's -retries log at `debug`. A brand new outage after a successful reconnect -warns again. - -### New constants - -In `zaptec/const.py`, alongside the existing `API_RETRY_*` constants: - -```python -STREAM_RECONNECT_INIT_DELAY = 1.0 -STREAM_RECONNECT_FACTOR = 2.0 -STREAM_RECONNECT_JITTER = 0.1 -STREAM_RECONNECT_MAX_DELAY = 300.0 # 5 minutes -``` - -Kept separate from `API_RETRY_*` since they govern a different thing (a -long-lived connection's reconnect cadence, not a single HTTP request's -retry count) even though the shape (exponential + jitter, capped) matches. - -### Reconnected signal - -No new logging plumbing needed: `stream_main()` already logs -`_LOGGER.info("Running service bus stream for %s", self.qual_id)` once it -successfully opens the receiver. That existing line now doubles as the -"reconnected" signal once a prior failure has warned — satisfying the -Home Assistant integration quality-scale guidance ("log a warning once -when unavailable, log once when reconnected") without adding a new log -statement. - -## Error handling summary - -| Condition | `stream_main()` behavior | Supervisor behavior | -|---|---|---| -| 403 fetching stream connection details | logs warning, returns | stops, no retry | -| `ServiceBusConnectionError` / other transient error | raises | logs once (warn), backs off, retries | -| Integration unload (`cancel_streams()`) | `CancelledError` propagates | `CancelledError` propagates, loop exits | -| Malformed individual stream message | already handled inside the `async for` loop (existing `except Exception: _LOGGER.exception("Couldn't process stream message")`, unchanged) — does not end the stream | n/a | - -## Testing - -- `stream_main()`: existing tests around `stream_update` routing are - unaffected. Add a test confirming a non-403 exception now propagates - out of `stream_main()` instead of being swallowed (behavior change). -- `_stream_supervisor()`: new tests — - - retries with backoff on a raised exception, calling `stream_main()` - again; - - stops (single call, no retry) when `stream_main()` returns normally; - - propagates `CancelledError` without retrying; - - resets backoff delay after a connection that stayed up past - `STREAM_RECONNECT_MAX_DELAY`; - - logs at `warning` only for the first failure of an outage, `debug` - for subsequent ones within the same outage. - -## Open questions / risks - -- `stream_main()` no longer catching its own exceptions means any *unexpected* - bug in message processing that somehow escapes the inner per-message - `try/except` would now also be treated as "transient, retry" by the - supervisor rather than silently logged once and left stopped. This is - considered acceptable — retrying is a reasonable default reaction to an - unexpected stream failure, and the per-message handler already isolates - normal message-processing errors from ending the stream at all. From 85c015fa1b5ffecf93d0b300515f0c01e9647adc Mon Sep 17 00:00:00 2001 From: rhammen <75572839+rhammen@users.noreply.github.com> Date: Tue, 1 Sep 2026 20:08:38 +0200 Subject: [PATCH 12/16] Drop the .superpowers/ gitignore entry Local-only tooling directory; excluded via .git/info/exclude instead. Co-Authored-By: Claude Opus 5 --- .gitignore | 1 - 1 file changed, 1 deletion(-) diff --git a/.gitignore b/.gitignore index 6e5a4fa7..161bb6ba 100644 --- a/.gitignore +++ b/.gitignore @@ -11,7 +11,6 @@ __pycache__ coverage.xml .ruff_cache htmlcov/ -.superpowers/ # Home Assistant configuration config/* From 49c35224181e7e0afb558482abc7cb2eea05182e Mon Sep 17 00:00:00 2001 From: rhammen <75572839+rhammen@users.noreply.github.com> Date: Tue, 1 Sep 2026 20:14:50 +0200 Subject: [PATCH 13/16] Report stream reconnects and reset the outage state on connect stream_main() gained an optional on_connect callback, fired once the service bus receiver is live. The supervisor uses it to log "connected after N reconnect attempt(s)" and to time the connection, so a stream that stayed up past STREAM_RECONNECT_STABLE_TIME resets the backoff and the warn-once state instead of relying on the max-delay heuristic. Co-Authored-By: Claude Opus 5 --- custom_components/zaptec/const.py | 3 ++ custom_components/zaptec/manager.py | 37 ++++++++++++----- custom_components/zaptec/zaptec/api.py | 7 +++- tests/test_manager.py | 56 +++++++++++++++++++++----- 4 files changed, 80 insertions(+), 23 deletions(-) diff --git a/custom_components/zaptec/const.py b/custom_components/zaptec/const.py index f939e33a..20ace25a 100644 --- a/custom_components/zaptec/const.py +++ b/custom_components/zaptec/const.py @@ -48,6 +48,9 @@ STREAM_RECONNECT_MAX_DELAY = 300.0 """Maximum delay in seconds between stream reconnect attempts (5 minutes).""" +STREAM_RECONNECT_STABLE_TIME = 60.0 +"""Uptime in seconds after which a stream connection counts as healthy again.""" + # This sets the delay after doing actions and the poll of updated values. # It was 0.3 and evidently that is a bit too fast for Zaptec cloud to handle. REQUEST_REFRESH_DELAY = 1 diff --git a/custom_components/zaptec/manager.py b/custom_components/zaptec/manager.py index 0f35104c..9afb9060 100644 --- a/custom_components/zaptec/manager.py +++ b/custom_components/zaptec/manager.py @@ -26,6 +26,7 @@ STREAM_RECONNECT_INIT_DELAY, STREAM_RECONNECT_JITTER, STREAM_RECONNECT_MAX_DELAY, + STREAM_RECONNECT_STABLE_TIME, ) from .coordinator import ZaptecUpdateCoordinator from .entity import KeyUnavailableError, ZaptecBaseEntity @@ -56,24 +57,38 @@ async def _stream_supervisor( cancellation still stops this immediately. """ delay = STREAM_RECONNECT_INIT_DELAY - warned = False - while True: + reconnects = 0 + connected_at: float | None = None + + def on_connect() -> None: + nonlocal connected_at connected_at = time.monotonic() + if reconnects: + _LOGGER.info( + "Stream for %s connected after %s reconnect attempt(s)", + install.qual_id, + reconnects, + ) + + while True: + connected_at = None try: - await install.stream_main(cb=cb, ssl_context=ssl_context) + await install.stream_main(cb=cb, ssl_context=ssl_context, on_connect=on_connect) except Exception: - if time.monotonic() - connected_at >= STREAM_RECONNECT_MAX_DELAY: - # The previous connection lived long enough to count this - # as a fresh outage rather than a continuation of the last. + # A connection that stayed up counts as recovered, so the next + # failure starts a fresh outage instead of continuing the last. + if connected_at is not None and ( + time.monotonic() - connected_at >= STREAM_RECONNECT_STABLE_TIME + ): delay = STREAM_RECONNECT_INIT_DELAY - warned = False - if not warned: + reconnects = 0 + if reconnects: + _LOGGER.debug("Stream for %s still reconnecting", install.qual_id, exc_info=True) + else: _LOGGER.warning( "Stream for %s disconnected, reconnecting", install.qual_id, exc_info=True ) - warned = True - else: - _LOGGER.debug("Stream for %s still reconnecting", install.qual_id, exc_info=True) + reconnects += 1 await asyncio.sleep(delay) delay = min(delay * STREAM_RECONNECT_FACTOR, STREAM_RECONNECT_MAX_DELAY) delay = random.normalvariate(delay, delay * STREAM_RECONNECT_JITTER) diff --git a/custom_components/zaptec/zaptec/api.py b/custom_components/zaptec/zaptec/api.py index be677b48..c7527150 100644 --- a/custom_components/zaptec/zaptec/api.py +++ b/custom_components/zaptec/zaptec/api.py @@ -386,7 +386,10 @@ def _stream_log(self, data: dict[str, Any]) -> None: _LOGGER.debug("@@@ EVENT %s", self.zaptec.redact(data)) async def stream_main( - self, cb: StreamCallback | None = None, ssl_context: ssl.SSLContext | None = None + self, + cb: StreamCallback | None = None, + ssl_context: ssl.SSLContext | None = None, + on_connect: Callable[[], None] | None = None, ) -> None: """Main stream handler.""" # Already running? @@ -432,6 +435,8 @@ async def stream_main( subscription_name=conf["Subscription"], ) _LOGGER.info("Running service bus stream for %s", self.qual_id) + if on_connect: + on_connect() # Store the receiver in order to close it and cancel this stream self._stream_receiver = receiver async with receiver: diff --git a/tests/test_manager.py b/tests/test_manager.py index 6919ba81..b7283b05 100644 --- a/tests/test_manager.py +++ b/tests/test_manager.py @@ -1,6 +1,7 @@ """Tests for custom_components.zaptec.manager.""" import asyncio +from collections.abc import Callable import logging from types import SimpleNamespace from unittest.mock import AsyncMock @@ -11,6 +12,7 @@ from custom_components.zaptec.manager import ( STREAM_RECONNECT_INIT_DELAY, STREAM_RECONNECT_MAX_DELAY, + STREAM_RECONNECT_STABLE_TIME, _stream_supervisor, ) @@ -101,29 +103,61 @@ async def test_stream_supervisor_logs_warning_once_then_debug( @pytest.mark.asyncio -async def test_stream_supervisor_resets_backoff_after_long_lived_connection( +async def test_stream_supervisor_resets_backoff_after_stable_connection( monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture ) -> None: - """A connection that outlived the max backoff delay counts as a fresh outage. + """A connection that stayed up past the stable time counts as a fresh outage. Verified via logging: a failure treated as a new outage warns again (see test_stream_supervisor_logs_warning_once_then_debug for the same-outage case, which stays at DEBUG). """ install = _fake_install() - install.stream_main.side_effect = [ConnectionError("1"), ConnectionError("2"), None] + now = 0.0 + # Fail, then hold a connection past the stable time before failing again. + attempts = iter([False, True]) + + async def stream_main(*, on_connect: Callable[[], None], **kwargs: object) -> None: + nonlocal now + connects = next(attempts, None) + if connects is None: + return # permanent stop, ends the supervisor loop + if connects: + on_connect() + now += STREAM_RECONNECT_STABLE_TIME + 1 + raise ConnectionError("boom") + + install.stream_main.side_effect = stream_main monkeypatch.setattr(asyncio, "sleep", AsyncMock()) - monkeypatch.setattr(manager_module, "STREAM_RECONNECT_MAX_DELAY", 100.0) - # 5 monotonic() calls: connected_at + failure-check per failed attempt (x2), - # then connected_at for the successful 3rd. Attempt 2's gap (0.0 -> 200.0) - # exceeds MAX_DELAY (100.0), so it counts as a new outage. - # Fallback (not bare next(clock)): Windows' ProactorEventLoop calls - # monotonic() more times during teardown, after the coroutine has returned. - clock = iter([0.0, 0.0, 0.0, 200.0, 500.0]) - monkeypatch.setattr(manager_module.time, "monotonic", lambda: next(clock, 500.0)) + monkeypatch.setattr(manager_module.time, "monotonic", lambda: now) with caplog.at_level(logging.DEBUG, logger="custom_components.zaptec.manager"): await _stream_supervisor(install, cb=AsyncMock(), ssl_context=None) warnings = [r for r in caplog.records if r.levelno == logging.WARNING] assert len(warnings) == 2 # noqa: PLR2004 # both failures counted as separate outages + + +@pytest.mark.asyncio +async def test_stream_supervisor_logs_reconnect_count_on_connect( + monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture +) -> None: + """Reconnecting logs how many attempts it took; the first connect stays quiet.""" + install = _fake_install() + outcomes = iter([ConnectionError("1"), ConnectionError("2")]) + + async def stream_main(*, on_connect: Callable[[], None], **kwargs: object) -> None: + outcome = next(outcomes, None) + if isinstance(outcome, Exception): + raise outcome + on_connect() + + install.stream_main.side_effect = stream_main + monkeypatch.setattr(asyncio, "sleep", AsyncMock()) + + with caplog.at_level(logging.DEBUG, logger="custom_components.zaptec.manager"): + await _stream_supervisor(install, cb=AsyncMock(), ssl_context=None) + + infos = [r for r in caplog.records if r.levelno == logging.INFO] + assert len(infos) == 1 + assert infos[0].args[1] == 2 # noqa: PLR2004 From d98b7f812c7ae9bde04816b64f04f60b9697cbdb Mon Sep 17 00:00:00 2001 From: rhammen <75572839+rhammen@users.noreply.github.com> Date: Tue, 1 Sep 2026 20:16:35 +0200 Subject: [PATCH 14/16] Keep the stream traceback only for unexpected failures The supervisor names the exception in its "disconnected" warning and attaches a traceback only when it isn't one of STREAM_TRANSIENT_ERRORS (service bus, network, timeout, request retry). Repeat failures still log the traceback at debug level. Co-Authored-By: Claude Opus 5 --- custom_components/zaptec/manager.py | 15 +++++++++++---- custom_components/zaptec/zaptec/__init__.py | 3 ++- custom_components/zaptec/zaptec/api.py | 11 +++++++++++ tests/test_manager.py | 19 +++++++++++++++++++ 4 files changed, 43 insertions(+), 5 deletions(-) diff --git a/custom_components/zaptec/manager.py b/custom_components/zaptec/manager.py index 9afb9060..48203e94 100644 --- a/custom_components/zaptec/manager.py +++ b/custom_components/zaptec/manager.py @@ -30,7 +30,7 @@ ) from .coordinator import ZaptecUpdateCoordinator from .entity import KeyUnavailableError, ZaptecBaseEntity -from .zaptec import Charger, Installation, Zaptec, ZaptecBase +from .zaptec import STREAM_TRANSIENT_ERRORS, Charger, Installation, Zaptec, ZaptecBase _LOGGER = logging.getLogger(__name__) @@ -74,7 +74,7 @@ def on_connect() -> None: connected_at = None try: await install.stream_main(cb=cb, ssl_context=ssl_context, on_connect=on_connect) - except Exception: + except Exception as err: # A connection that stayed up counts as recovered, so the next # failure starts a fresh outage instead of continuing the last. if connected_at is not None and ( @@ -82,11 +82,18 @@ def on_connect() -> None: ): delay = STREAM_RECONNECT_INIT_DELAY reconnects = 0 + # Only an unexpected failure is worth a traceback at warning level. + unexpected = not isinstance(err, STREAM_TRANSIENT_ERRORS) if reconnects: - _LOGGER.debug("Stream for %s still reconnecting", install.qual_id, exc_info=True) + _LOGGER.debug( + "Stream for %s still reconnecting (%r)", install.qual_id, err, exc_info=True + ) else: _LOGGER.warning( - "Stream for %s disconnected, reconnecting", install.qual_id, exc_info=True + "Stream for %s disconnected (%r), reconnecting", + install.qual_id, + err, + exc_info=unexpected, ) reconnects += 1 await asyncio.sleep(delay) diff --git a/custom_components/zaptec/zaptec/__init__.py b/custom_components/zaptec/zaptec/__init__.py index a71ebc9a..955b653f 100644 --- a/custom_components/zaptec/zaptec/__init__.py +++ b/custom_components/zaptec/zaptec/__init__.py @@ -2,7 +2,7 @@ from __future__ import annotations -from .api import Charger, Installation, Zaptec, ZaptecBase +from .api import STREAM_TRANSIENT_ERRORS, Charger, Installation, Zaptec, ZaptecBase from .const import MISSING, RETRYABLE_HTTP_STATUSES, Missing from .exceptions import ( AuthenticationError, @@ -20,6 +20,7 @@ __all__ = [ "MISSING", "RETRYABLE_HTTP_STATUSES", + "STREAM_TRANSIENT_ERRORS", "ZCONST", "AuthenticationError", "Charger", diff --git a/custom_components/zaptec/zaptec/api.py b/custom_components/zaptec/zaptec/api.py index c7527150..a633b972 100644 --- a/custom_components/zaptec/zaptec/api.py +++ b/custom_components/zaptec/zaptec/api.py @@ -64,6 +64,17 @@ TDict = dict[str, TValue] StreamCallback = Callable[[dict], Awaitable[None]] +STREAM_TRANSIENT_ERRORS: tuple[type[Exception], ...] = ( + ServiceBusError, + OSError, # includes ConnectionError + TimeoutError, + aiohttp.ClientError, + RequestConnectionError, + RequestRetryError, + RequestTimeoutError, +) +"""Errors a dropped stream is expected to fail with, e.g. a network outage.""" + class TLogExc(Protocol): """Protocol for logging exceptions.""" diff --git a/tests/test_manager.py b/tests/test_manager.py index b7283b05..49fde9c1 100644 --- a/tests/test_manager.py +++ b/tests/test_manager.py @@ -161,3 +161,22 @@ async def stream_main(*, on_connect: Callable[[], None], **kwargs: object) -> No infos = [r for r in caplog.records if r.levelno == logging.INFO] assert len(infos) == 1 assert infos[0].args[1] == 2 # noqa: PLR2004 + + +@pytest.mark.asyncio +async def test_stream_supervisor_logs_traceback_only_for_unexpected_errors( + monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture +) -> None: + """An expected disconnect warns without a traceback; anything else keeps one.""" + monkeypatch.setattr(asyncio, "sleep", AsyncMock()) + + for error, has_traceback in ((ConnectionError("boom"), False), (ValueError("boom"), True)): + install = _fake_install() + install.stream_main.side_effect = [error, None] + caplog.clear() + + with caplog.at_level(logging.DEBUG, logger="custom_components.zaptec.manager"): + await _stream_supervisor(install, cb=AsyncMock(), ssl_context=None) + + (warning,) = [r for r in caplog.records if r.levelno == logging.WARNING] + assert bool(warning.exc_info) is has_traceback From ce6380142f7fc3fdde047173df608ee478b135af Mon Sep 17 00:00:00 2001 From: rhammen <75572839+rhammen@users.noreply.github.com> Date: Tue, 1 Sep 2026 20:27:20 +0200 Subject: [PATCH 15/16] Widen the expected stream errors and cover the reset paths STREAM_TRANSIENT_ERRORS missed the failure issue #417 is actually about: a non-403 RequestError out of live_stream_connection_details() was classified unexpected. It now lists ZaptecApiError instead of three of its subclasses, plus MessageAlreadySettled, which derives from ValueError rather than ServiceBusError. on_connect now fires from inside the receiver context, and tests cover the backoff reset, the quiet first connect, and an attempt that fails before connecting at all. Co-Authored-By: Claude Opus 5 --- custom_components/zaptec/manager.py | 1 - custom_components/zaptec/zaptec/api.py | 12 +++--- tests/test_manager.py | 58 +++++++++++++++++++++++++- 3 files changed, 62 insertions(+), 9 deletions(-) diff --git a/custom_components/zaptec/manager.py b/custom_components/zaptec/manager.py index 48203e94..7e02ec9a 100644 --- a/custom_components/zaptec/manager.py +++ b/custom_components/zaptec/manager.py @@ -82,7 +82,6 @@ def on_connect() -> None: ): delay = STREAM_RECONNECT_INIT_DELAY reconnects = 0 - # Only an unexpected failure is worth a traceback at warning level. unexpected = not isinstance(err, STREAM_TRANSIENT_ERRORS) if reconnects: _LOGGER.debug( diff --git a/custom_components/zaptec/zaptec/api.py b/custom_components/zaptec/zaptec/api.py index a633b972..f0563a14 100644 --- a/custom_components/zaptec/zaptec/api.py +++ b/custom_components/zaptec/zaptec/api.py @@ -18,7 +18,7 @@ import aiohttp from aiolimiter import AsyncLimiter from azure.servicebus.aio import ServiceBusClient -from azure.servicebus.exceptions import ServiceBusError +from azure.servicebus.exceptions import MessageAlreadySettled, ServiceBusError import pydantic from .const import ( @@ -46,6 +46,7 @@ RequestError, RequestRetryError, RequestTimeoutError, + ZaptecApiError, ) from .redact import Redactor from .utils import mc_nbfx_decoder, to_under @@ -66,12 +67,11 @@ STREAM_TRANSIENT_ERRORS: tuple[type[Exception], ...] = ( ServiceBusError, + MessageAlreadySettled, # a ValueError, not a ServiceBusError OSError, # includes ConnectionError TimeoutError, aiohttp.ClientError, - RequestConnectionError, - RequestRetryError, - RequestTimeoutError, + ZaptecApiError, ) """Errors a dropped stream is expected to fail with, e.g. a network outage.""" @@ -446,11 +446,11 @@ async def stream_main( subscription_name=conf["Subscription"], ) _LOGGER.info("Running service bus stream for %s", self.qual_id) - if on_connect: - on_connect() # Store the receiver in order to close it and cancel this stream self._stream_receiver = receiver async with receiver: + if on_connect: + on_connect() async for msg in receiver: # For the exception in case it fails before setting the value binmsg = "" diff --git a/tests/test_manager.py b/tests/test_manager.py index 49fde9c1..d66ead3a 100644 --- a/tests/test_manager.py +++ b/tests/test_manager.py @@ -128,14 +128,18 @@ async def stream_main(*, on_connect: Callable[[], None], **kwargs: object) -> No raise ConnectionError("boom") install.stream_main.side_effect = stream_main - monkeypatch.setattr(asyncio, "sleep", AsyncMock()) + sleep_mock = AsyncMock() + monkeypatch.setattr(asyncio, "sleep", sleep_mock) monkeypatch.setattr(manager_module.time, "monotonic", lambda: now) + monkeypatch.setattr(manager_module.random, "normalvariate", lambda mu, sigma: mu) # noqa: ARG005 with caplog.at_level(logging.DEBUG, logger="custom_components.zaptec.manager"): await _stream_supervisor(install, cb=AsyncMock(), ssl_context=None) warnings = [r for r in caplog.records if r.levelno == logging.WARNING] assert len(warnings) == 2 # noqa: PLR2004 # both failures counted as separate outages + (last_delay,), _ = sleep_mock.await_args_list[-1] + assert last_delay == pytest.approx(STREAM_RECONNECT_INIT_DELAY) # backoff restarted too @pytest.mark.asyncio @@ -160,7 +164,7 @@ async def stream_main(*, on_connect: Callable[[], None], **kwargs: object) -> No infos = [r for r in caplog.records if r.levelno == logging.INFO] assert len(infos) == 1 - assert infos[0].args[1] == 2 # noqa: PLR2004 + assert "after 2 reconnect attempt(s)" in infos[0].getMessage() @pytest.mark.asyncio @@ -180,3 +184,53 @@ async def test_stream_supervisor_logs_traceback_only_for_unexpected_errors( (warning,) = [r for r in caplog.records if r.levelno == logging.WARNING] assert bool(warning.exc_info) is has_traceback + + +@pytest.mark.asyncio +async def test_stream_supervisor_stays_quiet_when_the_first_connect_succeeds( + monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture +) -> None: + """Connecting without ever having failed reports no reconnect count.""" + install = _fake_install() + + async def stream_main(*, on_connect: Callable[[], None], **kwargs: object) -> None: + on_connect() + + install.stream_main.side_effect = stream_main + monkeypatch.setattr(asyncio, "sleep", AsyncMock()) + + with caplog.at_level(logging.DEBUG, logger="custom_components.zaptec.manager"): + await _stream_supervisor(install, cb=AsyncMock(), ssl_context=None) + + assert not [r for r in caplog.records if r.levelno == logging.INFO] + + +@pytest.mark.asyncio +async def test_stream_supervisor_does_not_reset_when_the_retry_never_connects( + monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture +) -> None: + """The uptime of an earlier connection must not carry over to a later attempt.""" + install = _fake_install() + now = 0.0 + # Connect and stay up past the stable time, then fail without connecting at all. + attempts = iter([True, False]) + + async def stream_main(*, on_connect: Callable[[], None], **kwargs: object) -> None: + nonlocal now + connects = next(attempts, None) + if connects is None: + return # permanent stop, ends the supervisor loop + if connects: + on_connect() + now += STREAM_RECONNECT_STABLE_TIME + 1 + raise ConnectionError("boom") + + install.stream_main.side_effect = stream_main + monkeypatch.setattr(asyncio, "sleep", AsyncMock()) + monkeypatch.setattr(manager_module.time, "monotonic", lambda: now) + + with caplog.at_level(logging.DEBUG, logger="custom_components.zaptec.manager"): + await _stream_supervisor(install, cb=AsyncMock(), ssl_context=None) + + warnings = [r for r in caplog.records if r.levelno == logging.WARNING] + assert len(warnings) == 1 # the second failure never connected, so it is the same outage From 7c2b4c7e62fbf34ed7851314db3e71f121625dc8 Mon Sep 17 00:00:00 2001 From: rhammen <75572839+rhammen@users.noreply.github.com> Date: Tue, 1 Sep 2026 21:38:23 +0200 Subject: [PATCH 16/16] Require 5 minutes of uptime before a stream counts as recovered Measured on a live installation, stream observations arrive every ~3 minutes while charging (median gap 164s) and in a two-message burst every ~68 minutes when idle. At 60s a connection could be declared healthy having carried nothing, so a stream flapping just above that reset the backoff and re-warned on every cycle. Co-Authored-By: Claude Opus 5 --- custom_components/zaptec/const.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/custom_components/zaptec/const.py b/custom_components/zaptec/const.py index 20ace25a..043f41e2 100644 --- a/custom_components/zaptec/const.py +++ b/custom_components/zaptec/const.py @@ -48,7 +48,7 @@ STREAM_RECONNECT_MAX_DELAY = 300.0 """Maximum delay in seconds between stream reconnect attempts (5 minutes).""" -STREAM_RECONNECT_STABLE_TIME = 60.0 +STREAM_RECONNECT_STABLE_TIME = 300.0 """Uptime in seconds after which a stream connection counts as healthy again.""" # This sets the delay after doing actions and the poll of updated values.