diff --git a/Dockerfile b/Dockerfile
index 8f7f9ee..9f4ff04 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -6,4 +6,4 @@ FROM ghcr.io/home-assistant/home-assistant:stable
 # repeats the install attempt on every container recreate. Baking
 # smartthings-local into the image keeps the dev container usable
 # offline and avoids relying on that runtime install path.
-RUN pip3 install --no-cache-dir "smartthings-local>=0.1.2"
+RUN pip3 install --no-cache-dir "smartthings-local>=0.1.6"
diff --git a/custom_components/localthings/config_flow.py b/custom_components/localthings/config_flow.py
index d965d7b..3b49039 100644
--- a/custom_components/localthings/config_flow.py
+++ b/custom_components/localthings/config_flow.py
@@ -493,27 +493,84 @@ _CERT_ALERTS = frozenset(
     }
 )
 
-# OpenSSL renders a received fatal alert into its error text as e.g.
-# "tlsv1 alert unknown ca", which DtlsCoapSession.connect() wraps in a
-# ConnectionError. Reading it back tells us what the appliance objected to.
-#
-# Deliberately not the library's diagnostic probe (stateless=False): that
-# mode commits association state on the device, and an orphaned association
-# makes the next attempt time out (RFC 6347 §4.2.8) -- a bad trade on a
-# path the user is about to retry.
+# Older smartthings-local (< 0.1.3) rendered a received fatal alert straight
+# into the handshake exception's text, e.g. "tlsv1 alert unknown ca" wrapped
+# in a ConnectionError -- reading it back told us what the appliance
+# objected to. 0.1.3's "redacted typed failures" removed that: connect()'s
+# exceptions now carry a fixed, non-sensitive message with the real OpenSSL
+# text neither included nor chained (see smartthings_local.errors --
+# "backend errors can contain remote endpoints, local paths, or credential
+# metadata"). _alert_name is kept as a harmless fallback for exception text
+# that does carry it; _resolve_alert below is what actually classifies a
+# failure against a current library.
 _ALERT_RE = re.compile(r"alert ([a-z0-9 ]+)")
 
 
 def _alert_name(exc: Exception) -> str | None:
-    """The TLS alert an appliance sent, if this failure carried one."""
+    """The TLS alert an appliance sent, if this failure's exception text
+    carried one (only ever true against smartthings-local < 0.1.3)."""
     match = _ALERT_RE.search(str(exc).lower())
     return match.group(1).strip().replace(" ", "_") if match else None
 
 
+def _diagnostic_alert(host: str, port: int, cert_pem: str, key_pem: str):
+    """One opt-in stateful handshake against `port`, using our real
+    credentials, so a fatal Alert can be classified from the raw record
+    itself rather than parsed out of an exception's text.
+
+    This is the library's diagnose_dtls_handshake -- deliberately not used
+    for the primary candidate scan (it commits association state on the
+    device, and an orphaned association makes the *next* attempt time out
+    per RFC 6347 §4.2.8). Here it only runs once every real candidate has
+    already failed, to explain a failure that's happening either way --
+    one more orphaned association is a fair trade for a message that says
+    why, on a path the user is about to retry regardless.
+
+    Imported lazily, like `_clienthello_probe`, so an install whose
+    smartthings-local predates this API degrades to a generic message
+    instead of failing to load the config flow at all.
+    """
+    from smartthings_local.protocol.dtls_probe import diagnose_dtls_handshake
+
+    return diagnose_dtls_handshake(
+        host,
+        port,
+        cert_pem=cert_pem,
+        key_pem=key_pem,
+        timeout=CLIENTHELLO_PROBE_TIMEOUT_S,
+        retries=CLIENTHELLO_PROBE_RETRIES,
+    )
+
+
+def _resolve_alert(exc: Exception, host: str, port: int, cert_pem: str, key_pem: str) -> str | None:
+    """The TLS alert `port`'s failed handshake carried, if any -- the
+    exception's own text first (cheap, and all an older library ever
+    offers), then one bounded diagnostic handshake against a current one
+    that redacts it (see _diagnostic_alert)."""
+    name = _alert_name(exc)
+    if name is not None:
+        return name
+    try:
+        result = _diagnostic_alert(host, port, cert_pem, key_pem)
+    except Exception:
+        return 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(
     host: str,
     scan: _PortScan,
     failures: list[tuple[int, Exception]],
+    alerts: dict[int, str] | None = None,
 ) -> CannotConnect:
     """Turn "no port worked" into the most specific thing we can honestly
     say, in rough order of how much the evidence tells us: an alert means
@@ -521,13 +578,21 @@ def _classify_handshake_failure(
     certificate); a confirmed DTLS port that then timed out is likely still
     holding a session from a previous attempt; otherwise the sweep's own
     shape is the evidence.
+
+    `alerts` is the per-port classification `_handshake_and_read` already
+    resolved (exception text, or a diagnostic handshake -- see
+    _resolve_alert); a caller with only raw failures (or an older library)
+    still gets `_alert_name`'s exception-text reading as a fallback.
     """
