fix(gateway): make intentional systemd stops exit cleanly

hermes_cli/gateway.py

@@ -8,6 +8,7 @@
 import os
 import shutil
 import signal
+import stat
 import subprocess
 import sys
 import textwrap
@@ -2156,6 +2157,7 @@ def generate_systemd_unit(system: bool = False, run_as_user: str | None = None)
 User={username}
 Group={group_name}
 ExecStart={python_path} -m hermes_cli.main{f" {profile_arg}" if profile_arg else ""} gateway run --replace
+ExecStop={python_path} -m hermes_cli.main{f" {profile_arg}" if profile_arg else ""} gateway _write-planned-stop --pid $MAINPID
 WorkingDirectory={working_dir}
 Environment="HOME={home_dir}"
 Environment="USER={username}"
@@ -2194,6 +2196,7 @@ def generate_systemd_unit(system: bool = False, run_as_user: str | None = None)
 [Service]
 Type=simple
 ExecStart={python_path} -m hermes_cli.main{f" {profile_arg}" if profile_arg else ""} gateway run --replace
+ExecStop={python_path} -m hermes_cli.main{f" {profile_arg}" if profile_arg else ""} gateway _write-planned-stop --pid $MAINPID
 WorkingDirectory={working_dir}
 Environment="PATH={sane_path}"
 Environment="VIRTUAL_ENV={venv_dir}"
@@ -2218,6 +2221,52 @@ def _normalize_service_definition(text: str) -> str:
     return "\n".join(line.rstrip() for line in text.strip().splitlines())
 
 
