fix(gateway): make adapter fatal-error handoff cancellation-proof; exit if a platform is stranded

The fatal-error notification runs on the failing adapter's own polling
task, and adapter.disconnect() inside the handler can cancel that task
(its current-task guard misses because _safe_adapter_disconnect runs the
close in a wrapper task). The CancelledError killed the handler between
the fatal log and the reconnect queue, leaving the platform permanently
dead inside a live gateway process. #68447 fixed this for telegram at
the adapter layer; this hardens the shared gateway dispatch so every
platform gets the same protection (qqbot #25505/#29005, photon #68693).

- _handle_adapter_fatal_error now runs the real handler in a detached
  task, awaited through asyncio.shield() so caller cancellation cannot
  tunnel into it (Task.cancel() also cancels the task's _fut_waiter).
- If a retryable platform still ends up neither reconnected nor queued,
  the gateway exits with failure so launchd/systemd KeepAlive restarts
  it instead of running indefinitely with a dead platform (#68693).

Fixes #68693

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

gateway/run.py

@@ -3328,6 +3328,10 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
         # Key: Platform enum, Value: {"config": platform_config, "attempts": int, "next_retry": float}
         self._failed_platforms: Dict[Platform, Dict[str, Any]] = {}
 
+        # Strong refs to detached fatal-error handler tasks (see
+        # _handle_adapter_fatal_error) so the event loop can't GC them mid-run.
+        self._fatal_handler_tasks: set = set()
+
         # Track pending /update prompt responses per session.
         # Key: session_key, Value: True when a prompt is waiting for user input.
         self._update_prompt_pending: Dict[str, bool] = {}
@@ -4378,7 +4382,65 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
 
         If the error is retryable (e.g. network blip, DNS failure), queue the
         platform for background reconnection instead of giving up permanently.
+
+        The notification arrives on the failing adapter's own polling task,
+        and the disconnect inside the handler can cancel that task mid-flight:
+        disconnect()'s current-task guard misses it because
+        _safe_adapter_disconnect runs the close in a wrapper task. A cancelled
+        handler dies between the fatal log and the reconnect queue, silently
+        stranding the platform (observed 2026-07-21: telegram popped from
+        adapters but never queued after a travel network outage). Run the real
+        work in a detached task that adapter teardown cannot cancel.
         """
+        tasks = getattr(self, "_fatal_handler_tasks", None)
+        if tasks is None:
+            tasks = self._fatal_handler_tasks = set()
+        task = asyncio.create_task(self._handle_adapter_fatal_error_detached(adapter))
+        tasks.add(task)
+        task.add_done_callback(tasks.discard)
+        # Await so callers that expect completion still get it — but through
+        # shield(): Task.cancel() on the caller also cancels the future it is
+        # awaiting (_fut_waiter), so a plain `await task` would tunnel the
+        # cancellation straight into the "detached" task. shield() absorbs
+        # it: the caller sees CancelledError, the handler runs to completion.
+        await asyncio.shield(task)
+
+    async def _handle_adapter_fatal_error_detached(
+        self, adapter: BasePlatformAdapter
+    ) -> None:
+        """Run the fatal handler; if the platform still ends up stranded
+        (not reconnected, not queued, not intentionally disabled), exit the
+        gateway with failure so the service manager restarts it instead of
+        leaving a silent partial outage."""
+        try:
+            await self._handle_adapter_fatal_error_impl(adapter)
+        except Exception:
+            logger.exception(
+                "Fatal-error handling for %s raised unexpectedly",
+                adapter.platform.value,
+            )
+        finally:
+            platform = adapter.platform
+            shutdown_event = getattr(self, "_shutdown_event", None)
+            stranded = (
+                adapter.fatal_error_retryable
+                and platform not in self.adapters
+                and platform not in getattr(self, "_failed_platforms", {})
+                and not (shutdown_event is not None and shutdown_event.is_set())
+            )
+            if stranded:
+                logger.error(
+                    "%s adapter was lost without entering the reconnection "
+                    "queue; exiting gateway so the service manager restarts it.",
+                    platform.value,
+                )
+                self._exit_reason = (
+                    f"{platform.value} adapter lost without reconnection queue"
+                )
+                self._exit_with_failure = True
+                await self.stop()
+
+    async def _handle_adapter_fatal_error_impl(self, adapter: BasePlatformAdapter) -> None:
         # Snapshot the current owner of this platform slot before doing
         # anything else. If it's neither this adapter nor empty, a different
         # adapter has already taken over (e.g. this is a delayed notification

tests/gateway/test_platform_reconnect.py

@@ -949,3 +949,71 @@ class TestSpawnSupervised:
         # _MAX_SUPERVISED_RESTARTS + 1; the reset lets it run to completion.
         assert calls["n"] == target
         assert calls["n"] > runner._MAX_SUPERVISED_RESTARTS + 1
+
+
+class TestFatalHandoffCancellationProof:
+    """The fatal-error handoff must survive cancellation of the notifying
+    task, and a retryable platform must never be silently stranded."""
+
+    @pytest.mark.asyncio
+    async def test_caller_cancellation_does_not_strand_platform(self):
+        """The fatal notification arrives on the failing adapter's own
+        polling task, and adapter.disconnect() inside the handler can cancel
+        that task mid-teardown. The platform must still reach the reconnect
+        queue (previously the CancelledError killed the handler between the
+        fatal log and the queue, stranding the platform until a manual
+        restart)."""
+        runner = _make_runner()
+        runner.stop = AsyncMock()
+
+        adapter = StubAdapter(succeed=True)
+        adapter._set_fatal_error("network_error", "DNS failure", retryable=True)
+        runner.adapters[Platform.TELEGRAM] = adapter
+
+        release = asyncio.Event()
+
+        async def slow_disconnect():
+            await release.wait()
+
+        adapter.disconnect = slow_disconnect  # hold the handler mid-teardown
+
+        caller = asyncio.create_task(runner._handle_adapter_fatal_error(adapter))
+        for _ in range(5):
+            await asyncio.sleep(0)  # let the handler reach the disconnect await
+        caller.cancel()  # what disconnect() does to the notifying task
+        with pytest.raises(asyncio.CancelledError):
+            await caller
+        release.set()  # teardown completes after the caller has died
+
+        for _ in range(200):
+            if Platform.TELEGRAM in runner._failed_platforms:
+                break
+            await asyncio.sleep(0.01)
+        assert Platform.TELEGRAM in runner._failed_platforms
+
+    @pytest.mark.asyncio
+    async def test_stranded_retryable_platform_exits_for_supervisor_restart(self):
+        """If a retryable platform ends up neither reconnected nor queued
+        (e.g. its config entry is gone so queueing is skipped), the gateway
+        must exit with failure so launchd/systemd KeepAlive restarts it,
+        instead of running indefinitely with a dead platform while healthy
+        peers mask the loss (#68693)."""
+        runner = _make_runner()
+
+        async def _stop():
+            runner._shutdown_event.set()
+
+        runner.stop = AsyncMock(side_effect=_stop)
+        runner.config = GatewayConfig(platforms={})  # queueing impossible
+
+        adapter = StubAdapter(succeed=True)
+        adapter._set_fatal_error("network_error", "DNS failure", retryable=True)
+        runner.adapters[Platform.TELEGRAM] = adapter
+        # A healthy peer keeps self.adapters non-empty, so the existing
+        # "no platforms remain" shutdown branches do not fire.
+        runner.adapters[Platform.FEISHU] = StubAdapter(platform=Platform.FEISHU)
+
+        await runner._handle_adapter_fatal_error(adapter)
+
+        assert runner._exit_with_failure is True
+        assert runner.stop.await_count == 1