-    alerts = [name for name in (_alert_name(exc) for _, exc in failures) if name]
-    cert_alerts = [name for name in alerts if name in _CERT_ALERTS]
+    resolved = dict(alerts or {})
+    for port, exc in failures:
+        resolved.setdefault(port, _alert_name(exc))
+    alert_names = [name for name in resolved.values() if name]
+    cert_alerts = [name for name in alert_names if name in _CERT_ALERTS]
     if cert_alerts:
         return CertRejected(f"{host} rejected our certificate (alert {cert_alerts[0]})")
-    if alerts:
-        return HandshakeFailed(f"{host} refused the DTLS handshake (alert {alerts[0]})")
+    if alert_names:
+        return HandshakeFailed(f"{host} refused the DTLS handshake (alert {alert_names[0]})")
     if scan.confirmed:
         return HandshakeTimeout(
             f"DTLS server confirmed on {host}:{scan.confirmed} but the handshake never completed"
@@ -626,6 +691,40 @@ 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
@@ -649,7 +748,8 @@ def _handshake_and_read(host: str, scan: _PortScan, cert_pem: str, key_pem: str)
             if sess is not None:
                 with contextlib.suppress(Exception):
                     sess.close()
-    raise _classify_handshake_failure(host, scan, failures)
+    alerts = _diagnose_failures(host, scan, failures, cert_pem, key_pem)
+    raise _classify_handshake_failure(host, scan, failures, alerts)
 
 
 def _probe_and_validate(
diff --git a/custom_components/localthings/coordinator.py b/custom_components/localthings/coordinator.py
index d0a7ada..c41785c 100644
--- a/custom_components/localthings/coordinator.py
+++ b/custom_components/localthings/coordinator.py
@@ -103,10 +103,15 @@ def _local_source_port(host: str) -> int:
     time per RFC 6347 §4.2.8, instead of holding it 5-15 min. See
     DTLS_LOCAL_PORT_BASE. Requires smartthings-local >= 0.1.1.
 
-    Must stay unique per device on this host too: the library's socket is
-    unconnected, so two devices sharing a port would mis-demux each other's
-    datagrams. Last IPv4 octet as offset for the common case; a stable
-    CRC32 fold otherwise.
+    Must stay unique per device on this host too. That used to be load-
+    bearing for demuxing: an unconnected socket handed every device's
+    datagrams to whichever recvfrom() happened to be listening on their
+    shared port. smartthings-local >= 0.1.3 connect()s its UDP socket
+    instead (see endpoint.py's open_connected_udp_socket), so the kernel
+    already filters incoming datagrams to each session's own resolved peer
+    -- but a distinct port per device keeps that guarantee from ever
+    depending on it, and keeps captures/logs unambiguous. Last IPv4 octet
+    as offset for the common case; a stable CRC32 fold otherwise.
     """
     try:
         offset = int(ipaddress.IPv4Address(host)) & 0xFF
@@ -203,7 +208,10 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
     # A block-level ACK timeout on the summary GET doesn't prove the session
     # is dead (see _poll_once) -- require this many in a row before treating
     # it as one, so one slow transfer doesn't tear down a working OBSERVE
-    # subscription.
+    # subscription. Only covers that ambiguous case: smartthings-local
+    # >= 0.1.6 raises a distinct SessionClosedError, not a TimeoutError, the
+    # moment a dead reader thread is confirmed, and _defer_reconnect_for
+    # never defers that -- see its docstring for what changed there.
     _POLL_TIMEOUT_LIMIT: int = 3
 
     # Named (not inline literals) so the write-settle window in
@@ -717,6 +725,14 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
         not that the session is dead (earlier blocks succeeded). Left open;
         `_async_update_data` decides whether repeated timeouts warrant a
         reconnect. Any other exception is unambiguous -- close immediately.
+
+        smartthings-local >= 0.1.6 tells those two cases apart itself now:
+        a reader thread that has actually died raises `SessionClosedError`
+        (a ConnectionError, not a TimeoutError) the moment the next request
+        notices, instead of the old behavior of quietly hanging out to this
+        call's own timeout and surfacing as an ambiguous `TimeoutError`.
+        See `_defer_reconnect_for` for what that changes about how soon a
+        confirmed-dead session gets reconnected.
         """
         if self._session is None:
             self._connect_session()
@@ -1164,7 +1180,38 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
             if self._session is None:
                 # _poll_once already connects on a real poll; only fires if
                 # the session was closed out from under us concurrently.
-                await self.hass.async_add_executor_job(self._connect_session)
+                try:
+                    await self.hass.async_add_executor_job(self._connect_session)
+                except Exception as e:
+                    # Not a poll failure -- the poll that reached this line
+                    # already succeeded, and observe mode is only an
+                    # optimization on top of it. Give up on push this cycle
+                    # the same way the two branches below do, rather than
+                    # letting this escape _async_update_data uncaught: none
+                    # of this integration's reconnect bookkeeping would run,
+                    # and the base coordinator logs an ERROR traceback in
+                    # place of the deliberately quiet "poll failed,
+                    # reconnecting" voice used everywhere else in this file.
+                    #
+                    # Not counted by _reconnect_is_frequent() (that window
+                    # records reconnects the poll path itself performed --
+                    # feeding it a secondary path's failure would push the
+                    # next routine poll reconnect over the warn threshold),
+                    # and nothing to _close_session(): _connect_session only
+                    # publishes self._session once connect() and
+                    # start_reader() have both already succeeded, so it's
+                    # still None here. No _resubscribe_due either -- that
+                    # flag means "a live session nothing has tried yet",
+                    # and setting it would re-enter this doomed handshake
+                    # every cycle; _last_observe_attempt_ts (stamped above)
+                    # already paces the retry to _RECOVERY_RETRY_S.
+                    self._log.info(
+                        "observe-mode reconnect failed (%s), staying on polling: %s",
+                        type(e).__name__,
+                        e,
+                    )
+                    self._observe.abandon_observe_attempt()
+                    return
             sess = self._session
             if sess is None:
                 return
@@ -1220,6 +1267,19 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
         `_POLL_TIMEOUT_LIMIT` consecutive timeouts pile up. Any other
         exception reconnects immediately.
 
+        That includes `SessionClosedError`, which is the point: before
+        smartthings-local 0.1.6, a reader thread that had actually died was
+        indistinguishable from a slow transfer -- both surfaced here only as
+        a `TimeoutError`, so this tolerance was the only thing standing
+        between a truly dead session and a reconnect, worst case about
+        `_POLL_TIMEOUT_LIMIT` poll cycles (~2 minutes at the default 30s
+        interval). 0.1.6 confirms reader death directly and raises a
+        ConnectionError subclass for it instead, which isn't a TimeoutError
+        and so skips this tolerance entirely -- a genuinely dead session now
+        reconnects on the very first occurrence. Intended (see this repo's
+        README, "Known device behavior"), not a regression, but a real
+        change in observed reconnect timing for that one failure mode.
+
         Never defers before first discovery (issue #254): deferring returns
         an empty dict, which the base coordinator treats as a successful
         first refresh -- and since platforms enumerate `bound` once, the
@@ -1322,9 +1382,37 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
             # cycle's snapshot so discovery sees every subdevice on the
             # first poll rather than waiting a cycle.
             async with self._session_lock:
-                resources = await self.hass.async_add_executor_job(
-                    self._enumerate_subdevices_blocking, resources
-                )
+                try:
+                    resources = await self.hass.async_add_executor_job(
+                        self._enumerate_subdevices_blocking, resources
+                    )
+                except Exception as e:
+                    # _connect_session() inside here only runs at all if the
+                    # session the poll above just used got closed out from
+                    # under us within this same cycle -- rare, but not
+                    # impossible, and unguarded before this. Losing the
+                    # subdevice probe isn't losing first discovery: `resources`
+                    # 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.
+                    #
+                    # 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
@@ -1626,9 +1714,31 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
             )
         norm_href = "/" + "/".join(path_segs)
         async with self._session_lock:
-            return await self.hass.async_add_executor_job(
-                self._raw_read_blocking, path_segs, norm_href
-            )
+            try:
+                return await self.hass.async_add_executor_job(
+                    self._raw_read_blocking, path_segs, norm_href
+                )
+            except Exception as e:
+                # Unlike async_raw_write_sequence, there's nothing to
+                # reconnect-and-retry here -- a live debug read either lands
+                # or it doesn't, and a service call is the one place on this
+                # 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,
+                    translation_key="debug_read_failed",
+                    translation_placeholders={"href": norm_href, "error": str(e)},
+                ) from e
 
     async def async_raw_write_sequence(
         self,
@@ -1736,9 +1846,26 @@ 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):
-                    vcode, vrep, _vbody = await self.hass.async_add_executor_job(
-                        self._raw_read_blocking, _href_to_path_segs(href), href
-                    )
+                    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
+                        )
+                    except Exception as e:
+                        # The write already landed -- see `results` above,
+                        # built before this wait ever started. A failed
+                        # confirmation read (the session dying in the gap
+                        # verify_after just waited out, say) must not lose
+                        # that outcome behind a raised exception here, and
+                        # 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 -- 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
@@ -1753,6 +1880,7 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
                             if read_ok
                             else None
                         ),
