refactor(gateway): keep the stale-owner replacement within code-health caps

ensure_gateway_runtime's replacement branch moves into _observe_replacing_stale (CC back
under target; the stopped owner reads as draining/"replaced", the same retiring path an idle
exit takes). gateway status notes (the incompatible-owner report, standalone-by-config) move
to hermes_cli/gateway_status_notes.py so gateway.py shrinks instead of growing; blind excepts
narrowed; the TUI verdict-echo test moves to the notice test file.

(cherry picked from commit 547d7da12556d2d0ed26b3644bc4a7e62c446574)
(cherry picked from commit 0ddf64ebd2fb856afc9cc029aa4f22e7b08eddc0)

hermes_cli/gateway.py

@@ -5601,27 +5601,9 @@ def _status_host_kind() -> str:
     return "windows" if is_windows() else "other"
 
 
-def _incompatible_owner_verdict() -> str | None:
-    """``incompatible: <reason>`` when the gateway answering for this home is one this Hermes's clients
-    (TUI, Desktop, chat) refuse, e.g. an older Hermes's daemon left running across an upgrade."""
-    try:
-        from hermes_cli.gateway_runtime import discover_gateway_endpoint
-        observed = discover_gateway_endpoint(get_hermes_home(), timeout=1.5)
-    except Exception:
-        return None
-    if observed.state != "incompatible":
-        return None
-    return f"incompatible: {observed.reason_code or 'unknown'}"
-
-
-def _print_incompatible_owner_hint() -> None:
-    print("  It was started by another Hermes version; TUI, Desktop and chat cannot attach to it.")
-    print("  Replace it with this version:")
-    print("    hermes gateway restart")
-
-
 def _cmd_status(args):
     from hermes_cli.gateway_profile_lifecycle import print_parked_status
+    from hermes_cli.gateway_status_notes import print_standalone_by_config, report_incompatible_owner
     deep = getattr(args, "deep", False)
     full = getattr(args, "full", False)
     system = getattr(args, "system", False)
@@ -5653,17 +5635,13 @@ def _cmd_status(args):
         else:
             _gw_windows().status(deep=deep)
         _print_gateway_process_mismatch(snapshot)
-        if stale := _incompatible_owner_verdict():
-            print(f"✗ The running gateway cannot serve this Hermes ({stale})")
-            _print_incompatible_owner_hint()
+        report_incompatible_owner(())
         _print_multiplex_standalone_reason()
         _print_served_ingress_urls()
     else:
         pids = list(snapshot.gateway_pids)
-        if pids and (stale := _incompatible_owner_verdict()):
-            # An older Hermes's gateway: alive, but no client of this version can attach to it.
-            print(f"✗ Gateway is running (PID: {', '.join(map(str, pids))}) but cannot serve this Hermes ({stale})")
-            _print_incompatible_owner_hint()
+        if pids and report_incompatible_owner(pids):
+            pass  # an older Hermes's gateway: alive, but every client of this version refuses it
         elif pids:
             print(f"✓ Gateway is running (PID: {', '.join(map(str, pids))})")
             print("  (Running manually, not as a system service)")
@@ -5689,20 +5667,7 @@ def _cmd_status(args):
 
     _print_duplicate_credential_warnings()
     _print_other_profiles_gateway_status()
-    _print_standalone_by_config()
-
-
-def _print_standalone_by_config() -> None:
-    """Default-profile status: name the profiles that opted out of the host multiplexer by config,
-    so the served set the host gateway reports is not mistaken for the installed roster."""
-    from hermes_cli.profiles import get_active_profile_name, profiles_to_serve
-    if (get_active_profile_name() or "default") != "default":
-        return
-    roster = {name for name, _home in profiles_to_serve(True, include_standalone=True)}
-    served = {name for name, _home in profiles_to_serve(True)}
-    names = sorted(roster - served - {"default"})
-    if names:
-        print(f"standalone by config (temporary compatibility shim): {', '.join(names)}")
+    print_standalone_by_config()
 
 
 def _cmd_list(args):

hermes_cli/gateway_runtime.py

@@ -222,6 +222,27 @@ def _update_fenced(homes) -> bool:
 
 
 _REPLACE_FAILED_DETAIL = "an older Hermes gateway still owns this profile after a stop; run: hermes gateway stop"
