fix(photon): dispatch fatal-error notification from a detached task

Follow-up to #69112. That PR hardened the shared gateway dispatch path in
gateway/run.py against the *caller* being cancelled. A second, self-referential
cancellation specific to PhotonAdapter sits one layer underneath it and survived
that fix.

PhotonAdapter is the only platform adapter that awaits _notify_fatal_error()
inline, on the same task that detected the fault. Both _monitor_sidecar_health
and _supervise_sidecar run as self._sidecar_health_task /
self._sidecar_supervisor_task, and the notification routes into
GatewayRunner._handle_adapter_fatal_error_impl, which tears the adapter down via
_safe_adapter_disconnect -> disconnect(). disconnect() then cancels
self._sidecar_health_task and awaits it -- which, when the health task is what
raised the notification, means disconnect() cancels its own caller several
plain-await frames up.

disconnect()'s `task is not asyncio.current_task()` guard does not catch this.
The current task where that guard evaluates is the wrapper
_await_adapter_cleanup_with_timeout creates around disconnect() via
asyncio.ensure_future, not the health task further up the chain, so the guard
passes and the cancel lands.

CancelledError stopped subclassing Exception in Python 3.8, so the
`except Exception` that wrapped the inline notify call never saw it. The health
task died silently mid-handoff: no log line, no "exception never retrieved"
warning (cancellation is normal asyncio), and no retry. The platform stayed
stranded until the gateway was restarted by hand.

Fix: dispatch the notification onto a new task, the same pattern
DiscordAdapter._handle_bot_task_done already uses for this reason. disconnect()
can then cancel the health/supervisor task freely without that cancellation
reaching the code still running the handoff, so the handoff always reaches the
reconnect queue. Both Photon fatal call sites are converted: the health-poll
path (observed wedging) and the sidecar-crash path (same shape, not yet
observed). gateway/run.py is untouched.

Observed twice on a self-hosted gateway, ~4h38m and ~52min of silent inbound
outage, both cleared only by a manual restart, both post-dating #69112's merge.
In each case the fatal log line appears and `queued for background reconnection`
never does.

Tests: new tests/plugins/platforms/photon/test_fatal_notify_self_cancel.py
covers the self-cancellation (fails with CancelledError without this change),
that the dispatch does not block its caller, that a failing notification warns
rather than raising, and a source guard against reintroducing either inline
await. Two assertions in test_overflow_recovery.py that drove these coroutines
directly now drain pending tasks before asserting delivery, since the
notification is deliberately no longer awaited inline.

Prepared with agent assistance and reviewed before submission.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

plugins/platforms/photon/adapter.py

@@ -659,6 +659,47 @@ class PhotonAdapter(BasePlatformAdapter):
             self._http_client = None
         self._mark_disconnected()
 
+    def _dispatch_fatal_notification(self) -> None:
+        """Notify the gateway of a fatal error from a detached task.
+
+        ``_monitor_sidecar_health`` and ``_supervise_sidecar`` run as
+        ``self._sidecar_health_task`` / ``self._sidecar_supervisor_task`` and
+        used to call ``await self._notify_fatal_error()`` directly, on their
+        own call stack. That routes straight into
+        ``GatewayRunner._handle_adapter_fatal_error_impl``, which tears the
+        adapter down via ``_safe_adapter_disconnect`` -> ``disconnect()``.
+        ``disconnect()`` cancels ``self._sidecar_health_task`` and awaits it
+        to make sure it's really gone -- but when the *health task itself* is
+        what's driving this whole notification (the common case: the health
+        poll is what detected the fatal condition), that task IS the one
+        awaiting several frames up the stack, so ``disconnect()`` cancels and
+        awaits its own caller. ``disconnect()``'s
+        ``task is not asyncio.current_task()`` guard doesn't catch this,
+        because the wrapper task ``_await_adapter_cleanup_with_timeout``
+        creates around ``disconnect()`` (via ``asyncio.ensure_future``) is
+        the current task there, not the health/supervisor task further up
+        the plain-await chain. The self-cancellation throws an unhandled
+        ``CancelledError`` through the health/supervisor task with nothing
+        to catch it -- ``CancelledError`` stopped subclassing ``Exception``
+        in Python 3.8, so the ``except Exception`` around the notify call
+        never saw it -- and the task dies silently mid-handoff: no log line,
+        no retry, the platform stranded with no automated way back.
+
+        Dispatching the notification onto a brand-new task instead -- the
+        same pattern ``DiscordAdapter._handle_bot_task_done`` already uses --
+        means ``disconnect()`` can cancel the health/supervisor task freely
+        without that cancellation ever reaching back into the code that's
+        still running the handoff, so the handoff always reaches the
+        reconnect queue.
+        """
+        asyncio.create_task(self._notify_fatal_error_logged())
+
+    async def _notify_fatal_error_logged(self) -> None:
+        try:
+            await self._notify_fatal_error()
+        except Exception as exc:  # pragma: no cover - defensive
+            logger.warning("[photon] fatal-error notification failed: %s", exc)
+
     # -- Inbound stream consumer ------------------------------------------
 
     async def _inbound_loop(self) -> None:
