fix(gateway): write planned-stop marker from systemd ExecStop

hermes_cli/gateway.py

@@ -7,6 +7,7 @@
 import asyncio
 import logging
 import os
+import shlex
 import shutil
 import signal
 import subprocess
@@ -2394,6 +2395,11 @@ def generate_systemd_unit(system: bool = False, run_as_user: str | None = None)
         # _stable_service_working_dir() for the full rationale.
         working_dir = str(hermes_home) if hermes_home else _remap_path_for_user(working_dir, home_dir)
         venv_dir = _remap_path_for_user(venv_dir, home_dir)
+        python_path = _remap_path_for_user(python_path, home_dir)
+        planned_stop_exec = (
+            f"{shlex.quote(python_path)} -m hermes_cli.main "
+            "gateway planned-stop-helper --pid $MAINPID"
+        )
         path_entries = [_remap_path_for_user(p, home_dir) for p in path_entries]
         path_entries.extend(_build_user_local_paths(Path(home_dir), path_entries))
         path_entries.extend(_build_wsl_interop_paths(path_entries))
@@ -2422,6 +2428,7 @@ def generate_systemd_unit(system: bool = False, run_as_user: str | None = None)
 RestartForceExitStatus={GATEWAY_SERVICE_RESTART_EXIT_CODE}
 KillMode=mixed
 KillSignal=SIGTERM
+ExecStop={planned_stop_exec}
 ExecReload=/bin/kill -USR1 $MAINPID
 TimeoutStopSec={restart_timeout}
 StandardOutput=journal
@@ -2433,6 +2440,10 @@ def generate_systemd_unit(system: bool = False, run_as_user: str | None = None)
 
     hermes_home = str(get_hermes_home().resolve())
     profile_arg = _profile_arg(hermes_home)
+    planned_stop_exec = (
+        f"{shlex.quote(python_path)} -m hermes_cli.main "
+        "gateway planned-stop-helper --pid $MAINPID"
+    )
     path_entries.extend(_build_user_local_paths(Path.home(), path_entries))
     path_entries.extend(_build_wsl_interop_paths(path_entries))
     path_entries.extend(common_bin_paths)
@@ -2455,6 +2466,7 @@ def generate_systemd_unit(system: bool = False, run_as_user: str | None = None)
 RestartForceExitStatus={GATEWAY_SERVICE_RESTART_EXIT_CODE}
 KillMode=mixed
 KillSignal=SIGTERM
+ExecStop={planned_stop_exec}
 ExecReload=/bin/kill -USR1 $MAINPID
 TimeoutStopSec={restart_timeout}
 StandardOutput=journal
@@ -2836,6 +2848,16 @@ def systemd_stop(system: bool = False):
     print(f"✓ {_service_scope_label(system).capitalize()} service stopped")
 
 
+def gateway_planned_stop_helper(pid: int) -> bool:
+    """Write the planned-stop marker for ``pid`` and return success."""
+    try:
+        from gateway.status import write_planned_stop_marker
+
+        return bool(write_planned_stop_marker(pid))
+    except Exception:
+        return False
+
+
 def systemd_restart(system: bool = False):
     system = _select_systemd_scope(system)
     if system:
@@ -6507,6 +6529,17 @@ def _gateway_command_inner(args):
             else:
                 print(f"✓ Stopped {get_service_name()} service")
 
+    elif subcmd == "planned-stop-helper":
+        pid = int(getattr(args, "pid", 0) or 0)
+        if pid <= 0:
+            print("✗ --pid must be a positive integer")
+            sys.exit(1)
+        ok = gateway_planned_stop_helper(pid)
+        if not ok:
+            print("✗ Failed to write planned-stop marker")
+            sys.exit(1)
+        return
+
     elif subcmd == "restart":
         # Defense: refuse self-targeting gateway restart from inside the gateway.
         # Prevents agent-initiated kill loops when combined with supervisor KeepAlive.

hermes_cli/subcommands/gateway.py

@@ -89,6 +89,17 @@ def build_gateway_parser(subparsers, *, cmd_gateway: Callable, cmd_proxy: Callab
         help="Stop ALL gateway processes across all profiles",
     )
 
+    gateway_stop_helper = gateway_subparsers.add_parser(
+        "planned-stop-helper",
+        help=argparse.SUPPRESS,
+    )
+    gateway_stop_helper.add_argument(
+        "--pid",
+        type=int,
+        required=True,
+        help=argparse.SUPPRESS,
+    )
+
     # gateway restart
     gateway_restart = gateway_subparsers.add_parser(
         "restart", help="Restart gateway service"

tests/hermes_cli/test_gateway_service.py

@@ -325,7 +325,9 @@ def test_user_unit_avoids_recursive_execstop_and_uses_extended_stop_timeout(self
         unit = gateway_cli.generate_systemd_unit(system=False)
 
         assert "ExecStart=" in unit
-        assert "ExecStop=" not in unit
+        assert "ExecStop=" in unit
+        assert "gateway planned-stop-helper --pid $MAINPID" in unit
+        assert "gateway stop" not in unit
         assert "ExecReload=/bin/kill -USR1 $MAINPID" in unit
         assert f"RestartForceExitStatus={GATEWAY_SERVICE_RESTART_EXIT_CODE}" in unit
         # TimeoutStopSec must exceed the default drain_timeout (60s) so
@@ -386,7 +388,9 @@ def test_system_unit_avoids_recursive_execstop_and_uses_extended_stop_timeout(se
         unit = gateway_cli.generate_systemd_unit(system=True)
 
         assert "ExecStart=" in unit
-        assert "ExecStop=" not in unit
+        assert "ExecStop=" in unit
+        assert "gateway planned-stop-helper --pid $MAINPID" in unit
+        assert "gateway stop" not in unit
         assert "ExecReload=/bin/kill -USR1 $MAINPID" in unit
         assert f"RestartForceExitStatus={GATEWAY_SERVICE_RESTART_EXIT_CODE}" in unit
         # TimeoutStopSec must exceed the default drain_timeout (60s) so
@@ -395,6 +399,30 @@ def test_system_unit_avoids_recursive_execstop_and_uses_extended_stop_timeout(se
         assert self._expected_timeout_stop_sec() in unit
         assert "WantedBy=multi-user.target" in unit
 
+    def test_planned_stop_helper_writes_marker_for_target_pid(self, monkeypatch):
+        calls = []
+
+        monkeypatch.setattr(
+            gateway_cli,
+            "gateway_planned_stop_helper",
+            lambda pid: calls.append(pid) or True,
+        )
+
+        gateway_cli.gateway_command(SimpleNamespace(gateway_command="planned-stop-helper", pid=321))
+
+        assert calls == [321]
+
+    def test_planned_stop_helper_exits_nonzero_on_marker_failure(self, monkeypatch, capsys):
+        monkeypatch.setattr(gateway_cli, "gateway_planned_stop_helper", lambda pid: False)
+
+        with pytest.raises(SystemExit) as exc_info:
+            gateway_cli.gateway_command(
+                SimpleNamespace(gateway_command="planned-stop-helper", pid=321)
+            )
+
+        assert exc_info.value.code == 1
+        assert "Failed to write planned-stop marker" in capsys.readouterr().out
+
 
 class TestGatewayStopCleanup:
     def test_stop_only_kills_current_profile_by_default(self, tmp_path, monkeypatch):