diff --git a/gateway/run.py b/gateway/run.py index 72f2b2626adad..6ab51716c58fd 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -27,6 +27,7 @@ import asyncio import concurrent.futures import dataclasses +import faulthandler import inspect import json import logging @@ -3292,6 +3293,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew _gateway_started_at: float = 0.0 _shutdown_watchdog_done: Optional["threading.Event"] = None _platform_lock_takeover_on_start: bool = False + _reconnect_watcher_task: Optional["asyncio.Task"] = None def __init__(self, config: Optional[GatewayConfig] = None): global _gateway_runner_ref @@ -4041,14 +4043,29 @@ async def _connect_adapter_with_timeout( timeout = self._platform_connect_timeout_secs() if timeout <= 0: return await adapter.connect(is_reconnect=is_reconnect) + # Use the detach-on-timeout pattern instead of plain asyncio.wait_for: + # asyncio.wait_for cancels the overdue task but then waits for it to + # exit. An adapter connect() that catches CancelledError can therefore + # block recovery forever (the watcher never reaches the next retry). + # Keep ownership of the old task through its done callback, but + # release the runner at the deadline (#70344). + task = asyncio.ensure_future( + adapter.connect(is_reconnect=is_reconnect) + ) try: - return await asyncio.wait_for( - adapter.connect(is_reconnect=is_reconnect), timeout=timeout - ) - except asyncio.TimeoutError as exc: - raise TimeoutError( - f"{platform.value} connect timed out after {timeout:g}s" - ) from exc + done, _pending = await asyncio.wait({task}, timeout=timeout) + except asyncio.CancelledError: + task.cancel() + task.add_done_callback(consume_detached_task_result) + raise + if task in done: + result = await task + return bool(result) + task.cancel() + task.add_done_callback(consume_detached_task_result) + raise TimeoutError( + f"{platform.value} connect timed out after {timeout:g}s" + ) async def _connect_initial_adapter_with_timeout(self, adapter, platform) -> bool: """Connect one cold-start adapter with tightly scoped replace intent. @@ -4725,6 +4742,10 @@ async def _handle_adapter_fatal_error_impl(self, adapter: BasePlatformAdapter) - "%s queued for background reconnection", adapter.platform.value, ) + # Ensure the reconnect watcher is alive — if it died (e.g. from + # exhausting its restart budget), respawn it so queued platforms + # are not permanently stranded (#70344). + self._ensure_reconnect_watcher_running() if not self.adapters and not self._failed_platforms: self._exit_reason = adapter.fatal_error_message or "All messaging adapters disconnected" @@ -7736,6 +7757,35 @@ async def start(self) -> bool: Returns True if at least one adapter connected successfully. """ logger.info("Starting Hermes Gateway...") + # Enable faulthandler at gateway start so that SIGUSR2 (or an + # internal watchdog) can dump all thread and task stacks to stderr + # for post-mortem diagnosis of event-loop freezes (#70344). + faulthandler.enable() + # Also dump stacks to a rotating file for off-line analysis when + # the gateway is running under a service manager that doesn't + # capture stderr. + # faulthandler.register() and SIGUSR2 are POSIX-only; skip the + # signal-triggered file dump on Windows (faulthandler.enable() + # above still covers fatal-error dumps there). + _sigusr2 = getattr(signal, "SIGUSR2", None) + if _sigusr2 is not None and hasattr(faulthandler, "register"): + try: + _log_dir = getattr(self.config, "log_dir", None) or os.path.join( + os.environ.get("HERMES_HOME", str(Path.home() / ".hermes")), + "logs", + ) + _faulthandler_path = os.path.join(_log_dir, "gateway_faulthandler.log") + os.makedirs(_log_dir, exist_ok=True) + _fh = open(_faulthandler_path, "a", encoding="utf-8") + faulthandler.register( + _sigusr2, + file=_fh, + all_threads=True, + chain=True, + ) + except Exception: + logger.debug("Could not set up faulthandler file logging", exc_info=True) + try: self._gateway_loop = asyncio.get_running_loop() except RuntimeError: @@ -8251,7 +8301,7 @@ async def start(self) -> bool: ) if connected_count == 0: - if startup_nonretryable_errors: + if startup_nonretryable_errors and not startup_retryable_errors: reason = "; ".join(startup_nonretryable_errors) logger.error("Gateway hit a non-retryable startup conflict: %s", reason) try: @@ -8263,6 +8313,27 @@ async def start(self) -> bool: self._request_clean_exit(reason) self._startup_restore_in_progress = False return True + if startup_nonretryable_errors: + # Mixed failure mode (NS-609): some platforms are fatally + # misconfigured (e.g. WhatsApp enabled but never paired) while + # others hit merely transient errors (e.g. Telegram TimedOut + # during polling startup). Exiting with + # GATEWAY_FATAL_CONFIG_EXIT_CODE here is wrong in both + # supervision worlds: under supervisors that honor the + # exit-78 contract (systemd RestartPreventExitStatus, s6 + # finish→125 since #51228) the gateway goes PERMANENTLY down + # over a network blip; under anything else it crash-loops. + # Either way the retryable platforms never get their retry. + # Log the fatal side loudly, then fall through to the + # degraded/retry path below: the reconnect watcher recovers + # the retryable platforms; the non-retryable ones remain + # fatal-parked and visible in runtime status. + logger.error( + "%d platform(s) fatally misconfigured and parked: %s. " + "Staying alive so retryable platforms can recover.", + len(startup_nonretryable_errors), + "; ".join(startup_nonretryable_errors), + ) if enabled_platform_count > 0: if startup_retryable_errors: # All enabled platforms hit retryable failures (network @@ -8455,7 +8526,12 @@ async def start(self) -> bool: len(self._failed_platforms), ", ".join(p.value for p in self._failed_platforms), ) - self._spawn_supervised(self._platform_reconnect_watcher, "platform_reconnect_watcher") + # Track the reconnect watcher task so _ensure_reconnect_watcher_running + # can detect if it dies and respawn it (#70344). + self._reconnect_watcher_task = asyncio.create_task( + self._platform_reconnect_watcher() + ) + self._background_tasks.add(self._reconnect_watcher_task) # Start background handoff watcher — picks up CLI sessions marked # handoff_state='pending' in state.db and re-binds them to the @@ -9011,6 +9087,30 @@ def _active_profile_name(self) -> str: # self state, so inheriting the mixin keeps every self._kanban_* call site # working unchanged while lifting ~1,000 LOC out of this file. + def _ensure_reconnect_watcher_running(self) -> None: + """Ensure the platform reconnect watcher background task is alive. + + If the tracked reconnect watcher task has died (e.g. from exhausting + its restart budget, or a terminal exception that _spawn_supervised + could not recover), respawns it so platforms queued for reconnection + are not permanently stranded. Called after queueing a retryable fatal + error in _handle_adapter_fatal_error (#70344). + """ + if not getattr(self, "_running", False): + return + task = getattr(self, "_reconnect_watcher_task", None) + if task is not None and not task.done(): + return # already alive + logger.warning( + "Reconnect watcher task is dead (done=%s) — respawning", + task.done() if task is not None else "N/A", + ) + self._reconnect_watcher_task = asyncio.create_task( + self._platform_reconnect_watcher() + ) + if getattr(self, "_background_tasks", None) is not None: + self._background_tasks.add(self._reconnect_watcher_task) + async def _platform_reconnect_watcher(self) -> None: """Background task that periodically retries connecting failed platforms. diff --git a/tests/gateway/test_platform_reconnect.py b/tests/gateway/test_platform_reconnect.py index fe3526a9a1a93..c57efa407eea0 100644 --- a/tests/gateway/test_platform_reconnect.py +++ b/tests/gateway/test_platform_reconnect.py @@ -1017,3 +1017,227 @@ async def _stop(): assert runner._exit_with_failure is True assert runner.stop.await_count == 1 + + +# ── _ensure_reconnect_watcher_running ────────────────────────────────── + + +class TestEnsureReconnectWatcherRunning: + """Verify _ensure_reconnect_watcher_running respawns the watcher when dead.""" + + @pytest.mark.asyncio + async def test_reconnect_watcher_alive_does_nothing(self): + """Task is alive => no-op.""" + runner = _make_runner() + runner._running = True + runner._background_tasks = set() + + async def _dummy(): + await asyncio.sleep(3600) + + runner._reconnect_watcher_task = asyncio.create_task(_dummy()) + runner._background_tasks.add(runner._reconnect_watcher_task) + + old_task = runner._reconnect_watcher_task + runner._ensure_reconnect_watcher_running() + + # Same task, not replaced + assert runner._reconnect_watcher_task is old_task + assert not runner._reconnect_watcher_task.done() + + old_task.cancel() + try: + await old_task + except asyncio.CancelledError: + pass + + @pytest.mark.asyncio + async def test_reconnect_watcher_dead_respawns(self): + """Watcher is done => respawn.""" + runner = _make_runner() + runner._running = True + runner._background_tasks = set() + runner._reconnect_watcher_task = asyncio.create_task(asyncio.sleep(0)) + await runner._reconnect_watcher_task # let it finish + + assert runner._reconnect_watcher_task.done() + + runner._ensure_reconnect_watcher_running() + + assert runner._reconnect_watcher_task is not None + assert not runner._reconnect_watcher_task.done() + assert runner._reconnect_watcher_task in runner._background_tasks + + runner._reconnect_watcher_task.cancel() + try: + await runner._reconnect_watcher_task + except asyncio.CancelledError: + pass + + @pytest.mark.asyncio + async def test_reconnect_watcher_not_running_respawns(self): + """No task at all => creates one.""" + runner = _make_runner() + runner._running = True + runner._background_tasks = set() + runner._reconnect_watcher_task = None + + runner._ensure_reconnect_watcher_running() + + assert runner._reconnect_watcher_task is not None + assert not runner._reconnect_watcher_task.done() + assert runner._reconnect_watcher_task in runner._background_tasks + + runner._reconnect_watcher_task.cancel() + try: + await runner._reconnect_watcher_task + except asyncio.CancelledError: + pass + + @pytest.mark.asyncio + async def test_not_running_noop(self): + """_running is False => no-op.""" + runner = _make_runner() + runner._running = False + runner._reconnect_watcher_task = None + runner._ensure_reconnect_watcher_running() + assert runner._reconnect_watcher_task is None + + +# ── _handle_adapter_fatal_error calls _ensure_reconnect_watcher ──────── + + +class TestFatalErrorCallsEnsureWatcher: + """Verify _handle_adapter_fatal_error calls _ensure_reconnect_watcher_running.""" + + @pytest.mark.asyncio + async def test_retryable_fatal_error_calls_ensure_watcher(self): + """A retryable fatal error queues the platform AND ensures watcher is alive.""" + runner = _make_runner() + runner._running = True + runner._background_tasks = set() + runner._failed_platforms = {} + runner._fatal_handler_tasks = set() + runner._reconnect_watcher_task = asyncio.create_task(asyncio.sleep(0)) + # Let the dummy watcher finish so _ensure_reconnect_watcher_running + # detects it's dead and respawns. + await runner._reconnect_watcher_task + + platform_config = PlatformConfig(enabled=True, token="test") + runner.config = GatewayConfig( + platforms={Platform.TELEGRAM: platform_config} + ) + + adapter = StubAdapter( + platform=Platform.TELEGRAM, + succeed=False, + fatal_error="network outage", + fatal_retryable=True, + ) + # Pre-set fatal error attributes so the handler can read them + # without going through connect() (#70344). + adapter._set_fatal_error( + "NETWORK_ERROR", "network outage", retryable=True + ) + # Populate adapters so the impl pops it and queues for reconnect + runner.adapters[Platform.TELEGRAM] = adapter + + call_count = {"ensure": 0} + + def tracking_ensure(): + call_count["ensure"] += 1 + + with patch.object( + runner, + "_ensure_reconnect_watcher_running", + side_effect=tracking_ensure, + ): + await runner._handle_adapter_fatal_error(adapter) + + assert Platform.TELEGRAM in runner._failed_platforms + assert call_count["ensure"] >= 1 + + @pytest.mark.asyncio + async def test_nonretryable_fatal_error_does_not_call_ensure(self): + """A non-retryable error must NOT queue the platform or call the watcher.""" + runner = _make_runner() + runner._running = True + runner._background_tasks = set() + runner._failed_platforms = {} + runner._fatal_handler_tasks = set() + runner._reconnect_watcher_task = None + + platform_config = PlatformConfig(enabled=True, token="test") + runner.config = GatewayConfig( + platforms={Platform.TELEGRAM: platform_config} + ) + + adapter = StubAdapter( + platform=Platform.TELEGRAM, + succeed=False, + fatal_error="bad token", + fatal_retryable=False, + ) + # Pre-set fatal error attributes so the handler can read them + # without going through connect() (#70344). + adapter._set_fatal_error( + "AUTH_FAILED", "bad token", retryable=False + ) + runner.adapters[Platform.TELEGRAM] = adapter + + ensure_called = False + + def noop_ensure(): + nonlocal ensure_called + ensure_called = True + + with patch.object(runner, "_ensure_reconnect_watcher_running", side_effect=noop_ensure): + await runner._handle_adapter_fatal_error(adapter) + + assert Platform.TELEGRAM not in runner._failed_platforms + assert not ensure_called + + +# ── _connect_adapter_with_timeout detach-on-timeout ──────────────────── + + +class TestConnectAdapterDetachOnTimeout: + """Verify _connect_adapter_with_timeout uses the detach pattern.""" + + @pytest.mark.asyncio + async def test_connect_timed_out_raises_timeouterror(self): + """A connect() that never finishes must raise TimeoutError.""" + runner = _make_runner() + + adapter = StubAdapter(succeed=True) + + async def _slow_connect(**kwargs): + await asyncio.sleep(3600) # never finishes + + with patch.object(adapter, "connect", side_effect=_slow_connect): + with patch.object( + runner, "_platform_connect_timeout_secs", return_value=0.01 + ): + with pytest.raises(TimeoutError, match="timed out"): + await runner._connect_adapter_with_timeout( + adapter, Platform.TELEGRAM + ) + + # After the TimeoutError, the slow connect coroutine should have been + # cancelled and detached, so the event loop can move on. + await asyncio.sleep(0) + + @pytest.mark.asyncio + async def test_connect_success_returns_true(self): + """A successful connect returns True.""" + runner = _make_runner() + adapter = StubAdapter(succeed=True) + + with patch.object( + runner, "_platform_connect_timeout_secs", return_value=30.0 + ): + result = await runner._connect_adapter_with_timeout( + adapter, Platform.TELEGRAM + ) + + assert result is True diff --git a/tests/gateway/test_runner_startup_failures.py b/tests/gateway/test_runner_startup_failures.py index e8bb72b12e4d6..93d81236bee2e 100644 --- a/tests/gateway/test_runner_startup_failures.py +++ b/tests/gateway/test_runner_startup_failures.py @@ -512,6 +512,60 @@ async def test_runner_exits_with_ex_config_on_nonretryable_startup_error(monkeyp assert state["gateway_state"] == "startup_failed" +@pytest.mark.asyncio +async def test_runner_stays_alive_on_mixed_retryable_and_nonretryable_errors( + monkeypatch, tmp_path, caplog +): + """Mixed startup failures — one platform fatally misconfigured, another + merely transiently failing — must NOT exit with EX_CONFIG (NS-609). + + Real-world shape: WhatsApp enabled but never paired (non-retryable + ``whatsapp_not_paired``) while Telegram hits a startup TimedOut + (retryable). Exiting 78 here either takes the gateway permanently down + (supervisors honoring the exit-78 contract) or crash-loops it (anything + else) — and in both cases Telegram never gets its retry even though + nothing is wrong with its config. The gateway must stay alive in + degraded mode, park the fatal platform, and queue the retryable one.""" + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + config = GatewayConfig( + platforms={ + Platform.DISCORD: PlatformConfig(enabled=True, token="***"), + Platform.TELEGRAM: PlatformConfig(enabled=True, token="***"), + }, + sessions_dir=tmp_path / "sessions", + ) + runner = GatewayRunner(config) + + def _make_adapter(platform, platform_config): + if platform == Platform.DISCORD: + return _NonRetryableFailureAdapter() + return _RetryableFailureAdapter() + + monkeypatch.setattr(runner, "_create_adapter", _make_adapter) + + import logging + with caplog.at_level(logging.ERROR): + ok = await runner.start() + + # Gateway stays alive — no clean-exit request, no EX_CONFIG. + assert ok is True + assert runner.should_exit_cleanly is False + assert runner.exit_code is None + state = read_runtime_status() + assert state["gateway_state"] in {"degraded", "running"} + # The retryable platform is queued for reconnection… + assert Platform.TELEGRAM in runner._failed_platforms + assert state["platforms"]["telegram"]["state"] == "retrying" + # …while the misconfigured one is parked as fatal, not retried. + assert Platform.DISCORD not in runner._failed_platforms + assert state["platforms"]["discord"]["state"] == "fatal" + # The fatal side is still surfaced loudly for the operator. + assert any( + "fatally misconfigured" in record.message + for record in caplog.records + ), "Expected an error log calling out the parked platform(s)" + + @pytest.mark.asyncio async def test_start_gateway_propagates_fatal_config_exit_code(monkeypatch, tmp_path): """A clean exit carrying GATEWAY_FATAL_CONFIG_EXIT_CODE must surface as a