fix(coordinator): two gaps a follow-up Opus review found in the split

A second review of the observe-mode phase split (previous commit) found
two real regressions it introduced, both in the same failure family it
was built to close:

- async_send_command's failed-retry branch closed the session, then
  raised without downgrading observe mode -- the downgrade only ran on
  the retry's success path. A retry that also fails still leaves the
  session dead, so mode was left claiming "Push" on a session that no
  longer exists, same as the bug this whole fix targets. Moved the
  downgrade to run right after the close, unconditionally on how the
  retry goes.

- _attempt_observe_mode's stale-session abandon (the identity-check
  branch added in the previous commit) didn't flag a resubscribe. A
  session swap discovered there means a fresh, never-tried session now
  exists, but _last_observe_attempt_ts was already stamped for the
  now-abandoned attempt -- so that new session sat unsubscribed for up
  to _RECOVERY_RETRY_S (600s) instead of being retried on the next
  cycle. Now sets _resubscribe_due, same as the two reconnect paths do.

Also closes two test-coverage gaps the same review surfaced by mutation
testing: no test asserted the lock actually holds during the subscribe
burst (only that it's released for the wait), and no test distinguished
the max() in _maybe_retry_observe_mode's throttle from using
_last_observe_attempt_ts alone -- both mutations left the full suite
green. Added one test for each, plus extended two existing tests for the
bug fixes above; all four confirmed via mutation testing (revert the
fix, watch the new/extended test fail; restore it, watch it pass).

One finding from the same review is intentionally left open: async_close
is the one self._session writer that doesn't take _session_lock, so a
close racing _attempt_observe_mode isn't covered by today's identity
check. This is pre-existing (the lock didn't cover any of
_attempt_observe_mode before this branch's earlier commits either), not
a regression from this branch's work, and is a shutdown/unload-path
question rather than the write-vs-observe-mode race this branch set out
to fix.

custom_components/localthings/coordinator.py

@@ -776,8 +776,18 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
             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:
+            stale_session = self._session is not sess
+            if not reached or stale_session:
                 self._observe.abandon_observe_attempt()
+                if stale_session:
+                    # A reconnect elsewhere replaced the session while this
+                    # attempt waited -- that session has never been tried,
+                    # so retry it next cycle instead of leaving it
+                    # unsubscribed for up to _RECOVERY_RETRY_S, which
+                    # _last_observe_attempt_ts (already stamped above, for
+                    # the now-abandoned session) would otherwise throttle
+                    # for (issue #294).
+                    self._resubscribe_due = True
                 return
             self._observe.enter_observe_mode(sess, subscribed)
 
@@ -1078,6 +1088,15 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
             except Exception as e:
                 self._log.warning("command failed for %s, reconnecting: %s", write_href, e)
                 await self.hass.async_add_executor_job(self._close_session)
+                # The session is dead the moment it's closed, so any OBSERVE
+                # subscriptions on it are too -- downgrade here, before the
+                # retry, so a retry that also fails doesn't leave mode
+                # claiming "Push" on a session that no longer exists
+                # (issue #294; the poll path handles the same fact for its
+                # own reconnect the same way, unconditionally on close).
+                if self._observe.mode == MODE_OBSERVE:
+                    self._observe.downgrade_to_poll()
+                    self._resubscribe_due = True
                 await asyncio.sleep(self._RECONNECT_PAUSE_S)
                 try:
                     await self.hass.async_add_executor_job(_do_put)
@@ -1097,14 +1116,6 @@ 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()
 
     # ------------------------------------------------------------------

tests/localthings/test_coordinator.py

@@ -755,6 +755,11 @@ async def test_attempt_observe_mode_discards_stale_commit_after_session_swap(
     with nothing left to notice -- the identity re-check under the lock
     right before committing must catch this and abandon instead.
 
+    The new session is never-tried, though, not just abandoned: it must
+    flag an immediate resubscribe rather than let _last_observe_attempt_ts
+    (stamped for the now-abandoned attempt) throttle it for up to
+    _RECOVERY_RETRY_S.
+
     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
@@ -762,6 +767,7 @@ async def test_attempt_observe_mode_discards_stale_commit_after_session_swap(
     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._resubscribe_due is False
 
     other = FakeObserveSession()
 
@@ -777,6 +783,7 @@ async def test_attempt_observe_mode_discards_stale_commit_after_session_swap(
     assert coordinator.observe_mode == MODE_POLL
     assert coordinator._observe.subscribed_hrefs == set()
     assert coordinator._observe._refresh_thread is None
+    assert coordinator._resubscribe_due is True
 
 
 async def test_maybe_retry_observe_mode_uses_most_recent_attempt_not_just_mode_change(
@@ -807,6 +814,30 @@ async def test_maybe_retry_observe_mode_uses_most_recent_attempt_not_just_mode_c
     mock_subscribe.assert_not_called()
 
 
+async def test_maybe_retry_observe_mode_also_respects_a_mode_change_outside_an_attempt(
+    hass: HomeAssistant, mock_entry, mock_coordinator_observe_session
+) -> None:
+    """The mirror of the case above: last_mode_change_ts can be the more
+    recent of the two as well, e.g. right after the poll or command path's
+    own downgrade (neither goes through _attempt_observe_mode, so neither
+    stamps _last_observe_attempt_ts). Dropping last_mode_change_ts from the
+    max() would let a device that was *just* downgraded get re-attempted
+    immediately instead of respecting _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
+
+    coordinator._last_observe_attempt_ts = time.monotonic() - _RECOVERY_RETRY_S - 1
+    coordinator._observe.last_mode_change_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:
@@ -833,6 +864,29 @@ async def test_attempt_observe_mode_releases_lock_before_the_grace_wait(
     assert locked_during_wait["value"] is False
 
 
+async def test_attempt_observe_mode_holds_lock_during_the_subscribe_burst(
+    hass: HomeAssistant, mock_entry, mock_coordinator_observe_session
+) -> None:
+    """The other half of the split above: the subscribe burst does touch
+    the session, so it must hold _session_lock -- that's what actually
+    stops a concurrent close from landing mid-subscribe (issue #294)."""
+    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_subscribe = {"value": None}
+    real_subscribe_hrefs = coordinator._observe.subscribe_hrefs
+
+    def _check_lock(session, hrefs):
+        locked_during_subscribe["value"] = coordinator._session_lock.locked()
+        return real_subscribe_hrefs(session, hrefs)
+
+    with patch.object(coordinator._observe, "subscribe_hrefs", side_effect=_check_lock):
+        await coordinator._attempt_observe_mode()
+
+    assert locked_during_subscribe["value"] is True
+
+
 async def test_sweep_mismatch_never_downgrades_a_live_observe_session(
     hass: HomeAssistant, mock_entry, mock_coordinator_observe_session
 ) -> None:
@@ -1170,7 +1224,13 @@ async def test_send_command_raises_after_reconnect_retry_also_fails(
 ) -> None:
     """If the command still fails on the reconnected session, the user must
     see it -- previously this was swallowed into a log line with no
-    feedback at all (issue #294)."""
+    feedback at all (issue #294).
+
+    Also covers a sibling bug the fix for that same issue introduced: the
+    session is closed the moment the first attempt fails, so any OBSERVE
+    subscriptions on it are already dead regardless of whether the retry
+    that follows succeeds -- a failed retry must still downgrade mode, or
+    it's left claiming "Push" on a session that no longer exists."""
     from homeassistant.exceptions import HomeAssistantError
 
     from custom_components.localthings.registry.discovery import BoundEntity
@@ -1180,6 +1240,18 @@ async def test_send_command_raises_after_reconnect_retry_also_fails(
     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})
@@ -1200,6 +1272,8 @@ async def test_send_command_raises_after_reconnect_retry_also_fails(
         fake.post = _post
         await coordinator.async_send_command(bound, 5)
 
+    assert coordinator.observe_mode == MODE_POLL
+
 
 async def test_send_command_reconnect_downgrades_observe_mode(
     hass: HomeAssistant,