NousResearch/hermes-agent

Discord gateway zombie state and reconnect fixes

Cluster 26656 · gateway. Data as of 2026-10-24 20:38 UTC

Nothing open on GitHub. Fix merged in #66946 on 18 Jul 2026. Check it covers your case.

open issues

open PRs

items grouped

Known?: Yes. 4 related items grouped here, 1 issue already closed.
Fix event: A fix was merged to main in PR #66946. Check it covers your case. No release data is tracked.
Tracker: The maintainers' tracker marks this resolved, deciding PR #53924. That is separate from the fix event.
Still open: Nothing open on GitHub.

Next step: read #66946, then check it covers your case.

Read #66946 Pick up a Lab task

These items address a silent zombie state in the Discord gateway where clients fail to reconnect after network outages. Fixes involve REST liveness probes, forced reconnection on probe failure, and recovery from websocket and event-loop stalls.

Open issues0

No open issues.

Open PRs0

No open pull requests.

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

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

merged to main

files: plugins/platforms/discord/adapter.py, tests/gateway/test_discord_liveness.py, website/docs/reference/environment-variables.md

Show the diff

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="[redacted:secret-assignment]"))
+
+
+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
+
+
[redacted:email]
+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()
+
+
[redacted:email]
+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()
+
+
[redacted:email]
+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()
+
+
[redacted:email]
+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()
+
+
[redacted:email]
+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. |

#66946 fix(gateway): recover Discord websocket and event-loop stalls

merged to main

files: agent/async_utils.py, gateway/config.py, gateway/platforms/base.py, gateway/run.py, gateway/systemd_notify.py, hermes_cli/config.py, hermes_cli/gateway.py, plugins/platforms/discord/adapter.py, scripts/release.py, tests/gateway/test_bounded_adapter_teardown.py, tests/gateway/test_config.py, tests/gateway/test_discord_approval_mentions.py, tests/gateway/test_discord_liveness.py, tests/gateway/test_gateway_shutdown.py, tests/gateway/test_multiplex_adapter_registry.py, tests/gateway/test_runner_fatal_adapter.py, tests/gateway/test_safe_adapter_disconnect.py, tests/gateway/test_shutdown_cache_cleanup.py, tests/gateway/test_systemd_notify.py, tests/gateway/test_systemd_watchdog_lifecycle.py (+6 more)

Show the diff

agent/async_utils.py

