Merge branch 'rebase-6054' into stage/6054

api/agent_health.py

@@ -23,6 +23,7 @@ confirmed outage.
 from __future__ import annotations
 
 import importlib
+import inspect
 import json
 import os
 import threading
@@ -251,6 +252,155 @@ 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,
+) -> tuple[inspect.Parameter, int, list[inspect.Parameter]] | None:
+    try:
+        signature = inspect.signature(get_running_pid, follow_wrapped=False)
+    except (TypeError, ValueError):
+        return None
+    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,
+            inspect.Parameter.KEYWORD_ONLY,
+        ):
+            continue
+        name = str(param.name or "").lower()
+        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_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}
+        try:
+            return get_running_pid(**kwargs, cleanup_stale=False)
+        except TypeError:
+            try:
+                return get_running_pid(**kwargs)
+            except TypeError:
+                return None
+
+    if pid_path_param.kind is inspect.Parameter.POSITIONAL_OR_KEYWORD:
+        kwargs = {pid_path_param.name: pid_path}
+        try:
+            return get_running_pid(**kwargs, cleanup_stale=False)
+        except TypeError:
+            try:
+                return get_running_pid(**kwargs)
+            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
+
+
+def get_active_profile_gateway_running_pid(profile: str | None = None) -> int | None:
+    """Return the confirmed active-profile PID without state fallbacks.
+
+    Update recovery needs process identity, not just a recent ``running`` state:
+    an old gateway can remain alive when the restart CLI fails before executing.
+    Use the same active-profile home targeted by the restart helper rather than
+    the default-root health path. Keep this fail-closed when status is unavailable.
+    """
+    try:
+        from api.profiles import get_active_hermes_home, get_hermes_home_for_profile
+
+        gateway_status = _gateway_status_module()
+        if profile is None:
+            gateway_home = get_active_hermes_home()
+        else:
+            gateway_home = get_hermes_home_for_profile(profile)
+        gateway_pid_path = Path(gateway_home) / _GATEWAY_PID_FILE
+        return _gateway_running_pid_strict_path(gateway_status, gateway_pid_path)
+    except Exception:
+        return None
+
+
 def _runtime_detail_subset(runtime_status: dict[str, Any] | None) -> dict[str, Any]:
     """Return only non-sensitive runtime fields for the browser.
 

api/gateway_restart.py

@@ -10,7 +10,13 @@ import sys
 import threading
 from pathlib import Path
 
-from api.profiles import get_active_hermes_home
+from api.profiles import (
+    _PROFILE_ID_RE,
+    _is_root_profile,
+    get_active_hermes_home,
+    get_active_profile_name,
+    get_hermes_home_for_profile,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -46,8 +52,31 @@ def _release_lock() -> None:
         pass
 
 
+def _gateway_restart_profile_context(profile: str | None = None) -> tuple[Path, str | None]:
+    """Return the HERMES_HOME and CLI profile arg for a gateway restart."""
+    if profile is None:
+        raw_profile = str(get_active_profile_name() or "default").strip()
+        active_home = Path(get_active_hermes_home())
+    else:
+        raw_profile = str(profile or "")
+        if not raw_profile or not _PROFILE_ID_RE.fullmatch(raw_profile):
+            raise ValueError(f"Invalid profile for gateway restart: {profile!r}")
+        active_home = Path(get_hermes_home_for_profile(raw_profile))
+
+    if (
+        raw_profile == "default"
+        and active_home.name == "default"
+        and active_home.parent.name == "profiles"
+    ):
+        return active_home, None
+    if not raw_profile or not _PROFILE_ID_RE.fullmatch(raw_profile) or _is_root_profile(raw_profile):
+        return active_home, "default"
+    return active_home, raw_profile
+
+
 def restart_active_profile_gateway(
     *,
+    profile: str | None = None,
     quick_timeout_seconds: float = 2.0,
     background_wait_seconds: float = 240.0,
 ) -> dict:
@@ -66,18 +95,30 @@ def restart_active_profile_gateway(
         }
 
     try:
-        active_home = get_active_hermes_home()
+        active_home, cli_profile = _gateway_restart_profile_context(profile)
         env = os.environ.copy()
         env["HERMES_HOME"] = str(active_home)
         hermes_cmd = _resolve_hermes_command()
+        cmd = [hermes_cmd]
+        if cli_profile is not None:
+            cmd.extend(["--profile", cli_profile])
+        cmd.extend(["gateway", "restart"])
 
-        logger.info(
-            "Restarting gateway service via CLI command: %s gateway restart (HERMES_HOME=%s)",
-            hermes_cmd,
-            active_home,
-        )
+        if cli_profile is None:
+            logger.info(
+                "Restarting gateway service via CLI command: %s gateway restart (HERMES_HOME=%s)",
+                hermes_cmd,
+                active_home,
+            )
+        else:
+            logger.info(
+                "Restarting gateway service via CLI command: %s --profile %s gateway restart (HERMES_HOME=%s)",
+                hermes_cmd,
+                cli_profile,
+                active_home,
+            )
         proc = subprocess.Popen(
-            [hermes_cmd, "gateway", "restart"],
+            cmd,
             stdout=subprocess.PIPE,
             stderr=subprocess.PIPE,
             text=True,

api/updates.py

@@ -24,7 +24,9 @@ from collections import OrderedDict
 from pathlib import Path
 from urllib.parse import urlparse
 
+from api.agent_health import get_active_profile_gateway_running_pid
 from api.gateway_restart import restart_active_profile_gateway
+from api.profiles import get_active_profile_name
 from api.config import REPO_ROOT, STREAMS, STREAMS_LOCK
 
 logger = logging.getLogger(__name__)
@@ -42,6 +44,7 @@ _cache_lock = threading.Lock()
 _check_in_progress = False
 _apply_lock = threading.Lock()   # prevents concurrent stash/pull/pop on same repo
 CACHE_TTL = 1800  # 30 minutes
+_AGENT_GATEWAY_RESTART_RETRY_DELAY_S = 1.0
 _GIT_DIAGNOSTIC_MAX_CHARS = 300
 _CREDENTIAL_IN_URL_RE = re.compile(r"([a-zA-Z][a-zA-Z0-9+.-]*://)([^/@\s'\"]+)@")
 _GITHUB_TOKEN_RE = re.compile(r"\b(?:gh[pousr]_[A-Za-z0-9_]{20,}|github_pat_[A-Za-z0-9_]{20,})\b")
@@ -1814,11 +1817,62 @@ def _ensure_gateway_restart_for_agent_update() -> tuple[bool, dict]:
         - ok is False when restart did not complete and callers must abort success.
         - restart_payload contains helper status fields for response shaping.
     """
