Merge pull request #6057 from nesquena/stage/6054

Release exp: fail-closed gateway-restart recovery for agent updates (#6054)

CHANGELOG.md

@@ -81,6 +81,8 @@
 
 ### Fixed
 
+- **Agent updates no longer report a false failure when the gateway restart is transient — and never mask a real failure.** During an agent update, the launchd/gateway restart subprocess can briefly exit non-zero while the old process is being replaced; that was surfaced as a whole-update failure. The updater now confirms the outcome by the actual gateway PID for the exact profile being updated (retrying the restart once), so a genuine process handoff is reported as success — while a real restart failure, an unknown/ambient PID, a wrong-profile confirmation, or a compatibility-wrapped status helper that can't confirm the specific PID all fail closed (never masked as success). Restart targeting and the PID check are pinned to the update's profile (root aliases normalized), and malformed profile names are rejected before launch. Thanks @franksong2702. (#6054, #6045)
+
 - **Gateway provider errors are reported accurately, and your prompt survives a failed turn.** On gateway-backed sessions, a terminal provider error (bad model id, exhausted credentials, etc.) mid-turn is now classified and shown as the real error instead of a generic empty turn, the error card is bound to the correct session, and the in-flight user prompt plus any partial output is persisted so it survives a reload — even when you'd just re-sent an identical prompt (e.g. "continue"). Thanks @rodboev. (#5969, #5940)
 
 - **"Fork from here" works during an active response.** The fork button was silently dead while the agent was generating — clicking it did nothing. You can now fork from any past (already-committed) message while a response streams; only the currently-streaming message and the just-sent (not-yet-committed) prompt are blocked, with a clear toast. Thanks @kertisalex. (#5994, #5993)

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')