refactor(observe): stop tearing down live sessions on data-drift or poll hiccups

Replace the breadth/silence downgrade heuristic with a simpler rule: a
still-live OBSERVE session is never torn down on a sweep/cache mismatch
(the 30s sweep already corrects the cache regardless of mode) -- only a
proven reconnect invalidates subscriptions. A mismatch instead triggers
extra hot/warm subpolls this cycle as a bounded fallback.

Also stop treating a summary-poll block-level ACK timeout as session
death: distinguish TimeoutError (transfer was progressing, session
likely alive) from ConnectionError (session actually closed), backed by
a consecutive-timeout counter so a genuinely dead channel still
recovers. This was causing a flaky/slow device (e.g. a dishwasher) to
flap observe<->poll every ~45s on nothing but a slow blockwise GET.

Also remove the dead is_active/active_when scaffolding (never wired
into the coordinator) and add DEBUG logging for each OBSERVE notify
received, including its href.

custom_components/localthings/coordinator.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 
 import asyncio
 import logging
+import threading
 import time
 from datetime import timedelta
 from typing import Any
@@ -24,7 +25,7 @@ from .registry.discovery import discover, BoundEntity
 from .registry import CAPABILITIES
 from .registry.adapter import flatten
 from .registry.identity import read_identity, DeviceIdentity
