From f57157a1284e98dbb1161c9de1c62e7bf7a164f1 Mon Sep 17 00:00:00 2001 From: StellarisW Date: Sat, 18 Jul 2026 19:22:50 +0530 Subject: [PATCH] fix(gateway): recover Discord websocket and event-loop stalls MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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: 王鑫 --- gateway/config.py | 67 +++ gateway/platforms/base.py | 50 +- gateway/run.py | 388 ++++++++++++--- gateway/systemd_notify.py | 176 +++++++ hermes_cli/config.py | 8 + hermes_cli/gateway.py | 77 ++- plugins/platforms/discord/adapter.py | 364 +++++++++++---- .../gateway/test_bounded_adapter_teardown.py | 42 ++ tests/gateway/test_config.py | 56 +++ .../gateway/test_discord_approval_mentions.py | 51 +- tests/gateway/test_discord_liveness.py | 442 ++++++++++++++++-- tests/gateway/test_gateway_shutdown.py | 20 + .../test_multiplex_adapter_registry.py | 245 ++++++++++ tests/gateway/test_runner_fatal_adapter.py | 135 +++++- tests/gateway/test_safe_adapter_disconnect.py | 44 ++ tests/gateway/test_shutdown_cache_cleanup.py | 3 + tests/gateway/test_systemd_notify.py | 154 ++++++ .../test_systemd_watchdog_lifecycle.py | 98 ++++ tests/hermes_cli/test_gateway_service.py | 19 +- .../hermes_cli/test_systemd_watchdog_unit.py | 123 +++++ .../docs/reference/environment-variables.md | 4 +- website/docs/user-guide/messaging/discord.md | 18 + website/docs/user-guide/messaging/index.md | 24 + .../current/user-guide/messaging/discord.md | 16 + 24 files changed, 2426 insertions(+), 198 deletions(-) create mode 100644 gateway/systemd_notify.py create mode 100644 tests/gateway/test_systemd_notify.py create mode 100644 tests/gateway/test_systemd_watchdog_lifecycle.py create mode 100644 tests/hermes_cli/test_systemd_watchdog_unit.py diff --git a/gateway/config.py b/gateway/config.py index af2007c9e..e526445f3 100644 --- a/gateway/config.py +++ b/gateway/config.py @@ -130,6 +130,49 @@ def _coerce_optional_positive_int(value: Any, key: str) -> Optional[int]: return parsed +_SYSTEMD_WATCHDOG_MAX_SECONDS = 2_147_483_647 + + +def coerce_systemd_watchdog_seconds( + value: Any, key: str = "gateway.systemd_watchdog_seconds" +) -> int: + """Return a bounded positive watchdog interval or zero when disabled. + + Runtime and service generation share this normalization so a value can + never enable ``Type=notify`` while disabling application heartbeats. + """ + if value is None: + return 0 + if isinstance(value, bool): + logger.warning("Ignoring invalid %s (expected a positive integer)", key) + return 0 + if isinstance(value, int): + parsed = value + elif isinstance(value, str): + raw = value.strip() + if not raw or not raw.isascii() or not raw.isdecimal(): + logger.warning("Ignoring invalid %s (expected a positive integer)", key) + return 0 + try: + parsed = int(raw, 10) + except (TypeError, ValueError, OverflowError): + logger.warning("Ignoring invalid %s (expected a positive integer)", key) + return 0 + else: + logger.warning("Ignoring invalid %s (expected a positive integer)", key) + return 0 + if parsed == 0: + return 0 + if not 0 < parsed <= _SYSTEMD_WATCHDOG_MAX_SECONDS: + logger.warning( + "Ignoring invalid %s (expected an integer from 1 to %d)", + key, + _SYSTEMD_WATCHDOG_MAX_SECONDS, + ) + return 0 + return parsed + + def _coerce_dict(value: Any) -> Dict[str, Any]: """Return *value* when it is a mapping, otherwise an empty dict.""" return value if isinstance(value, dict) else {} @@ -768,6 +811,10 @@ class GatewayConfig: # gateway behaves exactly as before — single HERMES_HOME, no profile stamping. multiplex_profiles: bool = False + # Opt-in systemd event-loop watchdog. Zero preserves Type=simple and + # disables sd_notify at runtime. + systemd_watchdog_seconds: int = 0 + # Unauthorized DM policy unauthorized_dm_behavior: str = "pair" # "pair" or "ignore" @@ -786,6 +833,11 @@ class GatewayConfig: # dict with: name, platform, profile, and optional guild_id/chat_id/thread_id. profile_routes: list = field(default_factory=list) + def __post_init__(self) -> None: + self.systemd_watchdog_seconds = coerce_systemd_watchdog_seconds( + self.systemd_watchdog_seconds + ) + def get_connected_platforms(self) -> List[Platform]: """Return list of platforms that are enabled and configured.""" connected = [] @@ -889,6 +941,7 @@ 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"] diff --git a/gateway/platforms/base.py b/gateway/platforms/base.py index 427a37afe..e39569acb 100644 --- a/gateway/platforms/base.py +++ b/gateway/platforms/base.py @@ -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, diff --git a/gateway/run.py b/gateway/run.py index b447d329e..fb3fe77f6 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -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() diff --git a/gateway/systemd_notify.py b/gateway/systemd_notify.py new file mode 100644 index 000000000..4fb2b9d5e --- /dev/null +++ b/gateway/systemd_notify.py @@ -0,0 +1,176 @@ +"""Minimal, optional systemd ``sd_notify`` support for the gateway.""" + +from __future__ import annotations + +import asyncio +import math +import os +import socket +from typing import Optional + + +def _notify_address(raw: str) -> str: + """Translate systemd's ``@abstract`` notation to Python's address form.""" + return "\0" + raw[1:] if raw.startswith("@") else raw + + +def notify(message: str) -> bool: + """Send one nonblocking sd_notify datagram when systemd configured it. + + Notification failures are deliberately non-fatal: a missing socket or an + older platform must never prevent the gateway from starting. + """ + address = os.environ.get("NOTIFY_SOCKET", "").strip() + if not address: + return False + if not isinstance(message, str) or not message: + return False + if not hasattr(socket, "AF_UNIX"): + return False + try: + payload = message.encode("utf-8") + with socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM) as sender: + # A full receiver buffer must not stall the gateway event loop. + sender.setblocking(False) + sender.connect(_notify_address(address)) + sender.send(payload) + return True + except (OSError, UnicodeError, ValueError): + return False + + +def watchdog_interval_seconds() -> Optional[float]: + """Return systemd's configured watchdog interval in seconds.""" + if not os.environ.get("NOTIFY_SOCKET", "").strip(): + return None + if not hasattr(socket, "AF_UNIX"): + return None + raw = os.environ.get("WATCHDOG_USEC", "").strip() + if not raw: + return None + try: + interval = float(raw) / 1_000_000.0 + except (TypeError, ValueError): + return None + if not math.isfinite(interval) or interval <= 0: + return None + return interval + + +class SystemdWatchdog: + """Feed systemd while the asyncio event loop continues to make progress.""" + + def __init__( + self, + *, + config_enabled: bool = True, + lag_tolerance_seconds: Optional[float] = None, + ) -> None: + self._config_enabled = bool(config_enabled) + self.interval_seconds = watchdog_interval_seconds() + self._lag_tolerance_seconds = lag_tolerance_seconds + self._task: Optional[asyncio.Task[None]] = None + self._unhealthy = False + self._stopping = False + self._stopping_notified = False + + @property + def enabled(self) -> bool: + return self._config_enabled and self.interval_seconds is not None + + @property + def unhealthy(self) -> bool: + return self._unhealthy + + @property + def task(self) -> Optional[asyncio.Task[None]]: + return self._task + + def _lag_tolerance(self) -> float: + interval = self.interval_seconds or 0.0 + configured = self._lag_tolerance_seconds + if configured is None: + return max(0.1, interval * 0.25) + try: + value = float(configured) + except (TypeError, ValueError): + return max(0.1, interval * 0.25) + if not math.isfinite(value): + return max(0.1, interval * 0.25) + return max(0.0, value) + + def start(self) -> bool: + """Start the loop-progress sampler when systemd watchdog is enabled.""" + if not self.enabled: + return False + if self._task is not None and not self._task.done(): + return True + try: + asyncio.get_running_loop() + except RuntimeError: + return False + self._stopping = False + self._unhealthy = False + self._stopping_notified = False + self._task = asyncio.create_task(self._run(), name="hermes-systemd-watchdog") + return True + + def ready(self, status: str = "Gateway running") -> bool: + """Tell systemd that startup completed and the gateway is ready.""" + if not self.enabled: + return False + safe_status = str(status or "Gateway running").replace("\n", " ") + return notify(f"READY=1\nSTATUS={safe_status}") + + def record_tick(self, *, scheduled_at: float, now: float) -> bool: + """Feed systemd only when the event loop woke within its lag budget.""" + if not self.enabled or self._stopping or self._unhealthy: + return False + try: + lag = float(now) - float(scheduled_at) + except (TypeError, ValueError): + lag = float("inf") + if not math.isfinite(lag) or lag > self._lag_tolerance(): + self._unhealthy = True + notify("STATUS=watchdog unhealthy: event loop progress is late") + return False + notify("WATCHDOG=1") + return True + + async def _run(self) -> None: + interval = self.interval_seconds + if interval is None: + return + cadence = max(0.01, interval / 2.0) + loop = asyncio.get_running_loop() + scheduled_at = loop.time() + cadence + try: + while not self._stopping and not self._unhealthy: + await asyncio.sleep(max(0.0, scheduled_at - loop.time())) + now = loop.time() + if not self.record_tick(scheduled_at=scheduled_at, now=now): + return + scheduled_at += cadence + if scheduled_at < now: + scheduled_at = now + cadence + except asyncio.CancelledError: + return + + async def stop(self) -> None: + """Stop feeding systemd and emit ``STOPPING=1`` at most once.""" + self._stopping = True + task = self._task + current = asyncio.current_task() + if task is not None and task is not current: + if not task.done(): + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + except Exception: + pass + self._task = None + if self.enabled and not self._stopping_notified: + notify("STOPPING=1") + self._stopping_notified = True diff --git a/hermes_cli/config.py b/hermes_cli/config.py index 0bbd1f703..f7d83e092 100644 --- a/hermes_cli/config.py +++ b/hermes_cli/config.py @@ -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 diff --git a/hermes_cli/gateway.py b/hermes_cli/gateway.py index 9868b976c..5b33d7646 100644 --- a/hermes_cli/gateway.py +++ b/hermes_cli/gateway.py @@ -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}" diff --git a/plugins/platforms/discord/adapter.py b/plugins/platforms/discord/adapter.py index 8257ccbe4..9356d2068 100644 --- a/plugins/platforms/discord/adapter.py +++ b/plugins/platforms/discord/adapter.py @@ -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: diff --git a/tests/gateway/test_bounded_adapter_teardown.py b/tests/gateway/test_bounded_adapter_teardown.py index abe20608d..df2049df3 100644 --- a/tests/gateway/test_bounded_adapter_teardown.py +++ b/tests/gateway/test_bounded_adapter_teardown.py @@ -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.""" diff --git a/tests/gateway/test_config.py b/tests/gateway/test_config.py index 27992663e..20211fb3c 100644 --- a/tests/gateway/test_config.py +++ b/tests/gateway/test_config.py @@ -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 diff --git a/tests/gateway/test_discord_approval_mentions.py b/tests/gateway/test_discord_approval_mentions.py index 786ecace3..7696058ef 100644 --- a/tests/gateway/test_discord_approval_mentions.py +++ b/tests/gateway/test_discord_approval_mentions.py @@ -83,5 +83,54 @@ def test_yaml_config_bridges_approval_mentions_to_env(monkeypatch): {"discord": {"approval_mentions": True}}, {"approval_mentions": True}, ) - assert os.environ["DISCORD_APPROVAL_MENTIONS"] == "true" + + +def test_yaml_config_seeds_websocket_health_with_primary_precedence(monkeypatch): + for key in ( + "HERMES_DISCORD_LIVENESS_INTERVAL_SECONDS", + "HERMES_DISCORD_LIVENESS_FAILURE_THRESHOLD", + ): + monkeypatch.delenv(key, raising=False) + + seeded = _apply_yaml_config( + {}, + { + "websocket_liveness_interval_seconds": 11, + "liveness_interval_seconds": 99, + "websocket_liveness_failure_threshold": 2, + "websocket_heartbeat_ack_max_age_seconds": 75, + "websocket_max_latency_seconds": 30, + }, + ) + + assert os.environ["HERMES_DISCORD_LIVENESS_INTERVAL_SECONDS"] == "11" + assert os.environ["HERMES_DISCORD_LIVENESS_FAILURE_THRESHOLD"] == "2" + assert seeded == { + "websocket_liveness_interval_seconds": 11, + "websocket_liveness_failure_threshold": 2, + "websocket_heartbeat_ack_max_age_seconds": 75, + "websocket_max_latency_seconds": 30, + } + + +def test_yaml_config_bridges_nested_discord_extra_websocket_health(monkeypatch): + for key in ( + "HERMES_DISCORD_LIVENESS_INTERVAL_SECONDS", + "HERMES_DISCORD_LIVENESS_FAILURE_THRESHOLD", + ): + monkeypatch.delenv(key, raising=False) + + _apply_yaml_config( + {"platforms": {"discord": {"extra": { + "websocket_liveness_interval_seconds": 13, + "websocket_liveness_failure_threshold": 4, + }}}}, + {"extra": { + "websocket_liveness_interval_seconds": 13, + "websocket_liveness_failure_threshold": 4, + }}, + ) + + assert os.environ["HERMES_DISCORD_LIVENESS_INTERVAL_SECONDS"] == "13" + assert os.environ["HERMES_DISCORD_LIVENESS_FAILURE_THRESHOLD"] == "4" diff --git a/tests/gateway/test_discord_liveness.py b/tests/gateway/test_discord_liveness.py index b54dd6bfd..9ebffe951 100644 --- a/tests/gateway/test_discord_liveness.py +++ b/tests/gateway/test_discord_liveness.py @@ -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 diff --git a/tests/gateway/test_gateway_shutdown.py b/tests/gateway/test_gateway_shutdown.py index af276ce6b..371dae052 100644 --- a/tests/gateway/test_gateway_shutdown.py +++ b/tests/gateway/test_gateway_shutdown.py @@ -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() diff --git a/tests/gateway/test_multiplex_adapter_registry.py b/tests/gateway/test_multiplex_adapter_registry.py index 4c8f6497e..2b2289437 100644 --- a/tests/gateway/test_multiplex_adapter_registry.py +++ b/tests/gateway/test_multiplex_adapter_registry.py @@ -1,8 +1,14 @@ """Phase 3: secondary-profile adapter registry + same-token conflict detection.""" import logging +import asyncio +from contextlib import contextmanager +from pathlib import Path +from unittest.mock import AsyncMock import pytest +import gateway.run as gateway_run +from gateway.config import GatewayConfig, Platform, PlatformConfig from gateway.run import GatewayRunner @@ -144,6 +150,245 @@ class 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.""" diff --git a/tests/gateway/test_runner_fatal_adapter.py b/tests/gateway/test_runner_fatal_adapter.py index 7fce3841f..c37eef576 100644 --- a/tests/gateway/test_runner_fatal_adapter.py +++ b/tests/gateway/test_runner_fatal_adapter.py @@ -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"] diff --git a/tests/gateway/test_safe_adapter_disconnect.py b/tests/gateway/test_safe_adapter_disconnect.py index 9a17aa047..9c1916e66 100644 --- a/tests/gateway/test_safe_adapter_disconnect.py +++ b/tests/gateway/test_safe_adapter_disconnect.py @@ -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) diff --git a/tests/gateway/test_shutdown_cache_cleanup.py b/tests/gateway/test_shutdown_cache_cleanup.py index 45a3461d0..cc033b789 100644 --- a/tests/gateway/test_shutdown_cache_cleanup.py +++ b/tests/gateway/test_shutdown_cache_cleanup.py @@ -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 diff --git a/tests/gateway/test_systemd_notify.py b/tests/gateway/test_systemd_notify.py new file mode 100644 index 000000000..f998fce5a --- /dev/null +++ b/tests/gateway/test_systemd_notify.py @@ -0,0 +1,154 @@ +"""Tests for the optional systemd event-loop watchdog protocol.""" + +from __future__ import annotations + +import asyncio +import socket + +import pytest + + +def test_notify_without_notify_socket_is_a_noop(monkeypatch): + monkeypatch.delenv("NOTIFY_SOCKET", raising=False) + + from gateway.systemd_notify import notify + + assert notify("READY=1") is False + + +def test_notify_sends_real_unix_datagram(tmp_path, monkeypatch): + address = str(tmp_path / "notify.sock") + receiver = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM) + receiver.bind(address) + receiver.settimeout(1.0) + monkeypatch.setenv("NOTIFY_SOCKET", address) + + from gateway.systemd_notify import notify + + assert notify("READY=1") is True + assert receiver.recv(4096) == b"READY=1" + receiver.close() + + +@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 == [] diff --git a/tests/gateway/test_systemd_watchdog_lifecycle.py b/tests/gateway/test_systemd_watchdog_lifecycle.py new file mode 100644 index 000000000..0ab4afcac --- /dev/null +++ b/tests/gateway/test_systemd_watchdog_lifecycle.py @@ -0,0 +1,98 @@ +"""Gateway lifecycle contract for the opt-in systemd watchdog.""" + +from __future__ import annotations + +import inspect +from unittest.mock import patch + +import pytest + +from gateway.config import GatewayConfig +from gateway.run import GatewayRunner, start_gateway +from tests.gateway.restart_test_helpers import make_restart_runner + + +class _FakeWatchdog: + instances: list["_FakeWatchdog"] = [] + + def __init__(self, *, config_enabled: bool = True): + self.config_enabled = config_enabled + self.calls: list[str] = [] + self.__class__.instances.append(self) + + def start(self) -> bool: + self.calls.append("start") + return self.config_enabled + + def ready(self, status: str) -> bool: + self.calls.append(f"ready:{status}") + return True + + async def stop(self) -> None: + self.calls.append("stop") + + +def _bare_runner(*, seconds: int, running: bool = True) -> GatewayRunner: + runner = object.__new__(GatewayRunner) + runner.config = GatewayConfig(systemd_watchdog_seconds=seconds) + runner._running = running + runner._systemd_watchdog = None + return runner + + +def test_runner_starts_watchdog_only_after_running(monkeypatch): + _FakeWatchdog.instances.clear() + monkeypatch.setattr("gateway.systemd_notify.SystemdWatchdog", _FakeWatchdog) + runner = _bare_runner(seconds=120, running=True) + + assert runner._start_systemd_watchdog() is True + + watchdog = _FakeWatchdog.instances[-1] + assert watchdog.config_enabled is True + assert watchdog.calls == ["start", "ready:Hermes Gateway running"] + + +def test_runner_does_not_start_watchdog_when_disabled_or_not_running(monkeypatch): + _FakeWatchdog.instances.clear() + monkeypatch.setattr("gateway.systemd_notify.SystemdWatchdog", _FakeWatchdog) + + assert _bare_runner(seconds=0)._start_systemd_watchdog() is False + assert _bare_runner(seconds=120, running=False)._start_systemd_watchdog() is False + assert _FakeWatchdog.instances == [] + + +def test_gateway_ready_follows_background_service_startup(): + source = inspect.getsource(start_gateway) + + housekeeping_started = source.index("housekeeping_thread.start()") + watchdog_started = source.index("start_watchdog()") + shutdown_wait = source.index("await runner.wait_for_shutdown()", watchdog_started) + + assert housekeeping_started < watchdog_started < shutdown_wait + + +@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", + ] diff --git a/tests/hermes_cli/test_gateway_service.py b/tests/hermes_cli/test_gateway_service.py index 6dd6a2152..7041ab9ae 100644 --- a/tests/hermes_cli/test_gateway_service.py +++ b/tests/hermes_cli/test_gateway_service.py @@ -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, diff --git a/tests/hermes_cli/test_systemd_watchdog_unit.py b/tests/hermes_cli/test_systemd_watchdog_unit.py new file mode 100644 index 000000000..16183d648 --- /dev/null +++ b/tests/hermes_cli/test_systemd_watchdog_unit.py @@ -0,0 +1,123 @@ +"""Generated service behavior for the opt-in systemd watchdog.""" + +from __future__ import annotations + +from gateway.config import GatewayConfig +from hermes_cli import gateway as gateway_cli + + +def test_default_user_unit_keeps_simple_service_without_watchdog(monkeypatch): + monkeypatch.setattr(gateway_cli, "load_gateway_config", GatewayConfig) + + unit = gateway_cli.generate_systemd_unit(system=False) + + assert "Type=simple" in unit + assert "Type=notify" not in unit + assert "NotifyAccess=" not in unit + assert "WatchdogSec=" not in unit + + +def test_positive_watchdog_config_generates_notify_unit(monkeypatch): + monkeypatch.setattr( + gateway_cli, + "load_gateway_config", + lambda: GatewayConfig.from_dict({"systemd_watchdog_seconds": 120}), + raising=False, + ) + + unit = gateway_cli.generate_systemd_unit(system=False) + + assert "Type=notify" in unit + assert "NotifyAccess=main" in unit + assert "WatchdogSec=120s" in unit + + +def test_positive_watchdog_config_generates_notify_system_unit(monkeypatch, tmp_path): + monkeypatch.setattr( + gateway_cli, + "load_gateway_config", + lambda: GatewayConfig(systemd_watchdog_seconds=30), + ) + monkeypatch.setattr( + gateway_cli, + "_system_service_identity", + lambda _user: ("hermes", "hermes", str(tmp_path)), + ) + + unit = gateway_cli.generate_systemd_unit(system=True, run_as_user="hermes") + + assert "Type=notify" in unit + assert "NotifyAccess=main" in unit + assert "WatchdogSec=30s" in unit + + +def test_user_unit_reads_watchdog_from_config_yaml(tmp_path, monkeypatch): + hermes_home = tmp_path / "home" + hermes_home.mkdir() + (hermes_home / "config.yaml").write_text( + "gateway:\n systemd_watchdog_seconds: 45\n", + encoding="utf-8", + ) + monkeypatch.setenv("HERMES_HOME", str(hermes_home)) + + unit = gateway_cli.generate_systemd_unit(system=False) + + assert "Type=notify" in unit + assert "NotifyAccess=main" in unit + assert "WatchdogSec=45s" in unit + + +def test_system_unit_reads_watchdog_from_target_home(tmp_path, monkeypatch): + caller_home = tmp_path / "caller" + target_home = tmp_path / "target" + caller_home.mkdir() + target_home.mkdir() + (caller_home / "config.yaml").write_text( + "gateway:\n systemd_watchdog_seconds: 0\n", + encoding="utf-8", + ) + (target_home / "config.yaml").write_text( + "gateway:\n systemd_watchdog_seconds: 75\n", + encoding="utf-8", + ) + monkeypatch.setenv("HERMES_HOME", str(caller_home)) + monkeypatch.setattr( + gateway_cli, + "_system_service_identity", + lambda _user: ("service", "service", str(tmp_path / "account")), + ) + monkeypatch.setattr( + gateway_cli, + "_hermes_home_for_target_user", + lambda _home: str(target_home), + ) + + unit = gateway_cli.generate_systemd_unit(system=True, run_as_user="service") + + assert "Type=notify" in unit + assert "WatchdogSec=75s" in unit + + +def test_managed_watchdog_override_controls_generated_unit(tmp_path, monkeypatch): + hermes_home = tmp_path / "home" + managed_home = tmp_path / "managed" + hermes_home.mkdir() + managed_home.mkdir() + (hermes_home / "config.yaml").write_text( + "gateway:\n systemd_watchdog_seconds: 120\n", + encoding="utf-8", + ) + (managed_home / "config.yaml").write_text( + "gateway:\n systemd_watchdog_seconds: 0\n", + encoding="utf-8", + ) + monkeypatch.setenv("HERMES_HOME", str(hermes_home)) + monkeypatch.setenv("HERMES_MANAGED_DIR", str(managed_home)) + + from hermes_cli import managed_scope + + managed_scope.invalidate_managed_cache() + unit = gateway_cli.generate_systemd_unit(system=False) + + assert "Type=simple" in unit + assert "WatchdogSec=" not in unit diff --git a/website/docs/reference/environment-variables.md b/website/docs/reference/environment-variables.md index aab4cb6a4..bb9f22882 100644 --- a/website/docs/reference/environment-variables.md +++ b/website/docs/reference/environment-variables.md @@ -682,8 +682,8 @@ Advanced per-platform knobs for throttling the outbound message batcher. Most us | `HERMES_TELEGRAM_DISABLE_FALLBACK_IPS` | Disable the hard-coded Cloudflare fallback IPs used when DNS fails (`true`/`false`). | | `HERMES_DISCORD_TEXT_BATCH_DELAY_SECONDS` | Grace window before flushing a queued Discord text chunk (default: `0.6`). | | `HERMES_DISCORD_TEXT_BATCH_SPLIT_DELAY_SECONDS` | Delay between split chunks when a Discord message exceeds the length limit (default: `2.0`). | -| `HERMES_DISCORD_LIVENESS_INTERVAL_SECONDS` | Internal bridge for `discord.liveness_interval_seconds` (config.yaml). Interval for the Discord REST liveness probe that detects zombie clients behind dead proxies/NATs (default: `60`; set to `0` to disable). Prefer setting `discord.liveness_interval_seconds` in `config.yaml`. | -| `HERMES_DISCORD_LIVENESS_FAILURE_THRESHOLD` | Internal bridge for `discord.liveness_failure_threshold` (config.yaml). Consecutive probe failures before forcing a Discord reconnect (default: `3`). Prefer setting `discord.liveness_failure_threshold` in `config.yaml`. | +| `HERMES_DISCORD_LIVENESS_INTERVAL_SECONDS` | Compatibility/manual override for `discord.websocket_liveness_interval_seconds`. Interval for sampling the active Discord Gateway WebSocket (default: `15`; set to `0` to disable). Prefer the `config.yaml` key. | +| `HERMES_DISCORD_LIVENESS_FAILURE_THRESHOLD` | Compatibility/manual override for `discord.websocket_liveness_failure_threshold`. Consecutive unhealthy WebSocket samples before forcing a reconnect (default: `2`). Prefer the `config.yaml` key. | | `HERMES_MATRIX_TEXT_BATCH_DELAY_SECONDS` / `_SPLIT_DELAY_SECONDS` | Matrix equivalents of the Telegram batch knobs. | | `HERMES_FEISHU_TEXT_BATCH_DELAY_SECONDS` / `_SPLIT_DELAY_SECONDS` / `_MAX_CHARS` / `_MAX_MESSAGES` | Feishu batcher tuning — delay, split delay, max chars per message, max messages per batch. | | `HERMES_FEISHU_MEDIA_BATCH_DELAY_SECONDS` | Feishu media flush delay. | diff --git a/website/docs/user-guide/messaging/discord.md b/website/docs/user-guide/messaging/discord.md index 5835430c2..6b55e1c49 100644 --- a/website/docs/user-guide/messaging/discord.md +++ b/website/docs/user-guide/messaging/discord.md @@ -82,6 +82,24 @@ With `group_sessions_per_user: false`: This guide walks you through the full setup process — from creating your bot on Discord's Developer Portal to sending your first message. +### Gateway WebSocket health + +Discord REST and the Gateway WebSocket are separate transports. A successful REST response (including `fetch_user()` returning HTTP 200) does not prove that the bot can still receive Gateway events. Hermes therefore combines the ready state, client/socket closure state, socket openness, heartbeat ACK age, and finite heartbeat latency. + +After the configured number of consecutive unhealthy samples, the adapter emits one retryable fatal event. The existing gateway reconnect watcher creates a fresh adapter; the Discord adapter does not start a second unbounded reconnect loop. + +Configure the non-secret thresholds in `config.yaml`: + +```yaml +discord: + websocket_liveness_interval_seconds: 15 + websocket_liveness_failure_threshold: 2 + websocket_heartbeat_ack_max_age_seconds: 60 + websocket_max_latency_seconds: 30 +``` + +The old `liveness_interval_seconds` and `liveness_failure_threshold` names remain compatibility aliases only; they no longer mean REST probing. + ## Step 1: Create a Discord Application 1. Go to the [Discord Developer Portal](https://discord.com/developers/applications) and sign in with your Discord account. diff --git a/website/docs/user-guide/messaging/index.md b/website/docs/user-guide/messaging/index.md index ac69b9ffd..1b194e163 100644 --- a/website/docs/user-guide/messaging/index.md +++ b/website/docs/user-guide/messaging/index.md @@ -154,6 +154,30 @@ hermes gateway status # Check default service status hermes gateway status --system # Linux only: inspect the system service explicitly ``` +### Optional Linux event-loop watchdog + +A systemd-managed gateway can opt into process recovery when Python's asyncio +event loop stops receiving scheduling time. This covers whole-process stalls +that also prevent platform-specific liveness tasks from running: + +```yaml title="~/.hermes/config.yaml" +gateway: + systemd_watchdog_seconds: 120 +``` + +Regenerate the service unit after changing this setting: + +```bash +hermes gateway install --force +``` + +A positive value makes the generated unit use `Type=notify`, +`NotifyAccess=main`, and the matching `WatchdogSec`. Hermes sends heartbeats +only while its event loop is making timely progress; systemd restarts the +process when they stop. The default `0` keeps the existing `Type=simple` +behavior. This setting is Linux/systemd-only and does not treat an ordinary +platform network disconnect as an event-loop failure. + ## Chat Commands (Inside Messaging) | Command | Description | diff --git a/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/user-guide/messaging/discord.md b/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/user-guide/messaging/discord.md index 77b23996b..ade7ce08b 100644 --- a/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/user-guide/messaging/discord.md +++ b/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/user-guide/messaging/discord.md @@ -82,6 +82,22 @@ Hermes 按会话键跟踪正在运行的 agent。 本指南将引导你完成完整的设置流程——从在 Discord 开发者门户创建机器人到发送第一条消息。 +### Discord Gateway WebSocket 健康 + +Discord REST 与 Gateway WebSocket 是独立传输。REST 请求成功不代表机器人仍能接收 Gateway 事件。Hermes 会组合检查 ready 状态、client/socket 关闭状态、socket 是否打开、heartbeat ACK 年龄和有限 heartbeat latency。 + +连续异常达到阈值后，适配器只上报一次可重试失败；现有 Gateway 重连器创建新适配器，不会启动第二个无限重连循环。 + +```yaml +discord: + websocket_liveness_interval_seconds: 15 + websocket_liveness_failure_threshold: 2 + websocket_heartbeat_ack_max_age_seconds: 60 + websocket_max_latency_seconds: 30 +``` + +旧的 `liveness_interval_seconds` / `liveness_failure_threshold` 仅作为迁移别名保留，不再表示 REST probe。 + ## 第一步：创建 Discord 应用 1. 前往 [Discord 开发者门户](https://discord.com/developers/applications) 并使用你的 Discord 账号登录。