@@ -66,3 +66,19 @@ def safe_schedule_threadsafe(
             coro.close()
         log.log(log_level, "%s: %s", log_message, exc)
         return None
+
+
+def consume_detached_task_result(task: "asyncio.Future[Any]") -> None:
+    """Retrieve a detached task's result without surfacing cancellation.
+
+    Used as an ``add_done_callback`` on tasks that were cancelled and
+    detached (e.g. an adapter close path that swallows ``CancelledError``
+    past its teardown deadline). Observing ``task.exception()`` prevents
+    "exception was never retrieved" noise on the event loop; cancellation
+    and any terminal error are deliberately swallowed — the task's owner
+    already gave up on it.
+    """
+    try:
+        task.exception()
+    except (asyncio.CancelledError, Exception):
+        pass

gateway/config.py

@@ -130,6 +130,49 @@ def _coerce_optional_positive_int(value: Any, key: str) -> Optional[int]:
     return parsed
 
 
+_SYSTEMD_WATCHDOG_MAX_SECONDS = 2_147_483_647
+
+
+def coerce_systemd_watchdog_seconds(
+    value: Any, key: str = "gateway.systemd_watchdog_seconds"
+) -> int:
+    """Return a bounded positive watchdog interval or zero when disabled.
+
+    Runtime and service generation share this normalization so a value can
+    never enable ``Type=notify`` while disabling application heartbeats.
+    """
+    if value is None:
+        return 0
+    if isinstance(value, bool):
+        logger.warning("Ignoring invalid %s (expected a positive integer)", key)
+        return 0
+    if isinstance(value, int):
+        parsed = value
+    elif isinstance(value, str):
+        raw = value.strip()
+        if not raw or not raw.isascii() or not raw.isdecimal():
+            logger.warning("Ignoring invalid %s (expected a positive integer)", key)
+            return 0
+        try:
+            parsed = int(raw, 10)
+        except (TypeError, ValueError, OverflowError):
+            logger.warning("Ignoring invalid %s (expected a positive integer)", key)
+            return 0
+    else:
+        logger.warning("Ignoring invalid %s (expected a positive integer)", key)
+        return 0
+    if parsed == 0:
+        return 0
+    if not 0 < parsed <= _SYSTEMD_WATCHDOG_MAX_SECONDS:
+        logger.warning(
+            "Ignoring invalid %s (expected an integer from 1 to %d)",
+            key,
+            _SYSTEMD_WATCHDOG_MAX_SECONDS,
+        )
+        return 0
+    return parsed
+
+
 def _coerce_dict(value: Any) -> Dict[str, Any]:
     """Return *value* when it is a mapping, otherwise an empty dict."""
     return value if isinstance(value, dict) else {}
@@ -768,6 +811,10 @@ class GatewayConfig:
     # gateway behaves exactly as before — single HERMES_HOME, no profile stamping.
     multiplex_profiles: bool = False
 
+    # Opt-in systemd event-loop watchdog. Zero preserves Type=simple and
+    # disables sd_notify at runtime.
+    systemd_watchdog_seconds: int = 0
+
     # Unauthorized DM policy
     unauthorized_dm_behavior: str = "pair"  # "pair" or "ignore"
 
@@ -786,6 +833,11 @@ class GatewayConfig:
     # dict with: name, platform, profile, and optional guild_id/chat_id/thread_id.
     profile_routes: list = field(default_factory=list)
 
+    def __post_init__(self) -> None:
+        self.systemd_watchdog_seconds = coerce_systemd_watchdog_seconds(
+            self.systemd_watchdog_seconds
+        )
+
     def get_connected_platforms(self) -> List[Platform]:
         """Return list of platforms that are enabled and configured."""
         connected = []
@@ -889,6 +941,7 @@ def to_dict(self) -> Dict[str, Any]:
             "thread_sessions_per_user": self.thread_sessions_per_user,
             "max_concurrent_sessions": self.max_concurrent_sessions,
             "multiplex_profiles": self.multiplex_profiles,
+            "systemd_watchdog_seconds": self.systemd_watchdog_seconds,
             "unauthorized_dm_behavior": self.unauthorized_dm_behavior,
             "streaming": self.streaming.to_dict(),
             "session_store_max_age_days": self.session_store_max_age_days,
@@ -951,6 +1004,15 @@ def from_dict(cls, data: Dict[str, Any]) -> "GatewayConfig":
         thread_sessions_per_user = data.get("thread_sessions_per_user")
         multiplex_profiles = data.get("multiplex_profiles")
         nested_gateway = data.get("gateway") if isinstance(data.get("gateway"), dict) else {}
+        if "systemd_watchdog_seconds" in data:
+            systemd_watchdog_raw = data.get("systemd_watchdog_seconds")
+            systemd_watchdog_key = "systemd_watchdog_seconds"
+        else:
+            systemd_watchdog_raw = nested_gateway.get("systemd_watchdog_seconds")
+            systemd_watchdog_key = "gateway.systemd_watchdog_seconds"
+        systemd_watchdog_seconds = coerce_systemd_watchdog_seconds(
+            systemd_watchdog_raw, systemd_watchdog_key
+        )
         if multiplex_profiles is None and isinstance(nested_gateway, dict):
             # Also honor gateway.multiplex_profiles written by
             # ``hermes config set gateway.multiplex_profiles true``.
@@ -1010,6 +1072,7 @@ def from_dict(cls, data: Dict[str, Any]) -> "GatewayConfig":
             group_sessions_per_user=_coerce_bool(group_sessions_per_user, True),
             thread_sessions_per_user=_coerce_bool(thread_sessions_per_user, False),
             multiplex_profiles=_coerce_bool(multiplex_profiles, False),
+            systemd_watchdog_seconds=systemd_watchdog_seconds,
             max_concurrent_sessions=max_concurrent_sessions,
             unauthorized_dm_behavior=unauthorized_dm_behavior,
             streaming=StreamingConfig.from_dict(data.get("streaming", {})),
@@ -1145,6 +1208,10 @@ def load_gateway_config() -> GatewayConfig:
                     gw_data["multiplex_profiles"] = gateway_section["multiplex_profiles"]
                 if "max_concurrent_sessions" in gateway_section:
                     gw_data["max_concurrent_sessions"] = gateway_section["max_concurrent_sessions"]
+                if "systemd_watchdog_seconds" in gateway_section:
+                    gw_data["systemd_watchdog_seconds"] = gateway_section[
+                        "systemd_watchdog_seconds"
+                    ]
 
             if "max_concurrent_sessions" in yaml_cfg:
                 gw_data["max_concurrent_sessions"] = yaml_cfg["max_concurrent_sessions"]

gateway/platforms/base.py

@@ -4161,6 +4161,34 @@ def _unwrap_ephemeral(self, response: Any) -> Tuple[Optional[str], int]:
             return response.text, int(ttl or 0)
         return response, 0
 
+    def _final_delivery_adapter(
+        self, source: Optional[SessionSource]
+    ) -> "BasePlatformAdapter":
+        """Return the runner's current adapter for a new final-response send.
+
+        A reconnect removes the failed adapter from the runner registry before
+        its in-flight message task completes. That task must keep its own
+        cleanup and partial-message ownership, but an as-yet-unsent final
+        response belongs on the replacement transport. This helper deliberately
+        does not migrate message IDs or route edits/deletes through the new
+        adapter: those operations remain owned by the old transport.
+        """
+        runner = getattr(self, "gateway_runner", None)
+        resolve = getattr(runner, "_adapter_for_source", None)
+        if not callable(resolve):
+            return self
+        try:
+            live_adapter = resolve(source)
+        except Exception:
+            logger.debug("[%s] Failed to resolve live adapter for final delivery", self.name)
+            return self
+        if (
+            not isinstance(live_adapter, BasePlatformAdapter)
+            or live_adapter.platform != self.platform
+        ):
+            return self
+        return live_adapter
+
     async def _send_with_retry(
         self,
         chat_id: str,
@@ -5091,11 +5119,20 @@ async def _stop_typing_task() -> None:
                         except OSError:
                             pass
 
-                # Send the text portion
+                # Send the text portion. A reconnect may have replaced this
+                # adapter while its in-flight handler was still producing a
+                # final response; that response is a new message, so resolve
+                # the current transport before sending it.
                 if text_content and not _tts_caption_delivered:
-                    logger.info("[%s] Sending response (%d chars) to %s", self.name, len(text_content), event.source.chat_id)
+                    delivery_adapter = self._final_delivery_adapter(event.source)
+                    logger.info(
+                        "[%s] Sending response (%d chars) to %s",
+                        delivery_adapter.name,
+                        len(text_content),
+                        event.source.chat_id,
+                    )
                     _reply_anchor = _reply_anchor_for_event(event)
-                    result = await self._send_with_retry(
+                    result = await delivery_adapter._send_with_retry(
                         chat_id=event.source.chat_id,
                         content=text_content,
                         reply_to=_reply_anchor,
@@ -5103,16 +5140,15 @@ async def _stop_typing_task() -> None:
                     )
                     _record_delivery(result)
 
-                    # Schedule auto-deletion of system-notice replies.
-                    # Detached so the handler returns immediately; errors
-                    # (permission denied, message too old) are swallowed.
+                    # Schedule auto-deletion on the adapter that owns the new
+                    # message ID, which may be the reconnect replacement.
                     if (
                         _ephemeral_ttl
                         and _ephemeral_ttl > 0
                         and result.success
                         and result.message_id
                     ):
-                        self._schedule_ephemeral_delete(
+                        delivery_adapter._schedule_ephemeral_delete(
                             chat_id=event.source.chat_id,
                             message_id=result.message_id,
                             ttl_seconds=_ephemeral_ttl,

gateway/run.py

@@ -44,7 +44,7 @@
 from contextvars import copy_context
 from pathlib import Path
 from datetime import datetime
-from typing import Callable, Dict, Optional, Any, List, Union
+from typing import Awaitable, Callable, Dict, Optional, Any, List, Union
 
 # account_usage imports the OpenAI SDK chain (~230 ms). Only needed by
 # /usage; we still import it at module top in the gateway because test
@@ -53,7 +53,7 @@
 # gateway is a long-running daemon, so its boot cost matters less than
 # preserving the established test-patch surface.
 from agent.account_usage import fetch_account_usage, render_account_usage_lines
-from agent.async_utils import safe_schedule_threadsafe
+from agent.async_utils import consume_detached_task_result, safe_schedule_threadsafe
 from agent.conversation_loop import INTERRUPT_WAITING_FOR_MODEL_PREFIX
 from agent.i18n import t
 from hermes_cli.config import cfg_get
@@ -2944,6 +2944,16 @@ async def _dispose_unused_adapter(adapter: "BasePlatformAdapter | None") -> None
         )
 
 
+# Max seconds between platform reconnect retries (primary watcher and
+# secondary-profile reconnects share this policy — tune in one place).
+_RECONNECT_BACKOFF_CAP = 300
+
+
+def _reconnect_backoff(attempt: int) -> int:
+    """Exponential reconnect backoff: 30s, 60s, 120s, ... capped at 5 min."""
+    return min(30 * (2 ** (attempt - 1)), _RECONNECT_BACKOFF_CAP)
+
+
 class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, GatewaySlashCommandsMixin):
     """
     Main gateway controller.
@@ -2969,6 +2979,8 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
     _restart_command_source: Optional[SessionSource] = None
     _stop_task: Optional[asyncio.Task] = None
     _restart_task: Optional[asyncio.Task] = None
+    _profile_failed_platforms: Optional[Dict[str, Dict[Platform, asyncio.Task]]] = None
+    _systemd_watchdog: Optional[Any] = None
     _session_model_overrides: Dict[str, Dict[str, str]] = {}
     _session_reasoning_overrides: Dict[str, Dict[str, Any]] = {}
     _startup_restore_in_progress: bool = False
@@ -3043,6 +3055,8 @@ def __init__(self, config: Optional[GatewayConfig] = None):
         self._exit_reason: Optional[str] = None
         self._exit_code: Optional[int] = None
         self._draining = False
+        self._profile_failed_platforms: Dict[str, Dict[Platform, asyncio.Task]] = {}
+        self._systemd_watchdog = None
         # External (NAS-driven) drain state — distinct from the shutdown
         # ``_draining`` flag above. Set by ``_drain_control_watcher`` when the
         # ``.drain_request.json`` marker is present: the gateway flips
@@ -3518,6 +3532,35 @@ def _sync_voice_mode_state_to_adapter(self, adapter) -> None:
                 if mode in {"voice_only", "all"} and key.startswith(prefix)
             )
 
+    async def _await_adapter_cleanup_with_timeout(
+        self, awaitable: Awaitable[Any], timeout: float
+    ) -> bool:
+        """Wait for adapter cleanup without letting cancellation swallowing hang us.
+
+        ``asyncio.wait_for`` cancels an overdue child but then waits for it to
+        exit. An adapter close path that catches ``CancelledError`` can therefore
+        block recovery forever. Keep ownership of the old task through its done
+        callback, but release the runner at the deadline.
+        """
+        if timeout <= 0:
+            await awaitable
+            return True
+
+        task = asyncio.ensure_future(awaitable)
+        try:
+            done, _pending = await asyncio.wait({task}, timeout=timeout)
+        except asyncio.CancelledError:
+            task.cancel()
+            task.add_done_callback(consume_detached_task_result)
+            raise
+        if task in done:
+            await task
+            return True
+
+        task.cancel()
+        task.add_done_callback(consume_detached_task_result)
+        return False
+
     async def _safe_adapter_disconnect(self, adapter, platform) -> None:
         """Call adapter.disconnect() defensively, swallowing any error.
 
@@ -3531,16 +3574,15 @@ async def _safe_adapter_disconnect(self, adapter, platform) -> None:
         """
         timeout = self._adapter_disconnect_timeout_secs()
         try:
-            if timeout <= 0:
-                await adapter.disconnect()
-            else:
-                await asyncio.wait_for(adapter.disconnect(), timeout=timeout)
-        except asyncio.TimeoutError:
-            logger.warning(
-                "Timed out after %.1fs while disconnecting %s adapter; continuing shutdown",
-                timeout,
-                platform.value if platform is not None else "adapter",
+            completed = await self._await_adapter_cleanup_with_timeout(
+                adapter.disconnect(), timeout
             )
+            if not completed:
+                logger.warning(
+                    "Timed out after %.1fs while disconnecting %s adapter; continuing shutdown",
+                    timeout,
+                    platform.value if platform is not None else "adapter",
+                )
         except Exception as e:
             logger.debug(
                 "Defensive %s disconnect after failed connect raised: %s",
@@ -3560,42 +3602,40 @@ async def _bounded_adapter_teardown(
         ``TimeoutStopSec``; the resulting SIGKILL skips ``atexit`` PID-file
         cleanup, so the next start dies with "PID file race lost" (#14128).
 
-        Each await is wrapped in the existing per-adapter timeout budget
-        (``HERMES_GATEWAY_ADAPTER_DISCONNECT_TIMEOUT``). On timeout we log
-        and force forward progress; the loop never hangs regardless of any
-        adapter's internal behavior. Never raises.
+        Each await uses the existing per-adapter timeout budget
+        (``HERMES_GATEWAY_ADAPTER_DISCONNECT_TIMEOUT``). On timeout the old
+        task is cancelled and detached, then teardown forces forward progress;
+        the loop never hangs even if an adapter swallows cancellation. Never
+        raises.
         """
         timeout = self._adapter_disconnect_timeout_secs()
         suffix = f" (profile: {profile})" if profile else ""
         started_at = time.monotonic()
         try:
-            if timeout <= 0:
-                await adapter.cancel_background_tasks()
-            else:
-                await asyncio.wait_for(
-                    adapter.cancel_background_tasks(), timeout=timeout
-                )
-        except asyncio.TimeoutError:
-            logger.warning(
-                "✗ %s background-task cancel timed out after %.1fs - forcing continue%s",
-                platform.value, timeout, suffix,
+            cancelled = await self._await_adapter_cleanup_with_timeout(
+                adapter.cancel_background_tasks(), timeout
             )
+            if not cancelled:
+                logger.warning(
+                    "✗ %s background-task cancel timed out after %.1fs - forcing continue%s",
+                    platform.value, timeout, suffix,
+                )
         except Exception as e:
             logger.debug("✗ %s background-task cancel error%s: %s", platform.value, suffix, e)
         try:
-            if timeout <= 0:
-                await adapter.disconnect()
-            else:
-                await asyncio.wait_for(adapter.disconnect(), timeout=timeout)
-            logger.info(
-                "✓ %s disconnected (%.2fs)%s",
-                platform.value, time.monotonic() - started_at, suffix,
-            )
-        except asyncio.TimeoutError:
-            logger.warning(
-                "✗ %s disconnect timed out after %.1fs - forcing continue%s",
-                platform.value, timeout, suffix,
+            disconnected = await self._await_adapter_cleanup_with_timeout(
+                adapter.disconnect(), timeout
             )
+            if disconnected:
+                logger.info(
+                    "✓ %s disconnected (%.2fs)%s",
+                    platform.value, time.monotonic() - started_at, suffix,
+                )
+            else:
+                logger.warning(
+                    "✗ %s disconnect timed out after %.1fs - forcing continue%s",
+                    platform.value, timeout, suffix,
+                )
         except Exception as e:
             logger.error(
                 "✗ %s disconnect error after %.2fs%s: %s",
@@ -4220,7 +4260,10 @@ async def _handle_adapter_fatal_error(self, adapter: BasePlatformAdapter) -> Non
             # the same object twice.
             self.adapters.pop(adapter.platform, None)
             self.delivery_router.adapters = self.adapters
-            await adapter.disconnect()
+            # A half-closed transport can wedge an adapter's native close()
+            # indefinitely. Reuse the shutdown-path timeout so this runtime
+            # fatal handler always reaches the reconnect queue.
+            await self._safe_adapter_disconnect(adapter, adapter.platform)
 
         # Queue retryable failures for background reconnection
         if adapter.fatal_error_retryable:
@@ -8138,8 +8181,6 @@ async def _platform_reconnect_watcher(self) -> None:
         automatically — auto-pausing a recovered platform was the cause of
         bots silently staying dead after a transient DNS failure.
         """
-        _BACKOFF_CAP = 300  # 5 minutes max between retries
-
         await asyncio.sleep(10)  # initial delay — let startup finish
         while self._running:
             if not self._failed_platforms:
@@ -8270,7 +8311,7 @@ async def _platform_reconnect_watcher(self) -> None:
                             error_code=adapter.fatal_error_code,
                             error_message=adapter.fatal_error_message or "failed to reconnect",
                         )
-                        backoff = min(30 * (2 ** (attempt - 1)), _BACKOFF_CAP)
+                        backoff = _reconnect_backoff(attempt)
                         info["attempts"] = attempt
                         info["next_retry"] = time.monotonic() + backoff
                         logger.info(
@@ -8308,7 +8349,7 @@ async def _platform_reconnect_watcher(self) -> None:
                         error_code=None,
                         error_message=str(e),
                     )
-                    backoff = min(30 * (2 ** (attempt - 1)), _BACKOFF_CAP)
+                    backoff = _reconnect_backoff(attempt)
                     info["attempts"] = attempt
                     info["next_retry"] = time.monotonic() + backoff
                     logger.warning(
@@ -8325,6 +8366,62 @@ async def _platform_reconnect_watcher(self) -> None:
                     return
                 await asyncio.sleep(1)
 
+    async def _cancel_secondary_profile_reconnect_tasks(self) -> None:
+        """Cancel profile-scoped reconnects before tearing down their registry.
+
+        A reconnect can be waiting in adapter setup while shutdown begins. It
+        must not republish an adapter after the secondary registry is drained.
+        Waiting is bounded by the same adapter-cleanup budget; if a task does
+        not finish in time, the stopped runner state still prevents it from
+        installing an adapter when it eventually resumes.
+        """
+        pending = self._profile_failed_platforms
+        if not isinstance(pending, dict):
+            return
+        current = asyncio.current_task()
+        tasks: list[asyncio.Task] = []
+        for profile_pending in pending.values():
+            if not isinstance(profile_pending, dict):
+                continue
+            for task in profile_pending.values():
+                if isinstance(task, asyncio.Task) and task is not current and not task.done():
+                    tasks.append(task)
+        for task in tasks:
+            task.cancel()
+        timeout = self._adapter_disconnect_timeout_secs()
+        if tasks and timeout > 0:
+            _done, unfinished = await asyncio.wait(tasks, timeout=timeout)
+            if unfinished:
+                logger.warning(
+                    "Timed out waiting for %d secondary profile reconnect task(s) during shutdown",
+                    len(unfinished),
+                )
+        pending.clear()
+
+    def _start_systemd_watchdog(self) -> bool:
+        """Start sd_notify only after a configured gateway is truly running."""
+        if not self._running or self.config.systemd_watchdog_seconds <= 0:
+            return False
+        if self._systemd_watchdog is not None:
+            return True
+
+        from gateway.systemd_notify import SystemdWatchdog
+
+        watchdog = SystemdWatchdog(config_enabled=True)
+        if not watchdog.start():
+            return False
+        self._systemd_watchdog = watchdog
+        watchdog.ready("Hermes Gateway running")
+        return True
+
+    async def _stop_systemd_watchdog(self) -> None:
+        """Stop heartbeats before any potentially long shutdown drain."""
+        watchdog = self._systemd_watchdog
+        if watchdog is None:
+            return
+        self._systemd_watchdog = None
+        await watchdog.stop()
+
     async def stop(
         self,
         *,
@@ -8419,6 +8516,12 @@ def _phase_elapsed() -> float:
             self._running = False
             self._draining = True
 
+            stop_watchdog = getattr(self, "_stop_systemd_watchdog", None)
+            if callable(stop_watchdog):
+                await stop_watchdog()
+
+            await self._cancel_secondary_profile_reconnect_tasks()
+
             # Notify all chats with active agents BEFORE draining.
             # Adapters are still connected here, so messages can be sent.
             await self._notify_active_sessions_of_shutdown()
@@ -8931,19 +9034,7 @@ async def _start_one_profile_adapters(
                     continue
                 claimed[(platform, fp)] = profile_name
 
-            # Stamp every inbound event from this adapter with its profile so
-            # the agent turn (and session key) resolve to the right home.
-            adapter.set_message_handler(
-                self._make_profile_message_handler(profile_name)
-            )
-            adapter.set_fatal_error_handler(self._handle_adapter_fatal_error)
-            adapter.set_session_store(self.session_store)
-            adapter.set_busy_session_handler(self._handle_active_session_busy_message)
-            adapter.set_topic_recovery_fn(self._recover_telegram_topic_thread_id)
-            adapter.set_authorization_check(
-                self._make_adapter_auth_check(adapter.platform, profile_name=profile_name)
-            )
-            adapter._busy_text_mode = self._busy_text_mode
+            self._configure_profile_adapter(adapter, profile_name, platform)
 
             try:
                 with _profile_runtime_scope(profile_home):
@@ -8960,6 +9051,192 @@ async def _start_one_profile_adapters(
                 await self._safe_adapter_disconnect(adapter, platform)
         return connected
 
+    def _configure_profile_adapter(
+        self,
+        adapter: BasePlatformAdapter,
+        profile_name: str,
+        platform: Platform,
+    ) -> None:
+        """Install the profile-scoped handlers shared by startup and reconnect."""
+        adapter.set_message_handler(self._make_profile_message_handler(profile_name))
+        adapter.set_fatal_error_handler(
+            self._make_profile_fatal_error_handler(profile_name, platform)
+        )
+        adapter.set_session_store(self.session_store)
+        adapter.set_busy_session_handler(self._handle_active_session_busy_message)
+        adapter.set_topic_recovery_fn(self._recover_telegram_topic_thread_id)
+        adapter.set_authorization_check(
+            self._make_adapter_auth_check(platform, profile_name=profile_name)
+        )
+        adapter._busy_text_mode = self._busy_text_mode
+
+    async def _run_secondary_profile_reconnect(
+        self, profile_name: str, platform: Platform
+    ) -> None:
+        """Reconnect a retryable secondary adapter under its own profile scope."""
+        attempts = 0
+        current_task = asyncio.current_task()
+        try:
+            while self._running:
+                adapter = None
+                try:
+                    from hermes_cli.profiles import get_profile_dir
+                    from gateway.config import load_gateway_config
+
+                    profile_home = get_profile_dir(profile_name)
+                    with _profile_runtime_scope(profile_home):
+                        profile_config = load_gateway_config().platforms.get(platform)
+                        if profile_config is None or not profile_config.enabled:
+                            return
+                        adapter = self._create_adapter(platform, profile_config)
+                        if adapter is None:
+                            logger.warning(
+                                "Secondary %s reconnect skipped: adapter unavailable (profile: %s)",
+                                platform.value,
+                                profile_name,
+                            )
+                            return
+                        self._configure_profile_adapter(
+                            adapter, profile_name, platform
+                        )
+                        success = await self._connect_adapter_with_timeout(
+                            adapter, platform, is_reconnect=True
+                        )
+
+                    if success and self._running:
+                        profile_map = self._profile_adapters.setdefault(profile_name, {})
+                        if platform not in profile_map:
+                            profile_map[platform] = adapter
+                            self._sync_voice_mode_state_to_adapter(adapter)
+                            logger.info(
+                                "✓ %s reconnected (profile: %s)",
+                                platform.value,
+                                profile_name,
+                            )
+                            return
+                        # A newer reconnect already won the slot while this
+                        # attempt was awaiting connect; do not replace it.
+                        await self._safe_adapter_disconnect(adapter, platform)
+                        return
+
+                    # Shutdown can begin while connect() is in flight. Do not
+                    # republish a newly connected adapter after the registry has
+                    # been drained; release its partial resources instead.
+                    if success:
+                        await self._safe_adapter_disconnect(adapter, platform)
+                        return
+
+                    await self._safe_adapter_disconnect(adapter, platform)
+                    if (
+                        getattr(adapter, "has_fatal_error", False)
+                        and not getattr(adapter, "fatal_error_retryable", True)
+                    ):
+                        return
+                except asyncio.CancelledError:
+                    if adapter is not None:
+                        await self._safe_adapter_disconnect(adapter, platform)
+                    raise
+                except Exception:
+                    if adapter is not None:
+                        await self._safe_adapter_disconnect(adapter, platform)
+                    logger.debug(
+                        "Secondary %s reconnect attempt failed (profile: %s)",
+                        platform.value,
+                        profile_name,
+                        exc_info=True,
+                    )
+
+                if not self._running:
+                    return
+                attempts += 1
+                backoff = _reconnect_backoff(attempts)
+                logger.info(
+                    "Secondary %s reconnect retry in %ds (profile: %s)",
+                    platform.value,
+                    backoff,
+                    profile_name,
+                )
+                await asyncio.sleep(backoff)
+        finally:
+            pending = self._profile_failed_platforms
+            if isinstance(pending, dict):
+                profile_pending = pending.get(profile_name)
+                task = profile_pending.get(platform) if isinstance(profile_pending, dict) else None
+                if not isinstance(task, asyncio.Task) or task is current_task:
+                    if isinstance(profile_pending, dict):
+                        profile_pending.pop(platform, None)
+                        if not profile_pending:
+                            pending.pop(profile_name, None)
+
+    def _schedule_secondary_profile_reconnect(
+        self, profile_name: str, platform: Platform, adapter: BasePlatformAdapter
+    ) -> None:
+        """Schedule one runner-owned reconnect without sharing primary secrets."""
+        if not self._running or not adapter.fatal_error_retryable:
+            return
+        pending = self._profile_failed_platforms
+        if not isinstance(pending, dict):
+            pending = {}
+            self._profile_failed_platforms = pending
+        profile_pending = pending.setdefault(profile_name, {})
+        if platform in profile_pending:
+            return
+        task = asyncio.create_task(
+            self._run_secondary_profile_reconnect(profile_name, platform),
+            name=f"secondary-reconnect:{profile_name}:{platform.value}",
+        )
+        profile_pending[platform] = task
+        background_tasks = getattr(self, "_background_tasks", None)
+        if not isinstance(background_tasks, set):
+            background_tasks = set()
+            self._background_tasks = background_tasks
+        background_tasks.add(task)
+        task.add_done_callback(background_tasks.discard)
+
+    def _make_profile_fatal_error_handler(
+        self, profile_name: str, platform: Platform
+    ) -> Callable[[BasePlatformAdapter], Awaitable[None]]:
+        """Route a secondary-profile fatal error to that profile's reconnect slot."""
+        async def _handler(adapter: BasePlatformAdapter) -> None:
+            await self._handle_profile_adapter_fatal_error(profile_name, platform, adapter)
+
+        return _handler
+
+    async def _handle_profile_adapter_fatal_error(
+        self,
+        profile_name: str,
+        platform: Platform,
+        adapter: BasePlatformAdapter,
+    ) -> None:
+        """Remove a failed multiplexed adapter without touching the primary slot.
+
+        Secondary adapters are owned by ``_profile_adapters`` rather than
+        ``self.adapters``. The primary-only fatal handler intentionally ignores
+        them; without this route, a fatal secondary Discord client stayed live
+        forever after its liveness sampler stopped.
+        """
+        profile_map = getattr(self, "_profile_adapters", {}).get(profile_name)
+        if not isinstance(profile_map, dict) or profile_map.get(platform) is not adapter:
+            logger.debug(
+                "Ignoring stale fatal error from secondary %s adapter (profile: %s)",
+                platform.value,
+                profile_name,
+            )
+            return
+        profile_map.pop(platform, None)
+        await self._safe_adapter_disconnect(adapter, platform)
+        if not self._running:
+            return
+        self._schedule_secondary_profile_reconnect(profile_name, platform, adapter)
+        logger.error(
+            "Fatal %s adapter error for multiplexed profile %s (%s)",
+            platform.value,
+            profile_name,
+            adapter.fatal_error_code or "unknown",
+        )
+        # Reconnect is scoped to the profile's own config and secret mapping;
+        # never rebuild a secondary adapter with the default profile's credentials.
+
     def _make_profile_message_handler(self, profile_name: str):
         """Return a message handler that stamps source.profile then delegates.
 
@@ -21661,7 +21938,7 @@ def restart_signal_handler():
         if runner.exit_code is not None:
             raise SystemExit(runner.exit_code)
         return True
-    
+
     # Start the background cron scheduler via the resolved provider so
     # scheduled jobs fire automatically. The built-in provider is the
     # historical in-process 60s ticker; an external provider (e.g. chronos)
@@ -21698,7 +21975,14 @@ def restart_signal_handler():
         name="gateway-housekeeping",
     )
     housekeeping_thread.start()
-    
+
+    # READY is emitted only after adapters, cron, and housekeeping have all
+    # reached their running boundary. Missing config/systemd runtime state
+    # leaves the watchdog disabled without changing gateway behavior.
+    start_watchdog = getattr(runner, "_start_systemd_watchdog", None)
+    if callable(start_watchdog):
+        start_watchdog()
+
     # Wait for shutdown
     await runner.wait_for_shutdown()

gateway/systemd_notify.py

@@ -0,0 +1,176 @@
+"""Minimal, optional systemd ``sd_notify`` support for the gateway."""
+
+from __future__ import annotations
+
+import asyncio
+import math
+import os
+import socket
+from typing import Optional
+
+
+def _notify_address(raw: str) -> str:
+    """Translate systemd's ``@abstract`` notation to Python's address form."""
+    return "\0" + raw[1:] if raw.startswith("@") else raw
+
+
+def notify(message: str) -> bool:
+    """Send one nonblocking sd_notify datagram when systemd configured it.
+
+    Notification failures are deliberately non-fatal: a missing socket or an
+    older platform must never prevent the gateway from starting.
+    """
+    address = os.environ.get("NOTIFY_SOCKET", "").strip()
+    if not address:
+        return False
+    if not isinstance(message, str) or not message:
+        return False
+    if not hasattr(socket, "AF_UNIX"):
+        return False
+    try:
+        payload = message.encode("utf-8")
+        with socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM) as sender:
+            # A full receiver buffer must not stall the gateway event loop.
+            sender.setblocking(False)
+            sender.connect(_notify_address(address))
+            sender.send(payload)
+        return True
+    except (OSError, UnicodeError, ValueError):
+        return False
+
+
+def watchdog_interval_seconds() -> Optional[float]:
+    """Return systemd's configured watchdog interval in seconds."""
+    if not os.environ.get("NOTIFY_SOCKET", "").strip():
+        return None
+    if not hasattr(socket, "AF_UNIX"):
+        return None
+    raw = os.environ.get("WATCHDOG_USEC", "").strip()
+    if not raw:
+        return None
+    try:
+        interval = float(raw) / 1_000_000.0
+    except (TypeError, ValueError):
+        return None
+    if not math.isfinite(interval) or interval <= 0:
+        return None
+    return interval
+
+
+class SystemdWatchdog:
+    """Feed systemd while the asyncio event loop continues to make progress."""
+
+    def __init__(
+        self,
+        *,
+        config_enabled: bool = True,
+        lag_tolerance_seconds: Optional[float] = None,
+    ) -> None:
+        self._config_enabled = bool(config_enabled)
+        self.interval_seconds = watchdog_interval_seconds()
+        self._lag_tolerance_seconds = lag_tolerance_seconds
+        self._task: Optional[asyncio.Task[None]] = None
+        self._unhealthy = False
+        self._stopping = False
+        self._stopping_notified = False
+
+    @property
+    def enabled(self) -> bool:
+        return self._config_enabled and self.interval_seconds is not None
+
+    @property
+    def unhealthy(self) -> bool:
+        return self._unhealthy
+
+    @property
+    def task(self) -> Optional[asyncio.Task[None]]:
+        return self._task
+
+    def _lag_tolerance(self) -> float:
+        interval = self.interval_seconds or 0.0
+        configured = self._lag_tolerance_seconds
+        if configured is None:
+            return max(0.1, interval * 0.25)
+        try:
+            value = float(configured)
+        except (TypeError, ValueError):
+            return max(0.1, interval * 0.25)
+        if not math.isfinite(value):
+            return max(0.1, interval * 0.25)
+        return max(0.0, value)
+
+    def start(self) -> bool:
+        """Start the loop-progress sampler when systemd watchdog is enabled."""
+        if not self.enabled:
+            return False
+        if self._task is not None and not self._task.done():
+            return True
+        try:
+            asyncio.get_running_loop()
+        except RuntimeError:
+            return False
+        self._stopping = False
+        self._unhealthy = False
+        self._stopping_notified = False
+        self._task = asyncio.create_task(self._run(), name="hermes-systemd-watchdog")
+        return True
+
+    def ready(self, status: str = "Gateway running") -> bool:
+        """Tell systemd that startup completed and the gateway is ready."""
+        if not self.enabled:
+            return False
+        safe_status = str(status or "Gateway running").replace("\n", " ")
+        return notify(f"READY=1\nSTATUS={safe_status}")
+
+    def record_tick(self, *, scheduled_at: float, now: float) -> bool:
+        """Feed systemd only when the event loop woke within its lag budget."""
+        if not self.enabled or self._stopping or self._unhealthy:
+            return False
+        try:
+            lag = float(now) - float(scheduled_at)
+        except (TypeError, ValueError):
+            lag = float("inf")
+        if not math.isfinite(lag) or lag > self._lag_tolerance():
+            self._unhealthy = True
+            notify("STATUS=watchdog unhealthy: event loop progress is late")
+            return False
+        notify("WATCHDOG=1")
+        return True
+
+    async def _run(self) -> None:
+        interval = self.interval_seconds
+        if interval is None:
+            return
+        cadence = max(0.01, interval / 2.0)
+        loop = asyncio.get_running_loop()
+        scheduled_at = loop.time() + cadence
+        try:
+            while not self._stopping and not self._unhealthy:
+                await asyncio.sleep(max(0.0, scheduled_at - loop.time()))
+                now = loop.time()
+                if not self.record_tick(scheduled_at=scheduled_at, now=now):
+                    return
+                scheduled_at += cadence
+                if scheduled_at < now:
+                    scheduled_at = now + cadence
+        except asyncio.CancelledError:
+            return
+
+    async def stop(self) -> None:
+        """Stop feeding systemd and emit ``STOPPING=1`` at most once."""
+        self._stopping = True
+        task = self._task
+        current = asyncio.current_task()
+        if task is not None and task is not current:
+            if not task.done():
+                task.cancel()
+            try:
+                await task
+            except asyncio.CancelledError:
+                pass
+            except Exception:
+                pass
+        self._task = None
+        if self.enabled and not self._stopping_notified:
+            notify("STOPPING=1")
+            self._stopping_notified = True

hermes_cli/config.py

@@ -2525,6 +2525,14 @@ def _ensure_hermes_home_managed(home: Path):
         "history_backfill": True,         # If True, prepend recent channel scrollback when bot is triggered (recovers messages missed while require_mention gated them out)
         "history_backfill_limit": 50,     # Max number of recent messages to scan when assembling the backfill block
         "reactions": True,             # Add 👀/✅/❌ reactions to messages during processing
+        # Discord Gateway transport health. These settings inspect the active
+        # WebSocket's ready/open/heartbeat state; they never use Discord REST as
+        # proof that Gateway events are still arriving. Set any value to 0 to
+        # disable this compatibility-safe probe during a rollback.
+        "websocket_liveness_interval_seconds": 15,
+        "websocket_liveness_failure_threshold": 2,
+        "websocket_heartbeat_ack_max_age_seconds": 60,
+        "websocket_max_latency_seconds": 30,
         "channel_prompts": {},         # Per-channel ephemeral system prompts (forum parents apply to child threads)
         # Opt-in DM role-based auth (#12136). By default, DISCORD_ALLOWED_ROLES
         # authorizes only guild messages in the role's own guild — DMs require

hermes_cli/gateway.py

@@ -29,6 +29,7 @@
 
 PROJECT_ROOT = Path(__file__).parent.parent.resolve()
 
+from gateway.config import coerce_systemd_watchdog_seconds, load_gateway_config
 from gateway.status import terminate_pid
 from gateway.restart import (
     DEFAULT_GATEWAY_RESTART_DRAIN_TIMEOUT,
@@ -2624,8 +2625,16 @@ def _hermes_home_for_target_user(target_home_dir: str) -> str:
       /root/.hermes/profiles/coder     → /home/alice/.hermes/profiles/coder
       /opt/custom-hermes               → /opt/custom-hermes  (kept as-is)
     """
-    current_hermes = get_hermes_home().resolve()
-    current_default = (Path.home() / ".hermes").resolve()
+    current_hermes_raw = os.environ.get("HERMES_HOME", "").strip()
+    current_hermes = (
+        Path(current_hermes_raw).expanduser()
+        if current_hermes_raw
+        else get_hermes_home()
+    )
+    # Keep explicit custom paths lexical. Resolving a non-existent custom path
+    # can rewrite it through host-specific path mappings, which would bake a
+    # different HERMES_HOME into the generated service unit.
+    current_default = Path.home() / ".hermes"
     target_default = Path(target_home_dir) / ".hermes"
 
     # Default ~/.hermes → remap to target user's default
@@ -2703,6 +2712,44 @@ def _stable_service_working_dir() -> str:
     return str(PROJECT_ROOT)
 
 
+def _systemd_watchdog_seconds(hermes_home: str | Path | None = None) -> int:
+    """Resolve the managed-overlay-aware watchdog setting for a service home."""
+    override_token = None
+    reset_home_override = None
+    if hermes_home is not None:
+        from hermes_constants import (
+            reset_hermes_home_override,
+            set_hermes_home_override,
+        )
+
+        override_token = [redacted:secret-assignment]
+        reset_home_override = reset_hermes_home_override
+    try:
+        config = load_gateway_config()
+        return coerce_systemd_watchdog_seconds(
+            getattr(config, "systemd_watchdog_seconds", 0)
+        )
+    except Exception:
+        logger.debug(
+            "Could not resolve effective systemd watchdog configuration",
+            exc_info=True,
+        )
+        return 0
+    finally:
+        if override_token is not None and reset_home_override is not None:
+            reset_home_override(override_token)
+
+
+def _systemd_watchdog_service_fields(
+    hermes_home: str | Path | None = None,
+) -> tuple[str, str]:
+    """Return systemd service fields for the effective gateway config."""
+    seconds = _systemd_watchdog_seconds(hermes_home)
+    if seconds <= 0:
+        return "simple", ""
+    return "notify", f"NotifyAccess=main\nWatchdogSec={seconds}s\n"
+
+
 def generate_systemd_unit(system: bool = False, run_as_user: str | None = None) -> str:
     python_path = get_python_path()
     working_dir = _stable_service_working_dir()
@@ -2732,18 +2779,19 @@ def generate_systemd_unit(system: bool = False, run_as_user: str | None = None)
         "/sbin",
         "/bin",
     ]
-    # systemd's TimeoutStopSec must exceed the gateway's drain_timeout so
-    # there's budget left for post-interrupt cleanup (tool subprocess kill,
-    # adapter disconnect, session DB close) before systemd escalates to
-    # SIGKILL on the cgroup — otherwise bash/sleep tool-call children left
-    # by a force-interrupted agent get reaped by systemd instead of us
-    # (#8202). 30s of headroom covers the worst case we've observed.
+    # Preserve 30s for post-drain cleanup before systemd escalates, with a
+    # 60s minimum for installs that use the default immediate drain. Positive
+    # drain values extend the deadline directly instead of inheriting a second
+    # 60s floor, so a configured 45s drain yields 75s rather than 90s.
     _drain_timeout = int(_get_restart_drain_timeout() or 0)
-    restart_timeout = max(60, _drain_timeout) + 30
+    restart_timeout = max(60, _drain_timeout + 30)
 
     if system:
         username, group_name, home_dir = _system_service_identity(run_as_user)
         hermes_home = _hermes_home_for_target_user(home_dir)
+        systemd_type, systemd_watchdog_directives = _systemd_watchdog_service_fields(
+            hermes_home
+        )
         profile_arg = _profile_arg_for_target_user(hermes_home, home_dir)
         # Remap all paths that may resolve under the calling user's home
         # (e.g. /root/) to the target user's home so the service can
@@ -2766,8 +2814,8 @@ def generate_systemd_unit(system: bool = False, run_as_user: str | None = None)
 StartLimitIntervalSec=0
 
 [Service]
-Type=simple
-User={username}
+Type={systemd_type}
+{systemd_watchdog_directives}User={username}
 Group={group_name}
 ExecStart={python_path} -m hermes_cli.main{f" {profile_arg}" if profile_arg else ""} gateway run
 WorkingDirectory={working_dir}
@@ -2794,6 +2842,9 @@ def generate_systemd_unit(system: bool = False, run_as_user: str | None = None)
 """
 
     hermes_home = str(get_hermes_home().resolve())
+    systemd_type, systemd_watchdog_directives = _systemd_watchdog_service_fields(
+        hermes_home
+    )
     profile_arg = _profile_arg(hermes_home)
     path_entries.extend(_build_user_local_paths(Path.home(), path_entries))
     path_entries.extend(_build_wsl_interop_paths(path_entries))
@@ -2806,8 +2857,8 @@ def generate_systemd_unit(system: bool = False, run_as_user: str | None = None)
 StartLimitIntervalSec=0
 
 [Service]
-Type=simple
-ExecStart={python_path} -m hermes_cli.main{f" {profile_arg}" if profile_arg else ""} gateway run
+Type={systemd_type}
+{systemd_watchdog_directives}ExecStart={python_path} -m hermes_cli.main{f" {profile_arg}" if profile_arg else ""} gateway run
 WorkingDirectory={working_dir}
 Environment="PATH={sane_path}"
 Environment="VIRTUAL_ENV={venv_dir}"

plugins/platforms/discord/adapter.py

@@ -14,6 +14,7 @@
 import inspect
 import json
 import logging
+import math
 import os
 import re
 import struct
@@ -25,6 +26,10 @@
 from contextlib import suppress
 from typing import Callable, Dict, List, Optional, Any, Tuple
 
+from agent.async_utils import (
+    consume_detached_task_result as _consume_background_task_result,
+)
+
 logger = logging.getLogger(__name__)
 
 
@@ -133,6 +138,25 @@ def _truncate_discord_component_text(text: str, limit: int) -> str:
     return _prefix_within_utf16_limit(str(text or ""), max(0, limit))
 
 
+def _abort_discord_websocket_transport(websocket: Any) -> bool:
+    """Abort the active aiohttp transport after a bounded close times out."""
+    socket = getattr(websocket, "socket", None)
+    response = getattr(socket, "_response", None)
+    connection = getattr(socket, "_conn", None)
+    if connection is None:
+        connection = getattr(response, "connection", None)
+    protocol = getattr(connection, "protocol", None)
+    writer = getattr(socket, "_writer", None)
+    transport = getattr(writer, "transport", None)
+    if transport is None:
+        transport = getattr(protocol, "transport", None)
+    abort = getattr(transport, "abort", None)
+    if not callable(abort):
+        return False
+    abort()
+    return True
+
+
 async def _wait_for_ready_or_bot_exit(
     ready_event: asyncio.Event,
     bot_task: asyncio.Task,
@@ -858,23 +882,33 @@ 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
+        # WebSocket-level liveness probe. Discord REST and Gateway are distinct
+        # transports: a REST 200 cannot prove that this client is still receiving
+        # Gateway events. Sample the current Discord WebSocket's ready/open/ACK
+        # state and heartbeat latency instead; after consecutive unhealthy samples
+        # use the existing retryable-fatal path so GatewayRunner rebuilds a fresh
+        # adapter. The values are compatibility inputs from config; zero disables
+        # the probe without changing the rest of the adapter lifecycle.
+        self._liveness_interval_seconds = self._finite_positive_config_float(
+            "websocket_liveness_interval_seconds",
+            15.0,
+            env_key="HERMES_DISCORD_LIVENESS_INTERVAL_SECONDS",
+        )
+        self._liveness_failure_threshold = self._config_int(
+            "websocket_liveness_failure_threshold",
+            2,
+            env_key="HERMES_DISCORD_LIVENESS_FAILURE_THRESHOLD",
+        )
+        self._heartbeat_ack_max_age_seconds = self._finite_positive_config_float(
+            "websocket_heartbeat_ack_max_age_seconds",
+            60.0,
         )
-        self._liveness_failure_threshold = env_int(
-            "HERMES_DISCORD_LIVENESS_FAILURE_THRESHOLD", 3
+        self._max_latency_seconds = self._finite_positive_config_float(
+            "websocket_max_latency_seconds",
+            30.0,
         )
         self._liveness_task: Optional[asyncio.Task] = None
+        self._liveness_notification_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.
@@ -902,6 +936,38 @@ def __init__(self, config: PlatformConfig):
         self._last_overflow_preview: Dict[tuple, str] = {}
         self._warned_fail_closed_default = False
 
+    def _config_value(
+        self, key: str, default: Any, *, env_key: Optional[str] = None
+    ) -> Any:
+        """Resolve a liveness value from profile config, legacy env, or default."""
+        extra = self.config.extra if isinstance(getattr(self.config, "extra", None), dict) else {}
+        value = extra.get(key)
+        if value is None and env_key:
+            value = os.getenv(env_key)
+        return default if value is None or value == "" else value
+
+    def _finite_positive_config_float(
+        self, key: str, default: float, *, env_key: Optional[str] = None
+    ) -> float:
+        """Resolve a finite positive liveness duration; invalid values disable it."""
+        try:
+            value = float(self._config_value(key, default, env_key=env_key))
+        except (TypeError, ValueError):
+            return 0.0
+        return value if math.isfinite(value) and value > 0 else 0.0
+
+    def _config_int(
+        self, key: str, default: int, *, env_key: Optional[str] = None
+    ) -> int:
+        """Resolve a positive liveness count; invalid values disable it."""
+        value = self._config_value(key, default, env_key=env_key)
+        if isinstance(value, bool):
+            return 0
+        try:
+            return int(value)
+        except (TypeError, ValueError):
+            return 0
+
     def _handle_bot_task_done(self, task: asyncio.Task) -> None:
         """Surface post-startup discord.py task exits to the gateway supervisor.
 
@@ -1289,90 +1355,189 @@ async def _cancel_bot_task(self) -> None:
         self._bot_task = None
 
     def _start_liveness_probe(self) -> None:
-        """Start the periodic REST liveness probe if configured.
+        """Start the periodic Discord Gateway WebSocket health probe.
 
-        Idempotent: if a task is already running we leave it alone so a
-        re-entrant ``connect()`` cannot fork two probes against the same client.
+        REST success does not prove Gateway event delivery. Sample the active
+        Gateway WebSocket's ready/open/ACK state instead.
         """
-        if self._liveness_interval_seconds <= 0 or self._liveness_failure_threshold <= 0:
+        if (
+            self._liveness_interval_seconds <= 0
+            or self._liveness_failure_threshold <= 0
+            or self._heartbeat_ack_max_age_seconds <= 0
+            or self._max_latency_seconds <= 0
+        ):
             return
         if self._liveness_task and not self._liveness_task.done():
             return
         self._liveness_task = asyncio.create_task(self._liveness_loop())
 
+    def _read_websocket_health(self, client: Any) -> tuple[bool, str]:
+        """Return current Discord Gateway health without making a REST request."""
+        try:
+            ready = bool(client.is_ready())
+        except Exception:
+            return False, "not_ready"
+        if not ready:
+            return False, "not_ready"
+
+        try:
+            if client.is_closed():
+                return False, "client_closed"
+        except Exception:
+            return False, "client_closed"
+
+        websocket = getattr(client, "ws", None)
+        try:
+            socket_open = bool(
+                websocket is not None and getattr(websocket, "open", False)
+            )
+        except Exception:
+            # A transport object that cannot report its open state is not a
+            # usable event stream. Treat it as unhealthy rather than letting
+            # the periodic liveness task crash silently.
+            return False, "socket_state_unavailable"
+        if not socket_open:
+            return False, "socket_closed"
+
+        keep_alive = getattr(websocket, "_keep_alive", None)
+        last_ack = getattr(keep_alive, "_last_ack", None)
+        if not isinstance(last_ack, (int, float)):
+            return False, "ack_unavailable"
+        ack_age = time.perf_counter() - last_ack
+        if not math.isfinite(ack_age) or ack_age > self._heartbeat_ack_max_age_seconds:
+            return False, "ack_stale"
+
+        latency = getattr(client, "latency", None)
+        if not isinstance(latency, (int, float)) or not math.isfinite(latency):
+            return False, "latency_non_finite"
+        if latency > self._max_latency_seconds:
+            return False, "latency_exceeded"
+        return True, "healthy"
+
     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.
-        """
+        """Force a reconnect after repeated unhealthy Discord Gateway samples."""
         interval = self._liveness_interval_seconds
         threshold = self._liveness_failure_threshold
-        fails = 0
+        failures = 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():
+            if not self._running or client is None or self._disconnecting:
                 return
-            user = getattr(client, "user", None)
-            if user is None:
+            try:
+                healthy, reason = self._read_websocket_health(client)
+            except Exception:
+                # Health sampling must fail closed: an unexpected discord.py
+                # attribute change cannot be allowed to kill this watchdog
+                # task and leave an apparently-running adapter unrecovered.
+                healthy = False
+                reason = "health_check_error"
+            if healthy:
+                failures = 0
                 continue
+
+            failures += 1
+            logger.warning(
+                "[%s] Discord Gateway WebSocket unhealthy (%s, %d/%d)",
+                self.name,
+                reason,
+                failures,
+                threshold,
+            )
+            if failures < threshold:
+                continue
+            # Mark intentional recovery before closing the client. Closing a
+            # healthy-looking but stale transport can complete Bot.start(); its
+            # done callback must not overwrite this more specific fatal reason.
+            self._disconnecting = True
+            logger.error(
+                "[%s] Discord Gateway WebSocket remained unhealthy (%s); forcing reconnect",
+                self.name,
+                reason,
+            )
+            self._set_fatal_error(
+                "discord_websocket_health_stale",
+                f"Discord Gateway WebSocket health check failed: {reason}",
+                retryable=True,
+            )
+            self._liveness_notification_task = asyncio.create_task(
+                self._notify_liveness_fatal_error(client)
+            )
+            return
+
+    async def _notify_liveness_fatal_error(self, client: Any) -> None:
+        """Close the failed client, then notify the runner outside the sampler.
+
+        The sampler must not await itself through ``disconnect()``. Running the
+        close and fatal callback in this sibling task also means the runner owns
+        the bounded full teardown before it creates a replacement adapter.
+        """
+        failed_websocket = getattr(client, "ws", None)
+        try:
+            close_task = asyncio.create_task(client.close())
             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,
-                )
+                done, _pending = await asyncio.wait({close_task}, timeout=1.0)
+                if close_task not in done:
+                    raise asyncio.TimeoutError
+                await close_task
+            except asyncio.TimeoutError:
+                logger.warning("[%s] Timed out closing unhealthy Discord client", self.name)
+                close_task.cancel()
+                close_task.add_done_callback(_consume_background_task_result)
+                closing_task = getattr(client, "_closing_task", None)
+                if isinstance(closing_task, asyncio.Task):
+                    closing_task.cancel()
+                    closing_task.add_done_callback(_consume_background_task_result)
+                    # discord.Client.close() caches this task. Clear the cache
+                    # before the runner's bounded disconnect makes another
+                    # cleanup attempt; the stale task remains owned by its
+                    # done callback until it actually exits.
+                    client._closing_task = None
                 try:
-                    await self._notify_fatal_error()
+                    if _abort_discord_websocket_transport(failed_websocket):
+                        logger.warning(
+                            "[%s] Aborted unresponsive Discord WebSocket transport",
+                            self.name,
+                        )
                 except Exception:
                     logger.debug(
-                        "[%s] Fatal-error handler raised", self.name, exc_info=True,
+                        "[%s] Error aborting unhealthy Discord WebSocket transport",
+                        self.name,
+                        exc_info=True,
                     )
-                return
+            except Exception:
+                logger.debug("[%s] Error closing unhealthy Discord client", self.name, exc_info=True)
+            # The runner's bounded teardown can execute ``disconnect()`` inside
+            # a timeout wrapper, which is a different task from this notifier.
+            # Drop the self-reference before notifying so disconnect() cannot
+            # cancel this in-flight fatal callback as though it were unrelated.
+            if self._liveness_notification_task is asyncio.current_task():
+                self._liveness_notification_task = None
+            await self._notify_fatal_error()
+        except Exception:
+            logger.debug("[%s] Fatal-error handler raised", self.name, exc_info=True)
 
     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()
+        """Cancel and await liveness tasks without awaiting the current task."""
+        current = asyncio.current_task()
+        for task_name in ("_liveness_task", "_liveness_notification_task"):
+            task = getattr(self, task_name, None)
+            if task is None:
+                continue
+            if task is current:
+                continue
+            if not task.done():
+                task.cancel()
             try:
-                await self._liveness_task
+                await task
             except asyncio.CancelledError:
                 pass
-        self._liveness_task = None
+            except Exception:
+                logger.debug("[%s] Liveness task shutdown failed", self.name, exc_info=True)
+            setattr(self, task_name, None)
 
     async def cancel_background_tasks(self) -> None:
         """Cancel background tasks, but first flush any pending text-batch sends.
@@ -8426,10 +8591,10 @@ def _apply_yaml_config(yaml_cfg: dict, discord_cfg: dict) -> dict | None:
     env-driven model and merely owns the YAML→env translation here, next to
     the adapter that consumes it.
 
-    Env vars take precedence over YAML — every assignment is guarded by
-    ``not os.getenv(...)`` so explicit env vars survive a config.yaml
-    update.  Returns ``None`` because no extras are seeded into
-    ``PlatformConfig.extra`` directly (everything flows through env).
+    ``PlatformConfig.extra`` is the per-adapter source of truth for liveness
+    settings, which keeps multiplexed profiles isolated. The legacy env bridge
+    remains only for existing callers that construct adapters without config
+    extras. Returns canonical WebSocket liveness settings to seed that extra.
     """
     if "require_mention" in discord_cfg and not os.getenv("DISCORD_REQUIRE_MENTION"):
         os.environ["DISCORD_REQUIRE_MENTION"] = str(discord_cfg["require_mention"]).lower()
@@ -8518,17 +8683,42 @@ 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
+    _websocket_extra_cfg = discord_cfg.get("extra")
+    if not isinstance(_websocket_extra_cfg, dict):
+        _websocket_extra_cfg = {}
+    # Public config keys win over the generic ``extra`` form used by nested
+    # platform configuration.
+    _websocket_liveness_cfg = {
+        **_websocket_extra_cfg,
+        **discord_cfg,
+    }
+    # WebSocket health knobs: REST 200 is deliberately not used as Gateway
+    # health. Accept legacy liveness_* aliases for compatibility during the
+    # migration; the websocket_* spelling is the public config surface.
+    _websocket_liveness_keys = (
+        (
+            "websocket_liveness_interval_seconds",
+            "liveness_interval_seconds",
+            "HERMES_DISCORD_LIVENESS_INTERVAL_SECONDS",
+        ),
+        (
+            "websocket_liveness_failure_threshold",
+            "liveness_failure_threshold",
+            "HERMES_DISCORD_LIVENESS_FAILURE_THRESHOLD",
+        ),
+        ("websocket_heartbeat_ack_max_age_seconds", None, None),
+        ("websocket_max_latency_seconds", None, None),
+    )
+    seeded = {}
+    for primary_key, legacy_key, env_key in _websocket_liveness_keys:
+        value = _websocket_liveness_cfg.get(primary_key)
+        if value is None and legacy_key:
+            value = _websocket_liveness_cfg.get(legacy_key)
+        if value is not None:
+            seeded[primary_key] = value
+            if env_key and not os.getenv(env_key):
+                os.environ[env_key] = str(value)
+    return seeded or None
 
 
 def _is_connected(config) -> bool:

scripts/release.py

@@ -51,6 +51,8 @@
 LEGACY_AUTHOR_MAP = {
     "[email protected]": "ragingbulld",  # PR #65606 salvage (non-finite API wait deadlines; #65746)
     "[email protected]": "zzpigpinggai",  # PR #66017 salvage of #63617 (OpenRouter explicit-provider picker visibility)
+    "[email protected]": "StellarisW",  # PR #66222 salvage (Discord WebSocket liveness + systemd watchdog; #26656 follow-up)
+    "[redacted:email]": "wxy-nlp",  # PR #66222 salvage (systemd event-loop watchdog co-author)
     "[redacted:email]": "sam7894604",  # PR #55803 salvage (discord: /reasoning slash choices)
     "[email protected]": "hydraxman",  # PR #62028 salvage (copilot xhigh) — regression-test commit authored under a bare-noreply local git identity; PR author is @hydraxman
     "[redacted:email]": "antydizajn",  # PR #36043 salvage (auxiliary: route custom:<name> through named-provider arm + Palantir Bearer auth)

tests/gateway/test_bounded_adapter_teardown.py

@@ -91,6 +91,48 @@ async def hang():
     adapter.disconnect.assert_awaited_once()
 
 
[redacted:email]
+async def test_teardown_continues_after_cancellation_swallowing_background_cancel(
+    bare_runner, monkeypatch, caplog
+):
+    """A stuck cancellation handler cannot prevent adapter disconnect.
+
+    This models a platform task that catches ``CancelledError`` while it is
+    unwinding.  The teardown deadline must release runner ownership promptly,
+    then proceed to disconnect instead of waiting for that old task forever.
+    """
+    monkeypatch.setenv("HERMES_GATEWAY_ADAPTER_DISCONNECT_TIMEOUT", "0.01")
+    adapter = MagicMock()
+    started = asyncio.Event()
+    release = asyncio.Event()
+    finished = asyncio.Event()
+
+    async def swallow_cancellation():
+        started.set()
+        while not release.is_set():
+            try:
+                await release.wait()
+            except asyncio.CancelledError:
+                continue
+        finished.set()
+
+    adapter.cancel_background_tasks = AsyncMock(side_effect=swallow_cancellation)
+    adapter.disconnect = AsyncMock(return_value=None)
+    operation = asyncio.create_task(
+        bare_runner._bounded_adapter_teardown(adapter, Platform.FEISHU)
+    )
+    await started.wait()
+    done, _pending = await asyncio.wait({operation}, timeout=0.2)
+    try:
+        assert operation in done
+        adapter.disconnect.assert_awaited_once()
+        assert "feishu background-task cancel timed out" in caplog.text
+    finally:
+        release.set()
+        await asyncio.wait({operation}, timeout=0.2)
+        await asyncio.wait_for(finished.wait(), timeout=0.2)
+
+
 @pytest.mark.asyncio
 async def test_teardown_swallows_exceptions(bare_runner):
     """Errors in either await must not propagate — shutdown continues."""

tests/gateway/test_config.py

@@ -4,6 +4,8 @@
 import os
 from unittest.mock import patch
 
+import pytest
+
 from agent.secret_scope import reset_secret_scope, set_secret_scope
 from hermes_constants import reset_hermes_home_override, set_hermes_home_override
 from gateway.config import (
@@ -304,6 +306,7 @@ def test_full_roundtrip(self):
             quick_commands={"limits": {"type": "exec", "command": "echo ok"}},
             group_sessions_per_user=False,
             thread_sessions_per_user=True,
+            systemd_watchdog_seconds=120,
         )
         d = config.to_dict()
         restored = GatewayConfig.from_dict(d)
@@ -314,6 +317,33 @@ def test_full_roundtrip(self):
         assert restored.quick_commands == {"limits": {"type": "exec", "command": "echo ok"}}
         assert restored.group_sessions_per_user is False
         assert restored.thread_sessions_per_user is True
+        assert restored.systemd_watchdog_seconds == 120
+
+    def test_systemd_watchdog_from_dict_disables_invalid_values(self):
+        invalid_values = [
+            None,
+            0,
+            -1,
+            True,
+            1.5,
+            float("nan"),
+            float("inf"),
+            "120.0",
+            "1e3",
+            "bad",
+            2_147_483_648,
+        ]
+
+        for raw in invalid_values:
+            config = GatewayConfig.from_dict({"systemd_watchdog_seconds": raw})
+            assert config.systemd_watchdog_seconds == 0
+
+    def test_systemd_watchdog_from_dict_accepts_nested_positive_integer(self):
+        config = GatewayConfig.from_dict(
+            {"gateway": {"systemd_watchdog_seconds": "45"}}
+        )
+
+        assert config.systemd_watchdog_seconds == 45
 
     def test_max_concurrent_sessions_from_dict_normalizes_disabled_values(self):
         assert GatewayConfig.from_dict({}).max_concurrent_sessions is None
@@ -478,6 +508,32 @@ def test_multiplex_profiles_from_nested_gateway_section(self, tmp_path, monkeypa
 
         assert config.multiplex_profiles is True
 
+    def test_discord_websocket_health_settings_seed_platform_extra(self, tmp_path, monkeypatch):
+        hermes_home = tmp_path / ".hermes"
+        hermes_home.mkdir()
+        (hermes_home / "config.yaml").write_text(
+            "discord:\n"
+            "  websocket_liveness_interval_seconds: 17\n"
+            "  websocket_liveness_failure_threshold: 4\n"
+            "  websocket_heartbeat_ack_max_age_seconds: 75\n"
+            "  websocket_max_latency_seconds: 30\n",
+            encoding="utf-8",
+        )
+        monkeypatch.setenv("HERMES_HOME", str(hermes_home))
+        for key in (
+            "HERMES_DISCORD_LIVENESS_INTERVAL_SECONDS",
+            "HERMES_DISCORD_LIVENESS_FAILURE_THRESHOLD",
+        ):
+            monkeypatch.delenv(key, raising=False)
+
+        config = load_gateway_config()
+
+        extra = config.platforms[Platform.DISCORD].extra
+        assert extra["websocket_liveness_interval_seconds"] == 17
+        assert extra["websocket_liveness_failure_threshold"] == 4
+        assert extra["websocket_heartbeat_ack_max_age_seconds"] == 75
+        assert extra["websocket_max_latency_seconds"] == 30
+
     def test_relay_platform_enabled_from_env_url(self, tmp_path, monkeypatch):
         """GATEWAY_RELAY_URL must enable Platform.RELAY in config.platforms so
         start_gateway()'s connect loop actually dials the connector. Registering

tests/gateway/test_discord_approval_mentions.py

@@ -83,5 +83,54 @@ def test_yaml_config_bridges_approval_mentions_to_env(monkeypatch):
         {"discord": {"approval_mentions": True}},
         {"approval_mentions": True},
     )
-
     assert os.environ["DISCORD_APPROVAL_MENTIONS"] == "true"
+
+
+def test_yaml_config_seeds_websocket_health_with_primary_precedence(monkeypatch):
+    for key in (
+        "HERMES_DISCORD_LIVENESS_INTERVAL_SECONDS",
+        "HERMES_DISCORD_LIVENESS_FAILURE_THRESHOLD",
+    ):
+        monkeypatch.delenv(key, raising=False)
+
+    seeded = _apply_yaml_config(
+        {},
+        {
+            "websocket_liveness_interval_seconds": 11,
+            "liveness_interval_seconds": 99,
+            "websocket_liveness_failure_threshold": 2,
+            "websocket_heartbeat_ack_max_age_seconds": 75,
+            "websocket_max_latency_seconds": 30,
+        },
+    )
+
+    assert os.environ["HERMES_DISCORD_LIVENESS_INTERVAL_SECONDS"] == "11"
+    assert os.environ["HERMES_DISCORD_LIVENESS_FAILURE_THRESHOLD"] == "2"
+    assert seeded == {
+        "websocket_liveness_interval_seconds": 11,
+        "websocket_liveness_failure_threshold": 2,
+        "websocket_heartbeat_ack_max_age_seconds": 75,
+        "websocket_max_latency_seconds": 30,
+    }
+
+
+def test_yaml_config_bridges_nested_discord_extra_websocket_health(monkeypatch):
+    for key in (
+        "HERMES_DISCORD_LIVENESS_INTERVAL_SECONDS",
+        "HERMES_DISCORD_LIVENESS_FAILURE_THRESHOLD",
+    ):
+        monkeypatch.delenv(key, raising=False)
+
+    _apply_yaml_config(
+        {"platforms": {"discord": {"extra": {
+            "websocket_liveness_interval_seconds": 13,
+            "websocket_liveness_failure_threshold": 4,
+        }}}},
+        {"extra": {
+            "websocket_liveness_interval_seconds": 13,
+            "websocket_liveness_failure_threshold": 4,
+        }},
+    )
+
+    assert os.environ["HERMES_DISCORD_LIVENESS_INTERVAL_SECONDS"] == "13"
+    assert os.environ["HERMES_DISCORD_LIVENESS_FAILURE_THRESHOLD"] == "4"

tests/gateway/test_discord_liveness.py

@@ -1,18 +1,17 @@
-"""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.
+"""Regression tests for Discord Gateway WebSocket liveness.
+
+A Discord REST response and the Gateway WebSocket are independent transports.
+A half-closed Gateway socket can leave ``Bot.start()`` alive while REST still
+returns 200, so health must come from the active WebSocket's ready/open/ACK and
+heartbeat-latency state rather than ``fetch_user()``.
 """
 
 from __future__ import annotations
 
 import asyncio
+import time
 from types import SimpleNamespace
-from unittest.mock import AsyncMock
+from unittest.mock import AsyncMock, Mock
 
 import pytest
 
@@ -26,7 +25,8 @@
 _ensure_discord_mock()
 
 import plugins.platforms.discord.adapter as discord_platform  # noqa: E402
-from gateway.config import PlatformConfig  # noqa: E402
+from gateway.config import Platform, PlatformConfig  # noqa: E402
+from gateway.run import GatewayRunner  # noqa: E402
 from plugins.platforms.discord.adapter import DiscordAdapter  # noqa: E402
 
 
@@ -43,6 +43,12 @@ def __init__(self, *, intents, proxy=None, allowed_mentions=None, **_):
         super().__init__(intents=intents, allowed_mentions=allowed_mentions)
         self._never = asyncio.Event()
         self._closed = False
+        self._gateway_ready = True
+        self.latency = 0.05
+        self.ws = _FakeWebSocket()
+
+    def is_ready(self):
+        return self._gateway_ready
 
     async def start(self, token):
         if "on_ready" in self._events:
@@ -58,10 +64,110 @@ async def close(self):
         self._never.set()
 
 
-def _make_adapter(monkeypatch, *, interval=0.01, threshold=1) -> DiscordAdapter:
+class _FakeKeepAlive:
+    def __init__(self, *, ack_age: float = 0.0):
+        self._last_ack = time.perf_counter() - ack_age
+
+
+class _FakeWebSocket:
+    def __init__(self, *, open: bool = True, ack_age: float = 0.0):
+        self.open = open
+        self._keep_alive = _FakeKeepAlive(ack_age=ack_age)
+
+
+def _set_websocket_health(
+    bot: _LiveBot,
+    *,
+    ready: bool = True,
+    socket_open: bool = True,
+    latency: float = 0.05,
+    ack_age: float = 0.0,
+) -> None:
+    bot._gateway_ready = ready
+    bot.latency = latency
+    bot.ws = _FakeWebSocket(open=socket_open, ack_age=ack_age)
+
+
+def _make_adapter(
+    monkeypatch,
+    *,
+    interval=0.01,
+    threshold=1,
+    max_ack_age=1.0,
+    max_latency=1.0,
+) -> 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="[redacted:secret-assignment]"))
+    return DiscordAdapter(
+        PlatformConfig(
+            enabled=True,
+            token="[redacted:secret-assignment]",
+            extra={
+                "websocket_heartbeat_ack_max_age_seconds": max_ack_age,
+                "websocket_max_latency_seconds": max_latency,
+            },
+        )
+    )
+
+
+class _BrokenWebSocket:
+    @property
+    def open(self):
+        raise RuntimeError("socket state unavailable")
+
+
[redacted:email](
+    ("key", "attribute", "raw"),
+    [
+        ("websocket_liveness_interval_seconds", "_liveness_interval_seconds", "nan"),
+        ("websocket_heartbeat_ack_max_age_seconds", "_heartbeat_ack_max_age_seconds", "inf"),
+        ("websocket_max_latency_seconds", "_max_latency_seconds", "-inf"),
+    ],
+)
+def test_nonfinite_liveness_config_disables_that_probe_dimension(monkeypatch, key, attribute, raw):
+    adapter = DiscordAdapter(
+        PlatformConfig(enabled=True, token="[redacted:secret-assignment]", extra={key: raw})
+    )
+
+    assert getattr(adapter, attribute) == 0.0
+
+
+def test_default_liveness_bounds_trigger_timed_recovery(monkeypatch):
+    for key in (
+        "HERMES_DISCORD_LIVENESS_INTERVAL_SECONDS",
+        "HERMES_DISCORD_LIVENESS_FAILURE_THRESHOLD",
+    ):
+        monkeypatch.delenv(key, raising=False)
+
+    adapter = DiscordAdapter(PlatformConfig(enabled=True, token="[redacted:secret-assignment]"))
+
+    assert adapter._liveness_interval_seconds == 15.0
+    assert adapter._liveness_failure_threshold == 2
+    assert adapter._heartbeat_ack_max_age_seconds == 60.0
+    assert adapter._max_latency_seconds == 30.0
+
+
+def test_platform_config_extra_overrides_process_liveness_bridge(monkeypatch):
+    monkeypatch.setenv("HERMES_DISCORD_LIVENESS_INTERVAL_SECONDS", "99")
+    monkeypatch.setenv("HERMES_DISCORD_LIVENESS_FAILURE_THRESHOLD", "9")
+
+    adapter = DiscordAdapter(
+        PlatformConfig(
+            enabled=True,
+            token="[redacted:secret-assignment]",
+            extra={
+                "websocket_liveness_interval_seconds": 7,
+                "websocket_liveness_failure_threshold": 2,
+                "websocket_heartbeat_ack_max_age_seconds": 45,
+                "websocket_max_latency_seconds": 12,
+            },
+        )
+    )
+
+    assert adapter._liveness_interval_seconds == 7
+    assert adapter._liveness_failure_threshold == 2
+    assert adapter._heartbeat_ack_max_age_seconds == 45
+    assert adapter._max_latency_seconds == 12
 
 
 async def _connect(adapter: DiscordAdapter, monkeypatch, bot_factory):
@@ -80,6 +186,14 @@ async def _connect(adapter: DiscordAdapter, monkeypatch, bot_factory):
     assert await adapter.connect() is True
 
 
+async def _wait_until(predicate, message: str, timeout: float = 2.0) -> None:
+    deadline = asyncio.get_running_loop().time() + timeout
+    while not predicate():
+        if asyncio.get_running_loop().time() >= deadline:
+            pytest.fail(message)
+        await asyncio.sleep(0.01)
+
+
 @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."""
@@ -116,33 +230,33 @@ def factory(**kwargs):
 
 
 @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."""
+async def test_liveness_probe_does_not_call_rest_while_websocket_is_healthy(monkeypatch):
+    """A fresh Gateway ACK is sufficient; REST is not a transport health probe."""
     adapter = _make_adapter(monkeypatch, interval=0.01, threshold=3)
 
     def factory(**kwargs):
         bot = _LiveBot(intents=kwargs["intents"], allowed_mentions=kwargs.get("allowed_mentions"))
+        _set_websocket_health(bot)
         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
+    adapter._client.fetch_user.assert_not_awaited()
     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)
+async def test_liveness_probe_forces_reconnect_when_rest_succeeds_but_gateway_ack_is_stale(monkeypatch):
+    """A REST response must not hide a stale Gateway heartbeat failure."""
+    adapter = _make_adapter(monkeypatch, interval=0.005, threshold=2, max_ack_age=0.01)
 
     def factory(**kwargs):
         bot = _LiveBot(intents=kwargs["intents"], allowed_mentions=kwargs.get("allowed_mentions"))
-        bot.fetch_user = AsyncMock(side_effect=TimeoutError("dead proxy"))
+        _set_websocket_health(bot, ack_age=3600)
+        bot.fetch_user = AsyncMock(return_value=SimpleNamespace(id=999))
         return bot
 
     handler = AsyncMock()
@@ -150,24 +264,294 @@ def factory(**kwargs):
     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")
+    # The sampler schedules the close + supervisor callback in a sibling task
+    # so the fatal path cannot cancel/await itself through disconnect().
+    await _wait_until(
+        lambda: handler.await_count,
+        "liveness recovery notification did not complete within 2s",
+    )
 
+    assert adapter._liveness_task and adapter._liveness_task.done()
     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_code == "discord_websocket_health_stale"
     assert adapter.fatal_error_retryable is True
+    wedged.fetch_user.assert_not_awaited()
+    handler.assert_awaited_once()
+
+    await adapter.disconnect()
+
+
[redacted:email]
+async def test_liveness_fatal_queues_primary_runner_reconnect_without_self_cancellation(monkeypatch):
+    adapter = _make_adapter(monkeypatch, interval=0.005, threshold=1, max_ack_age=0.01)
+
+    def factory(**kwargs):
+        bot = _LiveBot(intents=kwargs["intents"], allowed_mentions=kwargs.get("allowed_mentions"))
+        _set_websocket_health(bot, ack_age=3600)
+        return bot
+
+    runner = GatewayRunner.__new__(GatewayRunner)
+    runner.adapters = {Platform.DISCORD: adapter}
+    runner._failed_platforms = {}
+    runner._running = True
+    runner.stop = AsyncMock()
+    runner.delivery_router = SimpleNamespace(adapters=runner.adapters)
+    runner.config = SimpleNamespace(platforms={Platform.DISCORD: adapter.config})
+    runner._update_platform_runtime_status = lambda *args, **kwargs: None
+    runner._adapter_disconnect_timeout_secs = lambda: 0.1
+    adapter.set_fatal_error_handler(runner._handle_adapter_fatal_error)
+    await _connect(adapter, monkeypatch, factory)
+
+    await _wait_until(
+        lambda: Platform.DISCORD in runner._failed_platforms,
+        "liveness fatal did not reach the runner reconnect queue",
+    )
+
+    assert adapter._liveness_notification_task is None or adapter._liveness_notification_task.done()
+    assert runner._failed_platforms[Platform.DISCORD]["attempts"] == 0
+    runner.stop.assert_not_awaited()
+
+
[redacted:email]
[redacted:email](
+    ("health", "expected_reason"),
+    [
+        ({"ready": False}, "not_ready"),
+        ({"socket_open": False}, "socket_closed"),
+        ({"latency": float("inf")}, "latency_non_finite"),
+    ],
+)
+async def test_liveness_probe_reports_gateway_health_failure_reason(monkeypatch, health, expected_reason):
+    adapter = _make_adapter(monkeypatch, interval=0.005, threshold=1)
+
+    def factory(**kwargs):
+        bot = _LiveBot(intents=kwargs["intents"], allowed_mentions=kwargs.get("allowed_mentions"))
+        _set_websocket_health(bot, **health)
+        bot.fetch_user = AsyncMock(return_value=SimpleNamespace(id=999))
+        return bot
+
+    handler = AsyncMock()
+    adapter.set_fatal_error_handler(handler)
+    await _connect(adapter, monkeypatch, factory)
+
+    await _wait_until(
+        lambda: handler.await_count,
+        "liveness loop did not surface a websocket health failure",
+    )
+
+    assert expected_reason in (adapter.fatal_error_message or "")
+    adapter._client.fetch_user.assert_not_awaited()
+    handler.assert_awaited_once()
+    await adapter.disconnect()
+
+
+
+
[redacted:email]
+async def test_liveness_probe_treats_websocket_state_read_error_as_unhealthy(monkeypatch):
+    adapter = _make_adapter(monkeypatch, interval=0.005, threshold=1)
+
+    def factory(**kwargs):
+        bot = _LiveBot(intents=kwargs["intents"], allowed_mentions=kwargs.get("allowed_mentions"))
+        bot.ws = _BrokenWebSocket()
+        return bot
+
+    handler = AsyncMock()
+    adapter.set_fatal_error_handler(handler)
+    await _connect(adapter, monkeypatch, factory)
+
+    await _wait_until(
+        lambda: handler.await_count,
+        "liveness loop did not surface a WebSocket state read error",
+    )
+
+    assert "socket_state_unavailable" in (adapter.fatal_error_message or "")
     handler.assert_awaited_once()
+    await adapter.disconnect()
 
+
[redacted:email]
+async def test_liveness_probe_recovers_when_health_reader_raises(monkeypatch):
+    adapter = _make_adapter(monkeypatch, interval=0.005, threshold=1)
+
+    def factory(**kwargs):
+        return _LiveBot(
+            intents=kwargs["intents"],
+            allowed_mentions=kwargs.get("allowed_mentions"),
+        )
+
+    handler = AsyncMock()
+    adapter.set_fatal_error_handler(handler)
+    await _connect(adapter, monkeypatch, factory)
+    monkeypatch.setattr(
+        adapter,
+        "_read_websocket_health",
+        lambda _client: (_ for _ in ()).throw(RuntimeError("unexpected state")),
+    )
+
+    await _wait_until(
+        lambda: handler.await_count,
+        "liveness loop did not recover from health-reader failure",
+    )
+
+    assert "health_check_error" in (adapter.fatal_error_message or "")
+    handler.assert_awaited_once()
     await adapter.disconnect()
 
 
[redacted:email]
+async def test_liveness_recovery_keeps_websocket_fatal_when_client_task_exits(monkeypatch):
+    """The close callback must not replace stale-ACK recovery with task-exited."""
+    adapter = _make_adapter(monkeypatch, interval=0.005, threshold=1, max_ack_age=0.01)
+
+    def factory(**kwargs):
+        bot = _LiveBot(intents=kwargs["intents"], allowed_mentions=kwargs.get("allowed_mentions"))
+        _set_websocket_health(bot, ack_age=3600)
+        return bot
+
+    handler = AsyncMock()
+    adapter.set_fatal_error_handler(handler)
+    await _connect(adapter, monkeypatch, factory)
+
+    await _wait_until(
+        lambda: handler.await_count,
+        "closed client task did not finish within 2s",
+    )
+
+    assert adapter._bot_task and adapter._bot_task.done()
+    assert adapter.fatal_error_code == "discord_websocket_health_stale"
+    assert handler.await_count == 1
+    await adapter.disconnect()
+
+
[redacted:email]
+async def test_liveness_recovery_not_blocked_by_hanging_client_close(monkeypatch):
+    """A wedged close must not prevent fatal notification/reconnect queueing."""
+    adapter = _make_adapter(monkeypatch, interval=60, threshold=1, max_ack_age=1.0)
+    monkeypatch.setenv("HERMES_GATEWAY_ADAPTER_DISCONNECT_TIMEOUT", "0.02")
+
+    def factory(**kwargs):
+        bot = _LiveBot(intents=kwargs["intents"], allowed_mentions=kwargs.get("allowed_mentions"))
+        _set_websocket_health(bot, ack_age=3600)
+        bot.fetch_user = AsyncMock(return_value=SimpleNamespace(id=999))
+        return bot
+
+    handler = AsyncMock()
+    adapter.set_fatal_error_handler(handler)
+    await _connect(adapter, monkeypatch, factory)
+    wedged = adapter._client
+    close_started = asyncio.Event()
+
+    async def hanging_close():
+        close_started.set()
+        await asyncio.Event().wait()
+
+    wedged.close = hanging_close
+    adapter._set_fatal_error(
+        "discord_websocket_health_stale",
+        "Discord Gateway WebSocket health check failed: ack_stale",
+        retryable=True,
+    )
+    notify_task = asyncio.create_task(adapter._notify_liveness_fatal_error(wedged))
+    await asyncio.wait_for(close_started.wait(), timeout=0.5)
+    await asyncio.wait_for(notify_task, timeout=2.0)
+    assert close_started.is_set() is True
+    assert handler.await_count == 1
+    assert adapter.fatal_error_code == "discord_websocket_health_stale"
+
+    # Restore a cooperative fake close so the test can release the bot task.
+    wedged.close = _LiveBot.close.__get__(wedged, _LiveBot)
+    await adapter.disconnect()
+
+
[redacted:email]
+async def test_liveness_close_timeout_aborts_aiohttp_transport_before_fatal_notification(
+    monkeypatch,
+):
+    """A close handshake timeout must abort the stale socket before reconnect."""
+    adapter = _make_adapter(monkeypatch, interval=60, threshold=1, max_ack_age=1.0)
+    handler = AsyncMock()
+    adapter.set_fatal_error_handler(handler)
+
+    close_started = asyncio.Event()
+    release_close = asyncio.Event()
+
+    async def hanging_close():
+        close_started.set()
+        while not release_close.is_set():
+            try:
+                await release_close.wait()
+            except asyncio.CancelledError:
+                # Model a close path that catches cancellation while unwinding.
+                continue
+
+    transport = Mock()
+    replacement_transport = Mock()
+    aiohttp_socket = SimpleNamespace(
+        close=hanging_close,
+        # aiohttp clears response.connection while cancellation unwinds close(),
+        # but its WebSocket writer still owns the underlying transport.
+        _response=SimpleNamespace(connection=None),
+        _conn=None,
+        _writer=SimpleNamespace(transport=transport),
+    )
+    gateway_websocket = SimpleNamespace(socket=aiohttp_socket)
+    replacement_websocket = SimpleNamespace(
+        socket=SimpleNamespace(
+            _response=SimpleNamespace(connection=None),
+            _conn=None,
+            _writer=SimpleNamespace(transport=replacement_transport),
+        )
+    )
+
+    class _StickyCloseClient:
+        def __init__(self):
+            self.ws = gateway_websocket
+            self._closing_task = None
+            self.close_attempts = 0
+
+        async def close(self):
+            if self._closing_task is not None:
+                return await self._closing_task
+
+            async def _close():
+                self.close_attempts += 1
+                if self.close_attempts == 1:
+                    # The library may publish a replacement WebSocket while the
+                    # old close handshake is still stuck. Recovery must never
+                    # abort the replacement transport.
+                    self.ws = replacement_websocket
+                    await hanging_close()
+
+            self._closing_task = asyncio.create_task(_close())
+            return await self._closing_task
+
+    client = _StickyCloseClient()
+
+    adapter._set_fatal_error(
+        "discord_websocket_health_stale",
+        "Discord Gateway WebSocket health check failed: socket_closed",
+        retryable=True,
+    )
+    notify_task = asyncio.create_task(adapter._notify_liveness_fatal_error(client))
+
+    await asyncio.wait_for(close_started.wait(), timeout=0.5)
+    done, _pending = await asyncio.wait({notify_task}, timeout=1.5)
+    finished_within_bound = notify_task in done
+    release_close.set()
+    if not notify_task.done():
+        await asyncio.wait_for(notify_task, timeout=0.5)
+
+    assert finished_within_bound is True
+    transport.abort.assert_called_once_with()
+    replacement_transport.abort.assert_not_called()
+    handler.assert_awaited_once()
+    assert client._closing_task is None
+    await client.close()
+    assert client.close_attempts == 2
+
+
 @pytest.mark.asyncio
 async def test_disconnect_cancels_liveness_task(monkeypatch):
     """``disconnect()`` must cancel the probe so the gateway can shut down

tests/gateway/test_gateway_shutdown.py

@@ -118,6 +118,26 @@ async def finish_agent():
     assert runner._shutdown_event.is_set() is True
 
 
[redacted:email]
+async def test_gateway_stop_cancels_secondary_reconnects_before_session_drain():
+    runner, _adapter = make_restart_runner()
+    order: list[str] = []
+
+    async def _cancel_secondary_reconnects() -> None:
+        order.append("secondary_reconnect_cancel")
+
+    async def _notify_sessions() -> None:
+        order.append("notify_sessions")
+
+    runner._cancel_secondary_profile_reconnect_tasks = _cancel_secondary_reconnects
+    runner._notify_active_sessions_of_shutdown = _notify_sessions
+
+    with patch("gateway.status.remove_pid_file"), patch("gateway.status.write_runtime_status"):
+        await runner.stop()
+
+    assert order[:2] == ["secondary_reconnect_cancel", "notify_sessions"]
+
+
 @pytest.mark.asyncio
 async def test_gateway_stop_interrupts_after_drain_timeout():
     runner, adapter = make_restart_runner()

tests/gateway/test_multiplex_adapter_registry.py

@@ -1,8 +1,14 @@
 """Phase 3: secondary-profile adapter registry + same-token conflict detection."""
 import logging
+import asyncio
+from contextlib import contextmanager
+from pathlib import Path
+from unittest.mock import AsyncMock
 
 import pytest
 
+import gateway.run as gateway_run
+from gateway.config import GatewayConfig, Platform, PlatformConfig
 from gateway.run import GatewayRunner
 
 
@@ -144,6 +150,245 @@ class _Evt:
         assert seen["profile"] == "writer"
 
 
+class _SecondaryRecoveryAdapter:
+    platform = Platform.DISCORD
+
+    def __init__(self, *, retryable=True):
+        self.fatal_error_retryable = retryable
+        self.fatal_error_code = "transport_stale" if retryable else "auth_failed"
+        self.fatal_error_message = "Gateway transport stale"
+        self.connected = False
+        self.disconnected = False
+
+    async def disconnect(self):
+        self.disconnected = True
+
+    def set_message_handler(self, handler):
+        self.message_handler = handler
+
+    def set_fatal_error_handler(self, handler):
+        self.fatal_error_handler = handler
+
+    def set_session_store(self, store):
+        self.session_store = store
+
+    def set_busy_session_handler(self, handler):
+        self.busy_session_handler = handler
+
+    def set_topic_recovery_fn(self, handler):
+        self.topic_recovery_fn = handler
+
+    def set_authorization_check(self, handler):
+        self.authorization_check = handler
+
+
+def _secondary_recovery_runner(*, running=True):
+    runner = GatewayRunner.__new__(GatewayRunner)
+    runner.config = GatewayConfig(multiplex_profiles=True)
+    runner._running = running
+    runner._profile_adapters = {}
+    runner._profile_failed_platforms = {}
+    runner._background_tasks = set()
+    runner.session_store = object()
+    runner._handle_active_session_busy_message = object()
+    runner._recover_telegram_topic_thread_id = object()
+    runner._busy_text_mode = "queue"
+    runner._make_adapter_auth_check = lambda platform, profile_name=None: object()
+    runner._adapter_disconnect_timeout_secs = lambda: 0
+    runner._sync_voice_mode_state_to_adapter = lambda adapter: None
+    return runner
+
+
+def _install_secondary_reconnect_context(monkeypatch, runner, adapter, scoped_homes=None):
+    @contextmanager
+    def fake_scope(profile_home):
+        if scoped_homes is not None:
+            scoped_homes.append(Path(profile_home))
+        yield
+
+    monkeypatch.setattr(gateway_run, "_profile_runtime_scope", fake_scope)
+    monkeypatch.setattr(
+        "hermes_cli.profiles.get_profile_dir", lambda name: Path("/profiles") / name
+    )
+    monkeypatch.setattr(
+        "gateway.config.load_gateway_config",
+        lambda: GatewayConfig(
+            multiplex_profiles=True,
+            platforms={
+                Platform.DISCORD: PlatformConfig(
+                    enabled=True, token="[redacted:secret-assignment]"
+                )
+            },
+        ),
+    )
+    monkeypatch.setattr(runner, "_create_adapter", lambda platform, config: adapter)
+
+
+class TestSecondaryProfileFatalRecovery:
+    @pytest.mark.asyncio
+    async def test_retryable_secondary_fatal_reconnects_with_its_profile_scope(
+        self, monkeypatch
+    ):
+        runner = _secondary_recovery_runner()
+        stale = _SecondaryRecoveryAdapter()
+        replacement = _SecondaryRecoveryAdapter()
+        runner._profile_adapters["reviewer"] = {Platform.DISCORD: stale}
+        scoped_homes: list[Path] = []
+        _install_secondary_reconnect_context(
+            monkeypatch, runner, replacement, scoped_homes
+        )
+
+        async def connect(adapter, platform, *, is_reconnect=False):
+            assert adapter is replacement
+            assert platform is Platform.DISCORD
+            assert is_reconnect is True
+            replacement.connected = True
+            return True
+
+        monkeypatch.setattr(runner, "_connect_adapter_with_timeout", connect)
+        await runner._handle_profile_adapter_fatal_error(
+            "reviewer", Platform.DISCORD, stale
+        )
+
+        assert stale.disconnected is True
+        assert Platform.DISCORD not in runner._profile_adapters["reviewer"]
+        tasks = list(runner._background_tasks)
+        assert len(tasks) == 1
+        await tasks[0]
+        assert runner._profile_adapters["reviewer"][Platform.DISCORD] is replacement
+        assert scoped_homes
+        assert all(path == Path("/profiles/reviewer") for path in scoped_homes)
+
+    @pytest.mark.asyncio
+    async def test_secondary_reconnect_cancellation_disposes_partial_adapter(
+        self, monkeypatch
+    ):
+        runner = _secondary_recovery_runner()
+        runner._profile_failed_platforms["reviewer"] = {}
+        partial = _SecondaryRecoveryAdapter()
+        _install_secondary_reconnect_context(monkeypatch, runner, partial)
+        connect_started = asyncio.Event()
+
+        async def connect(adapter, platform, *, is_reconnect=False):
+            connect_started.set()
+            await asyncio.Event().wait()
+
+        monkeypatch.setattr(runner, "_connect_adapter_with_timeout", connect)
+        task = asyncio.create_task(
+            runner._run_secondary_profile_reconnect("reviewer", Platform.DISCORD)
+        )
+        runner._profile_failed_platforms["reviewer"][Platform.DISCORD] = task
+        await connect_started.wait()
+        task.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await task
+
+        assert partial.disconnected is True
+        assert runner._profile_failed_platforms == {}
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("connect_result", [True, False], ids=["success", "failure"])
+    async def test_secondary_reconnect_does_not_publish_after_shutdown(
+        self, monkeypatch, connect_result
+    ):
+        runner = _secondary_recovery_runner()
+        runner._profile_failed_platforms["reviewer"] = {}
+        replacement = _SecondaryRecoveryAdapter()
+        _install_secondary_reconnect_context(monkeypatch, runner, replacement)
+        connect_started = asyncio.Event()
+        release_connect = asyncio.Event()
+
+        async def connect(adapter, platform, *, is_reconnect=False):
+            connect_started.set()
+            await release_connect.wait()
+            return connect_result
+
+        monkeypatch.setattr(runner, "_connect_adapter_with_timeout", connect)
+        task = asyncio.create_task(
+            runner._run_secondary_profile_reconnect("reviewer", Platform.DISCORD)
+        )
+        runner._profile_failed_platforms["reviewer"][Platform.DISCORD] = task
+        await connect_started.wait()
+        runner._running = False
+        release_connect.set()
+        await asyncio.wait_for(task, timeout=0.2)
+
+        assert runner._profile_adapters == {}
+        assert replacement.disconnected is True
+        assert runner._profile_failed_platforms == {}
+
+    @pytest.mark.asyncio
+    async def test_shutdown_cancels_secondary_reconnect_before_registry_teardown(self):
+        runner = _secondary_recovery_runner()
+        runner._profile_failed_platforms["reviewer"] = {}
+        runner._adapter_disconnect_timeout_secs = lambda: 0.1
+        started = asyncio.Event()
+        partial = _SecondaryRecoveryAdapter()
+
+        async def reconnect():
+            started.set()
+            try:
+                await asyncio.Event().wait()
+            except asyncio.CancelledError:
+                await runner._safe_adapter_disconnect(partial, Platform.DISCORD)
+                raise
+
+        task = asyncio.create_task(reconnect())
+        runner._profile_failed_platforms["reviewer"][Platform.DISCORD] = task
+        await started.wait()
+        await runner._cancel_secondary_profile_reconnect_tasks()
+
+        assert task.cancelled()
+        assert partial.disconnected is True
+        assert runner._profile_failed_platforms == {}
+
+    @pytest.mark.asyncio
+    async def test_secondary_fatal_during_shutdown_does_not_schedule_reconnect(self):
+        runner = _secondary_recovery_runner(running=False)
+        adapter = _SecondaryRecoveryAdapter()
+        runner._profile_adapters = {"reviewer": {Platform.DISCORD: adapter}}
+        scheduled = []
+        runner._schedule_secondary_profile_reconnect = lambda *args: scheduled.append(args)
+
+        await runner._handle_profile_adapter_fatal_error(
+            "reviewer", Platform.DISCORD, adapter
+        )
+
+        assert adapter.disconnected is True
+        assert Platform.DISCORD not in runner._profile_adapters["reviewer"]
+        assert scheduled == []
+
+    def test_secondary_reconnect_scheduler_is_noop_after_shutdown(self, monkeypatch):
+        runner = _secondary_recovery_runner(running=False)
+        created = []
+
+        def create_task(coro, *, name):
+            coro.close()
+            created.append(name)
+            return AsyncMock()
+
+        monkeypatch.setattr(asyncio, "create_task", create_task)
+        runner._schedule_secondary_profile_reconnect(
+            "reviewer", Platform.DISCORD, _SecondaryRecoveryAdapter()
+        )
+
+        assert created == []
+        assert runner._profile_failed_platforms == {}
+
+    @pytest.mark.asyncio
+    async def test_nonretryable_secondary_fatal_is_not_restarted(self):
+        runner = _secondary_recovery_runner()
+        adapter = _SecondaryRecoveryAdapter(retryable=False)
+        runner._profile_adapters = {"reviewer": {Platform.DISCORD: adapter}}
+
+        await runner._handle_profile_adapter_fatal_error(
+            "reviewer", Platform.DISCORD, adapter
+        )
+
+        assert adapter.disconnected is True
+        assert runner._background_tasks == set()
+
+
 class TestSecondaryProfileConfigHandling:
     """Secondary config errors degrade only when the profile is safe to skip."""

tests/gateway/test_runner_fatal_adapter.py

@@ -4,8 +4,9 @@
 import pytest
 
 from gateway.config import GatewayConfig, Platform, PlatformConfig
-from gateway.platforms.base import BasePlatformAdapter
+from gateway.platforms.base import BasePlatformAdapter, MessageEvent, SendResult
 from gateway.run import GatewayRunner
+from gateway.session import SessionSource, build_session_key
 
 
 class _FatalAdapter(BasePlatformAdapter):
@@ -47,6 +48,34 @@ async def get_chat_info(self, chat_id):
         return {"id": chat_id}
 
 
+class _ReplacementDeliveryAdapter(BasePlatformAdapter):
+    def __init__(self):
+        super().__init__(
+            PlatformConfig(enabled=True, token="token", typing_indicator=False),
+            Platform.DISCORD,
+        )
+        self.sent: list[str] = []
+        self.connected = True
+
+    async def connect(self, *, is_reconnect: bool = False) -> bool:
+        return True
+
+    async def disconnect(self) -> None:
+        self.connected = False
+
+    async def send(self, chat_id, content, reply_to=None, metadata=None):
+        if not self.connected:
+            return SendResult(success=False, error="Not connected")
+        self.sent.append(content)
+        return SendResult(success=True, message_id=f"m-{len(self.sent)}")
+
+    async def send_typing(self, chat_id, metadata=None) -> None:
+        return None
+
+    async def get_chat_info(self, chat_id):
+        return {"id": chat_id}
+
+
 @pytest.mark.asyncio
 async def test_runner_requests_clean_exit_for_nonretryable_startup_conflict(monkeypatch, tmp_path):
     config = GatewayConfig(
@@ -101,6 +130,50 @@ async def test_runner_queues_retryable_runtime_fatal_for_reconnection(monkeypatc
     assert runner._failed_platforms[Platform.WHATSAPP]["attempts"] == 0
 
 
[redacted:email]
+async def test_retryable_fatal_queues_reconnect_after_cancellation_swallowing_disconnect(
+    monkeypatch, tmp_path
+):
+    """A wedged old adapter cannot block runner-owned reconnect recovery."""
+    monkeypatch.setenv("HERMES_GATEWAY_ADAPTER_DISCONNECT_TIMEOUT", "0.01")
+    config = GatewayConfig(
+        platforms={Platform.WHATSAPP: PlatformConfig(enabled=True, token="token")},
+        sessions_dir=tmp_path / "sessions",
+    )
+    runner = GatewayRunner(config)
+    adapter = _RuntimeRetryableAdapter()
+    adapter._set_fatal_error("transport_stale", "transport stale", retryable=True)
+    runner.adapters = {Platform.WHATSAPP: adapter}
+    runner.delivery_router.adapters = runner.adapters
+
+    started = asyncio.Event()
+    release = asyncio.Event()
+    finished = asyncio.Event()
+
+    async def swallow_cancellation():
+        started.set()
+        while not release.is_set():
+            try:
+                await release.wait()
+            except asyncio.CancelledError:
+                continue
+        finished.set()
+
+    monkeypatch.setattr(adapter, "disconnect", swallow_cancellation)
+    operation = asyncio.create_task(runner._handle_adapter_fatal_error(adapter))
+    await started.wait()
+    done, _pending = await asyncio.wait({operation}, timeout=0.2)
+    try:
+        assert operation in done
+        assert runner.adapters == {}
+        assert Platform.WHATSAPP in runner._failed_platforms
+        assert runner._failed_platforms[Platform.WHATSAPP]["attempts"] == 0
+    finally:
+        release.set()
+        await asyncio.wait({operation}, timeout=0.2)
+        await asyncio.wait_for(finished.wait(), timeout=0.2)
+
+
 @pytest.mark.asyncio
 async def test_concurrent_fatal_notifications_disconnect_same_adapter_once(monkeypatch, tmp_path):
     """
@@ -192,3 +265,63 @@ async def test_stale_fatal_notification_from_superseded_adapter_is_ignored(monke
     assert runner.adapters[Platform.WHATSAPP] is new_adapter
     assert Platform.WHATSAPP not in runner._failed_platforms
     runner.stop.assert_not_awaited()
+
+
[redacted:email]
[redacted:email]("profile", [None, "reviewer"], ids=["primary", "secondary"])
+async def test_inflight_final_reply_uses_replacement_adapter_after_reconnect(
+    tmp_path, profile
+):
+    config = GatewayConfig(
+        platforms={Platform.DISCORD: PlatformConfig(enabled=True, token="token")},
+        sessions_dir=tmp_path / "sessions",
+    )
+    runner = GatewayRunner(config)
+    old_adapter = _ReplacementDeliveryAdapter()
+    replacement = _ReplacementDeliveryAdapter()
+    old_adapter.gateway_runner = runner
+    replacement.gateway_runner = runner
+    if profile:
+        runner.adapters = {}
+        runner._profile_adapters = {profile: {Platform.DISCORD: old_adapter}}
+    else:
+        runner.adapters = {Platform.DISCORD: old_adapter}
+    runner.delivery_router.adapters = runner.adapters
+
+    handler_started = asyncio.Event()
+    release_handler = asyncio.Event()
+
+    async def handler(_event):
+        await old_adapter.send("channel-1", "partial preview")
+        handler_started.set()
+        await release_handler.wait()
+        return "complete final reply"
+
+    old_adapter.set_message_handler(handler)
+    event = MessageEvent(
+        text="long-running request",
+        source=SessionSource(
+            platform=Platform.DISCORD,
+            chat_id="channel-1",
+            chat_type="dm",
+            user_id="user-1",
+            profile=profile,
+        ),
+        message_id="inbound-1",
+    )
+    task = asyncio.create_task(
+        old_adapter._process_message_background(event, build_session_key(event.source))
+    )
+    await handler_started.wait()
+
+    await old_adapter.disconnect()
+    if profile:
+        runner._profile_adapters[profile][Platform.DISCORD] = replacement
+    else:
+        runner.adapters = {Platform.DISCORD: replacement}
+    runner.delivery_router.adapters = runner.adapters
+    release_handler.set()
+    await task
+
+    assert old_adapter.sent == ["partial preview"]
+    assert replacement.sent == ["complete final reply"]

tests/gateway/test_safe_adapter_disconnect.py

@@ -77,3 +77,47 @@ async def hang():
 
     adapter.disconnect.assert_awaited_once()
     assert "Timed out after 0.0s while disconnecting feishu adapter" in caplog.text
+
+
[redacted:email]
+async def test_safe_disconnect_detaches_cancellation_swallowing_disconnect(
+    bare_runner, monkeypatch, caplog
+):
+    """A disconnect that catches cancellation cannot block fatal recovery.
+
+    ``asyncio.wait_for`` cancels its child at the deadline but then waits for
+    it to finish.  A half-closed transport can catch that cancellation while
+    unwinding, so the runner must detach the old close task and continue to the
+    reconnect queue instead of waiting indefinitely.
+    """
+    monkeypatch.setenv("HERMES_GATEWAY_ADAPTER_DISCONNECT_TIMEOUT", "0.01")
+    adapter = MagicMock()
+    started = asyncio.Event()
+    release = asyncio.Event()
+    finished = asyncio.Event()
+
+    async def swallow_cancellation():
+        started.set()
+        while not release.is_set():
+            try:
+                await release.wait()
+            except asyncio.CancelledError:
+                continue
+        finished.set()
+
+    adapter.disconnect = AsyncMock(side_effect=swallow_cancellation)
+    operation = asyncio.create_task(
+        bare_runner._safe_adapter_disconnect(adapter, Platform.FEISHU)
+    )
+    await started.wait()
+    done, _pending = await asyncio.wait({operation}, timeout=0.2)
+    try:
+        assert operation in done
+        assert "Timed out after 0.0s while disconnecting feishu adapter" in caplog.text
+    finally:
+        # The implementation must detach rather than abandon the old task.
+        # Release it here so this test leaves no cancellation-swallowing task
+        # behind when it runs against the pre-fix implementation.
+        release.set()
+        await asyncio.wait({operation}, timeout=0.2)
+        await asyncio.wait_for(finished.wait(), timeout=0.2)

tests/gateway/test_shutdown_cache_cleanup.py

@@ -81,6 +81,9 @@ async def _cleanup_agent_resources_off_loop(self, agent, *, context=""):
     async def _notify_active_sessions_of_shutdown(self):
         pass
 
+    async def _cancel_secondary_profile_reconnect_tasks(self):
+        pass
+
     async def _drain_active_agents(self, timeout):
         return {}, False

tests/gateway/test_systemd_notify.py

@@ -0,0 +1,154 @@
+"""Tests for the optional systemd event-loop watchdog protocol."""
+
+from __future__ import annotations
+
+import asyncio
+import socket
+
+import pytest
+
+
+def test_notify_without_notify_socket_is_a_noop(monkeypatch):
+    monkeypatch.delenv("NOTIFY_SOCKET", raising=False)
+
+    from gateway.systemd_notify import notify
+
+    assert notify("READY=1") is False
+
+
+def test_notify_sends_real_unix_datagram(tmp_path, monkeypatch):
+    address = str(tmp_path / "notify.sock")
+    receiver = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
+    receiver.bind(address)
+    receiver.settimeout(1.0)
+    monkeypatch.setenv("NOTIFY_SOCKET", address)
+
+    from gateway.systemd_notify import notify
+
+    assert notify("READY=1") is True
+    assert receiver.recv(4096) == b"READY=1"
+    receiver.close()
+
+
[redacted:email](
+    not hasattr(socket, "AF_UNIX"), reason="Unix datagram sockets are unavailable"
+)
+def test_notify_supports_systemd_abstract_socket(monkeypatch):
+    name = "\0hermes-test-notify"
+    receiver = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
+    receiver.bind(name)
+    receiver.settimeout(1.0)
+    monkeypatch.setenv("NOTIFY_SOCKET", "@hermes-test-notify")
+
+    try:
+        from gateway.systemd_notify import notify
+
+        assert notify("WATCHDOG=1") is True
+        assert receiver.recv(4096) == b"WATCHDOG=1"
+    finally:
+        receiver.close()
+
+
+def test_notify_uses_nonblocking_datagram_send(monkeypatch):
+    calls: list[object] = []
+
+    class _Sender:
+        def __enter__(self):
+            return self
+
+        def __exit__(self, *_args):
+            return None
+
+        def setblocking(self, value):
+            calls.append(("setblocking", value))
+
+        def connect(self, address):
+            calls.append(("connect", address))
+
+        def send(self, payload):
+            calls.append(("send", payload))
+
+    import gateway.systemd_notify as notify_mod
+
+    monkeypatch.setenv("NOTIFY_SOCKET", "/tmp/hermes-test-notify")
+    monkeypatch.setattr(notify_mod.socket, "socket", lambda *_args: _Sender())
+
+    assert notify_mod.notify("READY=1") is True
+    assert calls[0] == ("setblocking", False)
+
+
[redacted:email]("raw", [None, "", "0", "-1", "not-a-number"])
+def test_watchdog_interval_is_disabled_for_missing_invalid_or_nonpositive_values(
+    monkeypatch, raw
+):
+    if raw is None:
+        monkeypatch.delenv("WATCHDOG_USEC", raising=False)
+    else:
+        monkeypatch.setenv("WATCHDOG_USEC", raw)
+    monkeypatch.setenv("NOTIFY_SOCKET", "/tmp/hermes-test-notify-does-not-exist")
+
+    from gateway.systemd_notify import watchdog_interval_seconds
+
+    assert watchdog_interval_seconds() is None
+
+
+def test_watchdog_latches_when_loop_progress_is_late(monkeypatch):
+    calls: list[str] = []
+    monkeypatch.setenv("NOTIFY_SOCKET", "/tmp/hermes-test-notify")
+    monkeypatch.setenv("WATCHDOG_USEC", "1000000")
+
+    import gateway.systemd_notify as notify_mod
+
+    monkeypatch.setattr(
+        notify_mod, "notify", lambda message: calls.append(message) or True
+    )
+    watchdog = notify_mod.SystemdWatchdog(lag_tolerance_seconds=0.1)
+
+    assert watchdog.record_tick(scheduled_at=10.0, now=10.05) is True
+    assert calls == ["WATCHDOG=1"]
+    assert watchdog.record_tick(scheduled_at=10.0, now=10.2) is False
+    assert watchdog.unhealthy is True
+    assert calls[-1].startswith("STATUS=watchdog unhealthy")
+    assert watchdog.record_tick(scheduled_at=10.0, now=10.3) is False
+
+
[redacted:email]
+async def test_watchdog_sends_ready_heartbeat_and_stopping(monkeypatch):
+    calls: list[str] = []
+    monkeypatch.setenv("NOTIFY_SOCKET", "/tmp/hermes-test-notify")
+    monkeypatch.setenv("WATCHDOG_USEC", "20000")
+
+    import gateway.systemd_notify as notify_mod
+
+    monkeypatch.setattr(
+        notify_mod, "notify", lambda message: calls.append(message) or True
+    )
+    watchdog = notify_mod.SystemdWatchdog(lag_tolerance_seconds=1.0)
+
+    assert watchdog.start() is True
+    assert watchdog.ready("Gateway running") is True
+    await asyncio.sleep(0.04)
+    await watchdog.stop()
+
+    assert any(message.startswith("READY=1") for message in calls)
+    assert "WATCHDOG=1" in calls
+    assert calls[-1] == "STOPPING=1"
+    assert watchdog.unhealthy is False
+
+
+def test_watchdog_config_disabled_ignores_systemd_environment(monkeypatch):
+    calls: list[str] = []
+    monkeypatch.setenv("NOTIFY_SOCKET", "/tmp/hermes-test-notify")
+    monkeypatch.setenv("WATCHDOG_USEC", "1000000")
+
+    import gateway.systemd_notify as notify_mod
+
+    monkeypatch.setattr(
+        notify_mod, "notify", lambda message: calls.append(message) or True
+    )
+    watchdog = notify_mod.SystemdWatchdog(config_enabled=False)
+
+    assert watchdog.enabled is False
+    assert watchdog.start() is False
+    assert watchdog.ready() is False
+    assert calls == []

tests/gateway/test_systemd_watchdog_lifecycle.py

@@ -0,0 +1,98 @@
+"""Gateway lifecycle contract for the opt-in systemd watchdog."""
+
+from __future__ import annotations
+
+import inspect
+from unittest.mock import patch
+
+import pytest
+
+from gateway.config import GatewayConfig
+from gateway.run import GatewayRunner, start_gateway
+from tests.gateway.restart_test_helpers import make_restart_runner
+
+
+class _FakeWatchdog:
+    instances: list["_FakeWatchdog"] = []
+
+    def __init__(self, *, config_enabled: bool = True):
+        self.config_enabled = config_enabled
+        self.calls: list[str] = []
+        self.__class__.instances.append(self)
+
+    def start(self) -> bool:
+        self.calls.append("start")
+        return self.config_enabled
+
+    def ready(self, status: str) -> bool:
+        self.calls.append(f"ready:{status}")
+        return True
+
+    async def stop(self) -> None:
+        self.calls.append("stop")
+
+
+def _bare_runner(*, seconds: int, running: bool = True) -> GatewayRunner:
+    runner = object.__new__(GatewayRunner)
+    runner.config = GatewayConfig(systemd_watchdog_seconds=seconds)
+    runner._running = running
+    runner._systemd_watchdog = None
+    return runner
+
+
+def test_runner_starts_watchdog_only_after_running(monkeypatch):
+    _FakeWatchdog.instances.clear()
+    monkeypatch.setattr("gateway.systemd_notify.SystemdWatchdog", _FakeWatchdog)
+    runner = _bare_runner(seconds=120, running=True)
+
+    assert runner._start_systemd_watchdog() is True
+
+    watchdog = _FakeWatchdog.instances[-1]
+    assert watchdog.config_enabled is True
+    assert watchdog.calls == ["start", "ready:Hermes Gateway running"]
+
+
+def test_runner_does_not_start_watchdog_when_disabled_or_not_running(monkeypatch):
+    _FakeWatchdog.instances.clear()
+    monkeypatch.setattr("gateway.systemd_notify.SystemdWatchdog", _FakeWatchdog)
+
+    assert _bare_runner(seconds=0)._start_systemd_watchdog() is False
+    assert _bare_runner(seconds=120, running=False)._start_systemd_watchdog() is False
+    assert _FakeWatchdog.instances == []
+
+
+def test_gateway_ready_follows_background_service_startup():
+    source = inspect.getsource(start_gateway)
+
+    housekeeping_started = source.index("housekeeping_thread.start()")
+    watchdog_started = source.index("start_watchdog()")
+    shutdown_wait = source.index("await runner.wait_for_shutdown()", watchdog_started)
+
+    assert housekeeping_started < watchdog_started < shutdown_wait
+
+
[redacted:email]
+async def test_gateway_stop_stops_watchdog_before_session_drain():
+    runner, _adapter = make_restart_runner()
+    order: list[str] = []
+
+    class _OrderingWatchdog:
+        async def stop(self) -> None:
+            order.append("watchdog_stop")
+
+    async def _notify_sessions() -> None:
+        order.append("notify_sessions")
+
+    runner._systemd_watchdog = _OrderingWatchdog()
+    runner._notify_active_sessions_of_shutdown = _notify_sessions
+
+    with (
+        patch("gateway.status.remove_pid_file"),
+        patch("gateway.status.write_runtime_status"),
+    ):
+        await runner.stop()
+
+    assert order[:2] == [
+        "watchdog_stop",
+        "notify_sessions",
+    ]

tests/hermes_cli/test_gateway_service.py

@@ -421,7 +421,7 @@ def test_passes_when_unit_exists(self, tmp_path, monkeypatch):
 
 class TestGeneratedSystemdUnits:
     def _expected_timeout_stop_sec(self) -> str:
-        timeout = int(max(60, DEFAULT_GATEWAY_RESTART_DRAIN_TIMEOUT) + 30)
+        timeout = int(max(60, DEFAULT_GATEWAY_RESTART_DRAIN_TIMEOUT + 30))
         return f"TimeoutStopSec={timeout}"
 
     def test_user_unit_avoids_recursive_execstop_and_uses_extended_stop_timeout(self, monkeypatch):
@@ -437,9 +437,8 @@ def test_user_unit_avoids_recursive_execstop_and_uses_extended_stop_timeout(self
         assert "ExecReload=/bin/kill -USR1 $MAINPID" in unit
         assert f"RestartForceExitStatus={GATEWAY_SERVICE_RESTART_EXIT_CODE}" in unit
         assert f"RestartPreventExitStatus={GATEWAY_FATAL_CONFIG_EXIT_CODE}" in unit
-        # TimeoutStopSec must exceed the default drain_timeout (60s) so
-        # systemd doesn't SIGKILL the cgroup before post-interrupt cleanup
-        # (tool subprocess kill, adapter disconnect) runs — issue #8202.
+        # The default drain is immediate, so keep a bounded 60-second stop
+        # budget without forcing every restart to wait 90 seconds.
         assert self._expected_timeout_stop_sec() in unit
         # ExecStopPost reaps any process the gateway didn't clean up itself,
         # so long-lived helpers (e.g. adb) can't be left orphaned in the
@@ -450,6 +449,13 @@ def test_user_unit_avoids_recursive_execstop_and_uses_extended_stop_timeout(self
         # tool-call children before systemd SIGKILLs the cgroup — #8202.
         assert "KillMode=mixed" in unit
 
+    def test_user_unit_adds_cleanup_headroom_to_positive_drain_timeout(self, monkeypatch):
+        monkeypatch.setattr(gateway_cli, "_get_restart_drain_timeout", lambda: 45)
+
+        unit = gateway_cli.generate_systemd_unit(system=False)
+
+        assert "TimeoutStopSec=75" in unit
+
     def test_user_unit_includes_resolved_node_directory_in_path(self, monkeypatch):
         monkeypatch.setattr(gateway_cli.shutil, "which", lambda cmd: "/home/test/.nvm/versions/node/v24.14.0/bin/node" if cmd == "node" else None)
 
@@ -549,9 +555,8 @@ def test_system_unit_avoids_recursive_execstop_and_uses_extended_stop_timeout(se
         assert "ExecReload=/bin/kill -USR1 $MAINPID" in unit
         assert f"RestartForceExitStatus={GATEWAY_SERVICE_RESTART_EXIT_CODE}" in unit
         assert f"RestartPreventExitStatus={GATEWAY_FATAL_CONFIG_EXIT_CODE}" in unit
-        # TimeoutStopSec must exceed the default drain_timeout (60s) so
-        # systemd doesn't SIGKILL the cgroup before post-interrupt cleanup
-        # (tool subprocess kill, adapter disconnect) runs — issue #8202.
+        # The default drain is immediate, so keep a bounded 60-second stop
+        # budget without forcing every restart to wait 90 seconds.
         assert self._expected_timeout_stop_sec() in unit
         assert "WantedBy=multi-user.target" in unit
         # ExecStopPost reaps any process the gateway didn't clean up itself,

tests/hermes_cli/test_systemd_watchdog_unit.py

@@ -0,0 +1,123 @@
+"""Generated service behavior for the opt-in systemd watchdog."""
+
+from __future__ import annotations
+
+from gateway.config import GatewayConfig
+from hermes_cli import gateway as gateway_cli
+
+
+def test_default_user_unit_keeps_simple_service_without_watchdog(monkeypatch):
+    monkeypatch.setattr(gateway_cli, "load_gateway_config", GatewayConfig)
+
+    unit = gateway_cli.generate_systemd_unit(system=False)
+
+    assert "Type=simple" in unit
+    assert "Type=notify" not in unit
+    assert "NotifyAccess=" not in unit
+    assert "WatchdogSec=" not in unit
+
+
+def test_positive_watchdog_config_generates_notify_unit(monkeypatch):
+    monkeypatch.setattr(
+        gateway_cli,
+        "load_gateway_config",
+        lambda: GatewayConfig.from_dict({"systemd_watchdog_seconds": 120}),
+        raising=False,
+    )
+
+    unit = gateway_cli.generate_systemd_unit(system=False)
+
+    assert "Type=notify" in unit
+    assert "NotifyAccess=main" in unit
+    assert "WatchdogSec=120s" in unit
+
+
+def test_positive_watchdog_config_generates_notify_system_unit(monkeypatch, tmp_path):
+    monkeypatch.setattr(
+        gateway_cli,
+        "load_gateway_config",
+        lambda: GatewayConfig(systemd_watchdog_seconds=30),
+    )
+    monkeypatch.setattr(
+        gateway_cli,
+        "_system_service_identity",
+        lambda _user: ("hermes", "hermes", str(tmp_path)),
+    )
+
+    unit = gateway_cli.generate_systemd_unit(system=True, run_as_user="hermes")
+
+    assert "Type=notify" in unit
+    assert "NotifyAccess=main" in unit
+    assert "WatchdogSec=30s" in unit
+
+
+def test_user_unit_reads_watchdog_from_config_yaml(tmp_path, monkeypatch):
+    hermes_home = tmp_path / "home"
+    hermes_home.mkdir()
+    (hermes_home / "config.yaml").write_text(
+        "gateway:\n  systemd_watchdog_seconds: 45\n",
+        encoding="utf-8",
+    )
+    monkeypatch.setenv("HERMES_HOME", str(hermes_home))
+
+    unit = gateway_cli.generate_systemd_unit(system=False)
+
+    assert "Type=notify" in unit
+    assert "NotifyAccess=main" in unit
+    assert "WatchdogSec=45s" in unit
+
+
+def test_system_unit_reads_watchdog_from_target_home(tmp_path, monkeypatch):
+    caller_home = tmp_path / "caller"
+    target_home = tmp_path / "target"
+    caller_home.mkdir()
+    target_home.mkdir()
+    (caller_home / "config.yaml").write_text(
+        "gateway:\n  systemd_watchdog_seconds: 0\n",
+        encoding="utf-8",
+    )
+    (target_home / "config.yaml").write_text(
+        "gateway:\n  systemd_watchdog_seconds: 75\n",
+        encoding="utf-8",
+    )
+    monkeypatch.setenv("HERMES_HOME", str(caller_home))
+    monkeypatch.setattr(
+        gateway_cli,
+        "_system_service_identity",
+        lambda _user: ("service", "service", str(tmp_path / "account")),
+    )
+    monkeypatch.setattr(
+        gateway_cli,
+        "_hermes_home_for_target_user",
+        lambda _home: str(target_home),
+    )
+
+    unit = gateway_cli.generate_systemd_unit(system=True, run_as_user="service")
+
+    assert "Type=notify" in unit
+    assert "WatchdogSec=75s" in unit
+
+
+def test_managed_watchdog_override_controls_generated_unit(tmp_path, monkeypatch):
+    hermes_home = tmp_path / "home"
+    managed_home = tmp_path / "managed"
+    hermes_home.mkdir()
+    managed_home.mkdir()
+    (hermes_home / "config.yaml").write_text(
+        "gateway:\n  systemd_watchdog_seconds: 120\n",
+        encoding="utf-8",
+    )
+    (managed_home / "config.yaml").write_text(
+        "gateway:\n  systemd_watchdog_seconds: 0\n",
+        encoding="utf-8",
+    )
+    monkeypatch.setenv("HERMES_HOME", str(hermes_home))
+    monkeypatch.setenv("HERMES_MANAGED_DIR", str(managed_home))
+
+    from hermes_cli import managed_scope
+
+    managed_scope.invalidate_managed_cache()
+    unit = gateway_cli.generate_systemd_unit(system=False)
+
+    assert "Type=simple" in unit
+    assert "WatchdogSec=" not in unit

website/docs/reference/environment-variables.md

@@ -682,8 +682,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_DISCORD_LIVENESS_INTERVAL_SECONDS` | Compatibility/manual override for `discord.websocket_liveness_interval_seconds`. Interval for sampling the active Discord Gateway WebSocket (default: `15`; set to `0` to disable). Prefer the `config.yaml` key. |
+| `HERMES_DISCORD_LIVENESS_FAILURE_THRESHOLD` | Compatibility/manual override for `discord.websocket_liveness_failure_threshold`. Consecutive unhealthy WebSocket samples before forcing a reconnect (default: `2`). Prefer the `config.yaml` key. |
 | `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. |

website/docs/user-guide/messaging/discord.md

@@ -82,6 +82,24 @@ With `group_sessions_per_user: false`:
 
 This guide walks you through the full setup process — from creating your bot on Discord's Developer Portal to sending your first message.
 
+### Gateway WebSocket health
+
+Discord REST and the Gateway WebSocket are separate transports. A successful REST response (including `fetch_user()` returning HTTP 200) does not prove that the bot can still receive Gateway events. Hermes therefore combines the ready state, client/socket closure state, socket openness, heartbeat ACK age, and finite heartbeat latency.
+
+After the configured number of consecutive unhealthy samples, the adapter emits one retryable fatal event. The existing gateway reconnect watcher creates a fresh adapter; the Discord adapter does not start a second unbounded reconnect loop.
+
+Configure the non-secret thresholds in `config.yaml`:
+
+```yaml
+discord:
+  websocket_liveness_interval_seconds: 15
+  websocket_liveness_failure_threshold: 2
+  websocket_heartbeat_ack_max_age_seconds: 60
+  websocket_max_latency_seconds: 30
+```
+
+The old `liveness_interval_seconds` and `liveness_failure_threshold` names remain compatibility aliases only; they no longer mean REST probing.
+
 ## Step 1: Create a Discord Application
 
 1. Go to the [Discord Developer Portal](https://discord.com/developers/applications) and sign in with your Discord account.

website/docs/user-guide/messaging/index.md

@@ -154,6 +154,30 @@ hermes gateway status       # Check default service status
 hermes gateway status --system         # Linux only: inspect the system service explicitly
 ```
 
+### Optional Linux event-loop watchdog
+
+A systemd-managed gateway can opt into process recovery when Python's asyncio
+event loop stops receiving scheduling time. This covers whole-process stalls
+that also prevent platform-specific liveness tasks from running:
+
+```yaml title="~/.hermes/config.yaml"
+gateway:
+  systemd_watchdog_seconds: 120
+```
+
+Regenerate the service unit after changing this setting:
+
+```bash
+hermes gateway install --force
+```
+
+A positive value makes the generated unit use `Type=notify`,
+`NotifyAccess=main`, and the matching `WatchdogSec`. Hermes sends heartbeats
+only while its event loop is making timely progress; systemd restarts the
+process when they stop. The default `0` keeps the existing `Type=simple`
+behavior. This setting is Linux/systemd-only and does not treat an ordinary
+platform network disconnect as an event-loop failure.
+
 ## Chat Commands (Inside Messaging)
 
 | Command | Description |

website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/user-guide/messaging/discord.md

@@ -82,6 +82,22 @@ Hermes 按会话键跟踪正在运行的 agent。
 
 本指南将引导你完成完整的设置流程——从在 Discord 开发者门户创建机器人到发送第一条消息。
 
+### Discord Gateway WebSocket 健康
+
+Discord REST 与 Gateway WebSocket 是独立传输。REST 请求成功不代表机器人仍能接收 Gateway 事件。Hermes 会组合检查 ready 状态、client/socket 关闭状态、socket 是否打开、heartbeat ACK 年龄和有限 heartbeat latency。
+
+连续异常达到阈值后，适配器只上报一次可重试失败；现有 Gateway 重连器创建新适配器，不会启动第二个无限重连循环。
+
+```yaml
+discord:
+  websocket_liveness_interval_seconds: 15
+  websocket_liveness_failure_threshold: 2
+  websocket_heartbeat_ack_max_age_seconds: 60
+  websocket_max_latency_seconds: 30
+```
+
+旧的 `liveness_interval_seconds` / `liveness_failure_threshold` 仅作为迁移别名保留，不再表示 REST probe。
+
 ## 第一步：创建 Discord 应用
 
 1. 前往 [Discord 开发者门户](https://discord.com/developers/applications) 并使用你的 Discord 账号登录。

1 PR closed without merging

From the close comments: merged via another change 1

#26684 fix(gateway/discord): force reconnect on REST liveness probe failure (#26656) · merged via another change → 6f1a176, #53924

Closed issues · 1

Closed#26656Discord adapter enters silent zombie state after network outage; never auto-reconnects

How this cluster was grouped

Link evidence (structural links throughout): every member is connected by structural evidence: shared changed files, or a closing/superseded reference.
Pairwise link density 0.50: the share of item pairs that carry a link, counting a structural link in full and a relevance-only link at 0.36. 1.00 means every pair is linked.

Pick up a Lab task

Check whether a fix covers the open issues here, with your own agent. Bring back evidence for maintainers.

Open the Lab

At a glance

Subsystem: gateway
Merged PRs: #53924, #66946
Closed issues: 1
Attempts: 1 PR closed without merging
Fix link checked: 18 Jul 2026

For your agent

The same facts, typed.

cluster/26656.json
cluster/26656.md