fix(gateway): shield fatal-error handler from carrier task cancellation

When an adapter escalates a retryable fatal error from inside one of its
own tasks (e.g. Telegram's _polling_error_task after exhausting polling
network retries), the gateway's _handle_adapter_fatal_error tears the
adapter down via disconnect() — which cancels that very task. The
propagating CancelledError killed the handler between popping the
adapter from the adapter map and queueing the platform in
_failed_platforms, leaving a zombie gateway: process alive, zero
connected platforms, zero pending retries, until a manual restart.

Run the handler as a detached task under asyncio.shield so carrier
cancellation no longer aborts teardown/queueing mid-flight. The carrier
still observes CancelledError (teardown semantics unchanged); only the
handler is protected. A done-callback consumes the detached task's
exception to avoid 'Task exception was never retrieved' noise.

Fixes #81335

gateway/platforms/base.py

@@ -27,6 +27,23 @@ from utils import normalize_proxy_url
 
 logger = logging.getLogger(__name__)
 
+
+def _consume_detached_handler_exception(task: "asyncio.Task") -> None:
+    """Done-callback retrieving a detached fatal-error handler's exception.
+
+    Prevents "Task exception was never retrieved" warnings for handler tasks
+    we deliberately let finish in the background after their awaiting
+    (carrier) task was cancelled — see ``_notify_fatal_error``.
+    """
+    if task.cancelled():
+        return
+    exc = task.exception()
+    if exc is not None:
+        logger.error(
+            "Detached fatal-error handler task failed: %s", exc, exc_info=exc
+        )
+
+
 # Audio file extensions Hermes recognizes for native audio delivery.
 # Keep Telegram's narrower attachment/voice sets below separate: formats such
 # as MPEG-2 Layer II are audio to Hermes but unsupported by sendAudio/sendVoice.
@@ -3444,7 +3461,27 @@ class BasePlatformAdapter(ABC):
             return
         result = handler(self)
         if asyncio.iscoroutine(result):
-            await result
+            # Run the handler as a detached, shielded task. The notification
+            # is frequently awaited from inside an adapter-owned task (e.g.
+            # the Telegram ``_polling_error_task``), and the gateway's fatal
+            # handler tears the adapter down via ``disconnect()`` — which
+            # cancels that very task. Without the shield the cancellation
+            # killed the handler mid-flight: the adapter was already popped
+            # from the gateway's adapter map but never queued for background
+            # reconnection, leaving a zombie gateway with no platforms and no
+            # pending retries (#81335).
+            task = asyncio.ensure_future(result)
+            try:
+                await asyncio.shield(task)
+            except asyncio.CancelledError:
+                # The carrier task was cancelled (typically by our own
+                # teardown running inside the handler). Let the handler
+                # finish detached so reconnect queueing / the shutdown
+                # decision completes, and consume its eventual exception to
+                # avoid "Task exception was never retrieved" noise.
+                if not task.done():
+                    task.add_done_callback(_consume_detached_handler_exception)
+                raise
 
     def _acquire_platform_lock(self, scope: str, identity: str, resource_desc: str) -> bool:
         """Acquire a scoped lock for this adapter. Returns True on success.

tests/gateway/test_notify_fatal_error_shield.py

@@ -0,0 +1,120 @@
+"""Regression test for #81335 — fatal-error handler must survive cancellation
+of the task that awaits ``_notify_fatal_error``.
+
+The Telegram adapter escalates exhausted polling retries from inside its own
+``_polling_error_task``. The gateway's fatal handler tears the adapter down
+via ``disconnect()``, which cancels that very task. The handler used to be
+killed mid-flight by the propagating ``CancelledError``: the adapter was
+already popped from the gateway's adapter map, but the platform was never
+queued for background reconnection — a zombie gateway.
+
+These tests model that carrier-cancellation race directly against
+``BasePlatformAdapter._notify_fatal_error``.
+"""
+
+import asyncio
+
+import pytest
+
+from gateway.platforms.base import BasePlatformAdapter
+
+
+class _FakeAdapter:
+    """Minimal stand-in exposing only what ``_notify_fatal_error`` touches."""
+
+    _notify_fatal_error = BasePlatformAdapter._notify_fatal_error
+
+    def __init__(self):
+        self._fatal_error_handler = None
+        self.handler_completed = False
+
+
+@pytest.mark.asyncio
+async def test_handler_survives_carrier_cancellation():
+    """Handler must run to completion even when the awaiting task is
+    cancelled from inside the handler (the disconnect() self-cancel race)."""
+    adapter = _FakeAdapter()
+    carrier_task = None
+
+    async def gateway_handler(a):
+        # Step 1: teardown — cancels the carrier task (what the real
+        # handler does indirectly via adapter.disconnect()).
+        carrier_task.cancel()
+        # Yield so the cancellation is delivered while we're still running.
+        await asyncio.sleep(0.05)
+        # Step 2: the part that never ran before the fix — queueing the
+        # platform for background reconnection.
+        a.handler_completed = True
+
+    adapter._fatal_error_handler = gateway_handler
+
+    async def carrier():
+        await adapter._notify_fatal_error()
+
+    carrier_task = asyncio.create_task(carrier())
+    with pytest.raises(asyncio.CancelledError):
+        await carrier_task
+
+    # Let the detached, shielded handler finish.
+    await asyncio.sleep(0.2)
+
+    assert carrier_task.cancelled()
+    assert adapter.handler_completed, (
+        "fatal-error handler was killed by carrier cancellation — platform "
+        "would never be queued for reconnection (zombie gateway, #81335)"
+    )
+
+
+@pytest.mark.asyncio
+async def test_carrier_cancellation_still_propagates():
+    """The carrier task itself must still observe CancelledError (teardown
+    semantics unchanged) — only the handler is shielded."""
+    adapter = _FakeAdapter()
+    carrier_task = None
+
+    async def gateway_handler(a):
+        carrier_task.cancel()
+        await asyncio.sleep(0.05)
+        a.handler_completed = True
+
+    adapter._fatal_error_handler = gateway_handler
+
+    async def carrier():
+        await adapter._notify_fatal_error()
+
+    carrier_task = asyncio.create_task(carrier())
+    with pytest.raises(asyncio.CancelledError):
+        await carrier_task
+    assert carrier_task.cancelled()
+
+
+@pytest.mark.asyncio
+async def test_uncancelled_path_unchanged():
+    """Normal path (no cancellation) behaves exactly as before."""
+    adapter = _FakeAdapter()
+
+    async def gateway_handler(a):
+        a.handler_completed = True
+
+    adapter._fatal_error_handler = gateway_handler
+    await adapter._notify_fatal_error()
+    assert adapter.handler_completed
+
+
+@pytest.mark.asyncio
+async def test_sync_handler_still_supported():
+    """Synchronous handlers (non-coroutine return) keep working."""
+    adapter = _FakeAdapter()
+
+    def gateway_handler(a):
+        a.handler_completed = True
+
+    adapter._fatal_error_handler = gateway_handler
+    await adapter._notify_fatal_error()
+    assert adapter.handler_completed
+
+
+@pytest.mark.asyncio
+async def test_no_handler_is_noop():
+    adapter = _FakeAdapter()
+    await adapter._notify_fatal_error()  # must not raise