+                        "read_error": read_error,
                     }
             response["verified"] = verified
 
diff --git a/custom_components/localthings/manifest.json b/custom_components/localthings/manifest.json
index 60c97e6..0a8f5ea 100644
--- a/custom_components/localthings/manifest.json
+++ b/custom_components/localthings/manifest.json
@@ -10,7 +10,7 @@
   "requirements": [
     "cbor2>=5.4.6",
     "pyOpenSSL>=23.0",
-    "smartthings-local>=0.1.2"
+    "smartthings-local>=0.1.6"
   ],
   "version": "0.21.2"
 }
diff --git a/custom_components/localthings/translations/cs.json b/custom_components/localthings/translations/cs.json
index 9873b61..1b833b5 100644
--- a/custom_components/localthings/translations/cs.json
+++ b/custom_components/localthings/translations/cs.json
@@ -1558,6 +1558,9 @@
     "command_failed": {
       "message": "Příkaz pro {href} selhal i po opětovném připojení: {error}"
     },
+    "debug_read_failed": {
+      "message": "Čtení z {href} selhalo: {error}"
+    },
     "debug_too_many_writes": {
       "message": "Zadejte 1 až 10 zápisů."
     },
diff --git a/custom_components/localthings/translations/de.json b/custom_components/localthings/translations/de.json
index 649bce1..a2d33ee 100644
--- a/custom_components/localthings/translations/de.json
+++ b/custom_components/localthings/translations/de.json
@@ -1558,6 +1558,9 @@
     "command_failed": {
       "message": "Der Befehl an {href} ist auch nach erneutem Verbinden fehlgeschlagen: {error}"
     },
