diff --git a/api/agent_health.py b/api/agent_health.py index 1a6c1d76..2bc3fa8e 100644 --- a/api/agent_health.py +++ b/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} @@ -296,20 +360,22 @@ def _gateway_running_pid_strict_path(gateway_status: Any, pid_path: Path) -> int except TypeError: try: return get_running_pid(**kwargs) - except TypeError: - pass - - if pid_path_param.kind in ( - inspect.Parameter.POSITIONAL_ONLY, - inspect.Parameter.POSITIONAL_OR_KEYWORD, - ): - try: - return get_running_pid(pid_path, cleanup_stale=False) - except TypeError: - try: - return get_running_pid(pid_path) except TypeError: return None + + 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(*args, cleanup_stale=False) + except TypeError: + pass + try: + return get_running_pid(*args) + except TypeError: + return None return None diff --git a/tests/test_update_banner_fixes.py b/tests/test_update_banner_fixes.py index 55d332d2..0ab7a051 100644 --- a/tests/test_update_banner_fixes.py +++ b/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,