Fix issues found by an Opus review of the smartthings-local upgrade

An independent review of the last two commits' diff turned up five real
problems and one CI-breaking one. Fixed all of them:

- tests/test_coordinator_error_handling.py assigned directly onto
  coordinator instance attributes (coordinator._poll_once = dict), which
  `ty check custom_components tests` -- what CI actually runs, not the
  narrower `ty check custom_components` this branch had only been
  spot-checked against -- flags as invalid-assignment. Switched to
  monkeypatch.setattr, matching every other new test on this branch.

- coordinator.py's subdevice-enumeration failure comment claimed the
  probe "retries naturally next cycle." It doesn't: _run_discovery sets
  self._discovered = True unconditionally later in the same cycle, which
  is what gates the whole block, so a failure here is a first-and-only
  attempt, not a retried one -- a composite appliance's sibling
  subdevices are missing for the config entry's lifetime until reload.
  (A separate flag to retry wouldn't actually fix that either: every
  platform's async_setup_entry enumerates coordinator.bound exactly
  once, so a later-successful enumeration still couldn't add entities
  without a reload.) Corrected the comment and raised debug to warning,
  since the effect is silent and permanent otherwise.

- config_flow.py's new diagnostic-handshake fallback (_resolve_alert)
  was being called once per failing candidate, inside _handshake_and_read's
  scan loop -- contradicting _diagnostic_alert's own docstring ("only
  runs once every real candidate has already failed"). Two real costs:
  up to CLIENTHELLO_PROBE_TIMEOUT_S extra latency per failing port on
  the sweep-fallback path (several candidates), and -- more seriously --
  the diagnostic commits DTLS association state on each port it touches,
  which can make _probe_and_validate's own CertRejected re-mint retry
  (a fresh _handshake_and_read call against that same scan) time out
  against the very port it just polluted, per the RFC 6347 §4.2.8
  concern already documented elsewhere in this file. Moved to a new
  _diagnose_failures helper called once, after the loop, against the
  single best (confirmed-live) candidate.

- _resolve_alert also ignored the alert's level: ProbeResult.alert is
  set for a received alert of either severity, but only a fatal one (2)
  means the appliance broke off the handshake over it -- a warning
  (e.g. close_notify) was being read as a rejection reason. Older
  library exception text never had this ambiguity (OpenSSL only renders
  an exception for a fatal alert), so this was a bug the redaction
  fallback introduced. Now filters to level == 2.

- async_raw_read's new HomeAssistantError wrapper reported a translated
  error but left a confirmed-dead session installed, so every
  subsequent read/write would keep failing identically for up to the
  next full poll interval. Now closes the session on any non-TimeoutError
  failure, matching _poll_once's own posture.

- async_raw_write_sequence's verify_after fallback (vcode, vrep = 0, {})
  is indistinguishable from a real 4.04 by raw_code alone. Added a
  read_error field so a caller can tell "couldn't verify" from "the
  device said no."

Tests: 8 new/extended (once-not-per-port diagnostic count, alert-level
filtering, the warning log + permanent-loss framing, session-closing on
both async_raw_read and the verify_after path, read_error surfacing).
Full suite (1527 tests), ruff, ruff format, and -- critically --
`ty check custom_components tests` (the CI-matching invocation) all pass.

custom_components/localthings/config_flow.py

