fix(gateway): let clients reach a restart-wait owner to answer a parked prompt

A graceful restart (Desktop's code-SHA restart, `hermes update --gateway`) waits for in-flight
turns, but a turn parked on approval/clarify can only finish when a human answers, and the
draining owner refused every client: identify dropped its capabilities, `gateway ensure`
returned a bare `draining` (exit 6), the WS boundary closed interactive tickets 1013 and every
submit raised `runtime_draining`. The wait therefore ran the full 1800 s cap.

The draining owner still owns the turn (#106742: one owner per session), so it stays reachable
during the restart's after-turn wait:

- request_restart publishes `drain_reason: restart_wait` and identify keeps the endpoint while it
  is set; drain_gateway_runtime (stop()) withdraws it, so the window closes before teardown.
- `gateway ensure` reports `draining`/`restart_wait` (still exit 6, never a fresh-start target)
  WITH the endpoint; Desktop's ensureLocalGateway and the Python client attach to it.
- The WS boundary admits interactive tickets while the window is open, so attach + respond()
  reach the parked prompt.
- New submits are committed queued (durable FIFO, next owner runs them exactly once); the claim
  path stays refused, so the draining owner never starts them.
- A turn nobody answers still ends interrupted at the cap and is never replayed (unchanged).

Refs #134889 (GW-6)
Refs #139574

apps/desktop/electron/local-gateway.test.ts

@@ -150,6 +150,16 @@ test('ensure consumes structured readiness without acquiring a child owner', asy
   await expect(ensureLocalGateway(async () => ({ code: 5, stdout: JSON.stringify({ state: 'starting', reason_code: 'deadline' }) }))).rejects.toThrow('starting')
 })
 
+test('a restart-wait owner is attached to answer a parked prompt; a plain drain stays refused (#139041)', async () => {
+  const endpoint = { profile_id: '/private/profile', instance_id: 'owner', authority_epoch: 1, runtime_protocol: 1, api_origin: 'http://127.0.0.1:1234', capabilities: ['session-authority-v1'], supervisor: 'none', code_sha: 'old' }
+  const restartStale = vi.fn(async () => true)
+  const wait = { code: 6, stdout: JSON.stringify({ state: 'draining', reason_code: 'restart_wait', endpoint, client_code_sha: 'new' }) }
+  const connection = await ensureLocalGateway(async () => wait, undefined, restartStale)
+  expect(connection.gatewayEndpoint).toEqual(endpoint)
+  expect(restartStale).not.toHaveBeenCalled() // already restarting: never a second restart
+  await expect(ensureLocalGateway(async () => ({ code: 6, stdout: JSON.stringify({ state: 'draining', reason_code: 'update_paused', endpoint: null }) }))).rejects.toThrow('Gateway draining (update_paused)')
+})
+
 test('an ensure child that never reached the protocol boundary is diagnosed from stderr, not as a JSON parse error', async () => {
   // `main`'s hermes has no `gateway ensure` subcommand: argparse usage on stderr, nothing on stdout.
   const legacy = { code: 2, stdout: '', stderr: "usage: hermes gateway [-h] ...\nhermes gateway: 'ensure' is not a `hermes gateway` command.\nRun `hermes gateway --help` to see all commands.\n" }

apps/desktop/electron/local-gateway.ts

@@ -183,7 +183,12 @@ export async function ensureLocalGateway(
     payload = parseGatewayEnsureOutput(result)
   }
 
-  if (result.code !== 0 || payload.state !== 'ready') {
+  // A graceful restart's after-turn wait still owns its turns (#139041): attach to it so a turn parked
+  // on approval/clarify can be answered; new input queues for the next owner, and the next dial after
+  // the owner exits reaches its replacement.
+  const restartWait = payload.state === 'draining' && payload.reason_code === 'restart_wait' && Boolean(payload.endpoint)
+
+  if (!restartWait && (result.code !== 0 || payload.state !== 'ready')) {
     // A refused replacement (another install's or a newer Hermes's gateway) names its remedy in
     // `detail`; a start that died carries its own output there. Either beats a pointer elsewhere.
     const detail = typeof payload.detail === 'string' && payload.detail.trim() ? ` ${payload.detail.trim().split(/\r?\n/)[0]}.` : ''

gateway/drain_control.py

@@ -26,6 +26,18 @@ from utils import atomic_json_write
 
 _log = logging.getLogger(__name__)
 
+# ``drain_reason`` a graceful restart publishes for its after-turn wait (#139041). The owner still
+# runs its in-flight turns, so a client may attach to answer a parked approval/clarify; new input is
+# committed queued for the next owner and never claimed here. ``stop()`` withdraws it.
+RESTART_WAIT_REASON = "restart_wait"
+
+
+def restart_wait_open(runner) -> bool:
+    """True while *runner* is in a restart's after-turn wait (draining, not yet stopping)."""
+    descriptor = getattr(runner, "session_runtime_descriptor", None) or {}
+    return (bool(getattr(runner, "_draining", False)) and bool(getattr(runner, "_running", False))
+            and descriptor.get("drain_reason") == RESTART_WAIT_REASON)
+
 _DRAIN_REQUEST_FILENAME = ".drain_request.json"
 # Drain-gated lifecycle actions complete in minutes; an hour bounds the wedge a leaked
 # marker can cause. Long drains refresh the marker instead of raising this.

gateway/run_api.py

@@ -273,10 +273,12 @@ class GatewayRuntimeAPI:
             return await self.app(scope, receive, send)
         original_receive = receive
         allow_draining = False
+        from gateway.drain_control import restart_wait_open
 
         async def receive_admitted():
             message = await original_receive()
-            if (descriptor['state'] != 'ready' or self.runner._draining) and not allow_draining:
+            if ((descriptor['state'] != 'ready' or self.runner._draining) and not allow_draining
+                    and not restart_wait_open(self.runner)):
                 return {'type': 'websocket.disconnect', 'code': 1013}
             return message
 
@@ -314,7 +316,10 @@ class GatewayRuntimeAPI:
                     return
                 return await self.app(scope, receive, send)
         allow_draining = not operator
-        if (descriptor['state'] != 'ready' or self.runner._draining) and not allow_draining:
+        # A restart's after-turn wait still owns its turns: an interactive client attaches to answer
+        # a parked approval/clarify (#139041); stop() closes this window.
+        if ((descriptor['state'] != 'ready' or self.runner._draining) and not allow_draining
+                and not restart_wait_open(self.runner)):
             await ws.close(code=1013)
             return
         from gateway.session_authorities import authority_for_profile_id

gateway/run_bootstrap.py

@@ -269,6 +269,7 @@ async def _start_gateway_start_control_socket(runner):
             unserve_profile_verb, serve_profile_verb,
         )
         from gateway.run_plugin_rewire import reload_plugins_verb
+        from gateway.drain_control import RESTART_WAIT_REASON
         descriptor = runner.session_runtime_descriptor
 
         def _identify_runtime():
@@ -277,7 +278,11 @@ async def _start_gateway_start_control_socket(runner):
                 "instance_id", "runtime_protocol", "authority_epoch", "state", "api_origin",
                 "served_profiles", "parked_profiles", "capabilities") if key in descriptor})
             if getattr(runner, '_draining', False):
-                payload.update(state='draining', capabilities=[])
+                payload.update(state='draining')
+                # The restart's after-turn wait still owns its turns: clients reach it to answer a
+                # parked approval/clarify (#139041), so the endpoint stays dialable until stop().
+                if descriptor.get('drain_reason') != RESTART_WAIT_REASON:
+                    payload['capabilities'] = []
             if payload.get('state') == 'draining' and descriptor.get('drain_reason'):
                 payload['drain_reason'] = descriptor['drain_reason']
             payload["supervisor"] = {"manual": "none", "desktop": "none"}.get(

gateway/run_runtime.py

@@ -440,6 +440,9 @@ async def drain_gateway_runtime(runner):
         return
     runner._draining = True
     descriptor.update(state='draining', capabilities=[])
+    from gateway.drain_control import RESTART_WAIT_REASON
+    if descriptor.get('drain_reason') == RESTART_WAIT_REASON:
+        del descriptor['drain_reason']  # stopping now: no client attaches to answer anymore
     from gateway.session_hosted_service import stop_hosted_service
     await stop_hosted_service(runner)
     # Withdraw the public ingress callback without disconnecting egress needed

gateway/run_shutdown.py

@@ -1683,6 +1683,11 @@ class GatewayShutdownMixin(GatewaySessionEndMixin):
         self._restart_task_started = True
         # Refuse new turns; keep ``_running`` True so the active turn can still deliver its final response.
         self._draining = True
+        # Clients may still reach this owner to answer a turn parked on approval/clarify (#139041).
+        descriptor = getattr(self, "session_runtime_descriptor", None)
+        if descriptor is not None and not descriptor.get("drain_reason"):
+            from gateway.drain_control import RESTART_WAIT_REASON
+            descriptor["drain_reason"] = RESTART_WAIT_REASON
         # The restart's after-turn wait is a drain window too: pollers of GET /v1/runs/{id} must see
         # the boundary from the moment new turns are refused, not only once stop() begins (#115133).
         self._mark_api_runs_shutdown_requested()

gateway/session_authority.py

@@ -138,15 +138,22 @@ class SessionAuthority:
                 and source.user_id != actor.subject and 'session:operator' not in actor.capabilities):
             raise RuntimeStoreError('permission_denied')
 
-    def _require_admission_open(self):
-        if self.runner._draining or self.retiring:
+    def _require_admission_open(self, *, queue_only=False):
+        """``queue_only``: the caller only COMMITS a queued row (never claims or runs it). A restart's
+        after-turn wait accepts those (#139041): the row is durable FIFO input the next owner runs
+        exactly once, while this owner's claim path stays refused."""
+        if self.retiring:
             raise RuntimeStoreError('runtime_draining')
+        if self.runner._draining:
+            from gateway.drain_control import restart_wait_open
+            if not (queue_only and restart_wait_open(self.runner)):
+                raise RuntimeStoreError('runtime_draining')
 
-    def _admission_gate(self, authorize=None):
+    def _admission_gate(self, authorize=None, *, queue_only=False):
         """The admission gate re-checked inside an off-loop write transaction (with the caller's
         own guard), so a drain that began while the write waited on the writer still refuses it."""
         def guard(conn):
-            self._require_admission_open()
+            self._require_admission_open(queue_only=queue_only)
             if authorize is not None:
                 authorize(conn)
         return guard
@@ -434,7 +441,7 @@ class SessionAuthority:
                 return await self._submit(actor, request, _authorize_write=_authorize_write)
 
     async def _submit(self, actor, request, *, _authorize_write=None):
-        self._require_admission_open()
+        self._require_admission_open(queue_only=True)
         if (request.intent != 'queue' or not {'text'} <= set(request.payload) <= {
                 'text', 'attachments', 'finite', 'unattended', 'surface', 'voice_context', 'interrupted',
                 'voice_turn', 'display_kind', 'title_preview', 'turn_author', 'api_text', 'model_override', 'clarify_batch'}
@@ -469,7 +476,7 @@ class SessionAuthority:
                 return admit_session_input(self.db, epoch=self.epoch, principal_id=actor.subject,
                                            session_id=request.ref.session_id, request_id=request.request_id,
                                            payload=payload, intent=request.intent,
-                                           _authorize_write=self._admission_gate(_authorize_write))
+                                           _authorize_write=self._admission_gate(_authorize_write, queue_only=True))
             except Exception:
                 self._release_refused_capture(request, payload)
                 raise

hermes_cli/gateway_client.py

@@ -205,7 +205,9 @@ async def connect_gateway(endpoint=None):
         # A client's own start: the daemon it may launch ends itself once idle (no client, adapter,
         # cron/kanban work or admission), and the next client starts a fresh one.
         result = await asyncio.to_thread(ensure_gateway_runtime, home, idle_exit=True)
-        if result.state != "ready" or result.endpoint is None:
+        # A restart's after-turn wait still owns its turns (#139041): attach to answer a parked prompt.
+        attachable = result.state == "ready" or (result.state == "draining" and result.reason_code == "restart_wait")
+        if not attachable or result.endpoint is None:
             detail = getattr(result, "detail", None)
             if result.reason_code == "runtime_exited" and detail:
                 # A multi-line startup report (redacted traceback tail + where the full log is).

hermes_cli/gateway_runtime.py

@@ -67,12 +67,16 @@ def _endpoint(payload: dict, home: Path, control_home: Path | None = None) -> Ga
     if len(matches) != 1 or not isinstance(matches[0].get("profile_id"), str) or not matches[0]["profile_id"]:
         return GatewayDiscovery("inaccessible", reason_code="profile_mismatch")
     state = payload.get("state")
-    if state in {"starting", "draining", "conflict"}:
+    # A graceful restart's after-turn wait still owns its in-flight turns (#139041): it stays
+    # ``draining`` (exit 6, never a fresh-start target) but carries its endpoint, so an interactive
+    # client can attach to answer a turn parked on approval/clarify. New input queues for the next owner.
+    restart_wait = state == "draining" and payload.get("drain_reason") == "restart_wait"
+    if state in {"starting", "draining", "conflict"} and not restart_wait:
         # An auto-started gateway ending through its idle exit names that, so its clients wait
         # for it to go and start a fresh one instead of reporting a refusal.
         idle = state == "draining" and payload.get("drain_reason") == "idle_exit"
         return GatewayDiscovery(state, reason_code="idle_exit" if idle else None)
-    if state != "ready":
+    if state != "ready" and not restart_wait:
         return GatewayDiscovery("inaccessible", reason_code="invalid_runtime_state")
     capabilities = payload.get("capabilities")
     if (not isinstance(capabilities, list) or not all(isinstance(c, str) for c in capabilities)
@@ -96,13 +100,13 @@ def _endpoint(payload: dict, home: Path, control_home: Path | None = None) -> Ga
     supervisor = payload.get("supervisor")
     if supervisor not in {"none", "systemd", "launchd", "windows", "external"}:
         return GatewayDiscovery("inaccessible", reason_code="invalid_supervisor")
-    return GatewayDiscovery("ready", GatewayEndpoint(
+    return GatewayDiscovery("draining" if restart_wait else "ready", GatewayEndpoint(
         profile_id=matches[0]["profile_id"], instance_id=instance,
         authority_epoch=epoch, runtime_protocol=1, api_origin=origin,
         supervisor=supervisor, capabilities=frozenset(capabilities),
         control_home=str(control_home) if control_home is not None else None,
         code_sha=code_sha if isinstance(code_sha := payload.get("code_sha"), str) and code_sha else None,
-    ))
+    ), reason_code="restart_wait" if restart_wait else None)
 
 
 def control_home_for(home: Path, endpoint: GatewayEndpoint | None) -> Path:

tests/gateway/test_restart_wait_prompt_reach.py

@@ -0,0 +1,79 @@
+"""A graceful restart's after-turn wait stays reachable for a turn parked on a human prompt (#139041).
+
+The draining owner still owns its in-flight turn, so a client must be able to attach and answer the
+pending approval/clarify. New input is committed queued for the next owner and never claimed here.
+"""
+import asyncio
+from types import SimpleNamespace
+
+import pytest
+
+from hermes_state import SessionDB
+from hermes_state_runtime import RuntimeStoreError, begin_runtime_epoch, list_session_admissions
+
+
+@pytest.mark.platforms("linux")
+def test_ensure_hands_out_the_restart_wait_owner_as_draining_with_its_endpoint(tmp_path):
+    from tests.hermes_cli.test_gateway_runtime import control_peer
+    from hermes_cli.gateway_runtime import discover_gateway_endpoint
+    from hermes_cli.gateway_runtime_cli import ensure_exit_code
+
+    async def probe():
+        home = tmp_path / "profile"
+        identity = str(home.resolve())
+        payload = {"protocol": 1, "runtime_protocol": 1, "instance_id": "owner", "authority_epoch": 1,
+                   "hermes_home": identity, "state": "draining", "drain_reason": "restart_wait",
+                   "api_origin": "http://127.0.0.1:41817", "supervisor": "none",
+                   "served_profiles": [{"profile_id": identity, "home": identity}],
+                   "capabilities": ["session-authority-v1"]}
+        async with control_peer(home, payload):
+            wait = await asyncio.to_thread(discover_gateway_endpoint, home, timeout=1)
+            # Still draining (exit 6, never a fresh-start target) but dialable for answering prompts.
+            assert (wait.state, wait.reason_code) == ("draining", "restart_wait")
+            assert wait.endpoint is not None and wait.endpoint.instance_id == "owner"
+            assert ensure_exit_code(wait) == 6
+            # Once stop() begins the reason is withdrawn: a plain refusal, nothing to dial.
+            del payload["drain_reason"]
+            stopping = await asyncio.to_thread(discover_gateway_endpoint, home, timeout=1)
+            assert stopping.state == "draining" and stopping.endpoint is None
+
+    asyncio.run(probe())
+
+
+@pytest.mark.asyncio
+async def test_restart_wait_commits_new_input_queued_and_never_claims_it(tmp_path, monkeypatch):
+    from gateway.config import Platform
+    from gateway.session import SessionSource
+    from gateway.session_authority import LiveSession, SessionAuthority
+    from gateway.session_contract import Principal, SessionRef, Submission
+
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+    db = SessionDB(tmp_path / "state.db")
+    db.create_session("s", source="telegram")
+    ran = []
+
+    async def handle(event):
+        ran.append(event)
+        return "ran"
+
+    runner = SimpleNamespace(_draining=True, _running=True, config=SimpleNamespace(multiplex_profiles=False),
+                             session_runtime_descriptor={"state": "draining", "drain_reason": "restart_wait"},
+                             _adapter_for_source=lambda source: None, _handle_message=handle)
+    authority = SessionAuthority(runner, profile_id="p", instance_id="owner", db=db,
+                                 epoch=begin_runtime_epoch(db, instance_id="owner"))
+    authority.sessions["s"] = LiveSession(SessionSource(platform=Platform.TELEGRAM, chat_id="c", user_id="human"), "route")
+    actor = Principal("human", "p", frozenset({"session:submit"}), "t")
+    try:
+        receipt = await authority.submit(actor, Submission("r1", SessionRef("p", "s"), {"text": "next"}, "queue"))
+        task = authority.sessions["s"].task
+        if task is not None:
+            await asyncio.wait_for(task, 10)
+        rows = list_session_admissions(db, session_id="s")
+        assert [(r["admission_id"], r["status"]) for r in rows] == [(receipt.admission_id, "queued")]
+        assert ran == []  # the draining owner never runs it; the next owner's FIFO does
+        # stop() withdraws the window: input is refused, as before.
+        del runner.session_runtime_descriptor["drain_reason"]
+        with pytest.raises(RuntimeStoreError, match="runtime_draining"):
+            await authority.submit(actor, Submission("r2", SessionRef("p", "s"), {"text": "late"}, "queue"))
+    finally:
+        db.close()