+# ``draining`` reasons after which the owner is going away by itself: its own idle exit, or the
+# older-Hermes owner this call just stopped. Ensure waits it out and starts the current owner.
+_RETIRING = frozenset({"idle_exit", "replaced"})
+
+
+def _observe_replacing_stale(home: Path, deadline: float, replaced: set[Path]) -> GatewayDiscovery:
+    """Discovery, except that a gateway an older Hermes left on this home is stopped once (like
+    ``hermes gateway stop``) and reported as retiring; a refusal keeps ``incompatible`` + its remedy."""
+    from hermes_cli.gateway_runtime_service import remaining
+    observed = discover_gateway_endpoint(home, timeout=remaining(deadline))
+    if observed.state != "incompatible" or observed.reason_code != "runtime_protocol":
+        return observed
+    if home in replaced:
+        return GatewayDiscovery("incompatible", reason_code="runtime_protocol", detail=_REPLACE_FAILED_DETAIL)
+    from hermes_cli import gateway_runtime_replace as replace
+    replaced.add(home)
+    refusal = replace.replace_incompatible_owner(home, deadline=deadline)
+    if refusal is not None:
+        return GatewayDiscovery("incompatible", reason_code="runtime_protocol",
+                                detail=replace.REFUSAL_DETAIL.get(refusal, _REPLACE_FAILED_DETAIL))
+    return GatewayDiscovery("draining", reason_code="replaced")
 
 
 def ensure_gateway_runtime(profile_home: str | Path, *, timeout: float = DEFAULT_ENSURE_TIMEOUT,
@@ -258,7 +279,7 @@ def ensure_gateway_runtime(profile_home: str | Path, *, timeout: float = DEFAULT
     # still held for a moment after its control socket closes, which reads as "starting"), then
     # start the replacement this call owes the client.
     retiring = False
-    replaced = False
+    replaced: set[Path] = set()
     delay = 0.025
     try:
         while True:
@@ -268,27 +289,13 @@ def ensure_gateway_runtime(profile_home: str | Path, *, timeout: float = DEFAULT
             # Poll BEFORE discovering: an exit observed first, then "absent", cannot be a child that
             # published its owner a moment later. A winning racer shows up as starting/ready instead.
             status = child.poll() if child is not None else None
-            observed = discover_gateway_endpoint(home, timeout=remaining(deadline))
+            observed = _observe_replacing_stale(home, deadline, replaced)
             if observed.reason_code == "control_timeout":
                 return GatewayDiscovery("starting", reason_code="deadline")
-            if observed.state == "draining" and observed.reason_code == "idle_exit":
+            if observed.state == "draining" and observed.reason_code in _RETIRING:
                 retiring, requested, child = True, False, None
                 time.sleep(min(0.1, remaining(deadline)))
                 continue
-            if observed.state == "incompatible" and observed.reason_code == "runtime_protocol":
-                if replaced:
-                    return GatewayDiscovery("incompatible", reason_code="runtime_protocol",
-                                            detail=_REPLACE_FAILED_DETAIL)
-                from hermes_cli import gateway_runtime_replace as replace
-                replaced = True
-                refusal = replace.replace_incompatible_owner(home, deadline=deadline)
-                if refusal is not None:
-                    return GatewayDiscovery("incompatible", reason_code="runtime_protocol",
-                                            detail=replace.REFUSAL_DETAIL.get(refusal, _REPLACE_FAILED_DETAIL))
-                # Like an idle exit: its lock outlives its socket for a moment ("starting"); wait it
-                # out, then start the current owner this call owes the client.
-                retiring, requested, child = True, False, None
-                continue
             if observed.state not in {"absent", "starting"}:
                 return observed
             if status is not None and observed.state == "absent":

hermes_cli/gateway_runtime_replace.py

@@ -116,7 +116,7 @@ def replace_incompatible_owner(home: Path, *, deadline: float) -> str | None:
         return "owner_unreachable"
     try:
         found = _identify(home, timeout=min(2.0, budget))
-    except Exception:
+    except (OSError, ValueError, TypeError):  # timeout, refused/denied peer, malformed reply
         return "owner_unreachable"
     if found is None:
         return None  # it is already gone: nothing to replace

hermes_cli/gateway_status_notes.py

@@ -0,0 +1,44 @@
+"""Notes `hermes gateway status` prints beside the running/not-running verdict."""
+from __future__ import annotations
+
+from hermes_cli.config import get_hermes_home
+
+
+def _incompatible_owner_verdict() -> str | None:
+    """``incompatible: <reason>`` when the gateway answering for this home is one this Hermes's clients
+    (TUI, Desktop, chat) refuse, e.g. an older Hermes's daemon left running across an upgrade."""
+    from hermes_cli.gateway_runtime import discover_gateway_endpoint
+    try:
+        observed = discover_gateway_endpoint(get_hermes_home(), timeout=1.5)
+    except (OSError, ValueError):  # discovery reports peers as verdicts; only a bad home lands here
+        return None
+    if observed.state != "incompatible":
+        return None
+    return f"incompatible: {observed.reason_code or 'unknown'}"
+
+
+def report_incompatible_owner(pids) -> bool:
+    """Print ✗ (with the fix) when the running gateway cannot serve this Hermes; True if printed.
+    ``pids`` names the scanned processes for the manual-run line; empty for a service-managed one."""
+    stale = _incompatible_owner_verdict()
+    if not stale:
+        return False
+    running = f"Gateway is running (PID: {', '.join(map(str, pids))})" if pids else "The running gateway"
+    print(f"✗ {running} but cannot serve this Hermes ({stale})")
+    print("  It was started by another Hermes version; TUI, Desktop and chat cannot attach to it.")
+    print("  Replace it with this version:")
+    print("    hermes gateway restart")
+    return True
+
+
+def print_standalone_by_config() -> None:
+    """Default-profile status: name the profiles that opted out of the host multiplexer by config,
+    so the served set the host gateway reports is not mistaken for the installed roster."""
+    from hermes_cli.profiles import get_active_profile_name, profiles_to_serve
+    if (get_active_profile_name() or "default") != "default":
+        return
+    roster = {name for name, _home in profiles_to_serve(True, include_standalone=True)}
+    served = {name for name, _home in profiles_to_serve(True)}
+    names = sorted(roster - served - {"default"})
+    if names:
+        print(f"standalone by config (temporary compatibility shim): {', '.join(names)}")

tests/hermes_cli/test_gateway.py

@@ -260,9 +260,10 @@ def _run_status_with_snapshot(monkeypatch, snapshot) -> str:
     monkeypatch.setattr(gateway, "_installed_service_kind_for", lambda probe: None)
     monkeypatch.setattr(gateway, "named_profile_served_by_running_multiplexer", lambda: False)
     for name in ("_print_runtime_health", "_print_multiplex_standalone_reason", "_print_served_ingress_urls",
-                 "_print_duplicate_credential_warnings", "_print_other_profiles_gateway_status",
-                 "_print_standalone_by_config"):
+                 "_print_duplicate_credential_warnings", "_print_other_profiles_gateway_status"):
         monkeypatch.setattr(gateway, name, lambda *a, **k: None)
+    monkeypatch.setattr("hermes_cli.gateway_status_notes.print_standalone_by_config", lambda: None)
+    monkeypatch.setattr("hermes_cli.gateway_status_notes._incompatible_owner_verdict", lambda: None)
     buf = io.StringIO()
     with redirect_stdout(buf):
         gateway._cmd_status(SimpleNamespace(deep=False, full=False, system=False))

ui-tui/src/__tests__/createGatewayEventHandler.notice.test.ts

@@ -5,7 +5,7 @@ import { createGatewayEventHandler } from '../app/createGatewayEventHandler.js'
 import { resetOverlayState } from '../app/overlayStore.js'
 import { resetServerRequestsForTests } from '../app/serverRequestStore.js'
 import { turnController } from '../app/turnController.js'
-import { resetTurnState } from '../app/turnStore.js'
+import { getTurnState, resetTurnState } from '../app/turnStore.js'
 import { getUiState, patchUiState, resetUiState } from '../app/uiStore.js'
 import type { Msg } from '../types.js'
 
@@ -388,4 +388,26 @@ describe('createGatewayEventHandler', () => {
       expect(openExternalUrlMock).not.toHaveBeenCalled()
     })
   })
