Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 16 additions & 0 deletions hermes_cli/gateway.py
Original file line number Diff line number Diff line change
Expand Up @@ -752,9 +752,25 @@ def _spawn_gateway_restart_watcher(old_pid: int, run_argv: list[str]) -> bool:
# been spawned inside a job object (Electron/Tauri parent), and
# without breakaway the respawned gateway would die when that job
# tears down. See _subprocess_compat.windows_detach_flags().
#
# Mark the respawn as a detached service launch and sever its stdin.
# _windows_gateway_should_absorb_console_controls() keys off
# HERMES_GATEWAY_DETACHED (and, failing that, an interactive stdin):
# without this marker the respawned gateway inherits the spawning
# process's console handle, decides it is "interactive", and SKIPS the
# SetConsoleCtrlHandler(NULL, TRUE) guard. It then dies the instant
# Windows broadcasts CTRL_CLOSE_EVENT / CTRL_LOGOFF_EVENT when the
# parent console (e.g. the post-update desktop shell) goes away —
# silently, with no shutdown log. Mirror gateway_windows._spawn_detached:
# set HERMES_GATEWAY_DETACHED=1 and redirect stdin to DEVNULL so the
# respawn installs the console-control guard and survives. (#21301-followup)
_env = dict(os.environ)
_env["HERMES_GATEWAY_DETACHED"] = "1"
_popen_kwargs = {
"stdin": subprocess.DEVNULL,
"stdout": subprocess.DEVNULL,
"stderr": subprocess.DEVNULL,
"env": _env,
}
if sys.platform == "win32":
try:
Expand Down
65 changes: 65 additions & 0 deletions tests/hermes_cli/test_gateway.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
import argparse
import signal
import sys
from pathlib import Path
from types import ModuleType, SimpleNamespace

import pytest
Expand Down Expand Up @@ -374,6 +375,70 @@ def fake_signal(sig, handler):
assert (gateway.signal.SIGINT, gateway.signal.SIG_IGN) in signal_calls


class TestGatewayRestartWatcher:
"""Regression tests for the post-update detached respawn watcher.

The watcher must hand the respawned gateway the same detach context that
gateway_windows._spawn_detached gives the autostart path — otherwise the
respawn inherits the spawning console, decides it is interactive, skips the
SetConsoleCtrlHandler guard, and dies on the next CTRL_CLOSE/LOGOFF event
(the silent post-update death this fix addresses).
"""

def _watcher_source(self):
import ast
import re
import textwrap

content = (
Path(gateway.__file__).read_text(encoding="utf-8")
)
start = content.index("watcher = textwrap.dedent(")
seg = content[start : start + 3500]
m = re.search(
r'textwrap\.dedent\(\s*"""(.*?)"""\s*\)\.strip\(\)', seg, re.DOTALL
)
assert m, "could not locate watcher template literal"
inner = textwrap.dedent(m.group(1)).strip()
# Must be valid Python — it is executed via `python -c`.
ast.parse(inner)
return inner

def test_watcher_marks_respawn_detached_and_severs_stdin(self):
src = self._watcher_source()
# The respawned gateway must be tagged as a detached service launch so
# _windows_gateway_should_absorb_console_controls() returns True.
assert 'HERMES_GATEWAY_DETACHED' in src
assert '"HERMES_GATEWAY_DETACHED"] = "1"' in src
# And its stdin must be severed so the isatty() fallback can't class it
# interactive even if the env marker were somehow dropped.
assert '"stdin": subprocess.DEVNULL' in src
# The env overlay must be passed to Popen, not just constructed.
assert '"env": _env' in src

def test_by_cmdline_spawns_watcher(self, monkeypatch):
captured = {}

def fake_popen(argv, **kwargs):
captured["argv"] = argv
captured["kwargs"] = kwargs
return SimpleNamespace(pid=4321)

monkeypatch.setattr(gateway.subprocess, "Popen", fake_popen)
ok = gateway.launch_detached_gateway_restart_by_cmdline(
999999, ["pythonw.exe", "-m", "hermes_cli.main", "gateway", "run"]
)
assert ok is True
# argv = [python, "-c", watcher, old_pid, *run_argv]
assert captured["argv"][1] == "-c"
assert captured["argv"][3] == "999999"
assert captured["argv"][-3:] == ["hermes_cli.main", "gateway", "run"]

def test_rejects_empty_inputs(self):
assert gateway.launch_detached_gateway_restart_by_cmdline(0, ["x"]) is False
assert gateway.launch_detached_gateway_restart_by_cmdline(123, []) is False


class TestSystemdLingerStatus:
def test_reports_enabled(self, monkeypatch):
monkeypatch.setattr(gateway, "is_linux", lambda: True)
Expand Down
Loading