@@ -738,10 +779,7 @@ class PhotonAdapter(BasePlatformAdapter):
                 message,
                 retryable=True,
             )
-            try:
-                await self._notify_fatal_error()
-            except Exception as exc:  # pragma: no cover - defensive
-                logger.warning("[photon] fatal-error notification failed: %s", exc)
+            self._dispatch_fatal_notification()
             break
 
     async def _on_inbound_line(self, line: str) -> None:
@@ -1250,10 +1288,7 @@ class PhotonAdapter(BasePlatformAdapter):
                 f"Photon sidecar exited unexpectedly (code {exit_code})",
                 retryable=True,
             )
-            try:
-                await self._notify_fatal_error()
-            except Exception as exc:  # pragma: no cover - defensive
-                logger.warning("[photon] fatal-error notification failed: %s", exc)
+            self._dispatch_fatal_notification()
 
     async def _stop_sidecar(self) -> None:
         proc = self._sidecar_proc

tests/plugins/platforms/photon/test_fatal_notify_self_cancel.py

@@ -0,0 +1,150 @@
+"""Photon's fatal-error notification must not be cancelled by its own teardown.
+
+`_monitor_sidecar_health` and `_supervise_sidecar` run as long-lived tasks on
+the adapter. When one of them detects a fatal condition, the gateway's handler
+tears the adapter down, and `disconnect()` cancels `_sidecar_health_task` and
+awaits it. If the notification is awaited inline on the health task's own call
+stack, `disconnect()` ends up cancelling its own ultimate caller: the
+`task is not asyncio.current_task()` guard in `disconnect()` compares against
+the wrapper task the gateway created around `disconnect()`, not the health task
+several plain-await frames further up, so the guard passes and the cancel lands.
+
+`CancelledError` stopped subclassing `Exception` in Python 3.8, so the
+`except Exception` that used to wrap the inline notify call never saw it. The
+health task died silently mid-handoff -- no log line, no retry -- and the
+platform stayed stranded until someone restarted the gateway by hand.
+
+PR #69112 hardened the shared dispatch path in `gateway/run.py` against a
+cancelled *caller*. These tests cover the Photon-specific self-cancellation one
+layer down, which that PR's scope could not reach.
+"""
+from __future__ import annotations
+
+import asyncio
+from typing import Any, Dict, List
+
+import pytest
+
+from gateway.config import PlatformConfig
+from plugins.platforms.photon.adapter import PhotonAdapter
+
+
+def _make_adapter(monkeypatch: pytest.MonkeyPatch) -> PhotonAdapter:
+    monkeypatch.setenv("PHOTON_PROJECT_ID", "test-project-id")
+    monkeypatch.setenv("PHOTON_PROJECT_SECRET", "test-project-secret")
+    return PhotonAdapter(PlatformConfig(enabled=True, token="", extra={}))
+
+
+class TestFatalNotifyIsDetached:
+    """The notification must outlive cancellation of the task that raised it."""
+
+    @pytest.mark.asyncio
+    async def test_health_task_cancellation_does_not_kill_notification(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        """A fatal handler that cancels the health task (exactly what
+        ``disconnect()`` does) must still see the notification delivered."""
+        adapter = _make_adapter(monkeypatch)
+        adapter._inbound_running = True
+        adapter._sidecar_health_interval = 0
+        delivered = asyncio.Event()
+
+        async def fake_notify() -> None:
+            # Mirror the real handler: tear the adapter down, cancelling the
+            # health task, then finish the handoff.
+            await adapter.disconnect()
+            delivered.set()
+
+        monkeypatch.setattr(adapter, "_notify_fatal_error", fake_notify)
+        monkeypatch.setattr(adapter, "_stop_sidecar", lambda: _noop())
+
+        async def degraded(_path: str, _payload: Dict[str, Any]) -> Dict[str, Any]:
+            return {"stream": {"ok": False, "state": "degraded", "degradedForMs": 4000,
+                               "lastIssue": "stream persistently failing"}}
+
+        monkeypatch.setattr(adapter, "_sidecar_call", degraded)
+
+        health = asyncio.create_task(adapter._monitor_sidecar_health())
+        adapter._sidecar_health_task = health
+
+        await asyncio.wait_for(delivered.wait(), timeout=5.0)
+        assert adapter.has_fatal_error
+        assert adapter.fatal_error_code == "UPSTREAM_STREAM_DEGRADED"
+        assert adapter.fatal_error_retryable
+
+        # With the dispatch detached, the health task reaches its own `break`
+        # and returns cleanly instead of being cancelled out from under the
+        # handoff. Either way it must not die with an unhandled exception --
+        # that was the silent failure that stranded the platform.
+        await asyncio.wait_for(asyncio.shield(health), timeout=2.0)
+        assert health.done()
+        if not health.cancelled():
+            assert health.exception() is None
+
+    @pytest.mark.asyncio
+    async def test_dispatch_does_not_await_on_caller_stack(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        """``_dispatch_fatal_notification`` must return without awaiting, so a
+        cancel aimed at the calling task cannot reach the handoff."""
+        adapter = _make_adapter(monkeypatch)
+        started = asyncio.Event()
+        finished = asyncio.Event()
+
+        async def slow_notify() -> None:
+            started.set()
+            await asyncio.sleep(0.05)
+            finished.set()
+
+        monkeypatch.setattr(adapter, "_notify_fatal_error", slow_notify)
+
+        async def caller() -> None:
+            adapter._dispatch_fatal_notification()  # must not block
+
+        task = asyncio.create_task(caller())
+        await task  # returns immediately even though notify sleeps
+        await asyncio.wait_for(started.wait(), timeout=2.0)
+
+        task.cancel()  # cancelling the caller must not touch the notification
+        await asyncio.wait_for(finished.wait(), timeout=2.0)
+
+    @pytest.mark.asyncio
+    async def test_notification_failure_is_logged_not_raised(
+        self, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
+    ) -> None:
+        """A failing notification must warn rather than surface as an
+        unretrieved task exception."""
+        adapter = _make_adapter(monkeypatch)
+
+        async def boom() -> None:
+            raise RuntimeError("gateway unreachable")
+
+        monkeypatch.setattr(adapter, "_notify_fatal_error", boom)
+
+        with caplog.at_level("WARNING"):
+            await adapter._notify_fatal_error_logged()
+
+        assert "fatal-error notification failed" in caplog.text
+
+
+class TestBothCallSitesDetached:
+    """Neither fatal path may await the notification inline."""
+
+    def test_no_inline_notify_awaits_remain(self) -> None:
+        """Guard against a future edit reintroducing the inline await."""
+        import inspect
+
+        from plugins.platforms.photon import adapter as photon_adapter
+
+        for name in ("_monitor_sidecar_health", "_supervise_sidecar"):
+            src = inspect.getsource(getattr(photon_adapter.PhotonAdapter, name))
+            assert "await self._notify_fatal_error()" not in src, (
+                f"{name} awaits _notify_fatal_error inline; use "
+                f"_dispatch_fatal_notification() so disconnect() cannot cancel "
+                f"its own caller"
+            )
+            assert "_dispatch_fatal_notification()" in src
+
+
+async def _noop() -> None:
+    return None

tests/plugins/platforms/photon/test_overflow_recovery.py

@@ -19,6 +19,7 @@ No Node sidecar is spawned and no ports are bound.
 """
 from __future__ import annotations
 
+import asyncio
 from typing import Any, Dict
 
 import pytest
@@ -174,6 +175,10 @@ async def test_unexpected_sidecar_exit_raises_retryable_fatal(
     # than crashing the whole gateway.
     assert adapter.fatal_error_retryable is True
     assert adapter._running is False
+    # The notification is dispatched onto its own task rather than awaited on
+    # the supervisor's stack, so that disconnect() cancelling the supervisor
+    # cannot kill the handoff. Let that task run before asserting delivery.
+    await _drain_pending_tasks()
     assert notified == [True]
 
 
@@ -233,4 +238,25 @@ async def test_degraded_stream_health_raises_retryable_fatal(
     assert adapter.has_fatal_error is True
     assert adapter.fatal_error_code == "UPSTREAM_STREAM_DEGRADED"
     assert adapter.fatal_error_retryable is True
+    # Dispatched detached (see _dispatch_fatal_notification) so the health
+    # task's own teardown cannot cancel the handoff; drain before asserting.
+    await _drain_pending_tasks()
     assert notified == [True]
+
+
+async def _drain_pending_tasks(limit: int = 50) -> None:
+    """Let detached fatal-notification tasks finish before asserting on them.
+
+    ``_dispatch_fatal_notification`` deliberately does not await the
+    notification (that is what kept ``disconnect()`` from cancelling its own
+    caller), so a test that drives ``_monitor_sidecar_health`` /
+    ``_supervise_sidecar`` directly returns before the notification has run.
+    """
+    for _ in range(limit):
+        pending = [
+            t for t in asyncio.all_tasks()
+            if t is not asyncio.current_task() and not t.done()
+        ]
+        if not pending:
+            return
+        await asyncio.wait(pending, timeout=1.0)