fix(coordinator): close the observe-mode race an Opus design review found

The command-retry fix (issue #294) added a self._close_session() call to
async_send_command that isn't synchronized against _attempt_observe_mode,
which reads self._session and subscribes to it without holding
_session_lock. A write's retry racing an in-flight subscribe attempt
could tear down the session mid-subscribe -- or worse, land the close
*after* the attempt's grace wait already succeeded, letting it commit
observe mode against a session that's already gone: mode claims "Push"
forever, with nothing left to notice the underlying socket is dead.

Split ObserveManager.try_enter_observe_mode into four pieces
(subscribe_hrefs / await_observe_notifies / enter_observe_mode /
abandon_observe_attempt), keeping try_enter_observe_mode as a thin
wrapper so its direct callers in test_observe.py are unaffected.
_attempt_observe_mode now holds _session_lock only for the subscribe
burst (each send is fire-and-forget, not a network round trip) and
re-checks self._session is sess under the lock right before committing
-- sess keeps the old session object alive, so identity can't be
recycled onto a new one, which is what makes the check sufficient
without a separate generation counter. The wait itself stays lock-free,
so a command write is never blocked behind it.

Two more bugs the same investigation turned up, fixed in the same pass
since they're direct consequences of the design above:

- async_send_command's own successful reconnect didn't downgrade observe
  mode the way the poll path's reconnect already does, leaving the same
  stale-commit problem reachable with zero concurrency at all -- just a
  write's retry succeeding while mode was observe. Replaced the poll
  path's local just_downgraded_from_observe with an instance flag both
  reconnect sites set, so either one triggers an immediate resubscribe.

- _maybe_retry_observe_mode's 600s throttle gated solely on
  last_mode_change_ts, which _set_mode only stamps on an actual
  transition -- a device that never succeeds at observe mode leaves that
  timestamp stuck at construction time, so the throttle opens once and
  never closes again, re-attempting on every single poll cycle instead
  of every 600s. Now gates on the more recent of that timestamp and a
  new _last_observe_attempt_ts, stamped on every attempt regardless of
  outcome.

custom_components/localthings/coordinator.py

@@ -204,6 +204,14 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
         self._consecutive_poll_timeouts = 0
         self._unbound_hrefs: list[str] = []
         self._reconnect_times: list[float] = []
+        # See _maybe_retry_observe_mode: last_mode_change_ts alone doesn't
+        # move on a failed attempt, so this tracks attempts too.
+        self._last_observe_attempt_ts = 0.0
+        # Set by both reconnect paths (poll and command) that hand back a
+        # session with zero OBSERVE registrations while mode was still
+        # observe; consumed once to trigger an immediate resubscribe
+        # instead of waiting out _RECOVERY_RETRY_S.
+        self._resubscribe_due = False
 
     # ------------------------------------------------------------------
     # Session management (all blocking — must run in executor)
@@ -731,29 +739,62 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
             ir.async_delete_issue(self.hass, DOMAIN, issue_id)
 
     async def _attempt_observe_mode(self) -> None:
-        """Called once, right after first discovery. Blocking (sleeps for
-        the whole grace period) — must run in an executor."""
+        """Called once, right after first discovery, after a reconnect
+        downgrades from observe, and periodically while polling. Two
+        phases: subscribing holds `_session_lock` (each send is fire-and-
+        forget, not a network round trip); the grace wait that follows
+        does not, so a concurrent command write isn't blocked for the
+        whole ~15s wait (issue #294) -- only the brief subscribe burst.
+
+        The wait can outlast a reconnect elsewhere (the poll path's own
+        recovery, or a command retry), which would otherwise let a stale
+        success commit observe mode against a session that's already been
+        replaced -- claiming "Push" with nothing left to notice it's dead.
+        `self._session is sess` re-checked under the lock right before
+        committing closes that: `sess` keeps the old object alive, so
+        identity can't be recycled onto a new one.
+        """
         hrefs = self._hot_hrefs + self._warm_hrefs
         if not hrefs:
             return
-        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)
-        sess = self._session
-        if sess is None:
+        self._last_observe_attempt_ts = time.monotonic()
+        async with self._session_lock:
+            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)
+            sess = self._session
+            if sess is None:
+                return
+            subscribed = await self.hass.async_add_executor_job(
+                self._observe.subscribe_hrefs, sess, hrefs
+            )
+        if not subscribed:
+            self._observe.abandon_observe_attempt()
             return
