diff --git a/gateway/run.py b/gateway/run.py index 5ed80459140c..a82bb50f0760 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -27,6 +27,7 @@ except ModuleNotFoundError: 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 @@ -4047,14 +4049,29 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew 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. @@ -4731,6 +4748,10 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew "%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" @@ -7742,6 +7763,29 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew 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. + 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) + faulthandler.register( + signal.SIGUSR2, + file=open(_faulthandler_path, "a"), + 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: @@ -8461,7 +8505,12 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew 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 @@ -9017,6 +9066,30 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew # 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 fe3526a9a1a9..c57efa407eea 100644 --- a/tests/gateway/test_platform_reconnect.py +++ b/tests/gateway/test_platform_reconnect.py @@ -1017,3 +1017,227 @@ class TestFatalHandoffCancellationProof: 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