fix(gateway/discord): REST liveness probe to detect zombie clients (#26656)

plugins/platforms/discord/adapter.py

@@ -103,7 +103,7 @@ def __init__(self, id: int) -> None:  # noqa: A002 - matches discord API
 from gateway.config import Platform, PlatformConfig
 
 from gateway.platforms.helpers import MessageDeduplicator, ThreadParticipationTracker, convert_table_to_bullets
-from utils import atomic_json_write, env_float
+from utils import atomic_json_write, env_float, env_int
 from gateway.platforms.base import (
     BasePlatformAdapter,
     MessageEvent,
@@ -781,6 +781,23 @@ def __init__(self, config: PlatformConfig):
         self._typing_tasks: Dict[str, asyncio.Task] = {}
         self._bot_task: Optional[asyncio.Task] = None
         self._post_connect_task: Optional[asyncio.Task] = None
+        # REST-level liveness probe.  discord.py's WS reconnect handles clean
+        # drops, but a dead proxy / NAT can wedge the socket without delivering
+        # a RST — sends time out forever and ``client.start()`` never exits, so
+        # the bot-task done callback never fires.  See #26656.  An out-of-band
+        # ``fetch_user`` exercises the same REST path as message delivery and
+        # lets us detect the zombie state, close the wedged client, and trip the
+        # existing retryable-fatal reconnect path.  Knobs are surfaced in
+        # config.yaml as ``discord.liveness_interval_seconds`` /
+        # ``discord.liveness_failure_threshold`` (bridged to these env vars by
+        # ``_apply_yaml_config``); set either to 0 to disable.
+        self._liveness_interval_seconds = env_float(
+            "HERMES_DISCORD_LIVENESS_INTERVAL_SECONDS", 60.0
+        )
+        self._liveness_failure_threshold = env_int(
+            "HERMES_DISCORD_LIVENESS_FAILURE_THRESHOLD", 3
+        )
+        self._liveness_task: Optional[asyncio.Task] = None
         # True while disconnect() is intentionally closing discord.py. The
         # bot task's done callback uses this to distinguish an operator/service
         # shutdown from a runtime websocket crash.
@@ -1139,6 +1156,7 @@ async def on_voice_state_update(member, before, after):
             await _wait_for_ready_or_bot_exit(self._ready_event, self._bot_task, timeout=30)
 
             self._running = True
+            self._start_liveness_probe()
             return True
 
         except asyncio.TimeoutError:
@@ -1169,9 +1187,98 @@ async def _cancel_bot_task(self) -> None:
                 pass
         self._bot_task = None
 
+    def _start_liveness_probe(self) -> None:
+        """Start the periodic REST liveness probe if configured.
+
+        Idempotent: if a task is already running we leave it alone so a
+        re-entrant ``connect()`` cannot fork two probes against the same client.
+        """
+        if self._liveness_interval_seconds <= 0 or self._liveness_failure_threshold <= 0:
+            return
+        if self._liveness_task and not self._liveness_task.done():
+            return
+        self._liveness_task = asyncio.create_task(self._liveness_loop())
+
+    async def _liveness_loop(self) -> None:
+        """Probe Discord REST periodically and force a reconnect on persistent failure.
+
+        See #26656.  ``client.start()`` reconnects internally on clean WS drops,
+        but when the underlying socket is wedged behind a dead proxy the WS never
+        sees a RST and the adapter sits in a silent zombie state — process alive,
+        ``client.start()`` spinning, sends timing out forever, and the bot-task
+        done callback never fires because the task never completes.  An
+        out-of-band ``fetch_user`` exercises the same REST path as message
+        delivery and lets us detect the wedge.  After ``threshold`` consecutive
+        failures we close the client, set a retryable fatal error, and hand
+        control back to the gateway's platform reconnect watcher.
+        """
+        interval = self._liveness_interval_seconds
+        threshold = self._liveness_failure_threshold
+        fails = 0
+        while self._running:
+            try:
+                await asyncio.sleep(interval)
+            except asyncio.CancelledError:
+                return
+            client = self._client
+            if not self._running or client is None or getattr(self, "_disconnecting", False):
+                return
+            if hasattr(client, "is_closed") and client.is_closed():
+                return
+            user = getattr(client, "user", None)
+            if user is None:
+                continue
+            try:
+                await client.fetch_user(user.id)
+                fails = 0
+            except asyncio.CancelledError:
+                return
+            except Exception as exc:
+                fails += 1
+                logger.warning(
+                    "[%s] Discord liveness probe failed (%d/%d): %s",
+                    self.name, fails, threshold, exc,
+                )
+                if fails < threshold:
+                    continue
+                logger.error(
+                    "[%s] Discord client appears dead, forcing reconnect", self.name,
+                )
+                try:
+                    await client.close()
+                except Exception:
+                    logger.debug(
+                        "[%s] Error closing wedged Discord client", self.name, exc_info=True,
+                    )
+                self._set_fatal_error(
+                    "liveness_probe_failed",
+                    f"Discord REST liveness probe failed {fails} times in a row",
+                    retryable=True,
+                )
+                try:
+                    await self._notify_fatal_error()
+                except Exception:
+                    logger.debug(
+                        "[%s] Fatal-error handler raised", self.name, exc_info=True,
+                    )
+                return
+
+    async def _cancel_liveness_task(self) -> None:
+        """Cancel and await the liveness probe task, if running."""
+        if self._liveness_task and not self._liveness_task.done():
+            self._liveness_task.cancel()
+            try:
+                await self._liveness_task
+            except asyncio.CancelledError:
+                pass
+        self._liveness_task = None
+
     async def disconnect(self) -> None:
         """Disconnect from Discord."""
         self._disconnecting = True
+        # Cancel the liveness probe first so it can't fire a spurious fatal
+        # error / reconnect while we're intentionally tearing the adapter down.
+        await self._cancel_liveness_task()
         # Cancel the bot task before closing the client.  If connect() timed out
         # and returned False, the background client.start() task may still be
         # running; calling client.close() alone is not enough to stop it because
@@ -1203,6 +1310,7 @@ async def disconnect(self) -> None:
         self._client = None
         self._ready_event.clear()
         self._post_connect_task = None
+        self._liveness_task = None
 
         self._release_platform_lock()
 
@@ -7136,6 +7244,16 @@ def _apply_yaml_config(yaml_cfg: dict, discord_cfg: dict) -> dict | None:
     if _discord_rtm is not None and not os.getenv("DISCORD_REPLY_TO_MODE"):
         _rtm_str = "off" if _discord_rtm is False else str(_discord_rtm).lower()
         os.environ["DISCORD_REPLY_TO_MODE"] = _rtm_str
+    # liveness probe knobs: detect zombie clients behind dead proxies/NATs and
+    # force a reconnect (#26656).  Bridged to the env vars the adapter reads in
+    # __init__; set either to 0 to disable.  config.yaml is the user-facing
+    # surface — these env vars are an internal mechanism only.
+    lis = discord_cfg.get("liveness_interval_seconds")
+    if lis is not None and not os.getenv("HERMES_DISCORD_LIVENESS_INTERVAL_SECONDS"):
+        os.environ["HERMES_DISCORD_LIVENESS_INTERVAL_SECONDS"] = str(lis)
+    lft = discord_cfg.get("liveness_failure_threshold")
+    if lft is not None and not os.getenv("HERMES_DISCORD_LIVENESS_FAILURE_THRESHOLD"):
+        os.environ["HERMES_DISCORD_LIVENESS_FAILURE_THRESHOLD"] = str(lft)
     return None  # all settings flow through env; nothing to merge into extras
 
 

tests/gateway/test_discord_liveness.py

@@ -0,0 +1,188 @@
+"""Regression tests for the Discord REST liveness probe (#26656).
+
+discord.py's WebSocket reconnect handles clean drops, but a wedged proxy /
+NAT can leave the underlying socket dead without ever delivering a RST —
+sends time out forever while ``client.start()`` happily spins and never
+exits, so the bot-task done callback never fires either.  The probe in
+``DiscordAdapter`` periodically hits Discord REST so we can detect the
+zombie state and trip the gateway's existing reconnect path.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+
+import pytest
+
+# Re-use the shared discord-stub bootstrap and FakeBot from the connect
+# test module so this file doesn't duplicate the (large) mock surface.
+from tests.gateway.test_discord_connect import (  # noqa: E402
+    FakeBot,
+    _ensure_discord_mock,
+)
+
+_ensure_discord_mock()
+
+import plugins.platforms.discord.adapter as discord_platform  # noqa: E402
+from gateway.config import PlatformConfig  # noqa: E402
+from plugins.platforms.discord.adapter import DiscordAdapter  # noqa: E402
+
+
+class _LiveBot(FakeBot):
+    """A FakeBot whose ``start()`` stays pending like a real discord.py client.
+
+    The default ``FakeBot.start()`` returns immediately, which would let the
+    bot-task done callback fire and set a spurious fatal error.  Real clients
+    keep ``start()`` running for the life of the connection; this models that
+    so the liveness probe is the only thing that can trip a fatal error.
+    """
+
+    def __init__(self, *, intents, proxy=None, allowed_mentions=None, **_):
+        super().__init__(intents=intents, allowed_mentions=allowed_mentions)
+        self._never = asyncio.Event()
+        self._closed = False
+
+    async def start(self, token):
+        if "on_ready" in self._events:
+            await self._events["on_ready"]()
+        # Stay alive until close() is called — mirrors a real client.
+        await self._never.wait()
+
+    def is_closed(self):
+        return self._closed
+
+    async def close(self):
+        self._closed = True
+        self._never.set()
+
+
+def _make_adapter(monkeypatch, *, interval=0.01, threshold=1) -> DiscordAdapter:
+    monkeypatch.setenv("HERMES_DISCORD_LIVENESS_INTERVAL_SECONDS", str(interval))
+    monkeypatch.setenv("HERMES_DISCORD_LIVENESS_FAILURE_THRESHOLD", str(threshold))
+    return DiscordAdapter(PlatformConfig(enabled=True, token="test-token"))
+
+
+async def _connect(adapter: DiscordAdapter, monkeypatch, bot_factory):
+    monkeypatch.setattr(
+        "gateway.status.acquire_scoped_lock",
+        lambda scope, identity, metadata=None: (True, None),
+    )
+    monkeypatch.setattr("gateway.status.release_scoped_lock", lambda scope, identity: None)
+    intents = SimpleNamespace(
+        message_content=False, dm_messages=False, guild_messages=False,
+        members=False, voice_states=False,
+    )
+    monkeypatch.setattr(discord_platform.Intents, "default", lambda: intents)
+    monkeypatch.setattr(discord_platform.commands, "Bot", bot_factory)
+    monkeypatch.setattr(adapter, "_resolve_allowed_usernames", AsyncMock())
+    assert await adapter.connect() is True
+
+
+@pytest.mark.asyncio
+async def test_liveness_probe_disabled_when_interval_zero(monkeypatch):
+    """interval<=0 must skip the probe entirely so users can opt out."""
+    adapter = _make_adapter(monkeypatch, interval=0)
+
+    bot_holder: dict = {}
+
+    def factory(**kwargs):
+        bot = _LiveBot(intents=kwargs["intents"], allowed_mentions=kwargs.get("allowed_mentions"))
+        bot.fetch_user = AsyncMock()
+        bot_holder["bot"] = bot
+        return bot
+
+    await _connect(adapter, monkeypatch, factory)
+    assert adapter._liveness_task is None
+    await asyncio.sleep(0.05)
+    bot_holder["bot"].fetch_user.assert_not_called()
+    await adapter.disconnect()
+
+
+@pytest.mark.asyncio
+async def test_liveness_probe_disabled_when_threshold_zero(monkeypatch):
+    """threshold<=0 must also skip the probe."""
+    adapter = _make_adapter(monkeypatch, interval=0.01, threshold=0)
+
+    def factory(**kwargs):
+        bot = _LiveBot(intents=kwargs["intents"], allowed_mentions=kwargs.get("allowed_mentions"))
+        bot.fetch_user = AsyncMock()
+        return bot
+
+    await _connect(adapter, monkeypatch, factory)
+    assert adapter._liveness_task is None
+    await adapter.disconnect()
+
+
+@pytest.mark.asyncio
+async def test_liveness_probe_pings_rest_while_healthy(monkeypatch):
+    """A healthy probe keeps the adapter running and never sets a fatal error."""
+    adapter = _make_adapter(monkeypatch, interval=0.01, threshold=3)
+
+    def factory(**kwargs):
+        bot = _LiveBot(intents=kwargs["intents"], allowed_mentions=kwargs.get("allowed_mentions"))
+        bot.fetch_user = AsyncMock(return_value=SimpleNamespace(id=999))
+        return bot
+
+    await _connect(adapter, monkeypatch, factory)
+    await asyncio.sleep(0.05)
+    assert adapter._client.fetch_user.await_count >= 1
+    assert adapter._running is True
+    assert adapter.has_fatal_error is False
+    await adapter.disconnect()
+
+
+@pytest.mark.asyncio
+async def test_liveness_probe_forces_reconnect_after_threshold(monkeypatch):
+    """Once the probe fails ``threshold`` times in a row, the adapter must
+    close the wedged client and surface a retryable fatal error so the
+    gateway's reconnect watcher (gateway/run.py) can rebuild it."""
+    adapter = _make_adapter(monkeypatch, interval=0.005, threshold=2)
+
+    def factory(**kwargs):
+        bot = _LiveBot(intents=kwargs["intents"], allowed_mentions=kwargs.get("allowed_mentions"))
+        bot.fetch_user = AsyncMock(side_effect=TimeoutError("dead proxy"))
+        return bot
+
+    handler = AsyncMock()
+    adapter.set_fatal_error_handler(handler)
+    await _connect(adapter, monkeypatch, factory)
+    wedged = adapter._client
+
+    # Wait for the loop to exit (it returns after threshold consecutive
+    # failures).  Bounded by a generous timeout so a regression doesn't hang CI.
+    for _ in range(200):
+        if adapter._liveness_task and adapter._liveness_task.done():
+            break
+        await asyncio.sleep(0.01)
+    else:
+        pytest.fail("liveness loop did not terminate within 2s")
+
+    assert wedged.is_closed() is True
+    assert adapter.has_fatal_error is True
+    assert adapter.fatal_error_code == "liveness_probe_failed"
+    assert adapter.fatal_error_retryable is True
+    handler.assert_awaited_once()
+
+    await adapter.disconnect()
+
+
+@pytest.mark.asyncio
+async def test_disconnect_cancels_liveness_task(monkeypatch):
+    """``disconnect()`` must cancel the probe so the gateway can shut down
+    cleanly without leaking a background task."""
+    adapter = _make_adapter(monkeypatch, interval=60, threshold=3)
+
+    def factory(**kwargs):
+        bot = _LiveBot(intents=kwargs["intents"], allowed_mentions=kwargs.get("allowed_mentions"))
+        bot.fetch_user = AsyncMock()
+        return bot
+
+    await _connect(adapter, monkeypatch, factory)
+    task = adapter._liveness_task
+    assert task is not None and not task.done()
+
+    await adapter.disconnect()
+    assert task.done()
+    assert adapter._liveness_task is None

website/docs/reference/environment-variables.md

@@ -569,6 +569,8 @@ Advanced per-platform knobs for throttling the outbound message batcher. Most us
 | `HERMES_TELEGRAM_DISABLE_FALLBACK_IPS` | Disable the hard-coded Cloudflare fallback IPs used when DNS fails (`true`/`false`). |
 | `HERMES_DISCORD_TEXT_BATCH_DELAY_SECONDS` | Grace window before flushing a queued Discord text chunk (default: `0.6`). |
 | `HERMES_DISCORD_TEXT_BATCH_SPLIT_DELAY_SECONDS` | Delay between split chunks when a Discord message exceeds the length limit (default: `2.0`). |
+| `HERMES_DISCORD_LIVENESS_INTERVAL_SECONDS` | Internal bridge for `discord.liveness_interval_seconds` (config.yaml). Interval for the Discord REST liveness probe that detects zombie clients behind dead proxies/NATs (default: `60`; set to `0` to disable). Prefer setting `discord.liveness_interval_seconds` in `config.yaml`. |
+| `HERMES_DISCORD_LIVENESS_FAILURE_THRESHOLD` | Internal bridge for `discord.liveness_failure_threshold` (config.yaml). Consecutive probe failures before forcing a Discord reconnect (default: `3`). Prefer setting `discord.liveness_failure_threshold` in `config.yaml`. |
 | `HERMES_MATRIX_TEXT_BATCH_DELAY_SECONDS` / `_SPLIT_DELAY_SECONDS` | Matrix equivalents of the Telegram batch knobs. |
 | `HERMES_FEISHU_TEXT_BATCH_DELAY_SECONDS` / `_SPLIT_DELAY_SECONDS` / `_MAX_CHARS` / `_MAX_MESSAGES` | Feishu batcher tuning — delay, split delay, max chars per message, max messages per batch. |
 | `HERMES_FEISHU_MEDIA_BATCH_DELAY_SECONDS` | Feishu media flush delay. |