fix: fail closed on wrapped gateway pid helpers

api/agent_health.py

@@ -252,14 +252,27 @@ def _gateway_running_pid(gateway_status: Any, pid_path: Path | None) -> int | No
         return get_running_pid()
 
 
+def _callable_code_declares_parameter(get_running_pid: Any, name: str) -> bool:
+    """Return True when a Python callable's own code declares *name*."""
+    target = getattr(get_running_pid, "__func__", get_running_pid)
+    code = getattr(target, "__code__", None)
+    if code is None:
+        return False
+    declared_count = int(getattr(code, "co_argcount", 0)) + int(
+        getattr(code, "co_kwonlyargcount", 0)
+    )
+    return name in code.co_varnames[:declared_count]
+
+
 def _declared_pid_path_parameter(
     get_running_pid: Any,
-) -> inspect.Parameter | None:
+) -> tuple[inspect.Parameter, int, list[inspect.Parameter]] | None:
     try:
-        signature = inspect.signature(get_running_pid)
+        signature = inspect.signature(get_running_pid, follow_wrapped=False)
     except (TypeError, ValueError):
         return None
-    for param in signature.parameters.values():
+    params = list(signature.parameters.values())
+    for idx, param in enumerate(params):
         if param.kind not in (
             inspect.Parameter.POSITIONAL_ONLY,
             inspect.Parameter.POSITIONAL_OR_KEYWORD,
@@ -267,17 +280,68 @@ def _declared_pid_path_parameter(
         ):
             continue
         name = str(param.name or "").lower()
-        if "path" in name and "pid" in name:
-            return param
+        if (
+            "path" in name
+            and "pid" in name
+            and _callable_code_declares_parameter(get_running_pid, param.name)
+        ):
+            positional_index = sum(
+                1
+                for earlier in params[:idx]
+                if earlier.kind
+                in (
+                    inspect.Parameter.POSITIONAL_ONLY,
+                    inspect.Parameter.POSITIONAL_OR_KEYWORD,
+                )
+            )
+            return param, positional_index, params
     return None
 
 
+def _positional_only_pid_args(
+    params: list[inspect.Parameter],
+    positional_index: int,
+    pid_path: Path,
+) -> list[Any] | None:
+    positional_params = [
+        param
+        for param in params
+        if param.kind
+        in (
+            inspect.Parameter.POSITIONAL_ONLY,
+            inspect.Parameter.POSITIONAL_OR_KEYWORD,
+        )
+    ]
+    if positional_index >= len(positional_params):
+        return None
+    args: list[Any] = []
+    for param in positional_params[:positional_index]:
+        if param.default is inspect.Parameter.empty:
+            return None
+        args.append(param.default)
+    args.append(pid_path)
+    return args
+
+
+def _accepts_cleanup_stale_keyword(params: list[inspect.Parameter]) -> bool:
+    return any(
+        param.name == "cleanup_stale"
+        and param.kind
+        in (
+            inspect.Parameter.KEYWORD_ONLY,
+            inspect.Parameter.POSITIONAL_OR_KEYWORD,
+        )
+        for param in params
+    )
+
+
 def _gateway_running_pid_strict_path(gateway_status: Any, pid_path: Path) -> int | None:
     """Read a PID from an explicit path only; never fall back to ambient state."""
     get_running_pid = gateway_status.get_running_pid
-    pid_path_param = _declared_pid_path_parameter(get_running_pid)
-    if pid_path_param is None:
+    pid_path_match = _declared_pid_path_parameter(get_running_pid)
+    if pid_path_match is None:
         return None
+    pid_path_param, positional_index, params = pid_path_match
 
     if pid_path_param.kind is inspect.Parameter.KEYWORD_ONLY:
         kwargs = {pid_path_param.name: pid_path}
@@ -297,17 +361,19 @@ def _gateway_running_pid_strict_path(gateway_status: Any, pid_path: Path) -> int
             try:
                 return get_running_pid(**kwargs)
             except TypeError:
-                pass
+                return None
 
-    if pid_path_param.kind in (
-        inspect.Parameter.POSITIONAL_ONLY,
-        inspect.Parameter.POSITIONAL_OR_KEYWORD,
-    ):
+    if pid_path_param.kind is inspect.Parameter.POSITIONAL_ONLY:
+        args = _positional_only_pid_args(params, positional_index, pid_path)
+        if args is None:
+            return None
+        if _accepts_cleanup_stale_keyword(params):
             try:
-            return get_running_pid(pid_path, cleanup_stale=False)
+                return get_running_pid(*args, cleanup_stale=False)
             except TypeError:
+                pass
         try:
-                return get_running_pid(pid_path)
+            return get_running_pid(*args)
         except TypeError:
             return None
     return None

tests/test_update_banner_fixes.py

@@ -22,6 +22,7 @@ import io
 import json
 import subprocess
 import types
+import functools
 
 import pytest
 
@@ -1145,6 +1146,180 @@ class TestAgentUpdateRequiresGatewayRestart:
         ]
         assert [call[1]["HERMES_HOME"] for call in calls["popen"]] == [str(root_home), str(root_home)]
 