-        await self.hass.async_add_executor_job(
-            self._observe.try_enter_observe_mode,
-            sess,
-            hrefs,
-            self._OBSERVE_GRACE_PERIOD_S,
+        reached = await self.hass.async_add_executor_job(
+            self._observe.await_observe_notifies, subscribed, self._OBSERVE_GRACE_PERIOD_S
         )
+        async with self._session_lock:
+            if not reached or self._session is not sess:
+                self._observe.abandon_observe_attempt()
+                return
+            self._observe.enter_observe_mode(sess, subscribed)
 
     async def _maybe_retry_observe_mode(self) -> None:
         """While in poll-only mode, periodically re-attempt observe mode
-        so a device that gains internet access recovers push automatically."""
-        if time.monotonic() - self._observe.last_mode_change_ts < _RECOVERY_RETRY_S:
+        so a device that gains internet access recovers push automatically.
+
+        Gated on the more recent of the two timestamps, not just
+        `last_mode_change_ts`: `_set_mode` only stamps that on an actual
+        transition, so a device that never successfully enters observe
+        mode would otherwise leave this throttle open forever after the
+        first `_RECOVERY_RETRY_S` window -- re-attempting (and paying the
+        subscribe-burst lock) on every single poll cycle instead of every
+        `_RECOVERY_RETRY_S`.
+        """
+        last_attempt = max(self._observe.last_mode_change_ts, self._last_observe_attempt_ts)
+        if time.monotonic() - last_attempt < _RECOVERY_RETRY_S:
             return
         await self._attempt_observe_mode()
 
@@ -805,7 +846,6 @@ 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)
@@ -862,7 +902,7 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
                             "poll and resubscribing on the new session"
                         )
                         self._observe.downgrade_to_poll()
-                        just_downgraded_from_observe = True
+                        self._resubscribe_due = True
 
         if not self._discovered:
             # One-time (issue #177): find sibling subdevices before the
@@ -895,7 +935,8 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
         for href, rep in resources.items():
             self._observe.apply(href, rep, source=source)
 
-        if first_cycle or just_downgraded_from_observe:
+        if first_cycle or self._resubscribe_due:
+            self._resubscribe_due = False
             await self._attempt_observe_mode()
         elif self._observe.mode == MODE_POLL:
             await self._maybe_retry_observe_mode()
@@ -1056,6 +1097,14 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
                 self._observe.mark_write_pending(
                     write_href, settle_s=self._POST_TIMEOUT_S + self._POLL_TIMEOUT_S
                 )
+                # A reconnect hands back a session with zero OBSERVE
+                # registrations, same as the poll path's own reconnect --
+                # keeping observe mode here would leave those subscriptions
+                # claimed on the closed session with no poll failure left
+                # to notice (issue #294).
+                if self._observe.mode == MODE_OBSERVE:
+                    self._observe.downgrade_to_poll()
+                    self._resubscribe_due = True
         await self.async_request_refresh()
 
     # ------------------------------------------------------------------

custom_components/localthings/observe.py

@@ -171,17 +171,16 @@ class ObserveManager:
             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,
-        success_fraction: float = SUCCESS_FRACTION,
-    ) -> bool:
-        """Blocking — subscribes to every href then waits up to
-        `grace_period_s`, returning early once `success_fraction` of hrefs
-        have notified. Caller must run this in an executor, never on the
-        event loop."""
+    def subscribe_hrefs(self, session, hrefs: list[str]) -> set[str]:
+        """Register OBSERVE on every href; returns the ones that took.
+        Blocking — run in an executor.
+
+        Split from the grace wait below (issue #294) so the coordinator can
+        hold its session lock for just these sends -- each is a fire-and-
+        forget UDP datagram (DtlsCoapSession.subscribe doesn't wait for the
+        device's ack), unlike the wait, which can block for the whole grace
+        period and must not hold a lock a command write is also waiting on.
+        """
         with self._notify_cond:
             self._notified.clear()
         subscribed: set[str] = set()
@@ -192,30 +191,65 @@ class ObserveManager:
                 subscribed.add(href)
             except Exception as e:
                 self.log.warning("subscribe %s failed: %s", href, e)
+        return subscribed
+
+    def await_observe_notifies(
+        self,
+        subscribed: set[str],
+        grace_period_s: float = GRACE_PERIOD_S,
+        success_fraction: float = SUCCESS_FRACTION,
+    ) -> bool:
+        """Blocking — waits up to `grace_period_s`, returning early once
+        `success_fraction` of `subscribed` have notified. Touches no
+        session; safe to run without holding a session lock."""
         if not subscribed:
-            self._stop_refresh_task()
-            self._set_mode(MODE_POLL)
-            self.subscribed_hrefs = set()
             return False
 
         def _fraction_reached() -> bool:
             return len(set(self._notified) & subscribed) / len(subscribed) >= success_fraction
 
         with self._notify_cond:
-            reached = self._notify_cond.wait_for(
-                _fraction_reached,
-                timeout=grace_period_s,
-            )
+            return self._notify_cond.wait_for(_fraction_reached, timeout=grace_period_s)
 
-        if reached:
-            self.subscribed_hrefs = subscribed
-            self._set_mode(MODE_OBSERVE)
-            self.start_refresh_task(session)
-            return True
+    def enter_observe_mode(self, session, subscribed: set[str]) -> None:
+        """Commit a successful attempt. Caller must have re-confirmed
+        `session` is still the live one under its session lock (issue
+        #294) -- committing against a session a reconnect already replaced
+        would claim observe mode with nothing left to notice it's dead."""
+        self.subscribed_hrefs = set(subscribed)
+        self._set_mode(MODE_OBSERVE)
+        self.start_refresh_task(session)
 
+    def abandon_observe_attempt(self) -> None:
+        """Drop a failed or stale attempt: no subscriptions worth keeping."""
         self._stop_refresh_task()
         self.subscribed_hrefs = set()
         self._set_mode(MODE_POLL)
+
+    def try_enter_observe_mode(
+        self,
+        session,
+        hrefs: list[str],
+        grace_period_s: float = GRACE_PERIOD_S,
+        success_fraction: float = SUCCESS_FRACTION,
+    ) -> bool:
+        """Blocking — subscribes to every href then waits up to
+        `grace_period_s`, returning early once `success_fraction` of hrefs
+        have notified. Caller must run this in an executor, never on the
+        event loop.
+
+        Single-threaded convenience wrapper around the phase split above
+        (subscribe_hrefs / await_observe_notifies / enter_observe_mode /
+        abandon_observe_attempt) for callers -- direct and most existing
+        tests -- that don't need the lock-scoping those phases exist for."""
+        subscribed = self.subscribe_hrefs(session, hrefs)
+        if not subscribed:
+            self.abandon_observe_attempt()
+            return False
+        if self.await_observe_notifies(subscribed, grace_period_s, success_fraction):
+            self.enter_observe_mode(session, subscribed)
+            return True
+        self.abandon_observe_attempt()
         return False
 
     def _set_mode(self, mode: str) -> None:

tests/localthings/test_coordinator.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import time
 from datetime import timedelta
 from unittest.mock import AsyncMock, patch
 
@@ -21,6 +22,7 @@ from custom_components.localthings.const import (
     SUMMARY_INTERVAL_S,
 )
 from custom_components.localthings.coordinator import (
+    _RECOVERY_RETRY_S,
     LocalThingsCoordinator,
     _local_source_port,
 )
@@ -30,7 +32,7 @@ from custom_components.localthings.registry.capabilities.common import (
     remote_control_required_for_write,
 )
 
-from .conftest import ENTRY_DATA, MOCK_MODEL, MOCK_SERIAL
+from .conftest import ENTRY_DATA, MOCK_MODEL, MOCK_SERIAL, FakeObserveSession
 from .conftest import _load_fridge_resources as _load_fridge
 
 
@@ -742,6 +744,95 @@ async def test_reconnect_from_observe_mode_resubscribes_immediately(
     assert coordinator.observe_mode == MODE_OBSERVE
 
 
+async def test_attempt_observe_mode_discards_stale_commit_after_session_swap(
+    hass: HomeAssistant, mock_entry, mock_coordinator_observe_session
+) -> None:
+    """A reconnect (the poll path's own, or a command's retry) can swap
+    self._session while this attempt's grace wait is in flight -- it runs
+    without holding _session_lock precisely so a write isn't blocked behind
+    it (issue #294). Committing observe mode against the now-stale local
+    `sess` reference would claim "Push" on a session that's already gone,
+    with nothing left to notice -- the identity re-check under the lock
+    right before committing must catch this and abandon instead.
+
+    Simulates the swap from inside await_observe_notifies itself rather
+    than via real concurrency: subscribe_hrefs (and its lock) has already
+    returned by the time that call runs, so this lands exactly in the
+    window the identity check exists to cover, deterministically."""
+    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]
+
+    other = FakeObserveSession()
+
+    def _swap_session_mid_wait(subscribed, grace_period_s, success_fraction=None):
+        coordinator._session = other
+        return True
+
+    with patch.object(
+        coordinator._observe, "await_observe_notifies", side_effect=_swap_session_mid_wait
+    ):
+        await coordinator._attempt_observe_mode()
+
+    assert coordinator.observe_mode == MODE_POLL
+    assert coordinator._observe.subscribed_hrefs == set()
+    assert coordinator._observe._refresh_thread is None
+
+
+async def test_maybe_retry_observe_mode_uses_most_recent_attempt_not_just_mode_change(
+    hass: HomeAssistant, mock_entry, mock_coordinator_observe_session
+) -> None:
+    """_set_mode only stamps last_mode_change_ts on an actual transition,
+    so a device that never successfully enters observe mode leaves that
+    timestamp stuck at construction time forever -- a failed attempt keeps
+    calling _set_mode(MODE_POLL) while already in MODE_POLL, a no-op.
+    Gating solely on that timestamp would make the 600s throttle open once
+    and then never close again, re-attempting (and paying the subscribe
+    burst) on every single poll cycle instead of every _RECOVERY_RETRY_S."""
+    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]
+    assert coordinator.observe_mode == MODE_POLL  # never notified during setup
+
+    # Simulate exactly the scenario above: mode_change_ts is old (as it
+    # would be forever, for a device that never gets push), but an attempt
+    # really did just run.
+    coordinator._observe.last_mode_change_ts = time.monotonic() - _RECOVERY_RETRY_S - 1
+    coordinator._last_observe_attempt_ts = time.monotonic()
+
+    with patch.object(fake, "subscribe") as mock_subscribe:
+        await coordinator._maybe_retry_observe_mode()
+
+    mock_subscribe.assert_not_called()
+
+
+async def test_attempt_observe_mode_releases_lock_before_the_grace_wait(
+    hass: HomeAssistant, mock_entry, mock_coordinator_observe_session
+) -> None:
+    """The subscribe burst holds _session_lock (it touches the session);
+    the grace wait after it must not, or a command write could stall
+    behind up to _OBSERVE_GRACE_PERIOD_S of an unrelated observe-mode-entry
+    attempt (issue #294). By construction, subscribe_hrefs's own
+    `async with self._session_lock:` has already exited by the time
+    await_observe_notifies is even called -- checked here rather than
+    inferred from timing."""
+    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]
+
+    locked_during_wait = {"value": None}
+
+    def _check_lock(subscribed, grace_period_s, success_fraction=None):
+        locked_during_wait["value"] = coordinator._session_lock.locked()
+        return True
+
+    with patch.object(coordinator._observe, "await_observe_notifies", side_effect=_check_lock):
+        await coordinator._attempt_observe_mode()
+
+    assert locked_during_wait["value"] is False
+
+
 async def test_sweep_mismatch_never_downgrades_a_live_observe_session(
     hass: HomeAssistant, mock_entry, mock_coordinator_observe_session
 ) -> None:
@@ -1110,6 +1201,69 @@ async def test_send_command_raises_after_reconnect_retry_also_fails(
         await coordinator.async_send_command(bound, 5)
 
 
+async def test_send_command_reconnect_downgrades_observe_mode(
+    hass: HomeAssistant,
+    mock_entry,
+    mock_coordinator_observe_session,
+) -> None:
+    """A command's own successful reconnect hands back a session with zero
+    OBSERVE registrations too, same as the poll path's reconnect -- must
+    downgrade the same way and flag a resubscribe, or observe mode stays
+    claimed against a session the write just replaced underneath it
+    (issue #294)."""
+    from custom_components.localthings.registry.discovery import BoundEntity
+    from custom_components.localthings.registry.entities import NumberDesc
+
+    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
+
+    fake.notify_on_subscribe = {"notified": True}
+    entered = await hass.async_add_executor_job(
+        coordinator._observe.try_enter_observe_mode,
+        fake,
+        hrefs,
+        0.02,
+        0.8,
+    )
+    assert entered is True
+    assert coordinator.observe_mode == MODE_OBSERVE
+
+    def _write_fn(payload, rep, href=None):
+        return (["test", "vs", "0"], {"value": payload})
+
+    desc = NumberDesc(key="test", field="value", write_fn=_write_fn)
+    bound = BoundEntity(href="/test/vs/0", capability=coordinator.bound[0].capability, desc=desc)
+
+    calls = {"n": 0}
+
+    def _post(*args, **kwargs):
+        calls["n"] += 1
+        if calls["n"] == 1:
+            raise ConnectionError("socket closed")
+        return (0x44, b"")
+
+    with (
+        patch.object(fake, "subscribe") as mock_subscribe,
+        patch(
+            "custom_components.localthings.coordinator.asyncio.sleep",
+            new=AsyncMock(),
+        ),
+    ):
+        fake.post = _post
+        await coordinator.async_send_command(bound, 5)
+
+        # _resubscribe_due is consumed by this same call's own trailing
+        # refresh (async_request_refresh is awaited, not fire-and-forget),
+        # so the visible effect is a resubscribe attempt, not a lingering
+        # flag value to assert on afterward.
+        assert mock_subscribe.called
+
+    assert coordinator.observe_mode == MODE_POLL
+
+
 async def test_second_write_to_same_href_lands_during_first_writes_settle_window(
     hass: HomeAssistant,
     mock_entry,