+    "debug_read_failed": {
+      "message": "Das Lesen von {href} ist fehlgeschlagen: {error}"
+    },
     "debug_too_many_writes": {
       "message": "Geben Sie zwischen 1 und 10 Schreibvorgänge an."
     },
diff --git a/custom_components/localthings/translations/en.json b/custom_components/localthings/translations/en.json
index b4e96c8..30839a7 100644
--- a/custom_components/localthings/translations/en.json
+++ b/custom_components/localthings/translations/en.json
@@ -1558,6 +1558,9 @@
     "command_failed": {
       "message": "The command to {href} failed even after reconnecting: {error}"
     },
+    "debug_read_failed": {
+      "message": "The read from {href} failed: {error}"
+    },
     "debug_too_many_writes": {
       "message": "Provide between 1 and 10 writes."
     },
diff --git a/custom_components/localthings/translations/es.json b/custom_components/localthings/translations/es.json
index 05bbe46..763d485 100644
--- a/custom_components/localthings/translations/es.json
+++ b/custom_components/localthings/translations/es.json
@@ -173,6 +173,9 @@
     "command_failed": {
       "message": "El comando para {href} falló incluso después de reconectar: {error}"
     },
+    "debug_read_failed": {
+      "message": "La lectura desde {href} falló: {error}"
+    },
     "debug_too_many_writes": {
       "message": "Proporciona entre 1 y 10 escrituras."
     },
diff --git a/custom_components/localthings/translations/it.json b/custom_components/localthings/translations/it.json
index f96e426..d785c6f 100644
--- a/custom_components/localthings/translations/it.json
+++ b/custom_components/localthings/translations/it.json
@@ -1558,6 +1558,9 @@
     "command_failed": {
       "message": "Il comando per {href} è fallito anche dopo la riconnessione: {error}"
     },