+def _ensure_planned_stop_marker_readable() -> None:
+    """Relax marker permissions so a different service user can consume it.
+
+    `hermes gateway stop --system` commonly runs under sudo/root while the
+    service itself runs as an unprivileged `User=`. The gateway only needs
+    read access to detect a planned stop, so make the marker world-readable
+    without otherwise changing marker semantics.
+    """
+    marker_path = get_hermes_home() / ".gateway-planned-stop.json"
+    try:
+        current_mode = stat.S_IMODE(marker_path.stat().st_mode)
+    except OSError:
+        return
+    target_mode = current_mode | stat.S_IRGRP | stat.S_IROTH
+    if target_mode != current_mode:
+        try:
+            os.chmod(marker_path, target_mode)
+        except OSError:
+            pass
+
+
+def _write_planned_stop_only(pid: int | str | None) -> bool:
+    """Write the planned-stop marker for `pid` and exit without signalling.
+
+    Used by systemd `ExecStop` so the service user can mark the stop as
+    intentional before PID 1 delivers SIGTERM to the main gateway process.
+    """
+    try:
+        target_pid = int(pid)
+    except (TypeError, ValueError):
+        return False
+    if target_pid <= 0:
+        return False
+
+    try:
+        from gateway.status import write_planned_stop_marker
+
+        wrote = write_planned_stop_marker(target_pid)
+    except Exception:
+        return False
+
+    if wrote:
+        _ensure_planned_stop_marker_readable()
+    return wrote
+
+
 def _normalize_launchd_plist_for_comparison(text: str) -> str:
     """Normalize launchd plist text for staleness checks.
 
@@ -2512,10 +2561,10 @@ def systemd_stop(system: bool = False):
     _require_service_installed("stop", system=system)
     _sync_hermes_home_from_systemd_unit(system=system)
     try:
-        from gateway.status import get_running_pid, write_planned_stop_marker
+        from gateway.status import get_running_pid
         pid = get_running_pid(cleanup_stale=False)
         if pid is not None:
-            write_planned_stop_marker(pid)
+            _write_planned_stop_only(pid)
     except Exception:
         pass
     try:
@@ -2926,10 +2975,10 @@ def launchd_stop():
     label = get_launchd_label()
     target = f"{_launchd_domain()}/{label}"
     try:
-        from gateway.status import get_running_pid, write_planned_stop_marker
+        from gateway.status import get_running_pid
         pid = get_running_pid(cleanup_stale=False)
         if pid is not None:
-            write_planned_stop_marker(pid)
+            _write_planned_stop_only(pid)
     except Exception:
         pass
     # bootout unloads the service definition so KeepAlive doesn't respawn
@@ -5003,6 +5052,12 @@ def _gateway_command_inner(args):
         gateway_setup()
         return
 
+    if subcmd == "_write-planned-stop":
+        pid = getattr(args, "pid", None)
+        if not _write_planned_stop_only(pid):
+            sys.exit(1)
+        return
+
     # Service management commands
     if subcmd == "install":
         if is_managed():

hermes_cli/main.py

@@ -9528,6 +9528,12 @@ def main():
     # gateway list
     gateway_subparsers.add_parser("list", help="List all profiles and their gateway status")
 
+    gateway_write_planned_stop = gateway_subparsers.add_parser(
+        "_write-planned-stop",
+        help=argparse.SUPPRESS,
+    )
+    gateway_write_planned_stop.add_argument("--pid", required=True)
+
     # gateway setup
     gateway_subparsers.add_parser("setup", help="Configure messaging platforms")
 

tests/hermes_cli/test_gateway.py

@@ -164,6 +164,28 @@ def fake_signal(sig, handler):
     assert calls == [(False, 0)]
     assert (gateway.signal.SIGINT, gateway.signal.SIG_IGN) in signal_calls
 
+def test_gateway_write_planned_stop_internal_command_dispatches(monkeypatch):
+    calls = []
+
+    monkeypatch.setattr(
+        gateway,
+        "_write_planned_stop_only",
+        lambda pid: calls.append(pid) or True,
+    )
+
+    gateway.gateway_command(SimpleNamespace(gateway_command="_write-planned-stop", pid="321"))
+
+    assert calls == ["321"]
+
+
+def test_gateway_write_planned_stop_internal_command_exits_nonzero_on_failure(monkeypatch):
+    monkeypatch.setattr(gateway, "_write_planned_stop_only", lambda pid: False)
+
+    with pytest.raises(SystemExit) as exc_info:
+        gateway.gateway_command(SimpleNamespace(gateway_command="_write-planned-stop", pid="0"))
+
+    assert exc_info.value.code == 1
+
 
 class TestSystemdLingerStatus:
     def test_reports_enabled(self, monkeypatch):

tests/hermes_cli/test_gateway_service.py

@@ -1,6 +1,8 @@
 """Tests for gateway service management helpers."""
 
 import os
+import pwd
+import stat
 import subprocess
 from pathlib import Path
 from types import SimpleNamespace
@@ -38,6 +40,43 @@ def test_wait_for_user_dbus_socket_accepts_private_socket(self, monkeypatch):
 
 
 class TestSystemdServiceRefresh:
+    def test_generate_systemd_unit_includes_execstop_helper(self, monkeypatch):
+        monkeypatch.setattr(
+            gateway_cli,
+            "_system_service_identity",
+            lambda run_as_user=None: ("hermes", "hermes", "/home/hermes"),
+        )
+        monkeypatch.setattr(
+            gateway_cli,
+            "_hermes_home_for_target_user",
+            lambda home_dir: f"{home_dir}/.hermes/profiles/karting",
+        )
+        monkeypatch.setattr(gateway_cli, "get_python_path", lambda: "/home/hermes/.local/bin/python")
+        monkeypatch.setattr(gateway_cli, "_detect_venv_dir", lambda: None)
+        monkeypatch.setattr(gateway_cli, "_profile_arg", lambda hermes_home: "--profile karting")
+
+        unit = gateway_cli.generate_systemd_unit(system=True, run_as_user="hermes")
+
+        assert (
+            "ExecStop=/home/hermes/.local/bin/python -m hermes_cli.main --profile karting "
+            "gateway _write-planned-stop --pid $MAINPID"
+        ) in unit
+
+    def test_write_planned_stop_only_makes_marker_world_readable(self, tmp_path, monkeypatch):
+        marker_path = tmp_path / ".gateway-planned-stop.json"
+
+        monkeypatch.setattr(gateway_cli, "get_hermes_home", lambda: tmp_path)
+
+        def fake_write_planned_stop_marker(pid):
+            marker_path.write_text('{"target_pid":123}', encoding="utf-8")
+            marker_path.chmod(0o600)
+            return True
+
+        monkeypatch.setattr(status, "write_planned_stop_marker", fake_write_planned_stop_marker)
+
+        assert gateway_cli._write_planned_stop_only(123) is True
+        assert stat.S_IMODE(marker_path.stat().st_mode) == 0o644
+
     def test_systemd_install_repairs_outdated_unit_without_force(self, tmp_path, monkeypatch):
         unit_path = tmp_path / "hermes-gateway.service"
         unit_path.write_text("old unit\n", encoding="utf-8")
@@ -125,8 +164,8 @@ def test_systemd_stop_marks_running_gateway_as_planned_stop(self, monkeypatch):
         monkeypatch.setattr(gateway_cli, "_require_service_installed", lambda action, system=False: None)
         monkeypatch.setattr(status, "get_running_pid", lambda cleanup_stale=True: 321)
         monkeypatch.setattr(
-            status,
-            "write_planned_stop_marker",
+            gateway_cli,
+            "_write_planned_stop_only",
             lambda pid: markers.append(pid) or True,
         )
 
@@ -148,8 +187,8 @@ def test_systemd_stop_timeout_prints_status_guidance(self, monkeypatch, capsys):
         monkeypatch.setattr(gateway_cli, "_require_service_installed", lambda action, system=False: None)
         monkeypatch.setattr(status, "get_running_pid", lambda cleanup_stale=True: 321)
         monkeypatch.setattr(
-            status,
-            "write_planned_stop_marker",
+            gateway_cli,
+            "_write_planned_stop_only",
             lambda pid: markers.append(pid) or True,
         )