-    restart_result = restart_active_profile_gateway()
+    target_profile = str(get_active_profile_name() or "default").strip() or "default"
+    gateway_pid_before_restart = get_active_profile_gateway_running_pid(profile=target_profile)
+    restart_result = restart_active_profile_gateway(profile=target_profile)
     status = str(restart_result.get("status") or "")
     if status in {"completed", "in_progress"}:
         return True, restart_result
-    return False, restart_result
+    if status != "failed":
+        return False, restart_result
+
+    # launchd can briefly fail to spawn the replacement gateway while it is
+    # rotating the supervised process (#6045). Retry exactly once after a
+    # bounded delay so an already-applied Agent update is not reported as a
+    # complete failure because of that transient process handoff.
+    time.sleep(_AGENT_GATEWAY_RESTART_RETRY_DELAY_S)
+    retry_result = restart_active_profile_gateway(profile=target_profile)
+    retry_status = str(retry_result.get("status") or "")
+    if retry_status in {"completed", "in_progress"}:
+        return True, {
+            **retry_result,
+            "retry_attempted": True,
+            "initial_failure": restart_result.get("message"),
+        }
+    if retry_status != "failed":
+        return False, {
+            **retry_result,
+            "retry_attempted": True,
+            "initial_failure": restart_result.get("message"),
+        }
+
+    # A restart command can still exit non-zero after launchd has recovered the
+    # service. Only accept that recovery when the confirmed local PID changed;
+    # a merely-alive old gateway has not loaded the updated Agent checkout.
+    time.sleep(_AGENT_GATEWAY_RESTART_RETRY_DELAY_S)
+    gateway_pid_after_retry = get_active_profile_gateway_running_pid(profile=target_profile)
+    if (
+        gateway_pid_before_restart is not None
+        and gateway_pid_after_retry is not None
+        and gateway_pid_after_retry != gateway_pid_before_restart
+    ):
+        return True, {
+            "status": "completed",
+            "message": "Gateway service recovered after a transient restart failure",
+            "retry_attempted": True,
+            "process_replaced": True,
+            "initial_failure": restart_result.get("message"),
+            "retry_failure": retry_result.get("message"),
+        }
+
+    initial_message = str(restart_result.get("message") or "Restart failed")
+    retry_message = str(retry_result.get("message") or "retry did not complete")
+    return False, {
+        **retry_result,
+        "message": f"{initial_message}; recovery retry did not complete: {retry_message}",
+        "retry_attempted": True,
+        "initial_failure": restart_result.get("message"),
+    }
 
 
 def _agent_gateway_restart_failure_message(target: str, restart_result: dict) -> str:

tests/test_health_restart.py

@@ -96,11 +96,104 @@ def test_restart_active_profile_gateway_success_uses_active_profile_home(monkeyp
 
     assert result["status"] == "completed"
     assert result["message"] == "Gateway service restarted successfully"
-    assert called["args"] == ["/mock/bin/hermes", "gateway", "restart"]
+    assert called["args"] == ["/mock/bin/hermes", "--profile", "default", "gateway", "restart"]
     assert called["env"]["HERMES_HOME"] == "/mock/hermes/home"
     assert gateway_restart._GATEWAY_RESTART_LOCK.locked() is False
 
 
+def test_restart_active_profile_gateway_pins_explicit_default_profile(monkeypatch):
+    gateway_restart._GATEWAY_RESTART_LOCK = threading.Lock()
+    called = {}
+
+    def fake_popen(args, stdout=None, stderr=None, text=True, env=None):
+        called["args"] = args
+        called["env"] = env
+        return MockPopen(args, stdout_text="ok", returncode=0, env=env)
+
+    monkeypatch.setattr(
+        gateway_restart,
+        "get_hermes_home_for_profile",
+        lambda profile: "/mock/hermes/default" if profile == "default" else "/mock/hermes/profiles/work",
+    )
+    monkeypatch.setattr(gateway_restart.shutil, "which", lambda cmd: "/mock/bin/hermes")
+    monkeypatch.setattr(gateway_restart.subprocess, "Popen", fake_popen)
+
+    result = gateway_restart.restart_active_profile_gateway(profile="default")
+
+    assert result["status"] == "completed"
+    assert called["args"] == ["/mock/bin/hermes", "--profile", "default", "gateway", "restart"]
+    assert called["env"]["HERMES_HOME"] == "/mock/hermes/default"
+
+
+def test_restart_active_profile_gateway_omits_profile_for_isolated_default_home(monkeypatch):
+    gateway_restart._GATEWAY_RESTART_LOCK = threading.Lock()
+    called = {}
+
+    def fake_popen(args, stdout=None, stderr=None, text=True, env=None):
+        called["args"] = args
+        called["env"] = env
+        return MockPopen(args, stdout_text="ok", returncode=0, env=env)
+
+    monkeypatch.setattr(
+        gateway_restart,
+        "get_hermes_home_for_profile",
+        lambda profile: "/mock/hermes/profiles/default",
+    )
+    monkeypatch.setattr(gateway_restart.shutil, "which", lambda cmd: "/mock/bin/hermes")
+    monkeypatch.setattr(gateway_restart.subprocess, "Popen", fake_popen)
+
+    result = gateway_restart.restart_active_profile_gateway(profile="default")
+
+    assert result["status"] == "completed"
+    assert called["args"] == ["/mock/bin/hermes", "gateway", "restart"]
+    assert called["env"]["HERMES_HOME"] == "/mock/hermes/profiles/default"
+
+
+def test_restart_active_profile_gateway_rejects_malformed_explicit_profile(monkeypatch):
+    gateway_restart._GATEWAY_RESTART_LOCK = threading.Lock()
+
+    def fail_popen(*args, **kwargs):
+        raise AssertionError("malformed explicit profile must not launch subprocess")
+
+    monkeypatch.setattr(gateway_restart.subprocess, "Popen", fail_popen)
+
+    for profile in ("", " default", "default ", "default\n", "../bad", "bad;echo"):
+        result = gateway_restart.restart_active_profile_gateway(profile=profile)
+
+        assert result["status"] == "failed"
+        assert "Invalid profile for gateway restart" in result["message"]
+    assert gateway_restart._GATEWAY_RESTART_LOCK.locked() is False
+
+
+def test_restart_active_profile_gateway_accepts_renamed_root_alias(monkeypatch):
+    gateway_restart._GATEWAY_RESTART_LOCK = threading.Lock()
+    called = {}
+
+    def fake_popen(args, stdout=None, stderr=None, text=True, env=None):
+        called["args"] = args
+        called["env"] = env
+        return MockPopen(args, stdout_text="ok", returncode=0, env=env)
+
+    monkeypatch.setattr(
+        gateway_restart,
+        "get_hermes_home_for_profile",
+        lambda profile: "/mock/hermes/root" if profile == "rootalias" else "/mock/hermes/other",
+    )
+    monkeypatch.setattr(
+        gateway_restart,
+        "_is_root_profile",
+        lambda profile: profile in {"default", "rootalias"},
+    )
+    monkeypatch.setattr(gateway_restart.shutil, "which", lambda cmd: "/mock/bin/hermes")
+    monkeypatch.setattr(gateway_restart.subprocess, "Popen", fake_popen)
+
+    result = gateway_restart.restart_active_profile_gateway(profile="rootalias")
+
+    assert result["status"] == "completed"
+    assert called["args"] == ["/mock/bin/hermes", "--profile", "default", "gateway", "restart"]
+    assert called["env"]["HERMES_HOME"] == "/mock/hermes/root"
+
+
 def test_restart_active_profile_gateway_failure_preserves_empty_output_contract(monkeypatch):
     gateway_restart._GATEWAY_RESTART_LOCK = threading.Lock()
 

tests/test_issue716_agent_heartbeat.py

@@ -131,6 +131,30 @@ def test_agent_health_payload_alive_uses_safe_runtime_details(monkeypatch):
     assert "pid" not in payload["details"]
 
 
+def test_active_profile_gateway_running_pid_uses_active_profile_path(monkeypatch, tmp_path):
+    from api import agent_health, profiles
+
+    active_home = tmp_path / "profiles" / "active"
+    gateway_status = _PathSensitiveGatewayStatus(active_home)
+    monkeypatch.setattr(agent_health, "_gateway_status_module", lambda: gateway_status)
+    monkeypatch.setattr(profiles, "get_active_hermes_home", lambda: active_home)
+
+    assert agent_health.get_active_profile_gateway_running_pid() == 98765
+    assert gateway_status.running_pid_path == active_home / "gateway.pid"
+
+
+def test_active_profile_gateway_running_pid_fails_closed_when_status_is_unavailable(monkeypatch):
+    from api import agent_health
+
+    monkeypatch.setattr(
+        agent_health,
+        "_gateway_status_module",
+        lambda: (_ for _ in ()).throw(ModuleNotFoundError("gateway.status")),
+    )
+
+    assert agent_health.get_active_profile_gateway_running_pid() is None
+
+
 def test_agent_health_payload_down_when_gateway_metadata_exists_but_no_process(monkeypatch):
     from api import agent_health
 

tests/test_update_banner_fixes.py

@@ -22,6 +22,7 @@ import io
 import json
 import subprocess
 import types
+import functools
 
 import pytest
 
@@ -729,7 +730,7 @@ class TestSuccessfulUpdateReturnsRestartScheduled:
         monkeypatch.setattr(upd, '_schedule_restart', lambda delay=2.0: None)
         monkeypatch.setattr(
             'api.updates.restart_active_profile_gateway',
-            lambda: {'status': 'completed', 'message': 'Gateway service restarted successfully'},
+            lambda **kwargs: {'status': 'completed', 'message': 'Gateway service restarted successfully'},
         )
 
         result = upd.apply_update('agent')
@@ -829,6 +830,549 @@ class TestApplyForceUpdate:
 class TestAgentUpdateRequiresGatewayRestart:
     """Agent updates must prove gateway restart before returning ok=True."""
 
+    def test_agent_gateway_restart_retries_one_transient_failure(self, monkeypatch):
+        import api.updates as upd
+
+        restart_results = iter([
+            {'status': 'failed', 'message': 'Restart failed: bad file descriptor'},
+            {'status': 'completed', 'message': 'Gateway service restarted successfully'},
+        ])
+        restart_calls = []
+        sleeps = []
+
+        def fake_restart(*, profile=None):
+            restart_calls.append(profile)
+            return next(restart_results)
+
+        monkeypatch.setattr(upd, 'restart_active_profile_gateway', fake_restart)
+        monkeypatch.setattr(upd.time, 'sleep', sleeps.append)
+        monkeypatch.setattr(upd, 'get_active_profile_gateway_running_pid', lambda *, profile=None: 101)
+
+        ok, result = upd._ensure_gateway_restart_for_agent_update()
+
+        assert ok is True
+        assert result['status'] == 'completed'
+        assert result['retry_attempted'] is True
+        assert 'bad file descriptor' in result['initial_failure']
+        assert restart_calls == ['default', 'default']
+        assert sleeps == [upd._AGENT_GATEWAY_RESTART_RETRY_DELAY_S]
+
+    def test_agent_gateway_restart_retry_busy_stays_fail_closed(self, monkeypatch):
+        import api.updates as upd
+
+        restart_results = iter([
+            {'status': 'failed', 'message': 'Restart failed: first'},
+            {'status': 'busy', 'message': 'Restart already in progress'},
+        ])
+        sleeps = []
+        gateway_pid_calls = []
+
+        monkeypatch.setattr(upd, 'restart_active_profile_gateway', lambda **kwargs: next(restart_results))
+        monkeypatch.setattr(upd.time, 'sleep', sleeps.append)
+        monkeypatch.setattr(
+            upd,
+            'get_active_profile_gateway_running_pid',
+            lambda *, profile=None: gateway_pid_calls.append(profile) or 101,
+        )
+
+        ok, result = upd._ensure_gateway_restart_for_agent_update()
+
+        assert ok is False
+        assert result['status'] == 'busy'
+        assert result['retry_attempted'] is True
+        assert 'first' in result['initial_failure']
+        assert sleeps == [upd._AGENT_GATEWAY_RESTART_RETRY_DELAY_S]
+        assert gateway_pid_calls == ['default']
+
+    def test_agent_gateway_restart_accepts_verified_process_replacement_after_retry_failure(self, monkeypatch):
+        import api.updates as upd
+
+        timeline = []
+        restart_results = iter([
+            {'status': 'failed', 'message': 'Restart failed: first'},
+            {'status': 'failed', 'message': 'Restart failed: retry'},
+        ])
+        sleeps = []
+        gateway_pids = iter([101, 202])
+
+        def fake_restart(*, profile=None):
+            timeline.append('restart')
+            return next(restart_results)
+
+        def fake_gateway_pid(*, profile=None):
+            pid = next(gateway_pids)
+            timeline.append(f'pid:{pid}')
+            return pid
+
+        monkeypatch.setattr(upd, 'restart_active_profile_gateway', fake_restart)
+        monkeypatch.setattr(upd.time, 'sleep', sleeps.append)
+        monkeypatch.setattr(upd, 'get_active_profile_gateway_running_pid', fake_gateway_pid)
+
+        ok, result = upd._ensure_gateway_restart_for_agent_update()
+
+        assert ok is True
+        assert result['status'] == 'completed'
+        assert result['retry_attempted'] is True
+        assert result['process_replaced'] is True
+        assert 'first' in result['initial_failure']
+        assert 'retry' in result['retry_failure']
+        assert timeline == ['pid:101', 'restart', 'restart', 'pid:202']
+        assert sleeps == [
+            upd._AGENT_GATEWAY_RESTART_RETRY_DELAY_S,
+            upd._AGENT_GATEWAY_RESTART_RETRY_DELAY_S,
+        ]
+
+    def test_agent_gateway_restart_fails_closed_after_retry_and_health_check(self, monkeypatch):
+        import api.updates as upd
+
+        restart_results = iter([
+            {'status': 'failed', 'message': 'Restart failed: first'},
+            {'status': 'failed', 'message': 'Restart failed: retry'},
+        ])
+        restart_calls = []
+        sleeps = []
+
+        def fake_restart(*, profile=None):
+            restart_calls.append(profile)
+            return next(restart_results)
+
+        monkeypatch.setattr(upd, 'restart_active_profile_gateway', fake_restart)
+        monkeypatch.setattr(upd.time, 'sleep', sleeps.append)
+        monkeypatch.setattr(upd, 'get_active_profile_gateway_running_pid', lambda *, profile=None: 101)
+
+        ok, result = upd._ensure_gateway_restart_for_agent_update()
+
+        assert ok is False
+        assert result['status'] == 'failed'
+        assert result['retry_attempted'] is True
+        assert 'Restart failed: first' in result['message']
+        assert 'Restart failed: retry' in result['message']
+        assert restart_calls == ['default', 'default']
+        assert sleeps == [
+            upd._AGENT_GATEWAY_RESTART_RETRY_DELAY_S,
+            upd._AGENT_GATEWAY_RESTART_RETRY_DELAY_S,
+        ]
+
+    def test_agent_gateway_restart_default_retry_cannot_use_sticky_named_profile(self, monkeypatch):
+        import api.updates as upd
+
+        default_restart_results = iter([
+            {'status': 'failed', 'message': 'Restart failed: default first'},
+            {'status': 'failed', 'message': 'Restart failed: default retry'},
+        ])
+        restart_profiles = []
+
+        def fake_restart(*, profile=None):
+            effective_profile = profile or 'sticky-work'
+            restart_profiles.append(effective_profile)
+            if effective_profile == 'sticky-work':
+                return {'status': 'completed', 'message': 'wrong profile restarted'}
+            return next(default_restart_results)
+
+        monkeypatch.setattr(upd, 'get_active_profile_name', lambda: 'default')
+        monkeypatch.setattr(upd, 'restart_active_profile_gateway', fake_restart)
+        monkeypatch.setattr(upd.time, 'sleep', lambda seconds: None)
+        monkeypatch.setattr(upd, 'get_active_profile_gateway_running_pid', lambda *, profile=None: 101)
+
+        ok, result = upd._ensure_gateway_restart_for_agent_update()
+
+        assert ok is False
+        assert restart_profiles == ['default', 'default']
+        assert result['status'] == 'failed'
+        assert 'default first' in result['message']
+        assert 'default retry' in result['message']
+
+    def test_agent_gateway_restart_real_profile_seam_unchanged_default_pid_fails(
+        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": []}
+
+        class FailedRestartProcess:
+            returncode = 7
+
+            def communicate(self, timeout=None):
+                return "", "restart failed"
+
+        class PathStrictGatewayStatus:
+            def get_running_pid(self, pid_path=None, cleanup_stale=False):
+                path = pathlib.Path(pid_path) if pid_path is not None else None
+                calls["pid_paths"].append(path)
+                if path == root_home / "gateway.pid":
+                    return 101
+                if path == sticky_home / "gateway.pid":
+                    return 202
+                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", lambda: PathStrictGatewayStatus())
+        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["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_legacy_implicit_sticky_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": [], "implicit_pid": 0}
+
+        class FailedRestartProcess:
+            returncode = 7
+
+            def communicate(self, timeout=None):
+                return "", "restart failed"
+
+        class LegacyImplicitStickyGatewayStatus:
+            def __init__(self):
+                self._pids = iter([201, 202])
+
+            def get_running_pid(self, cleanup_stale=False):
+                calls["implicit_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", LegacyImplicitStickyGatewayStatus)
+        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["implicit_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_kwargs_wrapper_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"
+
+        class AmbientKwargsGatewayStatus:
+            def __init__(self):
+                self._pids = iter([201, 202])
+
+            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", AmbientKwargsGatewayStatus)
+        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_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,
+        tmp_path,
+    ):
+        from api import agent_health, gateway_restart, profiles
+        import api.updates as upd
+
+        base_home = tmp_path / ".hermes"
+        isolated_home = base_home / "profiles" / "default"
+        isolated_home.mkdir(parents=True)
+        calls = {"popen": [], "pid_paths": []}
+
+        class FailedRestartProcess:
+            returncode = 7
+
+            def communicate(self, timeout=None):
+                return "", "restart failed"
+
+        class PathStrictGatewayStatus:
+            def get_running_pid(self, pid_path=None, cleanup_stale=False):
+                path = pathlib.Path(pid_path) if pid_path is not None else None
+                calls["pid_paths"].append(path)
+                if path == isolated_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, "_INITIAL_HERMES_HOME", str(isolated_home))
+        monkeypatch.setattr(profiles, "_INITIAL_ISOLATED_PROFILE_OPT_IN", "1")
+        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", lambda: PathStrictGatewayStatus())
+        monkeypatch.setattr(upd.time, 'sleep', lambda seconds: None)
+
+        ok, result = upd._ensure_gateway_restart_for_agent_update()
+
+        assert ok is False
+        assert result["status"] == "failed"
+        assert calls["pid_paths"] == [isolated_home / "gateway.pid", isolated_home / "gateway.pid"]
+        assert [call[0] for call in calls["popen"]] == [
+            ["/mock/bin/hermes", "gateway", "restart"],
+            ["/mock/bin/hermes", "gateway", "restart"],
+        ]
+        assert [call[1]["HERMES_HOME"] for call in calls["popen"]] == [
+            str(isolated_home),
+            str(isolated_home),
+        ]
+
     def test_apply_update_agent_requires_gateway_restart(self, tmp_path, monkeypatch):
         import api.updates as upd
 
@@ -850,8 +1394,8 @@ class TestAgentUpdateRequiresGatewayRestart:
                 return 'Already up to date.', True
             return '', True
 
-        def fake_gateway_restart():
-            gateway_restarts.append('called')
+        def fake_gateway_restart(*, profile=None):
+            gateway_restarts.append(profile)
             return {'status': 'completed', 'message': 'Gateway service restarted successfully'}
 
         monkeypatch.setattr(upd, '_run_git', fake_run)
@@ -865,7 +1409,7 @@ class TestAgentUpdateRequiresGatewayRestart:
         assert result['target'] == 'agent'
         assert result['restart_scheduled'] is True
         assert result['gateway_restart'] == 'completed'
-        assert gateway_restarts == ['called']
+        assert gateway_restarts == ['default']
 
     def test_apply_update_agent_stash_conflict_success_invokes_gateway_restart(self, tmp_path, monkeypatch):
         import api.updates as upd
@@ -900,8 +1444,8 @@ class TestAgentUpdateRequiresGatewayRestart:
                 return 'Updating', True
             return '', True
 
-        def fake_gateway_restart():
-            gateway_restarts.append('called')
+        def fake_gateway_restart(*, profile=None):
+            gateway_restarts.append(profile)
             return {'status': 'in_progress', 'message': 'Gateway service restart initiated (in progress)'}
 
         monkeypatch.setattr(upd, '_run_git', fake_run)
@@ -916,7 +1460,7 @@ class TestAgentUpdateRequiresGatewayRestart:
         assert result['target'] == 'agent'
         assert result['restart_scheduled'] is True
         assert result['gateway_restart'] == 'in_progress'
-        assert gateway_restarts == ['called']
+        assert gateway_restarts == ['default']
 
     def test_apply_update_agent_without_gateway_restart_result_fails(self, tmp_path, monkeypatch):
         import api.updates as upd
@@ -943,8 +1487,8 @@ class TestAgentUpdateRequiresGatewayRestart:
         monkeypatch.setattr(upd, 'REPO_ROOT', tmp_path)
         monkeypatch.setattr(upd, '_AGENT_DIR', tmp_path)
         monkeypatch.setattr(upd, '_schedule_restart', lambda delay=2.0: (_ for _ in ()).throw(AssertionError('must not restart')))
-        monkeypatch.setattr('api.updates.restart_active_profile_gateway', lambda: (
-            restart_calls.append('called'),
+        monkeypatch.setattr('api.updates.restart_active_profile_gateway', lambda **kwargs: (
+            restart_calls.append(kwargs.get('profile')),
             {'status': 'busy', 'message': 'Restart already in progress. Please wait a moment and try again.'},
         )[1])
 
@@ -954,7 +1498,7 @@ class TestAgentUpdateRequiresGatewayRestart:
         assert result['target'] == 'agent'
         assert result['gateway_restart'] == 'busy'
         assert 'hermes gateway restart' in result['message']
-        assert restart_calls == ['called']
+        assert restart_calls == ['default']
 
     def test_apply_force_update_agent_uses_gateway_restart_status(self, tmp_path, monkeypatch):
         import api.updates as upd
@@ -978,7 +1522,7 @@ class TestAgentUpdateRequiresGatewayRestart:
         monkeypatch.setattr(upd, 'REPO_ROOT', tmp_path)
         monkeypatch.setattr(upd, '_AGENT_DIR', tmp_path)
         monkeypatch.setattr(upd, '_schedule_restart', lambda delay=2.0: None)
-        monkeypatch.setattr('api.updates.restart_active_profile_gateway', lambda: {'status': 'completed', 'message': 'Gateway service restarted successfully'})
+        monkeypatch.setattr('api.updates.restart_active_profile_gateway', lambda **kwargs: {'status': 'completed', 'message': 'Gateway service restarted successfully'})
 
         result = upd.apply_force_update('agent')
         assert result['ok'] is True
@@ -1008,7 +1552,7 @@ class TestAgentUpdateRequiresGatewayRestart:
         monkeypatch.setattr(upd, '_schedule_restart', lambda delay=2.0: (_ for _ in ()).throw(AssertionError('must not restart')))
         monkeypatch.setattr(
             'api.updates.restart_active_profile_gateway',
-            lambda: {'status': 'busy', 'message': 'Restart already in progress. Please wait a moment and try again.'},
+            lambda **kwargs: {'status': 'busy', 'message': 'Restart already in progress. Please wait a moment and try again.'},
         )
 
         result = upd.apply_force_update('agent')