Fix stale gateway PID recovery and add systemd startup cleanup

gateway/status.py

@@ -216,10 +216,11 @@ def _cleanup_invalid_pid_path(pid_path: Path, *, cleanup_stale: bool) -> None:
     if not cleanup_stale:
         return
     try:
-        if pid_path == _get_pid_path():
-            remove_pid_file()
-        else:
-            pid_path.unlink(missing_ok=True)
+        # By the time we get here, the PID record has already been proven
+        # invalid, stale, or non-gateway.  Force-unlink the file directly
+        # instead of delegating to remove_pid_file(), whose same-PID guard is
+        # only correct for the live atexit handoff path.
+        pid_path.unlink(missing_ok=True)
     except Exception:
         pass
 

hermes_cli/gateway.py

@@ -6,6 +6,7 @@
 
 import asyncio
 import os
+import shlex
 import shutil
 import signal
 import subprocess
@@ -1194,6 +1195,27 @@ def generate_systemd_unit(system: bool = False, run_as_user: str | None = None)
     common_bin_paths = ["/usr/local/sbin", "/usr/local/bin", "/usr/sbin", "/usr/bin", "/sbin", "/bin"]
     restart_timeout = max(60, int(_get_restart_drain_timeout() or 0))
 
+    def _stale_pid_exec_start_pre(target_home: str) -> str:
+        pid_file = Path(target_home) / "gateway.pid"
+        script = (
+            "import json, os, pathlib; "
+            f"p=pathlib.Path(r'{str(pid_file)}'); "
+            "d=None; "
+            "exists=p.exists(); "
+            "exists and (d:=json.loads(p.read_text())); "
+            "pid=(int(d.get('pid')) if isinstance(d, dict) else int(d)) if exists else None; "
+            "(os.kill(pid, 0) if pid is not None else None)"
+        )
+        # If the helper itself fails to parse/read the PID file, treat it as stale
+        # and remove it before startup.  Live processes survive because os.kill(pid, 0)
+        # succeeds and exits 0.
+        return (
+            "/bin/sh -lc "
+            + shlex.quote(
+                f"{python_path} -c {shlex.quote(script)} >/dev/null 2>&1 || rm -f {shlex.quote(str(pid_file))}"
+            )
+        )
+
     if system:
         username, group_name, home_dir = _system_service_identity(run_as_user)
         hermes_home = _hermes_home_for_target_user(home_dir)
@@ -1221,6 +1243,7 @@ def generate_systemd_unit(system: bool = False, run_as_user: str | None = None)
 Type=simple
 User={username}
 Group={group_name}
+ExecStartPre={_stale_pid_exec_start_pre(hermes_home)}
 ExecStart={python_path} -m hermes_cli.main{f" {profile_arg}" if profile_arg else ""} gateway run --replace
 WorkingDirectory={working_dir}
 Environment="HOME={home_dir}"
@@ -1256,6 +1279,7 @@ def generate_systemd_unit(system: bool = False, run_as_user: str | None = None)
 
 [Service]
 Type=simple
+ExecStartPre={_stale_pid_exec_start_pre(hermes_home)}
 ExecStart={python_path} -m hermes_cli.main{f" {profile_arg}" if profile_arg else ""} gateway run --replace
 WorkingDirectory={working_dir}
 Environment="PATH={sane_path}"

tests/gateway/test_status.py

@@ -105,6 +105,24 @@ def test_get_running_pid_accepts_explicit_pid_path_without_cleanup(self, tmp_pat
         assert status.get_running_pid(pid_path, cleanup_stale=False) == os.getpid()
         assert pid_path.exists()
 
+    def test_get_running_pid_cleans_up_stale_foreign_gateway_pid_file(self, tmp_path, monkeypatch):
+        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+        pid_path = tmp_path / "gateway.pid"
+        pid_path.write_text(json.dumps({
+            "pid": 999999,
+            "kind": "hermes-gateway",
+            "argv": ["python", "-m", "hermes_cli.main", "gateway", "run", "--replace"],
+            "start_time": 123,
+        }))
+
+        def _dead_pid(pid, sig):
+            raise ProcessLookupError
+
+        monkeypatch.setattr(status.os, "kill", _dead_pid)
+
+        assert status.get_running_pid() is None
+        assert not pid_path.exists()
+
 
 class TestGatewayRuntimeStatus:
     def test_write_runtime_status_overwrites_stale_pid_on_restart(self, tmp_path, monkeypatch):

tests/hermes_cli/test_gateway_service.py

@@ -88,6 +88,8 @@ 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 "ExecStartPre=" in unit
+        assert "gateway.pid" in unit
         assert "ExecStop=" not in unit
         assert "ExecReload=/bin/kill -USR1 $MAINPID" in unit
         assert f"RestartForceExitStatus={GATEWAY_SERVICE_RESTART_EXIT_CODE}" in unit
@@ -104,6 +106,8 @@ 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 "ExecStartPre=" in unit
+        assert "gateway.pid" in unit
         assert "ExecStop=" not in unit
         assert "ExecReload=/bin/kill -USR1 $MAINPID" in unit
         assert f"RestartForceExitStatus={GATEWAY_SERVICE_RESTART_EXIT_CODE}" in unit