+
+  it('echoes the canonical bootstrap verdict so a refused gateway is never a silent retry loop', () => {
+    const appended: Msg[] = []
+    const onEvent = createGatewayEventHandler(buildCtx(appended))
+
+    onEvent({
+      payload: {
+        cwd: '',
+        python: 'gateway ensure',
+        stderr_tail:
+          'Command failed: python gateway_bootstrap.py --start\ngateway incompatible: runtime_protocol\n' +
+          "gateway incompatible: another Hermes install's gateway owns this profile; stop it there: hermes gateway stop"
+      },
+      type: 'gateway.start_timeout'
+    } as any)
+
+    const messages = getTurnState().activity.map(a => a.text)
+
+    expect(messages).toContain('gateway incompatible: runtime_protocol')
+    expect(messages.some(m => m.includes('another Hermes install'))).toBe(true)
+    expect(messages.some(m => m.startsWith('Command failed'))).toBe(false)
+  })
 })

ui-tui/src/__tests__/createGatewayEventHandler.test.ts

@@ -785,28 +785,6 @@ describe('createGatewayEventHandler', () => {
     expect(messages.some(m => m.includes('[startup] timed out'))).toBe(false)
   })
 
-  it('echoes the canonical bootstrap verdict so a refused gateway is never a silent retry loop', () => {
-    const appended: Msg[] = []
-    const onEvent = createGatewayEventHandler(buildCtx(appended))
-
-    onEvent({
-      payload: {
-        cwd: '',
-        python: 'gateway ensure',
-        stderr_tail:
-          'Command failed: python gateway_bootstrap.py --start\ngateway incompatible: runtime_protocol\n' +
-          "gateway incompatible: another Hermes install's gateway owns this profile; stop it there: hermes gateway stop"
-      },
-      type: 'gateway.start_timeout'
-    } as any)
-
-    const messages = getTurnState().activity.map(a => a.text)
-
-    expect(messages).toContain('gateway incompatible: runtime_protocol')
-    expect(messages.some(m => m.includes('another Hermes install'))).toBe(true)
-    expect(messages.some(m => m.startsWith('Command failed'))).toBe(false)
-  })
-
   it('prefers raw text over Rich-rendered ANSI on message.complete (#16391)', () => {
     const appended: Msg[] = []
     const onEvent = createGatewayEventHandler(buildCtx(appended))