+    "debug_read_failed": {
+      "message": "La lettura da {href} è fallita: {error}"
+    },
     "debug_too_many_writes": {
       "message": "Specificare da 1 a 10 scritture."
     },
diff --git a/custom_components/localthings/translations/ko.json b/custom_components/localthings/translations/ko.json
index c6230ff..540d48e 100644
--- a/custom_components/localthings/translations/ko.json
+++ b/custom_components/localthings/translations/ko.json
@@ -1558,6 +1558,9 @@
     "command_failed": {
       "message": "재연결 후에도 {href} 명령이 실패했습니다: {error}"
     },
+    "debug_read_failed": {
+      "message": "{href}에서 읽기가 실패했습니다: {error}"
+    },
     "debug_too_many_writes": {
       "message": "1~10개의 쓰기를 지정하세요."
     },
diff --git a/custom_components/localthings/translations/nl.json b/custom_components/localthings/translations/nl.json
index 12875d3..4ed576e 100644
--- a/custom_components/localthings/translations/nl.json
+++ b/custom_components/localthings/translations/nl.json
@@ -1558,6 +1558,9 @@
     "command_failed": {
       "message": "Het commando naar {href} is ook na opnieuw verbinden mislukt: {error}"
     },
+    "debug_read_failed": {
+      "message": "Het lezen van {href} is mislukt: {error}"
+    },
     "debug_too_many_writes": {
       "message": "Geef tussen de 1 en 10 schrijfacties op."
     },
diff --git a/requirements-dev.txt b/requirements-dev.txt
index 6aa6f59..088db5f 100644
--- a/requirements-dev.txt
+++ b/requirements-dev.txt
@@ -6,7 +6,7 @@ pytest-homeassistant-custom-component>=0.13.316
 
 # Integration runtime deps, needed to import the component under test
 # (also declared in custom_components/localthings/manifest.json).
-smartthings-local>=0.1.2
+smartthings-local>=0.1.6
 cbor2>=5.4.6
 pyOpenSSL>=23.0
 cryptography>=41.0
diff --git a/tests/localthings/test_config_flow.py b/tests/localthings/test_config_flow.py
index a3d8e85..c531a99 100644
--- a/tests/localthings/test_config_flow.py
+++ b/tests/localthings/test_config_flow.py
@@ -238,6 +238,11 @@ class FakeSession:
 
     instances: ClassVar[list[FakeSession]] = []
     reject_certs: ClassVar[set[str]] = set()
+    # smartthings-local >= 0.1.3 ("redacted typed failures") no longer puts
+    # the alert in connect()'s exception text -- set True to model that, so
+    # a test can check the diagnostic-handshake fallback (_resolve_alert)
+    # instead of the legacy _alert_name text-parsing path.
+    redact_rejection: ClassVar[bool] = False
 
     def __init__(self, host, port, cert_pem=None, key_pem=None, **kwargs):
         self.host, self.port, self.cert_pem = host, port, cert_pem
@@ -245,6 +250,10 @@ class FakeSession:
 
     def connect(self):
         if self.cert_pem in FakeSession.reject_certs:
+            if FakeSession.redact_rejection:
+                from smartthings_local.errors import SessionError
+
+                raise SessionError()
             raise ConnectionError(
                 "DTLS handshake error: [('SSL routines', '', 'sslv3 alert bad certificate')]"
             )
@@ -268,6 +277,7 @@ def fake_dtls(monkeypatch):
 
     FakeSession.instances = []
     FakeSession.reject_certs = set()
