KrustyPlanet/hermes-agent
6
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

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

Browse source 
Replace REST-based Discord liveness probe with local WebSocket/heartbeat
state detection. REST success doesn't prove Gateway event delivery — a
half-closed WebSocket can leave Bot.start() alive while REST returns 200.
Now samples ready/open/ACK state and heartbeat latency; consecutive
unhealthy samples emit one retryable fatal code so GatewayRunner rebuilds
the adapter through the existing reconnect path.

Also fixes three lifecycle gaps in the recovery path:
1. asyncio.wait_for() can remain blocked if adapter cleanup swallows
   cancellation — now uses bounded asyncio.wait() with task detachment.
2. Multiplexed secondary-profile adapters had no profile-scoped reconnect
   owner — now uses one runner-owned reconnect slot per profile.
3. An in-flight turn could send its final text through the disconnected
   adapter after a replacement was registered — now resolves the live
   same-profile replacement for unsent final responses only (message IDs
   never migrate, edits/deletes stay on the old transport).

Adds an opt-in Linux/systemd event-loop watchdog (gateway.systemd_watchdog_seconds,
default 0) for the failure mode where the whole asyncio loop stops making
progress and no in-process liveness task can run. stdlib-only sd_notify,
Type=notify/WatchdogSec generation, READY/STOPPING lifecycle.

Co-authored-by: 王鑫 <wx.xw@bytedance.com>
This commit is contained in:
StellarisW 2026-07-18 19:22:50 +05:30 • committed by kshitij
parent c48a801b7c
commit f57157a128
24 changed files with 2426 additions and 198 deletions
Download patch file Download diff file Expand all files Collapse all files

67
gateway/config.py
View file

@ -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 @@ class GatewayConfig:
"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 @@ class 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 @@ class 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"]

50
gateway/platforms/base.py
View file

@ -4161,6 +4161,34 @@ class BasePlatformAdapter(ABC):
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 @@ class BasePlatformAdapter(ABC):
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 @@ class BasePlatformAdapter(ABC):
)
_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,

388
gateway/run.py
View file

