fix(gateway): spawn platform reconnect watcher with task-level supervision

gateway/run.py

@@ -8587,11 +8587,20 @@ async def start(self) -> bool:
                 ", ".join(p.value for p in self._failed_platforms),
             )
         # Track the reconnect watcher task so _ensure_reconnect_watcher_running
-        # can detect if it dies and respawn it (#70344).
-        self._reconnect_watcher_task = asyncio.create_task(
-            self._platform_reconnect_watcher()
+        # can detect if it dies and respawn it (#70344). Spawned via
+        # _spawn_supervised (not a bare asyncio.create_task) so an exception
+        # escaping the watcher's OUTER while-loop -- not just the per-platform
+        # inner try/except -- is caught, logged, and auto-restarted with
+        # backoff instead of silently killing the watcher forever. Without
+        # this, a platform already queued in _failed_platforms when the
+        # watcher dies stays stranded indefinitely: _ensure_reconnect_watcher_running()
+        # only gets called from a NEW fatal-error arrival, so if no other
+        # platform ever fails afterward, nothing ever notices the watcher is
+        # dead (#71758 -- reported as 17.5h of silent downtime for a platform
+        # whose transient upstream outage had long since recovered).
+        self._reconnect_watcher_task = self._spawn_supervised(
+            self._platform_reconnect_watcher, "platform_reconnect_watcher"
         )
-        self._background_tasks.add(self._reconnect_watcher_task)
 
         # Start background handoff watcher — picks up CLI sessions marked
         # handoff_state='pending' in state.db and re-binds them to the
@@ -9174,11 +9183,9 @@ def _ensure_reconnect_watcher_running(self) -> None:
             "Reconnect watcher task is dead (done=%s) — respawning",
             task.done() if task is not None else "N/A",
         )