@@ -554,7 +554,16 @@ def _resolve_alert(exc: Exception, host: str, port: int, cert_pem: str, key_pem:
         result = _diagnostic_alert(host, port, cert_pem, key_pem)
     except Exception:
         return None
-    return result.alert[1] if result.alert else None
+    if result.alert is None:
+        return None
+    level, name = result.alert
+    # ProbeResult.alert is set for a *received* alert record of either
+    # level -- fatal (2) means the appliance actually broke off the
+    # handshake over it; a warning (1, e.g. close_notify on an otherwise
+    # ordinary close) is not evidence of a rejection and must not be read
+    # as one. The old exception-text path never had this ambiguity: an
+    # OpenSSL exception only ever rendered for a fatal alert.
+    return name if level == 2 else None
 
 
 def _classify_handshake_failure(
@@ -682,12 +691,45 @@ def _read_device(sess, host: str, port: int) -> dict:
     }
 
 
+def _diagnose_failures(
+    host: str,
+    scan: _PortScan,
+    failures: list[tuple[int, Exception]],
+    cert_pem: str,
+    key_pem: str,
+) -> dict[int, str]:
+    """At most one diagnostic handshake (see _diagnostic_alert) across every
+    port `_handshake_and_read` just gave up on -- not one per port.
+
+    Called only after that loop has fully exhausted `scan.candidates`, never
+    interleaved with it: `_diagnostic_alert`'s own docstring says the extra
+    orphaned association it costs is a fair trade "on a path the user is
+    about to retry regardless" -- true for the retry `_probe_and_validate`
+    itself makes on a CertRejected (a fresh `_handshake_and_read` call
+    against this same `scan`), but only if that retry's real handshake
+    attempts are the ones landing on a clean slate. Running the diagnostic
+    per candidate mid-loop would pollute exactly the port(s) that retry is
+    about to reattempt; running several of them multiplies both the latency
+    (each is its own bounded handshake) and the pollution for no extra
+    classification value, since _classify_handshake_failure only ever needs
+    one alert to decide.
+
+    Targets a confirmed-live port over an unconfirmed sweep candidate --
+    the one actually worth spending the extra handshake on.
+    """
+    if not failures:
+        return {}
+    by_port = dict(failures)
+    port = next((p for p in scan.confirmed if p in by_port), next(iter(by_port)))
+    alert = _resolve_alert(by_port[port], host, port, cert_pem, key_pem)
+    return {port: alert} if alert is not None else {}
+
+
 def _handshake_and_read(host: str, scan: _PortScan, cert_pem: str, key_pem: str) -> dict:
     """Handshake each candidate in turn, returning the first device that answers."""
     from smartthings_local.protocol.dtls_session import DtlsCoapSession
 
     failures: list[tuple[int, Exception]] = []
-    alerts: dict[int, str] = {}
     for port in scan.candidates:
         sess = None
         try:
@@ -702,13 +744,11 @@ def _handshake_and_read(host: str, scan: _PortScan, cert_pem: str, key_pem: str)
         except Exception as exc:
             failures.append((port, exc))
             _LOGGER.debug("port %d failed: %s", port, exc)
-            alert = _resolve_alert(exc, host, port, cert_pem, key_pem)
-            if alert is not None:
-                alerts[port] = alert
         finally:
             if sess is not None:
                 with contextlib.suppress(Exception):
                     sess.close()
+    alerts = _diagnose_failures(host, scan, failures, cert_pem, key_pem)
     raise _classify_handshake_failure(host, scan, failures, alerts)
 
 

custom_components/localthings/coordinator.py

@@ -1395,9 +1395,24 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
                     # keeps the value _poll_once already returned, so
                     # discovery below still runs on the master's own data,
                     # same posture _poll_subdevice_seed takes for one sibling
-                    # going quiet. self.subdevices/_discovered are still
-                    # unset, so the probe retries naturally next cycle.
-                    self._log.debug("subdevice enumeration failed: %s", e)
+                    # going quiet.
+                    #
+                    # Not a one-cycle blip, though: `_run_discovery` a few
+                    # lines below sets `self._discovered = True`
+                    # unconditionally this same cycle, which is what gates
+                    # this whole block -- there is no next cycle where this
+                    # is retried. A composite appliance whose enumeration
+                    # fails here loses its sibling subdevices' entities for
+                    # this config entry's lifetime (a reload probes again).
+                    # warning, not debug, because of that: it's silent and
+                    # permanent otherwise, with nothing in the log pointing
+                    # at why a device is missing entities it should have.
+                    self._log.warning(
+                        "subdevice enumeration failed on first discovery; "
+                        "any sibling subdevices will be missing until this "
+                        "config entry is reloaded: %s",
+                        e,
+                    )
 
         source = "sweep" if self._discovered else "poll"
         first_cycle = not self._discovered
@@ -1710,6 +1725,14 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
                 # path a raw session exception would otherwise reach a user
                 # untranslated (write_resource already goes through
                 # HomeAssistantError; this brings read_resource in line).
+                if not isinstance(e, TimeoutError):
+                    # Same TimeoutError-vs-anything-else split as
+                    # _poll_once: a block-ACK timeout alone doesn't prove
+                    # the session is dead, but anything else does -- and
+                    # leaving a confirmed-dead one installed would fail
+                    # every read/write identically until the next real
+                    # poll cycle's own reconnect notices.
+                    await self.hass.async_add_executor_job(self._close_session)
                 self._log.warning("debug read failed for %s: %s", norm_href, e)
                 raise HomeAssistantError(
                     translation_domain=DOMAIN,
@@ -1823,6 +1846,7 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
             verified: dict[str, Any] = {}
             async with self._session_lock:
                 for href in dict.fromkeys(r["href"] for r in results):
+                    read_error: str | None = None
                     try:
                         vcode, vrep, _vbody = await self.hass.async_add_executor_job(
                             self._raw_read_blocking, _href_to_path_segs(href), href
@@ -1836,9 +1860,12 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
                         # one href's failure shouldn't stop the rest of the
                         # batch from being checked. Same "couldn't verify"
                         # posture as a 4.04/empty read below: held stays
-                        # None, not False.
+                        # None, not False -- but raw_code 0 alone is also
+                        # what a 4.04 produces, so read_error is what tells
+                        # the two apart for a caller inspecting the response.
                         self._log.debug("raw write verification read failed for %s: %s", href, e)
                         vcode, vrep = 0, {}
+                        read_error = str(e)
                     # None, not False, when the re-read brought back nothing
                     # to compare: every comparison against an empty rep is
                     # False, which would report a 4.04 as a revert -- the one
@@ -1853,6 +1880,7 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
                             if read_ok
                             else None
                         ),
+                        "read_error": read_error,
                     }
             response["verified"] = verified
 

tests/localthings/test_config_flow.py

@@ -528,6 +528,39 @@ async def test_rejected_reused_leaf_is_reminted_against_a_redacted_library(
     assert diagnosed == [49154]
 
 
+def test_diagnostic_handshake_runs_once_not_once_per_failing_port(monkeypatch) -> None:
+    """_diagnostic_alert commits association state on the device (see its
+    own docstring) -- running it once per failing candidate instead of once
+    overall would both add latency (each is its own bounded handshake) and
+    multiply that pollution right before _probe_and_validate might retry a
+    real handshake against these very same ports. Three candidates fail
+    here; the diagnostic must run exactly once, against the confirmed-live
+    port, not three times against every candidate in scan order."""
+    from custom_components.localthings import config_flow
+
+    FakeSession.instances = []
+    FakeSession.reject_certs = {MOCK_LEAF_CERT_PEM}
+    FakeSession.redact_rejection = True
+    monkeypatch.setattr("smartthings_local.protocol.dtls_session.DtlsCoapSession", FakeSession)
+
+    diagnosed: list[int] = []
+
+    class _DiagnosticResult:
+        alert = (2, "bad_certificate")
+
+    def _diagnostic_alert(host, port, cert_pem, key_pem):
+        diagnosed.append(port)
+        return _DiagnosticResult()
+
+    monkeypatch.setattr(config_flow, "_diagnostic_alert", _diagnostic_alert)
+
+    scan = _scan(confirmed=[49153, 49154], candidates=[49153, 49154, 49155])
+    with pytest.raises(config_flow.CertRejected):
+        config_flow._handshake_and_read(MOCK_HOST, scan, MOCK_LEAF_CERT_PEM, "KEY")
+
+    assert diagnosed == [49153]  # confirmed-live, and only once
+
+
 async def test_unconfirmed_port_failure_is_not_reminted(
     hass: HomeAssistant, monkeypatch, fake_dtls
 ) -> None:
@@ -653,6 +686,26 @@ def test_resolve_alert_falls_back_to_the_diagnostic_handshake() -> None:
     assert name == "bad_certificate"
 
 
+def test_resolve_alert_ignores_a_non_fatal_alert() -> None:
+    """ProbeResult.alert is set for a *received* alert of either level, but
+    only a fatal one (2) means the appliance actually broke off the
+    handshake over it -- a warning-level alert (e.g. close_notify on an
+    otherwise ordinary close) is not evidence of a rejection. The old
+    exception-text path never had this ambiguity: OpenSSL's exception only
+    ever rendered for a fatal alert, so nothing pre-0.1.3 could confuse the
+    two -- the diagnostic-handshake fallback must not introduce the mix-up."""
+    from smartthings_local.errors import SessionError
+
+    from custom_components.localthings import config_flow
+
+    class _Result:
+        alert = (1, "close_notify")  # warning level, not fatal
+
+    with patch.object(config_flow, "_diagnostic_alert", lambda *a, **k: _Result()):
+        name = config_flow._resolve_alert(SessionError(), MOCK_HOST, 49154, "CERT", "KEY")
+    assert name is None
+
+
 def test_resolve_alert_is_none_when_the_diagnostic_handshake_also_fails() -> None:
     """A best-effort extra probe: its own failure must not raise out of
     _resolve_alert, it just leaves the caller with no alert to report."""

tests/test_coordinator_error_handling.py

@@ -8,6 +8,7 @@ or getting translated into a HomeAssistantError for a service caller.
 
 from __future__ import annotations
 
+import pytest
 from homeassistant.core import HomeAssistant
 from pytest_homeassistant_custom_component.common import MockConfigEntry
 
@@ -39,7 +40,7 @@ def _coordinator(hass: HomeAssistant) -> LocalThingsCoordinator:
 
 
 async def test_subdevice_enumeration_failure_does_not_abort_first_discovery(
-    hass: HomeAssistant,
+    hass: HomeAssistant, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
 ) -> None:
     """_enumerate_subdevices_blocking's own _connect_session() call only
     fires if the session the poll above just used got closed out from under
@@ -48,20 +49,28 @@ async def test_subdevice_enumeration_failure_does_not_abort_first_discovery(
     through this integration's own logging, matching what already happens
     for the main poll's own reconnect.
 
+    Not a one-cycle blip once caught, though: `_discovered` flips True this
+    same cycle regardless (gating first discovery, not subdevice success),
+    so this is the *only* attempt a composite appliance's siblings ever get
+    without a config-entry reload -- logged at warning for exactly that
+    reason, not debug.
+
     Empty resources keep _run_discovery from binding anything (hot/warm
     hrefs stay empty), so _attempt_observe_mode's own session touch never
     runs either -- this test is purely about the enumeration failure not
     escaping _async_update_data.
     """
     coordinator = _coordinator(hass)
-    coordinator._poll_once = dict
+    monkeypatch.setattr(coordinator, "_poll_once", dict)
 
     def _boom(_resources):
         raise ConnectionError("session closed")
 
-    coordinator._enumerate_subdevices_blocking = _boom
+    monkeypatch.setattr(coordinator, "_enumerate_subdevices_blocking", _boom)
 
-    result = await coordinator._async_update_data()
+    with caplog.at_level("WARNING"):
+        result = await coordinator._async_update_data()
 
     assert coordinator._discovered is True
     assert result == {}
+    assert "subdevice enumeration failed" in caplog.text

tests/test_services.py

@@ -297,6 +297,7 @@ async def test_write_resource_verify_after_reports_held(hass, coordinator, devic
     verified = response["verified"]["/mode/vs/0"]
     assert verified["held"] is True
     assert verified["rep"] == {"x.field": "target"}
+    assert verified["read_error"] is None
 
 
 async def test_write_resource_verify_after_reports_reverted(hass, coordinator, device_id):
@@ -355,6 +356,9 @@ async def test_write_resource_verify_after_survives_a_failed_confirmation_read(
     verified = response["verified"]["/mode/vs/0"]
     assert verified["held"] is None
     assert verified["rep"] == {}
+    # raw_code 0 alone is indistinguishable from a real 4.04 -- read_error
+    # is what tells a caller this was an unreachable session, not a reply.
+    assert verified["read_error"] == "session closed"
 
 
 async def test_write_resource_no_verified_key_when_verify_after_is_zero(
@@ -656,6 +660,47 @@ async def test_read_resource_failure_is_surfaced_as_a_home_assistant_error(
         await _call_read(hass, device_id, href="/mode/vs/0")
 
 
+async def test_read_resource_failure_closes_a_confirmed_dead_session(
+    hass, coordinator, device_id, monkeypatch
+):
+    """A non-timeout failure is unambiguous (same TimeoutError-vs-anything-
+    else split as _poll_once) -- leaving a confirmed-dead session installed
+    would fail every subsequent read/write identically until the next real
+    poll cycle's own reconnect notices, up to a full update_interval later."""
+    coordinator._session = _FakeSession()
+
+    def _boom(path_segs, href):
+        raise ConnectionError("session closed")
+
+    monkeypatch.setattr(coordinator, "_raw_read_blocking", _boom)
+
+    with pytest.raises(HomeAssistantError):
+        await _call_read(hass, device_id, href="/mode/vs/0")
+
+    assert coordinator._session is None
+
+
+async def test_read_resource_timeout_does_not_close_the_session(
+    hass, coordinator, device_id, monkeypatch
+):
+    """The other half of the same distinction: a block-ACK TimeoutError
+    alone doesn't prove the session is dead (see _poll_once), so unlike
+    any other failure it must not tear down a session that might still be
+    perfectly fine."""
+    fake = _FakeSession()
+    coordinator._session = fake
+
+    def _boom(path_segs, href):
+        raise TimeoutError("GET timeout")
+
+    monkeypatch.setattr(coordinator, "_raw_read_blocking", _boom)
+
+    with pytest.raises(HomeAssistantError):
+        await _call_read(hass, device_id, href="/mode/vs/0")
+
+    assert coordinator._session is fake
+
+
 async def test_read_resource_without_href_returns_cached_snapshot_and_does_not_get(
     hass, coordinator, device_id
 ):