@ -44,7 +44,7 @@ from collections import OrderedDict
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
@ -2975,6 +2975,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
@ -3055,6 +3057,8 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
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
@ -3536,6 +3540,43 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
if mode in {"voice_only", "all"} and key.startswith(prefix)
)
@staticmethod
def _consume_detached_adapter_cleanup_result(task: asyncio.Future[Any]) -> None:
"""Retrieve a detached cleanup task result without surfacing cancellation."""
try:
task.exception()
except (asyncio.CancelledError, Exception):
pass
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(self._consume_detached_adapter_cleanup_result)
raise
if task in done:
await task
return True
task.cancel()
task.add_done_callback(self._consume_detached_adapter_cleanup_result)
return False
async def _safe_adapter_disconnect(self, adapter, platform) -> None:
"""Call adapter.disconnect() defensively, swallowing any error.
@ -3549,16 +3590,15 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
"""
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",
@ -3578,42 +3618,40 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
``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()
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:
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,
)
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",
@ -4238,7 +4276,10 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
# 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:
@ -8455,6 +8496,62 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
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,
*,
@ -8595,6 +8692,12 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
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()
@ -9107,19 +9210,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
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):
@ -9136,6 +9227,192 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
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 = min(30 * (2 ** (attempts - 1)), 300)
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.
@ -21837,7 +22114,7 @@ async def start_gateway(config: Optional[GatewayConfig] = None, replace: bool =
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)
@ -21874,7 +22151,14 @@ async def start_gateway(config: Optional[GatewayConfig] = None, replace: bool =
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()

176
gateway/systemd_notify.py Normal file
View file

@ -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

8
hermes_cli/config.py
View file

@ -2532,6 +2532,14 @@ DEFAULT_CONFIG = {
"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

77
hermes_cli/gateway.py
View file

@ -29,6 +29,7 @@ if os.name == "posix":
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 = set_hermes_home_override(hermes_home)
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 @@ Wants=network-online.target
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 @@ WantedBy=multi-user.target
"""
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 @@ Wants=network-online.target
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}"

364
plugins/platforms/discord/adapter.py
View file

@ -14,6 +14,7 @@ import hashlib
import inspect
import json
import logging
import math
import os
import re
import struct
@ -133,6 +134,31 @@ 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
def _consume_background_task_result(task: asyncio.Task) -> None:
"""Retrieve a detached cleanup task result without surfacing cancellation."""
with suppress(asyncio.CancelledError, Exception):
task.exception()
async def _wait_for_ready_or_bot_exit(
ready_event: asyncio.Event,
bot_task: asyncio.Task,
@ -858,23 +884,33 @@ class DiscordAdapter(BasePlatformAdapter):
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 = env_int(
"HERMES_DISCORD_LIVENESS_FAILURE_THRESHOLD", 3
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._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 +938,38 @@ class DiscordAdapter(BasePlatformAdapter):
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 +1357,189 @@ class DiscordAdapter(BasePlatformAdapter):
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())
async def _liveness_loop(self) -> None:
"""Probe Discord REST periodically and force a reconnect on persistent failure.
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"
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.
"""
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:
"""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):
if not self._running or client is None or self._disconnecting:
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,
)
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:
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 client.close()
if _abort_discord_websocket_transport(failed_websocket):
logger.warning(
"[%s] Aborted unresponsive Discord WebSocket transport",
self.name,
)
except Exception:
logger.debug(
"[%s] Error closing wedged Discord client", self.name, exc_info=True,
"[%s] Error aborting unhealthy Discord WebSocket transport",
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
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 +8593,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 +8685,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:

42
tests/gateway/test_bounded_adapter_teardown.py
View file

@ -91,6 +91,48 @@ async def test_teardown_bounds_hanging_cancel(bare_runner, monkeypatch, caplog):
adapter.disconnect.assert_awaited_once()
@pytest.mark.asyncio
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."""

56
tests/gateway/test_config.py
View file

@ -4,6 +4,8 @@ import logging
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 @@ class TestGatewayConfigRoundtrip:
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 @@ class TestGatewayConfigRoundtrip:
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 @@ class TestLoadGatewayConfig:
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

51
tests/gateway/test_discord_approval_mentions.py
View file

@ -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"

442
tests/gateway/test_discord_liveness.py
View file

@ -1,18 +1,17 @@
"""Regression tests for the Discord REST liveness probe (#26656).
"""Regression tests for Discord Gateway WebSocket liveness.
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.
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 @@ from tests.gateway.test_discord_connect import ( # noqa: E402
_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 @@ class _LiveBot(FakeBot):
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 @@ class _LiveBot(FakeBot):
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="test-token"))
return DiscordAdapter(
PlatformConfig(
enabled=True,
token="test-token",
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")
@pytest.mark.parametrize(
("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="test-token", 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="test-token"))
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="test-token",
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 @@ async def test_liveness_probe_disabled_when_threshold_zero(monkeypatch):
@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 @@ async def test_liveness_probe_forces_reconnect_after_threshold(monkeypatch):
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()
@pytest.mark.asyncio
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()
@pytest.mark.asyncio
@pytest.mark.parametrize(
("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()
@pytest.mark.asyncio
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()
@pytest.mark.asyncio
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()
@pytest.mark.asyncio
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()
@pytest.mark.asyncio
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()
@pytest.mark.asyncio
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

20
tests/gateway/test_gateway_shutdown.py
View file

@ -118,6 +118,26 @@ async def test_gateway_stop_drains_running_agents_before_disconnect():
assert runner._shutdown_event.is_set() is True
@pytest.mark.asyncio
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()

245
tests/gateway/test_multiplex_adapter_registry.py
View file

@ -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 TestProfileMessageHandler:
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="profile-token"
)
},
),
)
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."""

135
tests/gateway/test_runner_fatal_adapter.py
View file

@ -4,8 +4,9 @@ from unittest.mock import AsyncMock
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 @@ class _RuntimeRetryableAdapter(BasePlatformAdapter):
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
@pytest.mark.asyncio
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()
@pytest.mark.asyncio
@pytest.mark.parametrize("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"]

44
tests/gateway/test_safe_adapter_disconnect.py
View file

@ -77,3 +77,47 @@ async def test_safe_disconnect_times_out_and_continues(bare_runner, monkeypatch,
adapter.disconnect.assert_awaited_once()
assert "Timed out after 0.0s while disconnecting feishu adapter" in caplog.text
@pytest.mark.asyncio
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)

3
tests/gateway/test_shutdown_cache_cleanup.py
View file

@ -81,6 +81,9 @@ class _FakeGateway:
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

154
tests/gateway/test_systemd_notify.py Normal file
View file

@ -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()
@pytest.mark.skipif(
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)
@pytest.mark.parametrize("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
@pytest.mark.asyncio
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 == []

98
tests/gateway/test_systemd_watchdog_lifecycle.py Normal file
View file

@ -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
@pytest.mark.asyncio
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",
]

19
tests/hermes_cli/test_gateway_service.py
View file

@ -421,7 +421,7 @@ class TestRequireServiceInstalled:
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 @@ class TestGeneratedSystemdUnits:
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 @@ class TestGeneratedSystemdUnits:
# 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 @@ class TestGeneratedSystemdUnits:
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,

123
tests/hermes_cli/test_systemd_watchdog_unit.py Normal file
View file

@ -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

4
website/docs/reference/environment-variables.md
View file

@ -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. |

18
website/docs/user-guide/messaging/discord.md
View file

@ -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.

24
website/docs/user-guide/messaging/index.md
View file

@ -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 |

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

@ -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 账号登录。