From e6eada236744d86ede16ca606c0555304631be87 Mon Sep 17 00:00:00 2001
From: Marc Billow <mbillow@me.com>
Date: Thu, 9 Jul 2026 18:43:20 -0500
Subject: [PATCH] feat(observe): detect missed notifications via sweep
 comparison

---
 custom_components/localthings/observe.py | 26 ++++++++++
 tests/localthings/test_observe.py        | 62 ++++++++++++++++++++++++
 2 files changed, 88 insertions(+)

diff --git a/custom_components/localthings/observe.py b/custom_components/localthings/observe.py
index 94f1d20..e73ad42 100644
--- a/custom_components/localthings/observe.py
+++ b/custom_components/localthings/observe.py
@@ -44,6 +44,7 @@ class ObserveManager:
         self._settle_lock = threading.Lock()
         self.subscribed_hrefs: set[str] = set()
         self._notified: set[str] = set()
+        self.fallback_hrefs: set[str] = set()
 
     def mark_write_pending(self, href: str, settle_s: float = DEFAULT_SETTLE_S) -> None:
         with self._settle_lock:
@@ -119,3 +120,28 @@ class ObserveManager:
             self.mode = mode
             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."""
+        if self.mode != MODE_OBSERVE:
+            return 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,
+                )
+                return True
+        return False
+
+    def downgrade_to_poll(self) -> None:
+        self.fallback_hrefs = set(self.subscribed_hrefs)
+        self.subscribed_hrefs = set()
+        self._set_mode(MODE_POLL)
diff --git a/tests/localthings/test_observe.py b/tests/localthings/test_observe.py
index c6fc68f..6127462 100644
--- a/tests/localthings/test_observe.py
+++ b/tests/localthings/test_observe.py
@@ -189,3 +189,65 @@ def test_try_enter_observe_mode_clears_stale_notifications_on_retry():
     assert entered is False
     assert mgr.mode == 'poll'
     assert mgr.subscribed_hrefs == set()
+
+
+def test_check_sweep_for_misses_false_when_not_in_observe_mode():
+    mgr = _manager()
+    assert mgr.check_sweep_for_misses({'/oven/vs/0': {'a': 2}}) is False
+
+
+def test_check_sweep_for_misses_detects_mismatch():
+    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}))
+
+    notifier = threading.Thread(target=_notify_during_grace_period, daemon=True)
+    notifier.start()
+
+    mgr.try_enter_observe_mode(session, [href], grace_period_s=0.02)
+    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}})
+
+    assert missed is True
+
+
+def test_check_sweep_for_misses_ignores_href_during_settle_window():
+    mgr = _manager()
+    session = _FakeSession()
+    href = '/oven/vs/0'
+    mgr.on_notification(href, cbor2.dumps({'a': 1}))
+    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}})
+
+    assert missed is False
+
+
+def test_downgrade_to_poll_moves_subscribed_to_fallback():
+    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}))
+
+    notifier = threading.Thread(target=_notify_during_grace_period, daemon=True)
+    notifier.start()
+
+    mgr.try_enter_observe_mode(session, [href], grace_period_s=0.02)
+    notifier.join()
+
+    mgr.downgrade_to_poll()
+
+    assert mgr.mode == 'poll'
+    assert mgr.subscribed_hrefs == set()
+    assert mgr.fallback_hrefs == {href}