+    FakeSession.redact_rejection = False
     monkeypatch.setattr(config_flow, "_fetch_samsung_uuid", lambda: "test-uuid")
     monkeypatch.setattr(
         config_flow,
@@ -480,6 +490,77 @@ async def test_rejected_reused_leaf_is_reminted(
     assert [s.cert_pem for s in FakeSession.instances] == [MOCK_LEAF_CERT_PEM, "FULLCHAIN"]
 
 
+async def test_rejected_reused_leaf_is_reminted_against_a_redacted_library(
+    hass: HomeAssistant, monkeypatch, fake_dtls
+) -> None:
+    """Same flow as test_rejected_reused_leaf_is_reminted, but against a
+    connect() failure shaped like smartthings-local >= 0.1.3 -- a fixed,
+    redacted exception with no alert text at all (see errors.py's
+    "Classified errors"). The re-mint decision has to come from
+    _resolve_alert's diagnostic-handshake fallback instead of _alert_name."""
+    from custom_components.localthings import config_flow
+
+    existing = MockConfigEntry(domain=DOMAIN, data=ENTRY_DATA, unique_id="localthings_other")
+    existing.add_to_hass(hass)
+    _patch_clienthello(monkeypatch, {49154})
+    FakeSession.reject_certs = {MOCK_LEAF_CERT_PEM}
+    FakeSession.redact_rejection = True
+
+    class _DiagnosticResult:
+        alert = (2, "bad_certificate")
+
+    diagnosed: list[int] = []
+
+    def _diagnostic_alert(host, port, cert_pem, key_pem):
+        diagnosed.append(port)
+        return _DiagnosticResult()
+
+    monkeypatch.setattr(config_flow, "_diagnostic_alert", _diagnostic_alert)
+
+    result = await hass.config_entries.flow.async_init(DOMAIN, context={"source": "user"})
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"], {CONF_HOST: MOCK_HOST}
+    )
+
+    assert result["type"] == FlowResultType.CREATE_ENTRY
+    assert result["data"][CONF_LEAF_CERT_PEM] == "FULLCHAIN"
+    assert [s.cert_pem for s in FakeSession.instances] == [MOCK_LEAF_CERT_PEM, "FULLCHAIN"]
+    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:
@@ -554,6 +635,92 @@ def test_cert_alert_is_reported_as_a_certificate_problem() -> None:
         assert err.error_key == "cert_rejected"
 
 
+def test_classify_handshake_failure_uses_a_resolved_alert_over_exception_text() -> None:
+    """_handshake_and_read passes its own resolved `alerts` mapping (built
+    via _resolve_alert, which is what actually classifies a failure against
+    smartthings-local >= 0.1.3's redacted exceptions) -- it must win even
+    when the exception text itself says nothing."""
+    from custom_components.localthings.config_flow import (
+        CertRejected,
+        _classify_handshake_failure,
+    )
+
+    err = _classify_handshake_failure(
+        MOCK_HOST,
+        _scan(confirmed=[49154]),
+        [(49154, RuntimeError("session operation failed"))],
+        {49154: "bad_certificate"},
+    )
+    assert isinstance(err, CertRejected)
+    assert err.error_key == "cert_rejected"
+
+
+def test_resolve_alert_prefers_exception_text_over_the_diagnostic_handshake() -> None:
+    """A library still stamping the alert into its exception text (< 0.1.3)
+    answers for free; the diagnostic handshake must not run at all then."""
+    from custom_components.localthings.config_flow import _resolve_alert
+
+    def _must_not_run(*args, **kwargs):
+        raise AssertionError("must not run the diagnostic handshake")
+
+    with patch("custom_components.localthings.config_flow._diagnostic_alert", _must_not_run):
+        name = _resolve_alert(
+            _openssl_alert("tlsv1 alert unknown ca"), MOCK_HOST, 49154, "CERT", "KEY"
+        )
+    assert name == "unknown_ca"
+
+
+def test_resolve_alert_falls_back_to_the_diagnostic_handshake() -> None:
+    """smartthings-local >= 0.1.3 redacts the exception text (see errors.py's
+    "Classified errors"), so the only way left to learn *why* a handshake
+    failed is the library's own classification of the raw alert record."""
+    from smartthings_local.errors import SessionError
+
+    from custom_components.localthings import config_flow
+
+    class _Result:
+        alert = (2, "bad_certificate")
+
+    with patch.object(config_flow, "_diagnostic_alert", lambda *a, **k: _Result()):
+        name = config_flow._resolve_alert(SessionError(), MOCK_HOST, 49154, "CERT", "KEY")
+    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."""
+    from smartthings_local.errors import SessionError, SessionTimeoutError
+
+    from custom_components.localthings import config_flow
+
+    def _boom(*args, **kwargs):
+        raise SessionTimeoutError()
+
+    with patch.object(config_flow, "_diagnostic_alert", _boom):
+        name = config_flow._resolve_alert(SessionError(), MOCK_HOST, 49154, "CERT", "KEY")
+    assert name is None
+
+
 def test_non_cert_alert_is_kept_distinct_from_a_cert_problem() -> None:
     """A cipher or version mismatch is also a deliberate refusal, but no
     amount of fiddling with CA credentials will fix it."""
diff --git a/tests/localthings/test_coordinator.py b/tests/localthings/test_coordinator.py
index cc289fc..6dfd2d0 100644
--- a/tests/localthings/test_coordinator.py
+++ b/tests/localthings/test_coordinator.py
@@ -13,6 +13,7 @@ from homeassistant.const import EVENT_HOMEASSISTANT_STOP
 from homeassistant.core import HomeAssistant
 from homeassistant.exceptions import ServiceValidationError
 from homeassistant.helpers import issue_registry as ir
+from smartthings_local.errors import SessionClosedError, SessionError, SessionTimeoutError
 
 from custom_components.localthings.const import (
     CONF_BYPASS_REMOTE_CONTROL,
@@ -531,6 +532,36 @@ async def test_total_poll_failure_downgrades_observe_mode_to_poll(
     assert coordinator.last_update_success is True
 
 
+def test_defer_reconnect_for_reconnects_immediately_on_a_confirmed_dead_session(
+    hass: HomeAssistant, mock_entry
+) -> None:
+    """smartthings-local >= 0.1.6 raises SessionClosedError -- a
+    ConnectionError, not a TimeoutError -- the moment a dead reader thread
+    is confirmed, instead of the old behavior of letting the request hang
+    out to its own timeout and surface as an ambiguous TimeoutError.
+    _defer_reconnect_for must never extend the block-ACK tolerance to a
+    failure this unambiguous; see its docstring for the full reasoning."""
+    coordinator = LocalThingsCoordinator(hass, mock_entry)
+    coordinator._discovered = True
+
+    assert coordinator._defer_reconnect_for(SessionClosedError()) is False
+
+
+def test_defer_reconnect_for_still_tolerates_an_ambiguous_timeout(
+    hass: HomeAssistant, mock_entry
+) -> None:
+    """The other half of the same distinction: a plain block-ACK timeout --
+    still a TimeoutError, including smartthings-local's own
+    SessionTimeoutError subclass -- keeps its multi-cycle tolerance rather
+    than being swept into the immediate-reconnect path above."""
+    coordinator = LocalThingsCoordinator(hass, mock_entry)
+    coordinator._discovered = True
+
+    for _ in range(coordinator._POLL_TIMEOUT_LIMIT - 1):
+        assert coordinator._defer_reconnect_for(SessionTimeoutError()) is True
+    assert coordinator._defer_reconnect_for(SessionTimeoutError()) is False
+
+
 async def test_poll_timeout_skips_reconnect_when_push_is_healthy(
     hass: HomeAssistant, mock_entry, mock_coordinator_observe_session
 ) -> None:
@@ -786,6 +817,38 @@ async def test_attempt_observe_mode_discards_stale_commit_after_session_swap(
     assert coordinator._resubscribe_due is True
 
 
+async def test_attempt_observe_mode_survives_a_failed_reconnect(
+    hass: HomeAssistant, mock_entry, mock_coordinator_observe_session
+) -> None:
+    """The session was closed out from under this attempt concurrently
+    (rare, but real -- see the docstring above), and the reconnect it tries
+    on the way back in fails too (smartthings-local >= 0.1.3's redacted
+    SessionError, or any other exception). That must not escape
+    _async_update_data uncaught: it should land in the same "give up on
+    push this cycle" state the subscribe-failed and stale-session branches
+    already produce, not skip this integration's own logging/state handling
+    entirely."""
+    await hass.config_entries.async_setup(mock_entry.entry_id)
+    await hass.async_block_till_done()
+    coordinator: LocalThingsCoordinator = hass.data[DOMAIN][mock_entry.entry_id]
+    coordinator._session = None
+    coordinator._reconnect_times = []
+
+    with patch.object(
+        coordinator,
+        "_connect_session",
+        side_effect=SessionError(),
+    ):
+        await coordinator._attempt_observe_mode()  # must not raise
+
+    assert coordinator.observe_mode == MODE_POLL
+    assert coordinator._observe.subscribed_hrefs == set()
+    assert coordinator._resubscribe_due is False
+    # Not the poll path's own reconnect-frequency window (see the fix's
+    # comment) -- this failure must not count toward it.
+    assert coordinator._reconnect_times == []
+
+
 async def test_maybe_retry_observe_mode_uses_most_recent_attempt_not_just_mode_change(
     hass: HomeAssistant, mock_entry, mock_coordinator_observe_session
 ) -> None:
diff --git a/tests/test_coordinator_error_handling.py b/tests/test_coordinator_error_handling.py
new file mode 100644
index 0000000..81f8d8b
--- /dev/null
+++ b/tests/test_coordinator_error_handling.py
@@ -0,0 +1,76 @@
+"""Regression tests for a handful of call sites in coordinator.py that used
+to let a smartthings-local exception (EndpointError, SessionError,
+SessionTimeoutError, SessionClosedError, ... -- or the equivalent bare
+ConnectionError/TimeoutError/OSError an older library version raised) escape
+uncaught instead of going through this integration's own reconnect/logging
+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
+
+from custom_components.localthings.const import (
+    CONF_HOST,
+    CONF_LEAF_CERT_PEM,
+    CONF_LEAF_KEY_PEM,
+    CONF_PORT,
+    DOMAIN,
+)
+from custom_components.localthings.coordinator import LocalThingsCoordinator
+
+ENTRY_DATA = {
+    CONF_HOST: "10.0.0.198",
+    CONF_PORT: 49154,
+    CONF_LEAF_CERT_PEM: "-----BEGIN CERTIFICATE-----\nTEST-LEAF\n-----END CERTIFICATE-----",
+    CONF_LEAF_KEY_PEM: "-----BEGIN PRIVATE KEY-----\nTEST-LEAF-KEY\n-----END PRIVATE KEY-----",
+}
+
+
+def _coordinator(hass: HomeAssistant) -> LocalThingsCoordinator:
+    entry = MockConfigEntry(
+        domain=DOMAIN,
+        data=ENTRY_DATA,
+        unique_id="localthings_ERRHANDLING-TEST",
+    )
+    entry.add_to_hass(hass)
+    return LocalThingsCoordinator(hass, entry)
+
+
+async def test_subdevice_enumeration_failure_does_not_abort_first_discovery(
+    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
+    it within the same cycle -- rare, but until this fix, unguarded: an
+    exception there escaped _async_update_data entirely instead of going
+    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)
+    monkeypatch.setattr(coordinator, "_poll_once", dict)
+
+    def _boom(_resources):
+        raise ConnectionError("session closed")
+
+    monkeypatch.setattr(coordinator, "_enumerate_subdevices_blocking", _boom)
+
+    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
diff --git a/tests/test_services.py b/tests/test_services.py
index 84b3b57..cfc0a69 100644
--- a/tests/test_services.py
+++ b/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):
@@ -324,6 +325,42 @@ async def test_write_resource_verify_after_reports_reverted(hass, coordinator, d
     assert verified["rep"] == {"x.field": "original"}
 
 
+async def test_write_resource_verify_after_survives_a_failed_confirmation_read(
+    hass, coordinator, device_id, monkeypatch
+):
+    """The write itself already landed (see `results`, built before
+    verify_after's wait even starts) by the time the confirmation read runs
+    -- a session dying in the gap verify_after waits out (smartthings-local's
+    redacted SessionClosedError/SessionTimeoutError, or any other exception)
+    must not lose that outcome behind a raised exception. Same "couldn't
+    verify" posture as a 4.04/empty read: `held` stays None, not False."""
+    fake = _FakeSession()
+    fake.queue_get("mode/vs/0", {"x.field": "target"})  # write's own follow-up read
+    coordinator._session = fake
+
+    def _boom(path_segs, href):
+        raise ConnectionError("session closed")
+
+    monkeypatch.setattr(coordinator, "_raw_read_blocking", _boom)
+
+    with patch(_SLEEP_TARGET, new_callable=AsyncMock):
+        response = await _call_write(
+            hass,
+            device_id,
+            writes=[{"href": "/mode/vs/0", "payload": {"x.field": "target"}}],
+            verify_after=30,
+        )
+
+    # The write's own results survive even though verification blew up.
+    assert response["results"][0]["accepted"] is True
+    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(
     hass, coordinator, device_id
 ):
@@ -605,6 +642,65 @@ async def test_read_resource_surfaces_a_collections_list_body(hass, coordinator,
     assert response["body"] == batch
 
 
+async def test_read_resource_failure_is_surfaced_as_a_home_assistant_error(
+    hass, coordinator, device_id, monkeypatch
+):
+    """A session/network failure during a live debug read (e.g.
+    smartthings-local's redacted SessionError, or any other exception) must
+    not reach the service caller raw and untranslated -- write_resource
+    already goes through HomeAssistantError on failure, and async_raw_read
+    must match that instead of letting the exception escape uncaught."""
+
+    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")
+
+
+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
 ):