fix(gateway): implement platform-aware PID termination

tests/gateway/test_runner_startup_failures.py

@@ -87,3 +87,42 @@ async def test_runner_allows_cron_only_mode_when_no_platforms_are_enabled(monkey
     assert runner.adapters == {}
     state = read_runtime_status()
     assert state["gateway_state"] == "running"
+
+
+@pytest.mark.asyncio
+async def test_start_gateway_replace_force_uses_terminate_pid(monkeypatch, tmp_path):
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+
+    calls = []
+
+    class _CleanExitRunner:
+        def __init__(self, config):
+            self.config = config
+            self.should_exit_cleanly = True
+            self.exit_reason = None
+            self.adapters = {}
+
+        async def start(self):
+            return True
+
+        async def stop(self):
+            return None
+
+    monkeypatch.setattr("gateway.status.get_running_pid", lambda: 42)
+    monkeypatch.setattr("gateway.status.remove_pid_file", lambda: None)
+    monkeypatch.setattr("gateway.status.release_all_scoped_locks", lambda: 0)
+    monkeypatch.setattr("gateway.status.terminate_pid", lambda pid, force=False: calls.append((pid, force)))
+    monkeypatch.setattr("gateway.run.os.getpid", lambda: 100)
+    monkeypatch.setattr("gateway.run.os.kill", lambda pid, sig: None)
+    monkeypatch.setattr("time.sleep", lambda _: None)
+    monkeypatch.setattr("tools.skills_sync.sync_skills", lambda quiet=True: None)
+    monkeypatch.setattr("hermes_logging.setup_logging", lambda hermes_home, mode: tmp_path)
+    monkeypatch.setattr("hermes_logging._add_rotating_handler", lambda *args, **kwargs: None)
+    monkeypatch.setattr("gateway.run.GatewayRunner", _CleanExitRunner)
+
+    from gateway.run import start_gateway
+
+    ok = await start_gateway(config=GatewayConfig(), replace=True, verbosity=None)
+
+    assert ok is True
+    assert calls == [(42, False), (42, True)]

tests/gateway/test_status.py

@@ -2,6 +2,7 @@
 
 import json
 import os
+from types import SimpleNamespace
 
 from gateway import status
 
@@ -104,6 +105,41 @@ class TestGatewayRuntimeStatus:
         assert payload["platforms"]["telegram"]["error_message"] == "another poller is active"
 
 
+class TestTerminatePid:
+    def test_force_uses_taskkill_on_windows(self, monkeypatch):
+        calls = []
+        monkeypatch.setattr(status, "_IS_WINDOWS", True)
+
+        def fake_run(cmd, capture_output=False, text=False, timeout=None):
+            calls.append((cmd, capture_output, text, timeout))
+            return SimpleNamespace(returncode=0, stdout="", stderr="")
+
+        monkeypatch.setattr(status.subprocess, "run", fake_run)
+
+        status.terminate_pid(123, force=True)
+
+        assert calls == [
+            (["taskkill", "/PID", "123", "/T", "/F"], True, True, 10)
+        ]
+
+    def test_force_falls_back_to_sigterm_when_taskkill_missing(self, monkeypatch):
+        calls = []
+        monkeypatch.setattr(status, "_IS_WINDOWS", True)
+
+        def fake_run(*args, **kwargs):
+            raise FileNotFoundError
+
+        def fake_kill(pid, sig):
+            calls.append((pid, sig))
+
+        monkeypatch.setattr(status.subprocess, "run", fake_run)
+        monkeypatch.setattr(status.os, "kill", fake_kill)
+
+        status.terminate_pid(456, force=True)
+
+        assert calls == [(456, status.signal.SIGTERM)]
+
+
 class TestScopedLocks:
     def test_acquire_scoped_lock_rejects_live_other_process(self, tmp_path, monkeypatch):
         monkeypatch.setenv("HERMES_GATEWAY_LOCK_DIR", str(tmp_path / "locks"))

tests/hermes_cli/test_gateway.py

@@ -1,6 +1,5 @@
 """Tests for hermes_cli.gateway."""
 
-import signal
 from types import SimpleNamespace
 from unittest.mock import patch, call
 
@@ -211,8 +210,7 @@ class TestWaitForGatewayExit:
         assert poll_count == 3
 
     def test_force_kills_after_grace_period(self, monkeypatch):
-        """When the process doesn't exit, SIGKILL the saved PID."""
-        import time as _time
+        """When the process doesn't exit, force-kill the saved PID."""
 
         # Simulate monotonic time advancing past force_after
         call_num = 0
@@ -224,8 +222,8 @@ class TestWaitForGatewayExit:
             return call_num * 2.0  # 2, 4, 6, 8, ...
 
         kills = []
-        def mock_kill(pid, sig):
-            kills.append((pid, sig))
+        def mock_terminate(pid, force=False):
+            kills.append((pid, force))
 
         # get_running_pid returns the PID until kill is sent, then None
         def mock_get_running_pid():
@@ -234,14 +232,13 @@ class TestWaitForGatewayExit:
         monkeypatch.setattr("time.monotonic", fake_monotonic)
         monkeypatch.setattr("time.sleep", lambda _: None)
         monkeypatch.setattr("gateway.status.get_running_pid", mock_get_running_pid)
-        monkeypatch.setattr("os.kill", mock_kill)
+        monkeypatch.setattr(gateway, "terminate_pid", mock_terminate)
 
         gateway._wait_for_gateway_exit(timeout=10.0, force_after=5.0)
-        assert (42, signal.SIGKILL) in kills
+        assert (42, True) in kills
 
     def test_handles_process_already_gone_on_kill(self, monkeypatch):
-        """ProcessLookupError during SIGKILL is not fatal."""
-        import time as _time
+        """ProcessLookupError during force-kill is not fatal."""
 
         call_num = 0
         def fake_monotonic():
@@ -249,13 +246,24 @@ class TestWaitForGatewayExit:
             call_num += 1
             return call_num * 3.0  # Jump past force_after quickly
 
-        def mock_kill(pid, sig):
+        def mock_terminate(pid, force=False):
             raise ProcessLookupError
 
         monkeypatch.setattr("time.monotonic", fake_monotonic)
         monkeypatch.setattr("time.sleep", lambda _: None)
         monkeypatch.setattr("gateway.status.get_running_pid", lambda: 99)
-        monkeypatch.setattr("os.kill", mock_kill)
+        monkeypatch.setattr(gateway, "terminate_pid", mock_terminate)
 
         # Should not raise — ProcessLookupError means it's already gone.
         gateway._wait_for_gateway_exit(timeout=10.0, force_after=2.0)
+
+    def test_kill_gateway_processes_force_uses_helper(self, monkeypatch):
+        calls = []
+
+        monkeypatch.setattr(gateway, "find_gateway_pids", lambda exclude_pids=None: [11, 22])
+        monkeypatch.setattr(gateway, "terminate_pid", lambda pid, force=False: calls.append((pid, force)))
+
+        killed = gateway.kill_gateway_processes(force=True)
+
+        assert killed == 2
+        assert calls == [(11, True), (22, True)]