+    def test_agent_gateway_restart_wrapped_kwargs_pid_change_fails_closed(
+        self,
+        monkeypatch,
+        tmp_path,
+    ):
+        from api import agent_health, gateway_restart, profiles
+        import api.updates as upd
+
+        root_home = tmp_path / ".hermes"
+        sticky_home = root_home / "profiles" / "work"
+        sticky_home.mkdir(parents=True)
+        calls = {"popen": [], "ambient_pid": 0}
+
+        class FailedRestartProcess:
+            returncode = 7
+
+            def communicate(self, timeout=None):
+                return "", "restart failed"
+
+        def declared_pid_reader(pid_path=None, *, cleanup_stale=True):
+            raise AssertionError("wrapped declaration must not be followed")
+
+        class WrappedKwargsGatewayStatus:
+            def __init__(self):
+                self._pids = iter([201, 202])
+
+            @functools.wraps(declared_pid_reader)
+            def get_running_pid(self, **kwargs):
+                calls["ambient_pid"] += 1
+                return next(self._pids)
+
+        def fake_popen(args, stdout=None, stderr=None, text=True, env=None):
+            calls["popen"].append((args, dict(env or {})))
+            return FailedRestartProcess()
+
+        monkeypatch.setattr(profiles, "_DEFAULT_HERMES_HOME", root_home)
+        monkeypatch.setattr(profiles, "_active_profile", "work")
+        monkeypatch.setattr(gateway_restart, "_GATEWAY_RESTART_LOCK", threading.Lock())
+        monkeypatch.setattr(gateway_restart.shutil, "which", lambda cmd: "/mock/bin/hermes")
+        monkeypatch.setattr(gateway_restart.subprocess, "Popen", fake_popen)
+        monkeypatch.setattr(agent_health, "_gateway_status_module", WrappedKwargsGatewayStatus)
+        monkeypatch.setattr(upd.time, 'sleep', lambda seconds: None)
+
+        profiles.set_request_profile("default")
+        try:
+            ok, result = upd._ensure_gateway_restart_for_agent_update()
+        finally:
+            profiles.clear_request_profile()
+
+        assert ok is False
+        assert result["status"] == "failed"
+        assert calls["ambient_pid"] == 0
+        assert [call[0] for call in calls["popen"]] == [
+            ["/mock/bin/hermes", "--profile", "default", "gateway", "restart"],
+            ["/mock/bin/hermes", "--profile", "default", "gateway", "restart"],
+        ]
+        assert [call[1]["HERMES_HOME"] for call in calls["popen"]] == [str(root_home), str(root_home)]
+
+    def test_agent_gateway_restart_wrapped_args_pid_change_fails_closed(
+        self,
+        monkeypatch,
+        tmp_path,
+    ):
+        from api import agent_health, gateway_restart, profiles
+        import api.updates as upd
+
+        root_home = tmp_path / ".hermes"
+        sticky_home = root_home / "profiles" / "work"
+        sticky_home.mkdir(parents=True)
+        calls = {"popen": [], "ambient_pid": 0}
+
+        class FailedRestartProcess:
+            returncode = 7
+
+            def communicate(self, timeout=None):
+                return "", "restart failed"
+
+        def declared_pid_reader(pid_path=None, *, cleanup_stale=True):
+            raise AssertionError("wrapped declaration must not be followed")
+
+        class WrappedArgsGatewayStatus:
+            def __init__(self):
+                self._pids = iter([201, 202])
+
+            @functools.wraps(declared_pid_reader)
+            def get_running_pid(self, *args):
+                calls["ambient_pid"] += 1
+                return next(self._pids)
+
+        def fake_popen(args, stdout=None, stderr=None, text=True, env=None):
+            calls["popen"].append((args, dict(env or {})))
+            return FailedRestartProcess()
+
+        monkeypatch.setattr(profiles, "_DEFAULT_HERMES_HOME", root_home)
+        monkeypatch.setattr(profiles, "_active_profile", "work")
+        monkeypatch.setattr(gateway_restart, "_GATEWAY_RESTART_LOCK", threading.Lock())
+        monkeypatch.setattr(gateway_restart.shutil, "which", lambda cmd: "/mock/bin/hermes")
+        monkeypatch.setattr(gateway_restart.subprocess, "Popen", fake_popen)
+        monkeypatch.setattr(agent_health, "_gateway_status_module", WrappedArgsGatewayStatus)
+        monkeypatch.setattr(upd.time, 'sleep', lambda seconds: None)
+
+        profiles.set_request_profile("default")
+        try:
+            ok, result = upd._ensure_gateway_restart_for_agent_update()
+        finally:
+            profiles.clear_request_profile()
+
+        assert ok is False
+        assert result["status"] == "failed"
+        assert calls["ambient_pid"] == 0
+        assert [call[0] for call in calls["popen"]] == [
+            ["/mock/bin/hermes", "--profile", "default", "gateway", "restart"],
+            ["/mock/bin/hermes", "--profile", "default", "gateway", "restart"],
+        ]
+        assert [call[1]["HERMES_HOME"] for call in calls["popen"]] == [str(root_home), str(root_home)]
+
+    def test_agent_gateway_restart_shifted_positional_only_pid_path_is_bound(
+        self,
+        monkeypatch,
+        tmp_path,
+    ):
+        from api import agent_health, gateway_restart, profiles
+        import api.updates as upd
+
+        root_home = tmp_path / ".hermes"
+        sticky_home = root_home / "profiles" / "work"
+        sticky_home.mkdir(parents=True)
+        calls = {"popen": [], "pid_paths": [], "ambient_pid": 0}
+
+        class FailedRestartProcess:
+            returncode = 7
+
+            def communicate(self, timeout=None):
+                return "", "restart failed"
+
+        class ShiftedPositionalOnlyGatewayStatus:
+            def get_running_pid(self, ambient=None, pid_path=None, /, *, cleanup_stale=True):
+                if pid_path is None:
+                    calls["ambient_pid"] += 1
+                    return 202
+                path = pathlib.Path(pid_path)
+                calls["pid_paths"].append(path)
+                if path == root_home / "gateway.pid":
+                    return 101
+                return None
+
+        def fake_popen(args, stdout=None, stderr=None, text=True, env=None):
+            calls["popen"].append((args, dict(env or {})))
+            return FailedRestartProcess()
+
+        monkeypatch.setattr(profiles, "_DEFAULT_HERMES_HOME", root_home)
+        monkeypatch.setattr(profiles, "_active_profile", "work")
+        monkeypatch.setattr(gateway_restart, "_GATEWAY_RESTART_LOCK", threading.Lock())
+        monkeypatch.setattr(gateway_restart.shutil, "which", lambda cmd: "/mock/bin/hermes")
+        monkeypatch.setattr(gateway_restart.subprocess, "Popen", fake_popen)
+        monkeypatch.setattr(agent_health, "_gateway_status_module", ShiftedPositionalOnlyGatewayStatus)
+        monkeypatch.setattr(upd.time, 'sleep', lambda seconds: None)
+
+        profiles.set_request_profile("default")
+        try:
+            ok, result = upd._ensure_gateway_restart_for_agent_update()
+        finally:
+            profiles.clear_request_profile()
+
+        assert ok is False
+        assert result["status"] == "failed"
+        assert calls["ambient_pid"] == 0
+        assert calls["pid_paths"] == [root_home / "gateway.pid", root_home / "gateway.pid"]
+        assert [call[0] for call in calls["popen"]] == [
+            ["/mock/bin/hermes", "--profile", "default", "gateway", "restart"],
+            ["/mock/bin/hermes", "--profile", "default", "gateway", "restart"],
+        ]
+        assert [call[1]["HERMES_HOME"] for call in calls["popen"]] == [str(root_home), str(root_home)]
+
     def test_agent_gateway_restart_isolated_default_home_omits_profile_flag(
         self,
         monkeypatch,