fix(gateway): mark direct systemd stops as planned

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:

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 = """

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(

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

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/<pid>/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())

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()

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

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 <mod>`).
-    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")

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")
+    )