fix(polling): require consecutive misses before declaring a window dead

src/ccgram/handlers/polling/polling_state.py

@@ -29,6 +29,7 @@
 from ...providers.base import StatusUpdate
 from ...topic_state_registry import topic_state
 from .polling_types import (
+    MAX_MISSING_POLLS,
     MAX_PROBE_FAILURES,
     PANE_COUNT_TTL,
     RC_DEBOUNCE_SECONDS,
@@ -325,6 +326,23 @@ def reset_probe_failures(self, window_id: str) -> None:
         if ws:
             ws.probe_failures = 0
 
+    def record_missing_poll(self, window_id: str) -> int:
+        """Increment the consecutive-missing counter and return the new count."""
+        ws = self.get_state(window_id)
+        ws.missing_polls += 1
+        return ws.missing_polls
+
+    def clear_missing_polls(self, window_id: str) -> None:
+        """Reset the consecutive-missing counter for a single window."""
+        ws = self._states.get(window_id)
+        if ws:
+            ws.missing_polls = 0
+
+    def is_confirmed_missing(self, window_id: str) -> bool:
+        """Check whether a window has been absent for enough consecutive polls."""
+        ws = self._states.get(window_id)
+        return bool(ws) and ws.missing_polls >= MAX_MISSING_POLLS
+
     def clear_seen_status(self, window_id: str) -> None:
         """Clear startup status tracking for a single window."""
         ws = self._states.get(window_id)

src/ccgram/handlers/polling/polling_types.py

@@ -38,6 +38,12 @@
 # Consecutive topic probe failure threshold.
 MAX_PROBE_FAILURES = 3
 
+# Consecutive poll cycles a bound window must be absent from the multiplexer
+# listing before it is treated as dead. TmuxManager.list_windows() returns an
+# empty list when get_session() hits a transient error, and drops individual
+# windows whose pane metadata raises, so a single miss is not proof of death.
+MAX_MISSING_POLLS = 3
+
 # Typing indicator throttle interval (seconds).
 TYPING_INTERVAL = 4.0
 
@@ -64,6 +70,7 @@ class WindowPollState:
     has_seen_status: bool = False
     startup_time: float | None = None
     probe_failures: int = 0
+    missing_polls: int = 0
     screen_buffer: ScreenBuffer | None = field(default=None, repr=False)
     pane_count_cache: tuple[int, float] | None = None
     unbound_timer: float | None = None

src/ccgram/handlers/polling/window_tick/__init__.py

@@ -75,11 +75,22 @@ async def tick_window(
         return
 
     if window is None:
+        # A single absent poll is not proof of death: list_windows() returns []
+        # when the multiplexer session lookup hits a transient error, and drops
+        # individual windows whose pane metadata raises. Require the window to
+        # be missing from MAX_MISSING_POLLS consecutive cycles before showing
+        # the recovery banner, otherwise one bad poll kills a live session's
+        # topic. Genuine deaths still arrive instantly via the event stream.
+        rt.poll_state.record_missing_poll(window_id)
+        if not rt.poll_state.is_confirmed_missing(window_id):
+            return
         await _handle_dead_window_notification(
             bot, user_id, thread_id, window_id, runtime=rt
         )
         return
 
+    rt.poll_state.clear_missing_polls(window_id)
+
     await discover_and_register_transcript(
         window_id,
         _window=window,

tests/ccgram/handlers/polling/test_window_tick.py

@@ -12,7 +12,7 @@
     terminal_poll_state,
     terminal_screen_buffer,
 )
-from ccgram.handlers.polling.polling_types import TickContext
+from ccgram.handlers.polling.polling_types import MAX_MISSING_POLLS, TickContext
 from ccgram.handlers.polling.window_tick import (
     _check_interactive_only,
     _handle_dead_window_notification,
@@ -65,7 +65,8 @@ async def test_dead_window_calls_handle_dead(self):
         with patch.object(
             window_tick, "_handle_dead_window_notification", new_callable=AsyncMock
         ) as mock_dead:
-            await tick_window(bot, 1, 100, "@0", None)
+            for _ in range(MAX_MISSING_POLLS):
+                await tick_window(bot, 1, 100, "@0", None)
             mock_dead.assert_called_once()
             args, kwargs = mock_dead.call_args
             assert args == (bot, 1, 100, "@0")
@@ -83,7 +84,8 @@ async def test_dead_window_skips_other_work(self):
                 window_tick, "_scan_window_panes", new_callable=AsyncMock
             ) as mock_scan,
         ):
-            await tick_window(bot, 1, 100, "@0", None)
+            for _ in range(MAX_MISSING_POLLS):
+                await tick_window(bot, 1, 100, "@0", None)
             mock_status.assert_not_called()
             mock_scan.assert_not_called()
 
@@ -97,6 +99,69 @@ async def test_already_dead_notified_returns_early(self):
             mock_dead.assert_not_called()
 
 
+class TestTickWindowMissingPollDebounce:
+    """A window absent from a single poll must not be declared dead.
+
+    list_windows() returns [] when the multiplexer session lookup hits a
+    transient error, and drops individual windows whose pane metadata raises,
+    so one miss is not proof of death.
+    """
+
+    async def test_single_miss_does_not_notify(self):
+        bot = AsyncMock(spec=Bot)
+        with patch.object(
+            window_tick, "_handle_dead_window_notification", new_callable=AsyncMock
+        ) as mock_dead:
+            await tick_window(bot, 1, 100, "@0", None)
+            mock_dead.assert_not_called()
+
+    async def test_misses_below_threshold_do_not_notify(self):
+        bot = AsyncMock(spec=Bot)
+        with patch.object(
+            window_tick, "_handle_dead_window_notification", new_callable=AsyncMock
+        ) as mock_dead:
+            for _ in range(MAX_MISSING_POLLS - 1):
+                await tick_window(bot, 1, 100, "@0", None)
+            mock_dead.assert_not_called()
+
+    async def test_reappearing_window_resets_the_counter(self):
+        bot = AsyncMock(spec=Bot)
+        window = _make_window("@0")
+        with (
+            patch.object(
+                window_tick, "_handle_dead_window_notification", new_callable=AsyncMock
+            ) as mock_dead,
+            patch.object(
+                window_tick, "discover_and_register_transcript", new_callable=AsyncMock
+            ),
+            patch.object(window_tick, "_update_status", new_callable=AsyncMock),
+            patch.object(window_tick, "_scan_window_panes", new_callable=AsyncMock),
+            patch.object(
+                window_tick, "_maybe_check_passive_shell", new_callable=AsyncMock
+            ),
+            patch.object(window_tick, "get_message_queue", return_value=None),
+        ):
+            for _ in range(MAX_MISSING_POLLS - 1):
+                await tick_window(bot, 1, 100, "@0", None)
+            await tick_window(bot, 1, 100, "@0", window)
+            for _ in range(MAX_MISSING_POLLS - 1):
+                await tick_window(bot, 1, 100, "@0", None)
+            mock_dead.assert_not_called()
+
+    async def test_threshold_reached_notifies_once(self):
+        bot = AsyncMock(spec=Bot)
+        with patch.object(
+            window_tick, "_handle_dead_window_notification", new_callable=AsyncMock
+        ) as mock_dead:
+            for _ in range(MAX_MISSING_POLLS + 2):
+                await tick_window(bot, 1, 100, "@0", None)
+            # The real handler marks dead-notified; it is mocked here, so the
+            # early-return guard never engages and every post-threshold tick
+            # calls through. What matters is that no call happened before the
+            # threshold was crossed.
+            assert mock_dead.call_count == 3
+
+
 class TestTickWindowPendingQueue:
     async def test_pending_queue_skips_status_update(self):
         bot = AsyncMock(spec=Bot)