diff --git a/hermes_cli/gateway.py b/hermes_cli/gateway.py index 5a3b4fbcf57..92feb96a948 100644 --- a/hermes_cli/gateway.py +++ b/hermes_cli/gateway.py @@ -1201,6 +1201,27 @@ def is_windows() -> bool: return sys.platform == 'win32' +def _windows_gateway_should_absorb_console_controls() -> bool: + """Return True for detached Windows gateway runs that should ignore Ctrl+C. + + Foreground ``hermes gateway run`` must remain interruptible from + PowerShell/CMD. Detached service-style launches opt in via + ``HERMES_GATEWAY_DETACHED=1``; older wrappers without the env marker are + treated as detached when no interactive stdin is attached. + """ + if not is_windows(): + return False + + detached = os.getenv("HERMES_GATEWAY_DETACHED", "").strip().lower() + if detached in {"1", "true", "yes", "on"}: + return True + + try: + return not bool(sys.stdin and sys.stdin.isatty()) + except (ValueError, OSError): + return True + + # ============================================================================= # Service Configuration # ============================================================================= @@ -3042,34 +3063,17 @@ def run_gateway(verbose: int = 0, quiet: bool = False, replace: bool = False): _guard_official_docker_root_gateway() sys.path.insert(0, str(PROJECT_ROOT)) - # On Windows, when the gateway is launched as a detached background - # process (via ``hermes gateway install`` → Scheduled Task / Startup - # folder / direct pythonw.exe spawn) there is no console attached. In - # that case Windows can still deliver CTRL_C_EVENT / CTRL_BREAK_EVENT - # to the process group under some circumstances (e.g. when *another* - # process in the same group sends one), which Python 3.11 translates - # into KeyboardInterrupt inside asyncio.run(). The outer handler below - # catches that and exits cleanly — silently killing the gateway. On - # detached boots we must absorb those spurious signals so the gateway - # stays alive; real user Ctrl+C still comes through prompt_toolkit / - # the asyncio signal handler when running in a real console. - # - # IMPORTANT lesson (May 2026): we originally gated this on "stdin is - # NOT a TTY" assuming only detached pythonw runs would be vulnerable. - # Wrong. When the user runs `hermes gateway start` from a PowerShell - # console, the gateway inherits that console and stdin IS a TTY — - # but it's STILL vulnerable to CTRL_C_EVENT broadcast by any sibling - # `hermes` invocation (like `hermes gateway status` 30 seconds later) - # because Windows routes console events to all processes sharing the - # console. Every hermes CLI process after that sibling fires is a - # potential drive-by killer. So on Windows, for `gateway run` - # specifically (never interactive by design), always install the - # SIGINT absorber regardless of TTY state. + # Detached Windows gateway runs must ignore console-control broadcasts + # from sibling CLI processes, but foreground `hermes gateway run` still + # needs to obey the banner's "Press Ctrl+C to stop" contract. + # Service-style launchers set HERMES_GATEWAY_DETACHED=1; older wrappers + # without the marker are handled by the non-TTY fallback. try: _stdin_is_tty = bool(sys.stdin and sys.stdin.isatty()) except (ValueError, OSError): _stdin_is_tty = False - if is_windows(): + _absorb_windows_console_controls = _windows_gateway_should_absorb_console_controls() + if _absorb_windows_console_controls: try: signal.signal(signal.SIGINT, signal.SIG_IGN) if hasattr(signal, "SIGBREAK"): @@ -3167,6 +3171,7 @@ def run_gateway(verbose: int = 0, quiet: bool = False, replace: bool = False): replace=replace, argv=sys.argv, stdin_is_tty=_stdin_is_tty, + absorb_windows_console_controls=_absorb_windows_console_controls, ) def _atexit_hook() -> None: diff --git a/hermes_cli/gateway_windows.py b/hermes_cli/gateway_windows.py index b4820ab311f..4a3059223c4 100644 --- a/hermes_cli/gateway_windows.py +++ b/hermes_cli/gateway_windows.py @@ -216,6 +216,7 @@ def _build_gateway_cmd_script( lines.append(f"cd /d {_quote_cmd_script_arg(working_dir)}") lines.append(f'set "HERMES_HOME={hermes_home}"') lines.append('set "PYTHONIOENCODING=utf-8"') + lines.append('set "HERMES_GATEWAY_DETACHED=1"') # VIRTUAL_ENV lets the gateway's own python detection find the venv # if someone imports hermes_constants-based logic during startup. venv_dir = str(Path(python_path).resolve().parent.parent) @@ -371,6 +372,7 @@ def _build_gateway_argv() -> tuple[list[str], str, dict[str, str]]: env_overlay = { "HERMES_HOME": hermes_home, "PYTHONIOENCODING": "utf-8", + "HERMES_GATEWAY_DETACHED": "1", "VIRTUAL_ENV": str(Path(python_exe).resolve().parent.parent), } return argv, working_dir, env_overlay diff --git a/tests/hermes_cli/test_gateway.py b/tests/hermes_cli/test_gateway.py index c213c99c8d2..cfeb25a463f 100644 --- a/tests/hermes_cli/test_gateway.py +++ b/tests/hermes_cli/test_gateway.py @@ -90,6 +90,66 @@ def test_run_gateway_root_guard_has_escape_hatch(monkeypatch): assert calls == [(True, 2)] +def test_run_gateway_windows_foreground_keeps_ctrl_c_enabled(monkeypatch): + calls = [] + + def fake_start_gateway(*, replace, verbosity): + calls.append((replace, verbosity)) + return object() + + class _TTY: + def isatty(self): + return True + + signal_calls = [] + + def fake_signal(sig, handler): + signal_calls.append((sig, handler)) + + _install_fake_gateway_run(monkeypatch, fake_start_gateway) + monkeypatch.setattr(gateway, "is_windows", lambda: True) + monkeypatch.setattr(gateway, "supports_systemd_services", lambda: False) + monkeypatch.setattr(gateway.sys, "stdin", _TTY()) + monkeypatch.delenv("HERMES_GATEWAY_DETACHED", raising=False) + monkeypatch.setattr(gateway.signal, "signal", fake_signal) + monkeypatch.setattr(gateway.asyncio, "run", lambda coro: True) + + gateway.run_gateway() + + assert calls == [(False, 0)] + assert (gateway.signal.SIGINT, gateway.signal.SIG_IGN) not in signal_calls + + +def test_run_gateway_windows_detached_absorbs_console_controls(monkeypatch): + calls = [] + + def fake_start_gateway(*, replace, verbosity): + calls.append((replace, verbosity)) + return object() + + class _TTY: + def isatty(self): + return True + + signal_calls = [] + + def fake_signal(sig, handler): + signal_calls.append((sig, handler)) + + _install_fake_gateway_run(monkeypatch, fake_start_gateway) + monkeypatch.setattr(gateway, "is_windows", lambda: True) + monkeypatch.setattr(gateway, "supports_systemd_services", lambda: False) + monkeypatch.setattr(gateway.sys, "stdin", _TTY()) + monkeypatch.setenv("HERMES_GATEWAY_DETACHED", "1") + monkeypatch.setattr(gateway.signal, "signal", fake_signal) + monkeypatch.setattr(gateway.asyncio, "run", lambda coro: True) + + gateway.run_gateway() + + assert calls == [(False, 0)] + assert (gateway.signal.SIGINT, gateway.signal.SIG_IGN) in signal_calls + + class TestSystemdLingerStatus: def test_reports_enabled(self, monkeypatch): monkeypatch.setattr(gateway, "is_linux", lambda: True)