-from .observe import ObserveManager, MODE_OBSERVE, MODE_POLL
+from .observe import ObserveManager, MODE_OBSERVE, MODE_POLL, GRACE_PERIOD_S
 
 from .const import (
     DOMAIN, CONF_HOST, CONF_PORT, CONF_LEAF_CERT_PEM, CONF_LEAF_KEY_PEM,
@@ -54,10 +55,30 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
     device_info: DeviceInfo
     device_serial: str
 
+    # Class-level knobs (not instance attrs) so tests can shrink real-time
+    # delays via `patch.object(LocalThingsCoordinator, ...)` without
+    # touching the production defaults these are computed from. Production
+    # code always sees these two values; only tests override them.
+    _SUBPOLL_STEP_S: float = SUMMARY_INTERVAL_S / 10  # 3.0 s
+    _OBSERVE_GRACE_PERIOD_S: float = GRACE_PERIOD_S
+    _RECONNECT_PAUSE_S: float = 5.0
+
+    # 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. A single slow transfer on an otherwise
+    # fine session shouldn't tear down a working OBSERVE subscription.
+    _POLL_TIMEOUT_LIMIT: int = 3
+
     def __init__(self, hass: HomeAssistant, entry: ConfigEntry) -> None:
+        # Per-device logger (module logger scoped to this device's host) so
+        # every log line — including the base DataUpdateCoordinator's own
+        # messages and ObserveManager's — identifies which device it's
+        # about. A bare module-level logger is shared across every
+        # configured device, which makes multi-device logs ambiguous.
+        self._log = logging.getLogger(f"{__name__}.{entry.data[CONF_HOST]}")
         super().__init__(
             hass,
-            _LOGGER,
+            self._log,
             name=f"{DOMAIN}_{entry.data[CONF_HOST]}",
             update_interval=timedelta(seconds=SUMMARY_INTERVAL_S),
         )
@@ -68,7 +89,9 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
         self.bound = []
         self._cache = StateCache(_NoOpDescriptor())
         self._cache.set_on_change(self._on_cache_changed)
-        self._observe = ObserveManager(self._cache, logger=_LOGGER)
+        self._observe = ObserveManager(self._cache, logger=self._log)
+        self._push_pending = False
+        self._push_pending_lock = threading.Lock()
         self.device_serial = entry.data[CONF_HOST]  # placeholder until first poll
         self.device_info = DeviceInfo(
             identifiers={(DOMAIN, entry.data[CONF_HOST])},
@@ -81,6 +104,7 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
         self._warm_hrefs: list[str] = []
         self.device_type_name: str | None = None
         self.one_ui_version: str = ''
+        self._consecutive_poll_timeouts = 0
         self._unbound_hrefs: list[str] = []
 
     # ------------------------------------------------------------------
@@ -106,11 +130,11 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
         sess.connect()
         sess.start_reader()
         self._session = sess
-        _LOGGER.debug("DTLS connected to %s:%d", host, port)
+        self._log.debug("DTLS connected to %s:%d", host, port)
         try:
             self._identity = read_identity(sess, None)
         except Exception as e:
-            _LOGGER.debug("read_identity failed: %s", e)
+            self._log.debug("read_identity failed: %s", e)
             self._identity = None
 
     def _close_session(self) -> None:
@@ -133,23 +157,48 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
         """StateCache.set_on_change callback. Runs on whatever thread
         applied the update (DTLS reader thread for observe notifications,
         an executor thread for poll/sweep) — never the event loop, so the
-        HA push must be scheduled thread-safely."""
+        HA push must be scheduled thread-safely. A sweep/poll cycle can
+        call apply_rep for dozens of hrefs in a tight loop; coalesce those
+        into a single push instead of one hass.add_job per href."""
         if not changed:
             return
+        with self._push_pending_lock:
+            if self._push_pending:
+                return
+            self._push_pending = True
         self.hass.add_job(self._push_cache_snapshot)
 
     @callback
     def _push_cache_snapshot(self) -> None:
+        with self._push_pending_lock:
+            self._push_pending = False
         if self.bound:
             self.async_set_updated_data(flatten(self.bound, self._cache.snapshot()))
 
     def _poll_once(self) -> dict[str, dict]:
-        """GET /device/0, return parsed resources. Blocking."""
+        """GET /device/0, return parsed resources. Blocking.
+
+        `sess.get()` raises `TimeoutError` when one block's ACK doesn't
+        arrive in time — the transfer was progressing (earlier blocks
+        succeeded) and just didn't finish before the deadline on a slow
+        device. That does NOT prove the session is dead, so it's left
+        open here; `_async_update_data` decides whether repeated timeouts
+        (or a lack of them) warrant a reconnect. Anything else (a
+        `ConnectionError` from an explicitly closed/broken session, a bad
+        response code) is unambiguous — close immediately.
+        """
         if self._session is None:
             self._connect_session()
         sess = self._session
         try:
-            code, payload = sess.get(_SEED_PATH, timeout=15.0)
+            # A slow device can still be mid-transfer (block 8, block 11)
+            # when a tighter deadline cuts it off — that's a poll that
+            # would have succeeded, not a dead session. 35s gives a slow
+            # blockwise transfer room to actually finish instead of
+            # generating a TimeoutError every cycle.
+            code, payload = sess.get(_SEED_PATH, timeout=35.0)
+        except TimeoutError:
+            raise
         except Exception as e:
             self._close_session()
             raise RuntimeError(f"poll GET failed: {e}") from e
@@ -181,24 +230,29 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
                         self._observe.apply(href, rep, source='poll')
                         results[href] = rep
             except Exception as e:
-                _LOGGER.debug("sub-poll %s: %s", href, e)
+                self._log.debug("sub-poll %s: %s", href, e)
         return results
 
     # ------------------------------------------------------------------
     # Sub-poll loop (runs between summary polls)
     # ------------------------------------------------------------------
 
-    async def _run_subpolls(self) -> None:
+    async def _run_subpolls(self, force: bool = False) -> None:
         """Poll hot/warm hrefs in the gaps between summary polls. Only
         runs in poll-only mode — in observe-primary mode those hrefs are
-        already covered by push notifications."""
-        if self._observe.mode == MODE_OBSERVE:
+        already covered by push notifications — unless `force` is set,
+        which this cycle's sweep found disagreeing with the cache on a
+        still-live observe session (see log_sweep_discrepancies): a
+        bounded, self-limiting fallback for a channel that's gone silent
+        without a reconnect, without tearing down subscriptions that
+        would otherwise recover on their own once notifies resume."""
+        if self._observe.mode == MODE_OBSERVE and not force:
             return
         hot = self._hot_hrefs
         warm = self._warm_hrefs
         if not hot and not warm:
             return
-        step = SUMMARY_INTERVAL_S / 10  # 3.0 s
+        step = self._SUBPOLL_STEP_S
         for i in range(1, 10):          # slots 1..9  (T+3 s … T+27 s)
             await asyncio.sleep(step)
             hrefs = list(hot) + (list(warm) if i % 2 == 0 else [])
@@ -208,7 +262,7 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
                         self._poll_hrefs_blocking, hrefs
                     )
                 except Exception as e:
-                    _LOGGER.debug("sub-poll batch failed: %s", e)
+                    self._log.debug("sub-poll batch failed: %s", e)
 
     # ------------------------------------------------------------------
     # Discovery (runs once on first successful poll)
@@ -222,12 +276,12 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
         reg = for_device(one_ui) if one_ui else None
         unbound: list[str] = []
         if reg is not None:
-            _LOGGER.debug("device type: %s (oneUiVersion=%r)", reg.name, one_ui)
+            self._log.debug("device type: %s (oneUiVersion=%r)", reg.name, one_ui)
             bound = discover(resources, reg.capabilities, reg.pattern_capabilities,
                               log=unbound.append)
             self.device_type_name = reg.name
         else:
-            _LOGGER.warning("unknown device type oneUiVersion=%r; using common caps", one_ui)
+            self._log.warning("unknown device type oneUiVersion=%r; using common caps", one_ui)
             bound = discover(resources, CAPABILITIES, log=unbound.append)
             self.device_type_name = None
         self.bound = bound
@@ -265,7 +319,7 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
         self._warm_hrefs = sorted(warm)
 
         self._discovered = True
-        _LOGGER.info(
+        self._log.info(
             "discovered %d entities (serial=%s) hot=%s warm=%s",
             len(bound), serial, self._hot_hrefs, self._warm_hrefs,
         )
@@ -307,7 +361,8 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
         if sess is None:
             return
         await self.hass.async_add_executor_job(
-            self._observe.try_enter_observe_mode, sess, hrefs
+            self._observe.try_enter_observe_mode, sess, hrefs,
+            self._OBSERVE_GRACE_PERIOD_S,
         )
 
     async def _maybe_retry_observe_mode(self) -> None:
@@ -317,6 +372,34 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
             return
         await self._attempt_observe_mode()
 
+    def _defer_reconnect_for(self, e: Exception) -> bool:
+        """True if this poll failure should NOT trigger a reconnect this
+        cycle.
+
+        A `TimeoutError` (see `_poll_once`) means one block's ACK didn't
+        arrive in time — not that the session is dead. A recent OBSERVE
+        notify is direct proof the channel is still live, so always defer
+        in that case. Otherwise, defer until `_POLL_TIMEOUT_LIMIT`
+        consecutive timeouts have piled up — a single slow transfer is
+        normal on a flaky device; a run of them is a real problem. Any
+        other exception (a `ConnectionError`, an explicitly closed
+        session) is unambiguous and always reconnects immediately.
+        """
+        if not isinstance(e, TimeoutError):
+            return False
+        if self._observe.mode == MODE_OBSERVE and self._observe.recently_notified():
+            # Recent push is proof of life — reset the counter too, so
+            # timeouts from an earlier quiet stretch don't carry over and
+            # trigger a reconnect once the device goes quiet again. The
+            # counter should mean "consecutive timeouts with no push
+            # activity to vouch for the session," not just "consecutive
+            # timeouts" — otherwise an intermittently-active device could
+            # still accumulate its way into a false reconnect.
+            self._consecutive_poll_timeouts = 0
+            return True
+        self._consecutive_poll_timeouts += 1
+        return self._consecutive_poll_timeouts < self._POLL_TIMEOUT_LIMIT
+
     # ------------------------------------------------------------------
     # DataUpdateCoordinator hook
     # ------------------------------------------------------------------
@@ -327,22 +410,32 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
             self._subpoll_task.cancel()
             self._subpoll_task = None
 
+        just_downgraded_from_observe = False
         async with self._session_lock:
             try:
                 resources = await self.hass.async_add_executor_job(self._poll_once)
+                self._consecutive_poll_timeouts = 0
             except Exception as e:
+                if self._defer_reconnect_for(e):
+                    self._log.debug(
+                        "poll failed (%s), not yet treated as session "
+                        "death; skipping this cycle: %s",
+                        type(e).__name__, e,
+                    )
+                    return flatten(self.bound, self._cache.snapshot())
+                self._consecutive_poll_timeouts = 0
                 # One reconnect attempt — pause briefly so the device can
                 # clean up its DTLS session state before we knock again.
-                _LOGGER.warning("poll failed, reconnecting: %s", e)
+                self._log.warning("poll failed, reconnecting: %s", e)
                 await self.hass.async_add_executor_job(self._close_session)
-                await asyncio.sleep(5.0)
+                await asyncio.sleep(self._RECONNECT_PAUSE_S)
                 try:
                     resources = await self.hass.async_add_executor_job(self._poll_once)
                 except Exception as e2:
-                    _LOGGER.error("poll failed after reconnect: %s", e2)
+                    self._log.error("poll failed after reconnect: %s", e2)
                     snapshot = self._cache.snapshot()
                     if snapshot:
-                        _LOGGER.debug("Full error:", exc_info=e2)
+                        self._log.debug("Full error:", exc_info=e2)
                         return flatten(self.bound, snapshot)
                     raise UpdateFailed(f"poll failed after reconnect: {e2}") from e2
                 else:
@@ -350,33 +443,50 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
                     # OBSERVE registrations. If we were in observe mode,
                     # that state is now stale — the refresh task is still
                     # pinned to the old (closed) session and nothing will
-                    # ever re-subscribe on the new one. Tear it down now so
-                    # the existing poll-mode retry path (_maybe_retry_observe_mode)
-                    # re-subscribes on the live session on its next cycle.
+                    # ever re-subscribe on the new one. Tear it down and
+                    # try to resubscribe immediately below rather than
+                    # waiting for the poll-mode retry timer — that timer
+                    # exists to throttle devices that never had observe
+                    # working at all, but a reconnect just proved this
+                    # session is healthy, so there's no reason to wait.
                     if self._observe.mode == MODE_OBSERVE:
-                        _LOGGER.debug(
+                        self._log.debug(
                             "reconnect while in observe mode; downgrading to "
-                            "poll so recovery re-subscribes on the new session"
+                            "poll and resubscribing on the new session"
                         )
                         self._observe.downgrade_to_poll()
+                        just_downgraded_from_observe = True
 
         source = 'sweep' if self._discovered else 'poll'
-        if self._observe.mode == MODE_OBSERVE and self._observe.check_sweep_for_misses(resources):
-            self._observe.downgrade_to_poll()
+        sweep_mismatch = False
+        if self._observe.mode == MODE_OBSERVE:
+            # A sweep/cache mismatch never tears down a still-live OBSERVE
+            # session (see log_sweep_discrepancies) — the sweep below
+            # re-applies the authoritative state to the cache regardless,
+            # so there's nothing to correct by downgrading. Only a
+            # reconnect (above) proves subscriptions are actually gone.
+            # Instead, a mismatch triggers extra hot/warm subpolls this
+            # cycle below, so a channel gone silent without a reconnect
+            # (e.g. lost internet on an otherwise-live local session)
+            # still gets fresher-than-30s data.
+            sweep_mismatch = self._observe.log_sweep_discrepancies(resources)
         for href, rep in resources.items():
             self._observe.apply(href, rep, source=source)
 
         if not self._discovered:
             self._run_discovery(resources)
             await self._attempt_observe_mode()
+        elif just_downgraded_from_observe:
+            await self._attempt_observe_mode()
         elif self._observe.mode == MODE_POLL:
             await self._maybe_retry_observe_mode()
 
         # Schedule sub-polls for hot/warm hrefs between summary polls
-        # (no-op in observe-primary mode; _run_subpolls checks the mode).
+        # (no-op in observe-primary mode unless this cycle's sweep found a
+        # mismatch; _run_subpolls checks the mode/force).
         if self._hot_hrefs or self._warm_hrefs:
             self._subpoll_task = self.hass.async_create_task(
-                self._run_subpolls(), name="localthings_subpoll"
+                self._run_subpolls(force=sweep_mismatch), name="localthings_subpoll"
             )
 
         return flatten(self.bound, self._cache.snapshot())
@@ -396,7 +506,7 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
         rep = self._cache.get(href or '') or {}
         result = write_fn(payload, rep, href)
         if result is None:
-            _LOGGER.warning("write_fn rejected payload %r for %s", payload, href)
+            self._log.warning("write_fn rejected payload %r for %s", payload, href)
             return
         path_segs, body = result
 
@@ -406,11 +516,11 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
                 raise RuntimeError("no session")
             self._observe.mark_write_pending(href)
             code, _ = sess.post(path_segs, cbor2.dumps(body), timeout=8.0)
-            _LOGGER.info("PUT %s → code %#04x", href, code)
+            self._log.info("PUT %s → code %#04x", href, code)
 
         try:
             await self.hass.async_add_executor_job(_do_put)
         except Exception as e:
-            _LOGGER.error("command failed for %s: %s", href, e)
+            self._log.error("command failed for %s: %s", href, e)
         else:
             await self.async_request_refresh()

custom_components/localthings/observe.py

@@ -28,8 +28,32 @@ MODE_POLL = 'poll'
 
 DEFAULT_SETTLE_S = 4.0
 GRACE_PERIOD_S = 15.0
+
+
+def _rep_diff(cached: dict, sweep: dict) -> dict:
+    """Fields present in BOTH reps whose values differ, as
+    {field: (cached_value, sweep_value)}.
+
+    The batch `/device/0` sweep and an individual GET/notify can return
+    genuinely different-shaped reps for the same href (e.g. Samsung's
+    batch interface omits `rt`/`if` baseline fields that a direct GET
+    includes) — that's a representation-shape difference, not a missed
+    state change. Only comparing shared fields keeps the diff (and the
+    caller's miss detection) about real content changes.
+    """
+    common = set(cached) & set(sweep)
+    return {k: (cached[k], sweep[k]) for k in common if cached[k] != sweep[k]}
+
+
 SUCCESS_FRACTION = 0.8
 
+# A notify within this long counts as proof the DTLS session is alive,
+# even if the summary poll's blockwise GET just timed out on a slow
+# device. Twice the coordinator's 30s summary interval: generous enough
+# to absorb jitter, tight enough that a channel that's actually gone
+# quiet doesn't get credit for a notify from a while ago.
+PUSH_HEALTH_WINDOW_S = 60.0
+
 
 class ObserveManager:
     """Per-device observe-mode state: mode, write-settle guard, and (later)
@@ -47,6 +71,7 @@ class ObserveManager:
         self._settle_lock = threading.Lock()
         self.subscribed_hrefs: set[str] = set()
         self._notified: set[str] = set()
+        self._last_notify_ts: float | None = None
         self.fallback_hrefs: set[str] = set()
         self._refresh_task: ObserveRefreshTask | None = None
         self._refresh_stop: threading.Event | None = None
@@ -84,8 +109,25 @@ class ObserveManager:
         if not isinstance(rep, dict):
             return
         self._notified.add(href)
+        self._last_notify_ts = time.monotonic()
+        self.log.debug("observe notify: %s", href)
         self.apply(href, rep, source='observe')
 
+    def recently_notified(self, window_s: float = PUSH_HEALTH_WINDOW_S) -> bool:
+        """True if any OBSERVE notify has arrived within `window_s`.
+
+        Used only to gate whether a poll (summary GET) failure should
+        close/reconnect the DTLS session — NOT to decide observe-mode
+        health in general (see `log_sweep_discrepancies` for why that
+        distinction matters: a notify recent enough to prove the session
+        itself is alive is a much stronger, narrower claim than "the
+        channel has been perfectly healthy").
+        """
+        return (
+            self._last_notify_ts is not None
+            and time.monotonic() - self._last_notify_ts < window_s
+        )
+
     def try_enter_observe_mode(
         self, session, hrefs: list[str],
         grace_period_s: float = GRACE_PERIOD_S,
@@ -130,25 +172,54 @@ class ObserveManager:
             self.last_mode_change_ts = time.monotonic()
             self.last_mode_change_wall = time.time()
 
-    def check_sweep_for_misses(self, sweep_resources: dict[str, dict]) -> bool:
-        """Compare a safety-net sweep result against the cache for every
-        subscribed href. A mismatch outside its settle window means
-        OBSERVE silently missed a change (it only notifies on change, so
-        any disagreement is a real miss, not a case of "nothing changed
-        yet"). Returns True if a miss was detected."""
+    def log_sweep_discrepancies(self, sweep_resources: dict[str, dict]) -> bool:
+        """Log any subscribed href where the safety-net sweep disagrees
+        with the cache, on fields present in both reps (see `_rep_diff`
+        for why only shared fields count). Returns True if any
+        discrepancy was found (False if not in observe mode).
+
+        This never changes mode or subscriptions. The sweep already
+        re-applies the full /device/0 result to the cache every cycle
+        regardless of mode, so a missed notify never leaves data stale
+        beyond one sweep interval — there's nothing to correct here. And
+        a still-live OBSERVE session should never be torn down over a
+        data-drift inference: some resources (e.g. an alarm's derived
+        "triggeredTime") appear to update without ever emitting a notify
+        even when the channel is otherwise perfectly healthy, so treating
+        a mismatch as proof of channel death produces false positives.
+        Downgrading also throws away working push coverage for every
+        *other* subscribed href just to react to one that isn't the
+        problem.
+
+        The return value lets the caller respond more cheaply: without
+        an activity signal there's no way to tell "idle device, healthy
+        channel, nothing to notify about" from "dead channel, nothing
+        gets through" — both look the same here. So instead of guessing,
+        the coordinator uses a discrepancy as a trigger for extra hot/warm
+        subpolls this cycle (see `_run_subpolls`'s `force` parameter) —
+        a bounded, self-limiting response to a channel that's gone silent
+        for a reason OTHER than a reconnect (e.g. the device's OBSERVE
+        relay loses its internet connection while the local DTLS session
+        stays up), without tearing down subscriptions that will recover
+        on their own once notifies resume.
+        """
         if self.mode != MODE_OBSERVE:
             return False
+        found = False
         for href in self.subscribed_hrefs:
             if href not in sweep_resources or self._is_settling(href):
                 continue
             cached = self.cache.get(href)
-            if cached is not None and cached != sweep_resources[href]:
-                self.log.warning(
-                    "observe missed a change on %s (sweep disagrees with cache)",
-                    href,
+            if cached is None:
+                continue
+            diff = _rep_diff(cached, sweep_resources[href])
+            if diff:
+                found = True
+                self.log.debug(
+                    "observe missed a change on %s (sweep disagrees with cache): %s",
+                    href, diff,
                 )
-                return True
-        return False
+        return found
 
     def downgrade_to_poll(self) -> None:
         self.fallback_hrefs = set(self.subscribed_hrefs)

custom_components/localthings/registry/__init__.py

@@ -6,8 +6,8 @@ to the matching capabilities, and flattens the result into HA-ready entity
 state. The DTLS/CoAP transport itself lives in the smartthings-local
 package, not here.
 """
-from .adapter import flatten, is_active, _key
+from .adapter import flatten, _key
 from .discovery import BoundEntity, discover
 from .registry import CAPABILITIES
 
-__all__ = ['CAPABILITIES', 'discover', 'BoundEntity', 'flatten', 'is_active', '_key']
+__all__ = ['CAPABILITIES', 'discover', 'BoundEntity', 'flatten', '_key']

custom_components/localthings/registry/adapter.py

@@ -22,13 +22,3 @@ def flatten(bound: list[BoundEntity], resources: dict) -> dict[str, Any]:
         elif b.desc.field:
             out[_key(b)] = b.desc.value_fn(rep.get(b.desc.field))
     return out
-
-
-def is_active(bound: list[BoundEntity], resources: dict) -> bool:
-    """Return True if any capability's active_when fires."""
-    for b in bound:
-        if b.capability.active_when:
-            rep = resources.get(b.href) or {}
-            if b.capability.active_when(rep):
-                return True
-    return False

custom_components/localthings/registry/capabilities/operational.py

@@ -28,10 +28,6 @@ def _int(v):
         return None
 
 
-def _active_when(rep):
-    return _SAMSUNG_STATE_TO_OCF.get(rep.get('x.com.samsung.da.state')) == 'active'
-
-
 def _parse_hms(v):
     if not v:
         return None
@@ -113,5 +109,4 @@ OPERATIONAL_STATE = Capability(
                        ['operational', 'state', 'vs', '0'],
                        {'x.com.samsung.da.state': p})),
     ),
-    active_when=_active_when,
 )

custom_components/localthings/registry/capabilities/oven.py

@@ -213,11 +213,6 @@ def _naturalsteam_write(p, rep, href=None):
     return ['mode', 'vs', '0'], {'x.com.samsung.da.options': opts}
 
 
-# ---------------------------------------------------------------------------
-def _active_when(rep):
-    return _SAMSUNG_STATE_TO_OCF.get(rep.get('x.com.samsung.da.state')) == 'active'
-
-
 # ---------------------------------------------------------------------------
 # Capabilities
 # ---------------------------------------------------------------------------
@@ -225,7 +220,6 @@ def _active_when(rep):
 OVEN_OPERATIONAL_STATE = Capability(
     href='/operational/state/vs/0',
     poll_tier='hot',
-    active_when=_active_when,
     entities=(
         SensorDesc(key='machine_state', field='x.com.samsung.da.state',
                    name='Machine state', icon='mdi:stove', value_fn=_to_ocf),

custom_components/localthings/registry/capability.py

@@ -16,7 +16,6 @@ class Capability:
     href_prefix: Optional[str] = None        # pattern caps only: bind only if href starts with this
     strip_prefix_in_key: bool = False         # strip href_prefix segs before building key_override
     match_fn: Optional[Callable[[dict, dict], bool]] = None  # match_fn(rep, resources) -> bool
-    # Rare optional hooks — only operational-state-style resources use these.
-    active_when: Optional[Callable[[dict], bool]] = None
+    # Rare optional hook — only operational-state-style resources use this.
     on_observation: Optional[Callable[[dict, dict], None]] = None
     project: Optional[Callable[[dict, dict], dict]] = None

tests/localthings/conftest.py

@@ -35,6 +35,32 @@ from custom_components.localthings.const import (
     CONF_CA_CERT_PEM, CONF_CA_KEY_PEM,
     CONF_HOST, CONF_LEAF_CERT_PEM, CONF_LEAF_KEY_PEM, CONF_PORT, DOMAIN,
 )
+from custom_components.localthings.coordinator import LocalThingsCoordinator
+
+
+@pytest.fixture(autouse=True)
+def _fast_coordinator_timers():
+    """Shrink the coordinator's real-time delays for every localthings test.
+
+    `_run_subpolls` and `_attempt_observe_mode` are, by design, blocking/
+    real-time operations in production (a real sub-poll cadence and a real
+    CoAP OBSERVE grace period). Tests that drive them through a full
+    `hass.config_entries.async_setup(...)` + `hass.async_block_till_done()`
+    would otherwise burn tens of real seconds per test waiting them out.
+
+    This patches only the *class* attributes coordinator.py exposes for
+    this purpose (`_SUBPOLL_STEP_S`, `_OBSERVE_GRACE_PERIOD_S`) — the
+    production defaults they're computed from (`SUMMARY_INTERVAL_S`,
+    `GRACE_PERIOD_S`) are untouched. Tests that need to exercise the real
+    grace-period race (e.g. test_observe.py's own `ObserveManager` tests)
+    still pass `grace_period_s=` explicitly and are unaffected.
+    """
+    with (
+        patch.object(LocalThingsCoordinator, '_SUBPOLL_STEP_S', 0.001),
+        patch.object(LocalThingsCoordinator, '_OBSERVE_GRACE_PERIOD_S', 0.02),
+        patch.object(LocalThingsCoordinator, '_RECONNECT_PAUSE_S', 0.01),
+    ):
+        yield
 
 FIXTURES = Path(__file__).resolve().parent.parent / 'fixtures'
 

tests/localthings/test_coordinator.py

@@ -12,10 +12,10 @@ from homeassistant.core import HomeAssistant
 from homeassistant.helpers import issue_registry as ir
 
 from custom_components.localthings.const import (
-    DOMAIN, SUMMARY_INTERVAL_S,
+    CONF_HOST, DOMAIN, SUMMARY_INTERVAL_S,
 )
 from custom_components.localthings.coordinator import LocalThingsCoordinator
-from custom_components.localthings.observe import MODE_OBSERVE, MODE_POLL
+from custom_components.localthings.observe import MODE_OBSERVE, MODE_POLL, PUSH_HEALTH_WINDOW_S
 
 from .conftest import ENTRY_DATA, MOCK_SERIAL
 
@@ -182,6 +182,41 @@ async def test_last_resources_backed_by_state_cache(
     assert resources  # fridge fixture is non-empty
 
 
+def test_logger_is_scoped_to_device_host(
+    hass: HomeAssistant, mock_entry
+) -> None:
+    """Every log line (coordinator's own, DataUpdateCoordinator's internal
+    messages, and ObserveManager's) must identify which device it's about —
+    a bare module-level logger is shared across every configured device and
+    makes multi-device logs ambiguous."""
+    coordinator = LocalThingsCoordinator(hass, mock_entry)
+
+    assert coordinator._log.name.endswith(ENTRY_DATA[CONF_HOST])
+    assert coordinator.logger is coordinator._log
+    assert coordinator._observe.log is coordinator._log
+
+
+def test_cache_changes_coalesce_into_a_single_push(
+    hass: HomeAssistant, mock_entry
+) -> None:
+    """A poll/sweep cycle applying many hrefs in a tight loop must not
+    schedule one hass.add_job push per href — only one push per burst."""
+    coordinator = LocalThingsCoordinator(hass, mock_entry)
+
+    with patch.object(hass, 'add_job') as mock_add_job:
+        coordinator._on_cache_changed(True, 'poll')
+        coordinator._on_cache_changed(True, 'poll')
+        coordinator._on_cache_changed(True, 'poll')
+
+        assert mock_add_job.call_count == 1
+
+        coordinator._push_cache_snapshot()
+
+        coordinator._on_cache_changed(True, 'poll')
+
+        assert mock_add_job.call_count == 2
+
+
 async def test_enters_poll_mode_when_no_notifies_arrive(
     hass: HomeAssistant, mock_entry, mock_coordinator_observe_session
 ) -> None:
@@ -259,8 +294,21 @@ async def test_reconnect_while_observe_mode_downgrades_to_poll(
     assert entered is True
     assert coordinator.observe_mode == MODE_OBSERVE
 
+    # Age out the notify so recently_notified() is False — a poll failure
+    # with a recent notify is now treated as evidence the session is
+    # still alive (see test_poll_failure_skips_reconnect_when_push_is_healthy)
+    # and would not trigger a reconnect at all. This test covers a
+    # genuinely dead channel: no recent push, poll fails, reconnect fires.
+    coordinator._observe._last_notify_ts -= PUSH_HEALTH_WINDOW_S + 1
+
     # Simulate the existing "poll failed, reconnecting" branch: _poll_once
     # fails once (triggering the reconnect/backoff path), then succeeds.
+    # No fresh notifies are supplied for the immediate resubscribe attempt
+    # this now triggers, so its grace period (shortened for tests by the
+    # `_fast_coordinator_timers` autouse fixture — see conftest.py) times
+    # out and mode stays 'poll' — this test only asserts the tear-down half of
+    # the fix; test_reconnect_from_observe_mode_resubscribes_immediately
+    # covers the successful-immediate-resubscribe half.
     with (
         patch(
             'custom_components.localthings.coordinator.LocalThingsCoordinator._poll_once',
@@ -277,6 +325,332 @@ async def test_reconnect_while_observe_mode_downgrades_to_poll(
     assert coordinator._observe.mode == MODE_POLL
 
 
+async def test_poll_timeout_skips_reconnect_when_push_is_healthy(
+    hass: HomeAssistant, mock_entry, mock_coordinator_observe_session
+) -> None:
+    """A poll (summary GET) TimeoutError with a recent OBSERVE notify must
+    not reconnect or downgrade at all — the notify proves the DTLS session
+    and its subscriptions are still alive, so the GET failure is just a
+    choked blockwise transfer on a slow device, not a dead channel. This
+    is the fix for a flaky device (e.g. a slow dishwasher) repeatedly
+    tearing down a healthy observe session on nothing but a poll hiccup."""
+    fake = mock_coordinator_observe_session
+    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]
+    hrefs = coordinator._hot_hrefs + coordinator._warm_hrefs
+
+    def _notify_during_grace_period():
+        time.sleep(0.005)
+        for href in hrefs:
+            fake.on_notification(href, cbor2.dumps({'notified': True}))
+
+    notifier = threading.Thread(target=_notify_during_grace_period, daemon=True)
+    notifier.start()
+    entered = await hass.async_add_executor_job(
+        coordinator._observe.try_enter_observe_mode,
+        fake, hrefs, 0.02, 0.8,
+    )
+    notifier.join()
+    assert entered is True
+    assert coordinator.observe_mode == MODE_OBSERVE
+
+    with (
+        patch(
+            'custom_components.localthings.coordinator.LocalThingsCoordinator._poll_once',
+            side_effect=TimeoutError('GET /device/0 block 11 timeout'),
+        ),
+        patch(
+            'custom_components.localthings.coordinator.LocalThingsCoordinator._close_session',
+        ) as mock_close,
+    ):
+        await coordinator.async_request_refresh()
+        await hass.async_block_till_done()
+
+    assert coordinator.observe_mode == MODE_OBSERVE
+    mock_close.assert_not_called()
+
+
+async def test_poll_timeout_reconnects_after_consecutive_limit_even_with_push(
+    hass: HomeAssistant, mock_entry, mock_coordinator_observe_session, fridge_resources
+) -> None:
+    """Push health only matters at the moment of the failing poll — if
+    notifies later go stale too (not supplied here, so recently_notified()
+    becomes False once the window elapses) a run of consecutive timeouts
+    must still escalate to a reconnect, so a genuinely dead channel isn't
+    stuck forever just because it looked healthy once."""
+    fake = mock_coordinator_observe_session
+    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]
+    hrefs = coordinator._hot_hrefs + coordinator._warm_hrefs
+
+    def _notify_during_grace_period():
+        time.sleep(0.005)
+        for href in hrefs:
+            fake.on_notification(href, cbor2.dumps({'notified': True}))
+
+    notifier = threading.Thread(target=_notify_during_grace_period, daemon=True)
+    notifier.start()
+    entered = await hass.async_add_executor_job(
+        coordinator._observe.try_enter_observe_mode,
+        fake, hrefs, 0.02, 0.8,
+    )
+    notifier.join()
+    assert entered is True
+    assert coordinator.observe_mode == MODE_OBSERVE
+
+    # No notify is recent any more, so every timeout below counts toward
+    # the consecutive-timeout limit instead of being deferred.
+    coordinator._observe._last_notify_ts -= PUSH_HEALTH_WINDOW_S + 1
+
+    # Call _async_update_data directly rather than via
+    # async_request_refresh() — the coordinator's built-in debouncer
+    # (REQUEST_REFRESH_DEFAULT_COOLDOWN, real wall-clock seconds) would
+    # otherwise coalesce these rapid-fire calls into far fewer than the
+    # consecutive-timeout count this test needs to exercise.
+    with patch(
+        'custom_components.localthings.coordinator.LocalThingsCoordinator._poll_once',
+        side_effect=TimeoutError('GET /device/0 block 11 timeout'),
+    ):
+        for _ in range(coordinator._POLL_TIMEOUT_LIMIT - 1):
+            await coordinator._async_update_data()
+            assert coordinator.observe_mode == MODE_OBSERVE
+
+    # The final consecutive timeout crosses the limit and triggers the
+    # existing reconnect path, which then succeeds and downgrades.
+    with (
+        patch(
+            'custom_components.localthings.coordinator.LocalThingsCoordinator._poll_once',
+            side_effect=[TimeoutError('GET /device/0 block 11 timeout'), fridge_resources],
+        ),
+        patch(
+            'custom_components.localthings.coordinator.asyncio.sleep',
+            new=AsyncMock(),
+        ),
+    ):
+        await coordinator._async_update_data()
+
+    assert coordinator.observe_mode == MODE_POLL
+
+
+async def test_poll_timeout_counter_resets_when_push_is_healthy_again(
+    hass: HomeAssistant, mock_entry, mock_coordinator_observe_session
+) -> None:
+    """Timeouts accrued during a quiet stretch must not survive into a
+    later burst of healthy push — otherwise an intermittently-active
+    device could reconnect on old timeouts even though push just proved
+    the session alive. The counter means "consecutive timeouts with no
+    push activity to vouch for the session," not just "consecutive
+    timeouts count.\""""
+    fake = mock_coordinator_observe_session
+    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]
+    hrefs = coordinator._hot_hrefs + coordinator._warm_hrefs
+
+    def _notify_during_grace_period():
+        time.sleep(0.005)
+        for href in hrefs:
+            fake.on_notification(href, cbor2.dumps({'notified': True}))
+
+    notifier = threading.Thread(target=_notify_during_grace_period, daemon=True)
+    notifier.start()
+    entered = await hass.async_add_executor_job(
+        coordinator._observe.try_enter_observe_mode,
+        fake, hrefs, 0.02, 0.8,
+    )
+    notifier.join()
+    assert entered is True
+    assert coordinator.observe_mode == MODE_OBSERVE
+
+    # Age the notify out so timeouts accrue toward the limit.
+    coordinator._observe._last_notify_ts -= PUSH_HEALTH_WINDOW_S + 1
+    with patch(
+        'custom_components.localthings.coordinator.LocalThingsCoordinator._poll_once',
+        side_effect=TimeoutError('GET /device/0 block 11 timeout'),
+    ):
+        for _ in range(coordinator._POLL_TIMEOUT_LIMIT - 1):
+            await coordinator._async_update_data()
+    assert coordinator._consecutive_poll_timeouts == coordinator._POLL_TIMEOUT_LIMIT - 1
+
+    # A fresh notify proves push is healthy again — the next timeout must
+    # not add to the pre-existing count, and must not reconnect.
+    fake.on_notification(hrefs[0], cbor2.dumps({'notified': True}))
+    with patch(
+        'custom_components.localthings.coordinator.LocalThingsCoordinator._poll_once',
+        side_effect=TimeoutError('GET /device/0 block 11 timeout'),
+    ):
+        await coordinator._async_update_data()
+
+    assert coordinator._consecutive_poll_timeouts == 0
+    assert coordinator.observe_mode == MODE_OBSERVE
+
+
+async def test_reconnect_from_observe_mode_resubscribes_immediately(
+    hass: HomeAssistant, mock_entry, mock_coordinator_observe_session, fridge_resources
+) -> None:
+    """A reconnect just proved the session is healthy, so the coordinator
+    must attempt resubscribe in the same update cycle rather than waiting
+    for the 600s poll-mode retry timer (_RECOVERY_RETRY_S) — that timer
+    exists to throttle devices that never had observe working at all, not
+    ones that just proved their session works."""
+    fake = mock_coordinator_observe_session
+    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]
+    hrefs = coordinator._hot_hrefs + coordinator._warm_hrefs
+
+    def _notify(delay: float = 0.005) -> threading.Thread:
+        def _run():
+            time.sleep(delay)
+            for href in hrefs:
+                fake.on_notification(href, cbor2.dumps({'notified': True}))
+        t = threading.Thread(target=_run, daemon=True)
+        t.start()
+        return t
+
+    notifier = _notify()
+    entered = await hass.async_add_executor_job(
+        coordinator._observe.try_enter_observe_mode,
+        fake, hrefs, 0.02, 0.8,
+    )
+    notifier.join()
+    assert entered is True
+    assert coordinator.observe_mode == MODE_OBSERVE
+
+    # Age out the notify so recently_notified() is False and the poll
+    # failure below actually triggers a reconnect (see
+    # test_poll_failure_skips_reconnect_when_push_is_healthy for the
+    # healthy-push case, which now skips reconnecting entirely).
+    coordinator._observe._last_notify_ts -= PUSH_HEALTH_WINDOW_S + 1
+
+    # If a stale retry timer (rather than an immediate resubscribe) were
+    # driving recovery, mode would still be 'poll' right after this single
+    # refresh call — the 600s gate wouldn't have elapsed yet.
+    with (
+        patch(
+            'custom_components.localthings.coordinator.LocalThingsCoordinator._poll_once',
+            side_effect=[RuntimeError('connection lost'), fridge_resources],
+        ),
+        patch(
+            'custom_components.localthings.coordinator.asyncio.sleep',
+            new=AsyncMock(),
+        ),
+    ):
+        notifier = _notify()
+        await coordinator.async_request_refresh()
+        await hass.async_block_till_done()
+        notifier.join()
+
+    assert coordinator.observe_mode == MODE_OBSERVE
+
+
+async def test_sweep_mismatch_never_downgrades_a_live_observe_session(
+    hass: HomeAssistant, mock_entry, mock_coordinator_observe_session
+) -> None:
+    """A sweep/cache mismatch is diagnostic-only (see test_observe.py) —
+    it must never tear down a still-live OBSERVE session. The sweep
+    already re-applies the authoritative state to the cache regardless of
+    mode, so a downgrade would only throw away working push coverage on
+    every other subscribed href without fixing anything."""
+    fake = mock_coordinator_observe_session
+    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]
+    hrefs = coordinator._hot_hrefs + coordinator._warm_hrefs
+    assert len(hrefs) >= 2
+
+    def _notify(delay: float = 0.005) -> threading.Thread:
+        def _run():
+            time.sleep(delay)
+            for href in hrefs:
+                fake.on_notification(href, cbor2.dumps({'notified': True}))
+        t = threading.Thread(target=_run, daemon=True)
+        t.start()
+        return t
+
+    notifier = _notify()
+    entered = await hass.async_add_executor_job(
+        coordinator._observe.try_enter_observe_mode,
+        fake, hrefs, 0.02, 0.8,
+    )
+    notifier.join()
+    assert entered is True
+    assert coordinator.observe_mode == MODE_OBSERVE
+
+    # A sweep result disagreeing with the cache on every subscribed href —
+    # previously treated as a device-wide miss and downgraded.
+    stale_sweep = {href: {'notified': False} for href in hrefs}
+
+    with patch(
+        'custom_components.localthings.coordinator.LocalThingsCoordinator._poll_once',
+        return_value=stale_sweep,
+    ):
+        await coordinator.async_request_refresh()
+        await hass.async_block_till_done()
+
+    assert coordinator.observe_mode == MODE_OBSERVE
+    # The mismatch is diagnostic-only, but the sweep result still won —
+    # the cache reflects the latest authoritative state either way.
+    assert coordinator.last_resources[hrefs[0]]['notified'] is False
+
+
+async def test_sweep_mismatch_forces_subpolls_on_a_live_observe_session(
+    hass: HomeAssistant, mock_entry, mock_coordinator_observe_session
+) -> None:
+    """A sweep/cache mismatch on a live observe session must not tear
+    anything down, but it should trigger extra hot/warm subpolls this
+    cycle — the bounded fallback for a channel gone silent without a
+    reconnect (e.g. lost internet on an otherwise-live local session)."""
+    fake = mock_coordinator_observe_session
+    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]
+    hrefs = coordinator._hot_hrefs + coordinator._warm_hrefs
+    assert len(hrefs) >= 2
+
+    def _notify(delay: float = 0.005) -> threading.Thread:
+        def _run():
+            time.sleep(delay)
+            for href in hrefs:
+                fake.on_notification(href, cbor2.dumps({'notified': True}))
+        t = threading.Thread(target=_run, daemon=True)
+        t.start()
+        return t
+
+    notifier = _notify()
+    entered = await hass.async_add_executor_job(
+        coordinator._observe.try_enter_observe_mode,
+        fake, hrefs, 0.02, 0.8,
+    )
+    notifier.join()
+    assert entered is True
+    assert coordinator.observe_mode == MODE_OBSERVE
+
+    stale_sweep = {href: {'notified': False} for href in hrefs}
+
+    with (
+        patch(
+            'custom_components.localthings.coordinator.LocalThingsCoordinator._poll_once',
+            return_value=stale_sweep,
+        ),
+        patch.object(
+            LocalThingsCoordinator, '_run_subpolls', new_callable=AsyncMock,
+        ) as mock_subpolls,
+    ):
+        await coordinator.async_request_refresh()
+        await hass.async_block_till_done()
+
+    mock_subpolls.assert_called_once_with(force=True)
+
+
 async def test_write_marks_href_pending_before_post(
     hass: HomeAssistant, mock_entry, mock_coordinator_observe_session
 ) -> None:

tests/localthings/test_observe.py

@@ -9,7 +9,9 @@ import pytest
 
 from smartthings_local.ocf.state_cache import StateCache
 
-from custom_components.localthings.observe import ObserveManager, MODE_POLL
+from custom_components.localthings.observe import (
+    ObserveManager, MODE_POLL, _rep_diff,
+)
 
 
 class _NullDescriptor:
@@ -156,6 +158,25 @@ def test_on_notification_ignores_malformed_cbor():
     assert mgr.cache.get('/oven/vs/0') is None
 
 
+def test_recently_notified_false_before_any_notify():
+    mgr = _manager()
+    assert mgr.recently_notified() is False
+
+
+def test_recently_notified_true_just_after_a_notify():
+    mgr = _manager()
+    mgr.on_notification('/oven/vs/0', cbor2.dumps({'a': 1}))
+    assert mgr.recently_notified() is True
+
+
+def test_recently_notified_false_after_window_elapses():
+    mgr = _manager()
+    mgr.on_notification('/oven/vs/0', cbor2.dumps({'a': 1}))
+    assert mgr.recently_notified(window_s=0.01) is True
+    time.sleep(0.02)
+    assert mgr.recently_notified(window_s=0.01) is False
+
+
 def test_try_enter_observe_mode_clears_stale_notifications_on_retry():
     """Regression test: verify second call to try_enter_observe_mode() doesn't
     get polluted by notifications from the first call. This happens in real
@@ -200,19 +221,65 @@ def test_try_enter_observe_mode_clears_stale_notifications_on_retry():
         mgr.close()
 
 
-def test_check_sweep_for_misses_false_when_not_in_observe_mode():
+def test_log_sweep_discrepancies_noop_when_not_in_observe_mode(caplog):
     mgr = _manager()
-    assert mgr.check_sweep_for_misses({'/oven/vs/0': {'a': 2}}) is False
+    with caplog.at_level('DEBUG'):
+        mgr.log_sweep_discrepancies({'/oven/vs/0': {'a': 2}})
+    assert 'observe missed a change' not in caplog.text
 
 
-def test_check_sweep_for_misses_detects_mismatch():
+def test_log_sweep_discrepancies_never_changes_mode():
+    """A sweep/cache mismatch is diagnostic-only — a still-live OBSERVE
+    session is never torn down over a data-drift inference. The 30s sweep
+    already re-applies the authoritative state to the cache regardless of
+    mode, so there's nothing a downgrade would fix; it would only throw
+    away working push coverage on every other subscribed href."""
+    mgr = _manager()
+    session = _FakeSession()
+    hrefs = ['/oven/vs/0', '/power/vs/0']
+
+    def _notify_during_grace_period():
+        time.sleep(0.005)
+        for href in hrefs:
+            mgr.on_notification(href, cbor2.dumps({'a': 1}))
+
+    notifier = threading.Thread(target=_notify_during_grace_period, daemon=True)
+    notifier.start()
+
+    try:
+        mgr.try_enter_observe_mode(session, hrefs, grace_period_s=0.02)
+        notifier.join()
+        assert mgr.mode == 'observe'
+
+        # Sweep sees values the cache never got via notify on BOTH hrefs.
+        mgr.log_sweep_discrepancies({hrefs[0]: {'a': 2}, hrefs[1]: {'a': 2}})
+
+        assert mgr.mode == 'observe'
+    finally:
+        mgr.close()
+
+
+def test_rep_diff_reports_only_shared_fields_that_differ():
+    """Fields present on only one side (a shape difference, e.g. batch
+    sweep vs. individual GET/notify returning different fields for the
+    same href) must not be reported — only shared fields with different
+    values."""
+    diff = _rep_diff({'a': 1, 'b': 2, 'c': 3}, {'a': 1, 'b': 5, 'd': 9})
+
+    assert diff == {'b': (2, 5)}
+
+
+def test_log_sweep_discrepancies_ignores_shape_only_differences(caplog):
+    """A sweep rep missing/adding fields the cache doesn't have (or vice
+    versa) is a representation-shape difference, not a missed change —
+    must not be logged as a discrepancy."""
     mgr = _manager()
     session = _FakeSession()
     href = '/oven/vs/0'
 
     def _notify_during_grace_period():
         time.sleep(0.005)
-        mgr.on_notification(href, cbor2.dumps({'a': 1}))
+        mgr.on_notification(href, cbor2.dumps({'a': 1, 'rt': ['x'], 'if': ['y']}))
 
     notifier = threading.Thread(target=_notify_during_grace_period, daemon=True)
     notifier.start()
@@ -222,15 +289,48 @@ def test_check_sweep_for_misses_detects_mismatch():
         notifier.join()
         assert mgr.mode == 'observe'
 
-        # Sweep sees a value the cache never got via notify -> observe missed it.
-        missed = mgr.check_sweep_for_misses({href: {'a': 2}})
+        # Sweep rep has the same 'a' value but lacks rt/if and adds 'href' —
+        # a shape difference (batch interface), not a missed content change.
+        with caplog.at_level('DEBUG'):
+            mgr.log_sweep_discrepancies({href: {'a': 1, 'href': href}})
+    finally:
+        mgr.close()
 
-        assert missed is True
+    assert 'observe missed a change' not in caplog.text
+
+
+def test_log_sweep_discrepancies_includes_the_diff(caplog):
+    """The log line must include what actually differed, not just that it
+    did."""
+    mgr = _manager()
+    session = _FakeSession()
+    hrefs = ['/oven/vs/0', '/power/vs/0']
+
+    def _notify_during_grace_period():
+        time.sleep(0.005)
+        for href in hrefs:
+            mgr.on_notification(href, cbor2.dumps({'a': 1, 'b': 2}))
+
+    notifier = threading.Thread(target=_notify_during_grace_period, daemon=True)
+    notifier.start()
+
+    try:
+        mgr.try_enter_observe_mode(session, hrefs, grace_period_s=0.02)
+        notifier.join()
+        assert mgr.mode == 'observe'
+
+        with caplog.at_level('DEBUG'):
+            mgr.log_sweep_discrepancies(
+                {hrefs[0]: {'a': 1, 'b': 5}, hrefs[1]: {'a': 1, 'b': 9}}
+            )
+
+        assert "{'b': (2, 5)}" in caplog.text
+        assert mgr.mode == 'observe'
     finally:
         mgr.close()
 
 
-def test_check_sweep_for_misses_ignores_href_during_settle_window():
+def test_log_sweep_discrepancies_ignores_href_during_settle_window(caplog):
     mgr = _manager()
     session = _FakeSession()
     href = '/oven/vs/0'
@@ -238,9 +338,10 @@ def test_check_sweep_for_misses_ignores_href_during_settle_window():
     mgr.try_enter_observe_mode(session, [href], grace_period_s=0.01)
     mgr.mark_write_pending(href, settle_s=5.0)
 
-    missed = mgr.check_sweep_for_misses({href: {'a': 2}})
+    with caplog.at_level('DEBUG'):
+        mgr.log_sweep_discrepancies({href: {'a': 2}})
 
-    assert missed is False
+    assert 'observe missed a change' not in caplog.text
 
 
 def test_downgrade_to_poll_moves_subscribed_to_fallback():

tests/test_capability.py

@@ -6,7 +6,6 @@ def test_capability_defaults():
     c = Capability(href='/kidslock/vs/0',
                    entities=(BinarySensorDesc(key='child_lock', field='x.com.samsung.da.kidsLock'),))
     assert c.poll_tier == 'cold'
-    assert c.active_when is None
     assert len(c.entities) == 1
 
 

tests/test_end_to_end_v2.py

@@ -2,7 +2,7 @@ import pytest
 from tests.conftest import _load_device
 from custom_components.localthings.registry.by_type import for_device, _type_key
 from custom_components.localthings.registry.discovery import discover
-from custom_components.localthings.registry.adapter import flatten, is_active
+from custom_components.localthings.registry.adapter import flatten
 
 
 @pytest.mark.parametrize('name,expected_type_key', [
@@ -24,5 +24,3 @@ def test_full_pipeline_v2(name, expected_type_key):
 
     state = flatten(bound, resources)
     assert state
-    # is_active returns bool; just verify it runs without error
-    is_active(bound, resources)

tests/test_operational_capability.py

@@ -6,8 +6,3 @@ def test_machine_state_maps_samsung_to_ocf():
     assert ms.value_fn('Run') == 'active'
     assert ms.value_fn('Pause') == 'pause'
     assert ms.value_fn('Ready') == 'idle'
-
-
-def test_active_when_true_only_when_running():
-    assert OPERATIONAL_STATE.active_when({'x.com.samsung.da.state': 'Run'}) is True
-    assert OPERATIONAL_STATE.active_when({'x.com.samsung.da.state': 'Ready'}) is False