diff --git a/gateway/run.py b/gateway/run.py index 30c4764bd8755..5f778198ba6ec 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -5226,24 +5226,15 @@ def _start_gateway_make_shutdown_signal_handler(runner, _signal_initiated_shutdo planned_stop_seen = [False] def shutdown_signal_handler(received_signal=None): - # Planned --replace takeover (sibling marked this PID): exit 0 so systemd won't revive us. - def _takeover() -> bool: - from gateway.status import consume_takeover_marker_for_self - return consume_takeover_marker_for_self() + from gateway.run_shutdown import _classify_shutdown_signal - # Planned stop: CLI marks first, else its SIGTERM looks like an external kill. SIGINT = Ctrl+C. - def _planned_stop() -> bool: - from gateway.status import consume_planned_stop_marker_for_self - return consume_planned_stop_marker_for_self() + planned_takeover, planned_stop = _classify_shutdown_signal(received_signal) # Fast (<10ms) sync snapshot: stdlib + /proc, no subprocesses (`ps aux` here once blocked ~3s). def _snapshot(): from gateway.shutdown_forensics import snapshot_shutdown_context return snapshot_shutdown_context(received_signal) - planned_takeover = bool(_best_effort(_takeover, "Takeover marker check failed: %s")) - planned_stop = received_signal == signal.SIGINT or ( - not planned_takeover and bool(_best_effort(_planned_stop, "Planned stop marker check failed: %s"))) # `hermes gateway stop` writes the marker, THEN signals: the planned-stop watcher can consume # the marker in between, and the CLI's own SIGTERM must not then read as an external kill. if planned_stop: diff --git a/gateway/run_shutdown.py b/gateway/run_shutdown.py index 240caab5e74bb..a98094134494c 100644 --- a/gateway/run_shutdown.py +++ b/gateway/run_shutdown.py @@ -13,6 +13,7 @@ import logging import os import shlex +import signal import sys import threading import time @@ -65,6 +66,36 @@ def _resolve_gateway_exit_verdict(runner, signal_initiated_shutdown: bool) -> bo raise SystemExit(GATEWAY_SERVICE_RESTART_EXIT_CODE) return True +def _classify_shutdown_signal(received_signal=None) -> tuple[bool, bool]: + """Return ``(planned_takeover, planned_stop)`` for a shutdown signal. + + Keep the marker decisions in one function shared by the real POSIX signal + handler and its Windows/file-watcher equivalent. In particular, a SIGTERM + preceded by systemd's ``ExecStop`` marker must take the planned-stop path, + not the unexpected-signal path that asks the service manager to restart us. + """ + planned_takeover = False + try: + from gateway.status import consume_takeover_marker_for_self + + planned_takeover = consume_takeover_marker_for_self() + except Exception as exc: + logger.debug("Takeover marker check failed: %s", exc) + + if received_signal == signal.SIGINT: + return planned_takeover, True + if planned_takeover: + return planned_takeover, False + + try: + from gateway.status import consume_planned_stop_marker_for_self + + return planned_takeover, consume_planned_stop_marker_for_self() + except Exception as exc: + logger.debug("Planned stop marker check failed: %s", exc) + return planned_takeover, False + + # Windows has no bash/setsid chain: a tiny detached Python watcher waits for the gateway PID to # exit (bounded), then spawns ``hermes gateway restart``. _WINDOWS_RESTART_WATCHER = """ diff --git a/gateway/status.py b/gateway/status.py index 3275cf5cfab66..cda3b24bca560 100644 --- a/gateway/status.py +++ b/gateway/status.py @@ -1948,13 +1948,16 @@ def _terminate_verified_owner( return None -def write_planned_stop_marker(target_pid: int) -> bool: - """Record that ``target_pid`` is being stopped intentionally: unexpected SIGTERM exits non-zero - so service managers revive the gateway; the CLI writes this first so a deliberate stop exits - cleanly.""" +def write_planned_stop_marker( + target_pid: int, + *, + trigger_watcher: bool = True, +) -> bool: + """Record an intentional stop; reserve for the signal handler when watcher triggering is off.""" return _write_marker(_get_planned_stop_marker_path(), { "target_pid": target_pid, "target_start_time": _get_process_start_time(target_pid), - "stopper_pid": os.getpid(), "written_at": _utc_now_iso(), + "stopper_pid": os.getpid(), "trigger_watcher": trigger_watcher, + "written_at": _utc_now_iso(), }) @@ -2012,9 +2015,14 @@ def consume_planned_stop_marker_for_self() -> bool: def planned_stop_marker_targets_self() -> bool: """Non-destructive watcher probe: True when a live planned-stop marker names us. Never unlinks a matching marker (the shutdown handler does the authoritative consume); malformed/expired ones - are still cleaned up; markers naming another PID are left alone.""" + are still cleaned up; markers naming another PID or reserved for the signal handler are left + alone.""" parsed = _read_live_pid_marker(_get_planned_stop_marker_path(), _PLANNED_STOP_MARKER_TTL_S) - return parsed is not None and _pid_marker_names_self(parsed[1], parsed[2]) + return ( + parsed is not None + and parsed[0].get("trigger_watcher", True) is not False + and _pid_marker_names_self(parsed[1], parsed[2]) + ) def get_running_pid( diff --git a/hermes_cli/gateway.py b/hermes_cli/gateway.py index c285babcaa433..0d609bbadd15a 100644 --- a/hermes_cli/gateway.py +++ b/hermes_cli/gateway.py @@ -3316,7 +3316,7 @@ def generate_systemd_unit(system: bool = False, run_as_user: str | None = None) python=python_path, home=hermes_home) cleanup = installation_command(project_root, module="gateway.cgroup_cleanup", python=python_path, home=hermes_home) - stop_mark = installation_command(project_root, module="gateway.systemd_stop_mark", + stop_mark = installation_command(project_root, module="hermes_systemd_planned_stop", python=python_path, home=hermes_home) return f"""[Unit] Description={SERVICE_DESCRIPTION} @@ -3340,7 +3340,7 @@ def generate_systemd_unit(system: bool = False, run_as_user: str | None = None) KillMode=mixed KillSignal=SIGTERM ExecReload=/bin/kill -USR1 $MAINPID -ExecStop=-{_systemd_command(stop_mark)} +ExecStop=-{_systemd_command(stop_mark)} $MAINPID ExecStopPost=-{_systemd_command(cleanup)} TimeoutStopSec={restart_timeout} StandardOutput=journal diff --git a/hermes_systemd_planned_stop.py b/hermes_systemd_planned_stop.py new file mode 100644 index 0000000000000..a3e6511d28ad9 --- /dev/null +++ b/hermes_systemd_planned_stop.py @@ -0,0 +1,99 @@ +"""Minimal systemd ExecStop helper for planned gateway shutdowns. + +This module intentionally imports only Python's standard library. It runs from +systemd while the application may be partially torn down, so it must not load +``gateway`` (whose package initializer imports the full application stack) or +any optional provider dependency. +""" + +from __future__ import annotations + +import json +import os +import sys +import tempfile +from collections.abc import Sequence +from datetime import datetime, timezone +from pathlib import Path + +_MARKER_FILENAME = ".gateway-planned-stop.json" + + +def _get_process_hermes_home() -> Path: + """Return the process-level HERMES_HOME used by the gateway.""" + value = os.environ.get("HERMES_HOME", "").strip() + if value: + return Path(value) + return Path.home() / ".hermes" + + +def _get_process_start_time(pid: int) -> int | None: + """Return the Linux process start-time fingerprint, when available.""" + try: + # Match gateway.status: field 22 in /proc//stat. + return int(Path(f"/proc/{pid}/stat").read_text(encoding="utf-8").split()[21]) + except (FileNotFoundError, IndexError, PermissionError, ValueError, OSError): + return None + + +def _write_json_file(path: Path, payload: dict[str, object]) -> None: + """Atomically write a compact, owner-readable JSON file.""" + path.parent.mkdir(parents=True, exist_ok=True) + fd, temporary_path = tempfile.mkstemp( + dir=str(path.parent), prefix=f".{path.stem}_", suffix=".tmp" + ) + try: + with os.fdopen(fd, "w", encoding="utf-8") as handle: + json.dump(payload, handle, separators=(",", ":")) + handle.flush() + os.fsync(handle.fileno()) + os.replace(temporary_path, path) + except BaseException: + try: + os.unlink(temporary_path) + except OSError: + pass + raise + + +def write_planned_stop_marker(target_pid: int) -> bool: + """Write the marker consumed by the gateway's SIGTERM handler.""" + try: + record: dict[str, object] = { + "target_pid": target_pid, + "target_start_time": _get_process_start_time(target_pid), + "stopper_pid": os.getpid(), + # The systemd ExecStop marker belongs to the real SIGTERM handler; + # the polling watcher must not consume it first. + "trigger_watcher": False, + "written_at": datetime.now(timezone.utc).isoformat(), + } + _write_json_file(_get_process_hermes_home() / _MARKER_FILENAME, record) + return True + except OSError: + return False + + +def main(argv: Sequence[str] | None = None) -> int: + args = list(sys.argv[1:] if argv is None else argv) + if len(args) != 1: + print("systemd planned-stop helper requires exactly one PID", file=sys.stderr) + return 2 + + try: + pid = int(args[0]) + except (TypeError, ValueError): + print("systemd planned-stop helper received an invalid PID", file=sys.stderr) + return 2 + if pid <= 0: + print("systemd planned-stop helper requires a positive PID", file=sys.stderr) + return 2 + + if write_planned_stop_marker(pid): + return 0 + print("systemd planned-stop helper could not write the marker", file=sys.stderr) + return 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/gateway/test_gateway_shutdown.py b/tests/gateway/test_gateway_shutdown.py index 1a4dbdceeee5a..b2459391a58a9 100644 --- a/tests/gateway/test_gateway_shutdown.py +++ b/tests/gateway/test_gateway_shutdown.py @@ -1,10 +1,14 @@ import asyncio +import os +import signal import subprocess from unittest.mock import AsyncMock, MagicMock, patch import pytest import gateway.run as gateway_run +import gateway.run_shutdown as gateway_shutdown +from gateway import status from gateway.config import HomeChannel, Platform from gateway.platforms.event import MessageEvent from gateway.restart import DEFAULT_GATEWAY_POST_INTERRUPT_GRACE_TIMEOUT, GATEWAY_SERVICE_RESTART_EXIT_CODE @@ -13,6 +17,20 @@ from tools import browser_tool_lifecycle as bt_lifecycle +def test_sigterm_with_planned_stop_marker_is_classified_as_planned(tmp_path, monkeypatch): + """The handler's SIGTERM path consumes and classifies the marker as planned.""" + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + assert status.write_planned_stop_marker(os.getpid()) is True + + planned_takeover, planned_stop = gateway_shutdown._classify_shutdown_signal( + signal.SIGTERM + ) + + assert planned_takeover is False + assert planned_stop is True + assert not (tmp_path / ".gateway-planned-stop.json").exists() + + @pytest.mark.asyncio async def test_cancel_background_tasks_cancels_inflight_message_processing(): _runner, adapter = make_restart_runner() diff --git a/tests/gateway/test_systemd_planned_stop.py b/tests/gateway/test_systemd_planned_stop.py new file mode 100644 index 0000000000000..3a0581fb78ba9 --- /dev/null +++ b/tests/gateway/test_systemd_planned_stop.py @@ -0,0 +1,61 @@ +"""Tests for the systemd ExecStop planned-stop marker helper.""" + +from __future__ import annotations + +import os +import subprocess +import sys +from pathlib import Path + +from gateway import status +import hermes_systemd_planned_stop as systemd_planned_stop + + +def test_main_writes_consumable_marker_for_target_pid(tmp_path, monkeypatch): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + + assert systemd_planned_stop.main([str(os.getpid())]) == 0 + # ExecStop publishes the marker before systemd sends SIGTERM. The generic + # watcher must leave it for the real signal handler to consume, otherwise + # the later SIGTERM would be misclassified as unexpected. + assert status.planned_stop_marker_targets_self() is False + assert (tmp_path / ".gateway-planned-stop.json").exists() + assert status.consume_planned_stop_marker_for_self() is True + + +def test_main_rejects_missing_invalid_and_nonpositive_pid(tmp_path, monkeypatch): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + + assert systemd_planned_stop.main([]) == 2 + assert systemd_planned_stop.main(["not-a-pid"]) == 2 + assert systemd_planned_stop.main(["0"]) == 2 + assert not (tmp_path / ".gateway-planned-stop.json").exists() + + +def test_main_reports_marker_write_failure(monkeypatch, capsys): + monkeypatch.setattr( + systemd_planned_stop, + "write_planned_stop_marker", + lambda _pid, **_kwargs: False, + ) + + assert systemd_planned_stop.main(["1234"]) == 1 + assert "could not write the marker" in capsys.readouterr().err + + +def test_helper_imports_without_optional_dependencies(): + """ExecStop must work even when application dependencies are unavailable.""" + project_root = Path(__file__).resolve().parents[2] + env = os.environ.copy() + env["PYTHONPATH"] = str(project_root) + + result = subprocess.run( + [sys.executable, "-S", "-c", "import hermes_systemd_planned_stop"], + cwd=project_root, + env=env, + capture_output=True, + text=True, + timeout=10, + ) + + assert result.returncode == 0, result.stderr diff --git a/tests/gateway/test_systemd_stop_mark.py b/tests/gateway/test_systemd_stop_mark.py index 2409a10ba6b3d..603c7bd9322ae 100644 --- a/tests/gateway/test_systemd_stop_mark.py +++ b/tests/gateway/test_systemd_stop_mark.py @@ -52,8 +52,10 @@ def _boom(_pid): from hermes_cli import gateway as gateway_cli unit = gateway_cli.generate_systemd_unit(system=False) - # Both modules run through the installation launcher (`hermes --run-module `). - assert "ExecStop=-" in unit and "gateway.systemd_stop_mark" in unit + # Both modules retain the installation-bound launcher/runtime contract. + stop_command = next(line for line in unit.splitlines() if line.startswith("ExecStop=")) + assert stop_command.startswith("ExecStop=-") and "hermes_systemd_planned_stop" in stop_command + assert stop_command.endswith(" $MAINPID") and "$$MAINPID" not in stop_command # The cgroup reaper still runs after the main process exits. assert "ExecStopPost=-" in unit and "gateway.cgroup_cleanup" in unit - assert unit.index("systemd_stop_mark") < unit.index("cgroup_cleanup") + assert unit.index("hermes_systemd_planned_stop") < unit.index("cgroup_cleanup") diff --git a/tests/hermes_cli/test_systemd_planned_stop_unit.py b/tests/hermes_cli/test_systemd_planned_stop_unit.py new file mode 100644 index 0000000000000..94555b1498e10 --- /dev/null +++ b/tests/hermes_cli/test_systemd_planned_stop_unit.py @@ -0,0 +1,48 @@ +"""Tests for planned-stop hooks in generated systemd units.""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +import hermes_cli.gateway as gateway_cli + + +def _assert_planned_stop_hook(unit: str) -> None: + command = next(line for line in unit.splitlines() if line.startswith("ExecStop=")) + assert command.startswith("ExecStop=-") + assert "hermes_systemd_planned_stop" in command + assert "gateway.systemd_stop_mark" not in command + # Keep the manager-provided PID outside generic argv escaping: quoting + # through _systemd_command would turn it into the literal $$MAINPID. + assert command.endswith(" $MAINPID") + assert "$$MAINPID" not in command + assert unit.index(command) < unit.index("ExecStopPost=") + + +def _runtime_owner(monkeypatch, managed_runtime: bool) -> None: + monkeypatch.setattr( + "hermes_cli._launchers.resolve_store_python", + lambda repo_root, **kwargs: Path("/store/python") if managed_runtime else None, + ) + monkeypatch.setattr(gateway_cli, "get_python_path", lambda: "/venv/bin/python") + + +@pytest.mark.parametrize("managed_runtime", [False, True]) +def test_user_unit_marks_direct_systemd_stop_as_planned(monkeypatch, managed_runtime): + _runtime_owner(monkeypatch, managed_runtime) + _assert_planned_stop_hook(gateway_cli.generate_systemd_unit(system=False)) + + +@pytest.mark.parametrize("managed_runtime", [False, True]) +def test_system_unit_marks_direct_systemd_stop_as_planned(monkeypatch, managed_runtime): + _runtime_owner(monkeypatch, managed_runtime) + monkeypatch.setattr( + gateway_cli, + "_system_service_identity", + lambda run_as_user=None: ("alice", "alice", "/home/alice", 1000), + ) + _assert_planned_stop_hook( + gateway_cli.generate_systemd_unit(system=True, run_as_user="alice") + )