import asyncio from unittest.mock import AsyncMock, MagicMock import pytest import gateway.run as gateway_run from gateway.config import GatewayConfig, Platform, PlatformConfig from gateway.platforms.base import BasePlatformAdapter, SendResult from gateway.restart import ( DEFAULT_GATEWAY_RESTART_DRAIN_TIMEOUT, GATEWAY_SERVICE_RESTART_EXIT_CODE, ) class StartupRaceAdapter(BasePlatformAdapter): def __init__( self, platform: Platform, *, on_connect=None, wait_for_disconnect: asyncio.Event | None = None, ): super().__init__(PlatformConfig(enabled=True, token="***"), platform) self.on_connect = on_connect self.wait_for_disconnect = wait_for_disconnect self.connected = False self.disconnected = False self.background_cancelled = False async def connect(self, *, is_reconnect: bool = False): if self.on_connect: self.on_connect() if self.wait_for_disconnect is not None: await self.wait_for_disconnect.wait() self.connected = True return True async def disconnect(self): self.disconnected = True async def cancel_background_tasks(self): self.background_cancelled = True await super().cancel_background_tasks() async def send(self, chat_id, content, reply_to=None, metadata=None): return SendResult(success=True, message_id="1") async def send_typing(self, chat_id, metadata=None): return None async def get_chat_info(self, chat_id): return {"id": chat_id} def make_startup_runner(tmp_path): runner = object.__new__(gateway_run.GatewayRunner) runner.config = GatewayConfig( platforms={ Platform.TELEGRAM: PlatformConfig(enabled=True, token="***"), Platform.SLACK: PlatformConfig(enabled=True, token="***"), }, sessions_dir=tmp_path / "sessions", ) runner.adapters = {} runner._running = False runner._shutdown_event = asyncio.Event() runner._exit_reason = None runner._exit_code = None runner._exit_cleanly = False runner._exit_with_failure = False runner._draining = False runner._restart_requested = False runner._restart_task_started = False runner._restart_detached = False runner._restart_via_service = False runner._restart_drain_timeout = DEFAULT_GATEWAY_RESTART_DRAIN_TIMEOUT runner._stop_task = None runner._running_agents = {} runner._running_agents_ts = {} runner._pending_messages = {} runner._pending_approvals = {} runner._background_tasks = set() runner._failed_platforms = {} runner._voice_mode = {} runner.hooks = MagicMock() runner.hooks.loaded_hooks = [] runner.hooks.discover_and_load = MagicMock() runner.hooks.emit = AsyncMock() runner.session_store = MagicMock() runner.delivery_router = MagicMock() runner.delivery_router.adapters = {} runner._update_runtime_status = MagicMock() runner._update_platform_runtime_status = MagicMock() runner._sync_voice_mode_state_to_adapter = MagicMock() runner._suspend_stuck_loop_sessions = MagicMock(return_value=0) runner._notify_active_sessions_of_shutdown = AsyncMock() runner._drain_active_agents = AsyncMock(return_value=({}, False)) runner._finalize_shutdown_agents = AsyncMock() runner._send_update_notification = AsyncMock(return_value=False) runner._schedule_update_notification_watch = MagicMock() runner._send_restart_notification = AsyncMock() runner.wait_for_shutdown = gateway_run.GatewayRunner.wait_for_shutdown.__get__( runner, gateway_run.GatewayRunner ) async def no_op_watcher(*args, **kwargs): await asyncio.Event().wait() runner._session_housekeeping_watcher = no_op_watcher runner._platform_reconnect_watcher = no_op_watcher runner._run_process_watcher = no_op_watcher runner._safe_adapter_disconnect = gateway_run.GatewayRunner._safe_adapter_disconnect.__get__( runner, gateway_run.GatewayRunner ) runner.request_restart = gateway_run.GatewayRunner.request_restart.__get__( runner, gateway_run.GatewayRunner ) runner.stop = gateway_run.GatewayRunner.stop.__get__(runner, gateway_run.GatewayRunner) return runner def patch_startup_side_effects(monkeypatch, tmp_path): monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path) monkeypatch.setattr("hermes_cli.plugins.discover_plugins", lambda: None) monkeypatch.setattr("agent.shell_hooks.register_from_config", lambda *args, **kwargs: None) monkeypatch.setattr("tools.process_registry.process_registry.recover_from_checkpoint", lambda: 0) @pytest.mark.asyncio async def test_startup_aborts_when_restart_begins_during_platform_connect(tmp_path, monkeypatch): patch_startup_side_effects(monkeypatch, tmp_path) runner = make_startup_runner(tmp_path) first_disconnected = asyncio.Event() telegram = StartupRaceAdapter( Platform.TELEGRAM, on_connect=lambda: runner.request_restart(detached=False, via_service=True), ) slack = StartupRaceAdapter(Platform.SLACK, wait_for_disconnect=first_disconnected) async def disconnect_and_release(): telegram.disconnected = True first_disconnected.set() telegram.disconnect = disconnect_and_release runner._create_adapter = MagicMock(side_effect=[telegram, slack]) result = await asyncio.wait_for(runner.start(), timeout=30) assert result is True assert telegram.disconnected is True assert telegram.background_cancelled is True assert slack.connected is False assert runner._running is False assert runner.adapters == {} assert runner._update_runtime_status.call_args_list[-1].args[0] == "stopped" assert not any( call.args[:1] == ("running",) for call in runner._update_runtime_status.call_args_list ) assert not any( call.args[:2] == (Platform.SLACK.value, "connected") for call in runner._update_platform_runtime_status.call_args_list ) def _patch_aborted_startup(monkeypatch, runner_cls): """Run start_gateway() against a runner that aborts before running mode.""" monkeypatch.setattr("gateway.status.get_running_pid", lambda: None) monkeypatch.setattr("gateway.status.acquire_gateway_runtime_lock", lambda: True) monkeypatch.setattr("gateway.status.write_pid_file", lambda: None) monkeypatch.setattr("gateway.status.remove_pid_file", lambda: None) monkeypatch.setattr("gateway.status.release_gateway_runtime_lock", lambda: None) monkeypatch.setattr("tools.skills_sync.sync_skills", lambda quiet=True: None) monkeypatch.setattr("hermes_logging.setup_logging", lambda hermes_home, mode: None) monkeypatch.setattr("gateway.run.GatewayRunner", runner_cls) @pytest.mark.asyncio async def test_start_gateway_does_not_start_cron_after_aborted_startup(tmp_path, monkeypatch): monkeypatch.setenv("HERMES_HOME", str(tmp_path)) cron_started = False export_shutdown_calls = 0 class ExportRuntime: def shutdown(self): nonlocal export_shutdown_calls export_shutdown_calls += 1 class AbortedStartupRunner: def __init__(self, config): self.config = config self.adapters = {} self._running = False self.should_exit_cleanly = True self.should_exit_with_failure = False self.exit_reason = None self.exit_code = GATEWAY_SERVICE_RESTART_EXIT_CODE self._gateway_health_export_runtime = ExportRuntime() async def start(self): return True async def wait_for_shutdown(self): return None def fail_if_cron_starts(*args, **kwargs): nonlocal cron_started cron_started = True _patch_aborted_startup(monkeypatch, AbortedStartupRunner) monkeypatch.setattr("gateway.run._start_cron_ticker", fail_if_cron_starts) monkeypatch.setattr("tools.mcp_tool_lifecycle.shutdown_mcp_servers", lambda: None) with pytest.raises(SystemExit) as exc: await gateway_run.start_gateway(config=GatewayConfig(), replace=False, verbosity=None) assert exc.value.code == GATEWAY_SERVICE_RESTART_EXIT_CODE assert cron_started is False assert export_shutdown_calls == 1 @pytest.mark.asyncio async def test_start_gateway_preserves_service_restart_fallback_after_aborted_startup( tmp_path, monkeypatch ): """A legacy service restart without an explicit exit code still exits with EX_TEMPFAIL.""" monkeypatch.setenv("HERMES_HOME", str(tmp_path)) cron_started = False class AbortedStartupRunner: def __init__(self, config): self.config = config self.adapters = {} self._running = False self._restart_requested = True self._restart_via_service = True self.should_exit_cleanly = False self.should_exit_with_failure = False self.exit_reason = None self.exit_code = None async def start(self): return True async def wait_for_shutdown(self): return None def fail_if_cron_starts(*args, **kwargs): nonlocal cron_started cron_started = True _patch_aborted_startup(monkeypatch, AbortedStartupRunner) monkeypatch.setattr("gateway.run._start_cron_ticker", fail_if_cron_starts) monkeypatch.setattr("tools.mcp_tool_lifecycle.shutdown_mcp_servers", lambda: None) with pytest.raises(SystemExit) as exc: await gateway_run.start_gateway( config=GatewayConfig(), replace=False, verbosity=None ) assert exc.value.code == GATEWAY_SERVICE_RESTART_EXIT_CODE assert cron_started is False @pytest.mark.asyncio @pytest.mark.parametrize( ("unexpected_signal", "expected_success"), [(True, False), (False, True)], ids=["unexpected-sigterm", "planned-stop"], ) async def test_start_gateway_classifies_startup_signal_exit( tmp_path, monkeypatch, unexpected_signal, expected_success ): """A startup SIGTERM is restartable unless a planned-stop marker classified it as intentional.""" monkeypatch.setenv("HERMES_HOME", str(tmp_path)) signal_state = None cron_started = False class AbortedStartupRunner: def __init__(self, config): self.config = config self.adapters = {} self._running = False self._restart_requested = False self._restart_via_service = False self.should_exit_cleanly = False self.should_exit_with_failure = False self.exit_reason = None self.exit_code = None async def start(self): if unexpected_signal: signal_state[0] = True return True async def wait_for_shutdown(self): return None def capture_signal_state(runner, state): nonlocal signal_state signal_state = state return lambda received_signal=None: None def fail_if_cron_starts(*args, **kwargs): nonlocal cron_started cron_started = True _patch_aborted_startup(monkeypatch, AbortedStartupRunner) monkeypatch.setattr( "gateway.run._start_gateway_make_shutdown_signal_handler", capture_signal_state ) monkeypatch.setattr("gateway.run._start_cron_ticker", fail_if_cron_starts) monkeypatch.setattr("tools.mcp_tool_lifecycle.shutdown_mcp_servers", lambda: None) result = await gateway_run.start_gateway( config=GatewayConfig(), replace=False, verbosity=None ) assert result is expected_success assert cron_started is False @pytest.mark.asyncio async def test_failure_exit_still_stops_cron_housekeeping_and_mcp(monkeypatch): """``should_exit_with_failure`` used to return before the cooperative teardown, leaking the cron ticker / housekeeping threads and open MCP connections for embedded callers (#12175).""" import threading stopped = [] cron_stop = threading.Event() threads = [threading.Thread(target=cron_stop.wait, name=n, daemon=True) for n in ("cron", "housekeeping")] for thread in threads: thread.start() watcher_stop = threading.Event() watcher = threading.Thread(target=watcher_stop.wait, daemon=True) watcher.start() async def fake_mcp_shutdown(*_args, **_kwargs): stopped.append("mcp") monkeypatch.setattr(gateway_run, "_shutdown_mcp_servers_nonblocking", fake_mcp_shutdown) monkeypatch.setattr(gateway_run, "_stop_cron_provider", lambda provider: stopped.append("provider")) monkeypatch.setattr("hermes_cli.nous_auth_keepalive.stop_nous_auth_keepalive", lambda: None) runner = MagicMock(should_exit_with_failure=True, exit_reason="boom", exit_code=None) result = await gateway_run._start_gateway_shutdown_tail( runner, None, cron_stop, object(), threads[0], threads[1], watcher_stop, watcher, [False]) assert result is False assert cron_stop.is_set() and watcher_stop.is_set() assert stopped == ["provider", "mcp"] for thread in threads + [watcher]: thread.join(timeout=2) assert not thread.is_alive()