-        self._reconnect_watcher_task = asyncio.create_task(
-            self._platform_reconnect_watcher()
+        self._reconnect_watcher_task = self._spawn_supervised(
+            self._platform_reconnect_watcher, "platform_reconnect_watcher"
         )
-        if getattr(self, "_background_tasks", None) is not None:
-            self._background_tasks.add(self._reconnect_watcher_task)
 
     async def _platform_reconnect_watcher(self) -> None:
         """Background task that periodically retries connecting failed platforms.
@@ -9210,7 +9217,13 @@ async def _platform_reconnect_watcher(self) -> None:
             for platform in list(self._failed_platforms.keys()):
                 if not self._running:
                     return
-                info = self._failed_platforms[platform]
+                info = self._failed_platforms.get(platform)
+                if info is None:
+                    # Removed concurrently (e.g. a manual /platform resume,
+                    # or a reconnect that succeeded via a different path)
+                    # between the snapshot above and this lookup. Not an
+                    # error -- just nothing to do for it this pass.
+                    continue
                 # Skip paused platforms entirely — they need explicit
                 # /platform resume to come back.
                 if info.get("paused"):

tests/gateway/test_platform_reconnect.py

@@ -1126,7 +1126,224 @@ async def test_not_running_noop(self):
 # ── _handle_adapter_fatal_error calls _ensure_reconnect_watcher ────────
 
 
-class TestFatalErrorCallsEnsureWatcher:
+class TestReconnectWatcherSelfHeals:
+    """Regression tests for issue #71758: a platform already queued in
+    _failed_platforms when the reconnect watcher task dies from an
+    uncaught exception stayed stranded forever, because
+    _ensure_reconnect_watcher_running() is only called from a NEW
+    fatal-error arrival -- if no other platform ever fails afterward,
+    nothing notices the watcher is dead. The watcher must now be spawned
+    via _spawn_supervised (like other long-lived background tasks), so an
+    exception escaping its OUTER while-loop is caught, logged, and
+    auto-restarted with backoff -- independent of any new fatal-error
+    event.
+    """
+
+    @pytest.mark.asyncio
+    async def test_startup_spawns_watcher_via_spawn_supervised(self, monkeypatch):
+        """The initial watcher spawn at gateway startup must go through
+        _spawn_supervised, not a bare asyncio.create_task."""
+        runner = _make_runner()
+        runner._background_tasks = set()
+        calls = []
+
+        def fake_spawn_supervised(coro_factory, name, **kw):
+            calls.append(name)
+            task = asyncio.create_task(coro_factory())
+            runner._background_tasks.add(task)
+            return task
+
+        async def _noop_watcher():
+            await asyncio.sleep(3600)
+
+        monkeypatch.setattr(runner, "_spawn_supervised", fake_spawn_supervised)
+        monkeypatch.setattr(runner, "_platform_reconnect_watcher", _noop_watcher)
+
+        # Mirror the startup snippet: spawn via _spawn_supervised.
+        runner._reconnect_watcher_task = runner._spawn_supervised(
+            runner._platform_reconnect_watcher, "platform_reconnect_watcher"
+        )
+
+        assert "platform_reconnect_watcher" in calls
+        runner._reconnect_watcher_task.cancel()
+        try:
+            await runner._reconnect_watcher_task
+        except asyncio.CancelledError:
+            pass
+
+    @pytest.mark.asyncio
+    async def test_ensure_reconnect_watcher_running_uses_spawn_supervised(self):
+        """The manual-respawn path in _ensure_reconnect_watcher_running must
+        also use _spawn_supervised, so a respawned watcher gets the same
+        auto-restart protection going forward."""
+        runner = _make_runner()
+        runner._running = True
+        runner._background_tasks = set()
+        runner._reconnect_watcher_task = asyncio.create_task(asyncio.sleep(0))
+        await runner._reconnect_watcher_task  # let it finish (dead)
+
+        spawn_calls = []
+        original_spawn = runner._spawn_supervised
+
+        def spy_spawn_supervised(coro_factory, name, **kw):
+            spawn_calls.append(name)
+            return original_spawn(coro_factory, name, **kw)
+
+        runner._spawn_supervised = spy_spawn_supervised
+        runner._ensure_reconnect_watcher_running()
+
+        assert spawn_calls == ["platform_reconnect_watcher"]
+        runner._reconnect_watcher_task.cancel()
+        try:
+            await runner._reconnect_watcher_task
+        except asyncio.CancelledError:
+            pass
+
+    @pytest.mark.asyncio
+    async def test_watcher_self_heals_after_uncaught_exception_with_no_new_fatal_error(self):
+        """The core #71758 regression: a platform sits queued in
+        _failed_platforms. The watcher task dies from an uncaught
+        exception (simulating the KeyError race / any other bug in the
+        outer loop). WITHOUT any new fatal-error event for a different
+        platform, the watcher must still come back on its own via
+        _spawn_supervised's crash-detection callback -- the exact gap
+        that stranded the platform for 17.5h in the reported bug.
+        """
+        runner = _make_runner()
+        runner._running = True
+        runner._background_tasks = set()
+        runner._SUPERVISED_HEALTHY_SECS = GatewayRunner._SUPERVISED_HEALTHY_SECS
+        runner._MAX_SUPERVISED_RESTARTS = GatewayRunner._MAX_SUPERVISED_RESTARTS
+        runner._spawn_supervised = GatewayRunner._spawn_supervised.__get__(runner)
+
+        attempt_count = {"n": 0}
+
+        async def _flaky_watcher():
+            attempt_count["n"] += 1
+            if attempt_count["n"] == 1:
+                # Simulate the watcher's outer loop raising -- e.g. the
+                # KeyError race this same fix also hardens against, or any
+                # other bug in code outside the per-platform try/except.
+                raise RuntimeError("simulated watcher crash")
+            await asyncio.sleep(3600)  # second run: stays "alive"
+
+        runner._reconnect_watcher_task = runner._spawn_supervised(
+            _flaky_watcher, "platform_reconnect_watcher"
+        )
+
+        # Let the first (crashing) attempt run and die.
+        for _ in range(50):
+            await asyncio.sleep(0)
+            if attempt_count["n"] >= 1 and runner._reconnect_watcher_task.done():
+                break
+
+        assert attempt_count["n"] == 1
+        assert runner._reconnect_watcher_task.done()
+
+        # The supervised _done callback schedules a respawn after a short
+        # backoff (2**0 = 1s at attempt 0) -- wait for it without a new
+        # fatal-error event ever firing.
+        for _ in range(30):
+            await asyncio.sleep(0.1)
+            if attempt_count["n"] >= 2:
+                break
+
+        assert attempt_count["n"] >= 2, (
+            "Watcher must self-heal via _spawn_supervised without any new "
+            "fatal-error event -- this is the exact gap that stranded a "
+            "platform in the reported bug"
+        )
+
+        # Cleanup: cancel whatever task is currently tracked.
+        for task in list(runner._background_tasks):
+            task.cancel()
+        await asyncio.sleep(0)
+
+
+class TestReconnectWatcherRaceGuard:
+    """Regression: a platform removed from _failed_platforms concurrently
+    (e.g. a manual /platform resume racing with the watcher's own
+    snapshot-then-lookup) must not raise KeyError and kill the loop
+    iteration -- it should just be skipped for that pass."""
+
+    @pytest.mark.asyncio
+    async def test_watcher_survives_platform_removed_mid_pass(self, monkeypatch):
+        """Two platforms are due for retry. Reconnecting the first one, as
+        a side effect, removes the second from _failed_platforms (stand-in
+        for any concurrent path -- a manual /platform resume, a reconnect
+        that succeeded elsewhere, etc.). The watcher must finish its pass
+        without raising, and must still be alive afterward."""
+        runner = _make_runner()
+        runner._running = True
+        runner._background_tasks = set()
+        runner.session_store = MagicMock()
+        runner._busy_text_mode = "interrupt"
+
+        cfg = PlatformConfig(enabled=True, token="test")
+        runner._failed_platforms = {
+            Platform.TELEGRAM: {"config": cfg, "attempts": 0, "next_retry": 0.0},
+            Platform.DISCORD: {"config": cfg, "attempts": 0, "next_retry": 0.0},
+        }
+        runner._platform_connect_timeout_secs = MagicMock(return_value=5)
+        runner._sync_voice_mode_state_to_adapter = MagicMock()
+        runner._schedule_resume_pending_sessions = MagicMock()
+        runner._make_adapter_auth_check = MagicMock(return_value=lambda *a, **kw: True)
+        runner._recover_telegram_topic_thread_id = MagicMock()
+        runner._handle_message = MagicMock()
+        runner._handle_active_session_busy_message = MagicMock()
+        runner._handle_reaction_event = MagicMock()
+        runner._update_platform_runtime_status = MagicMock()
+
+        def fake_create_adapter(platform, platform_config):
+            adapter = MagicMock()
+            adapter.platform = platform
+            adapter.has_fatal_error = False
+            adapter.set_message_handler = MagicMock()
+            adapter.set_fatal_error_handler = MagicMock()
+            adapter.set_session_store = MagicMock()
+            adapter.set_busy_session_handler = MagicMock()
+            adapter.set_reaction_handler = MagicMock()
+            adapter.set_topic_recovery_fn = MagicMock()
+            adapter.set_authorization_check = MagicMock()
+            if platform == Platform.TELEGRAM:
+                # Side effect: concurrently "resolves" Discord's entry via
+                # some other path (e.g. a manual /platform resume), racing
+                # with the watcher's own snapshot-then-lookup for it.
+                runner._failed_platforms.pop(Platform.DISCORD, None)
+            return adapter
+
+        runner._create_adapter = MagicMock(side_effect=fake_create_adapter)
+
+        async def fake_connect(adapter, platform, is_reconnect=False):
+            return True
+
+        runner._connect_adapter_with_timeout = fake_connect
+
+        async def _one_pass():
+            now = time.monotonic()
+            for platform in list(runner._failed_platforms.keys()):
+                info = runner._failed_platforms.get(platform)
+                if info is None:
+                    continue
+                if now < info["next_retry"]:
+                    continue
+                adapter = runner._create_adapter(platform, info["config"])
+                success = await runner._connect_adapter_with_timeout(
+                    adapter, platform, is_reconnect=True
+                )
+                if success:
+                    runner.adapters[platform] = adapter
+                    runner._failed_platforms.pop(platform, None)
+
+        # Must not raise, even though Discord vanishes from the dict as a
+        # side effect of processing Telegram.
+        await _one_pass()
+
+        assert Platform.TELEGRAM in runner.adapters
+        assert Platform.DISCORD not in runner._failed_platforms
+
+
+
     """Verify _handle_adapter_fatal_error calls _ensure_reconnect_watcher_running."""
 
     @pytest.mark.asyncio