Merge pull request #7818 from nesquena/stage/0924-o

Gateway-backend turns survive a WebUI restart (#7785)

CHANGELOG.md

@@ -44,6 +44,15 @@
 
 ### Fixed
 
+- **Gateway-backend turns survive a WebUI restart.** With the Gateway runs API enabled
+  (`HERMES_WEBUI_CHAT_BACKEND=gateway` + `HERMES_WEBUI_GATEWAY_USE_RUNS_API=true`), the Gateway
+  runs the turn, but restarting the WebUI still marked it interrupted, because the Gateway `run_id`
+  only lived in process memory. The run is now saved on the session as soon as the Gateway admits
+  it, with an `Idempotency-Key` so the Gateway keeps a durable record. On startup the WebUI
+  reattaches: it follows `GET /v1/runs/{id}` until the run settles and writes the real final
+  answer back. Text streamed before the restart isn't replayed; the settled answer replaces it.
+  Stop on a reattached turn goes to the Gateway that owns the run. Thanks @carlotestor. (#7785)
+
 - **Fewer self-inflicted console errors: git badge on CLI/subagent sessions, pollers after a
   profile switch, and the PWA startup preload.** Three separate sources of noise in DevTools, all
   caused by the WebUI itself. `/api/git-info` returned 404 for any session without a WebUI sidecar

api/gateway_chat.py

@@ -47,6 +47,21 @@ _STREAM_RUN_IDS: dict[str, str] = {}
 _STREAM_RUN_LIFECYCLE: dict[str, dict[str, Any]] = {}
 _STREAM_RUN_STARTING_CONDITION = threading.Condition()
 GATEWAY_RUN_ID_WAIT_TIMEOUT = 5.0
+# stream_id -> (base_url, api_key) of the Gateway the stream's run lives on, for Stop
+# and approval replies; a reattached run's Gateway need not be the process profile's.
+_STREAM_ENDPOINTS: dict[str, tuple[str, str]] = {}
+
+
+def gateway_run_endpoint(run_id: str) -> tuple[str, str]:
+    """URL and key of the Gateway that owns run_id; the process Gateway if no live stream holds it."""
+    run_id = str(run_id or "").strip()
+    with _STREAM_RUN_STARTING_CONDITION:
+        for stream_id, mapped in list(_STREAM_RUN_IDS.items()):
+            if run_id and mapped == run_id and stream_id in _STREAM_ENDPOINTS:
+                return _STREAM_ENDPOINTS[stream_id]
+    from api.config import get_config
+
+    return _gateway_base_url(get_config()), _gateway_api_key()
 
 
 def _mark_gateway_run_starting(stream_id: str) -> None:
@@ -565,24 +580,66 @@ def _gateway_runs_approval_event(payload: dict) -> dict | None:
     }
 
 
+def _relay_gateway_run_approval(session_id, run_id, payload, base_url, api_key, *, put_gateway_event) -> None:
+    """Auto-approve or surface one runs-API approval request as a WebUI approval card."""
+    approval_data = _gateway_runs_approval_event(payload)
+    if not approval_data:
+        return
+    approval_data["run_id"] = run_id
+    from api.config import gateway_supports_approval_identity_v1
+    identity_v1 = bool(approval_data.get("_gateway_raw_approval_id_present")) and gateway_supports_approval_identity_v1(base_url, api_key)
+    approval_data["_gateway_agent_identity_v1"] = identity_v1
+    auto_approved, head, total = _settle_gateway_run_approval(
+        session_id,
+        approval_data,
+        base_url,
+        api_key,
+    )
+    if not auto_approved:
+        put_gateway_event("approval", {**(head or approval_data), "pending_count": total})
+
+
+def _admit_gateway_run(url_runs, headers, run_body, stream_id) -> str:
+    """POST /v1/runs; the same stream and body return the originally admitted run id."""
+    req = urllib.request.Request(
+        url_runs,
+        data=json.dumps(run_body).encode("utf-8"),
+        # Durable run record on the gateway: GET /v1/runs/{id} survives either side restarting.
+        headers={**headers, "Idempotency-Key": f"webui-{stream_id}"},
+        method="POST",
+    )
+    with urllib.request.urlopen(req, timeout=30) as resp:
+        run_data = json.loads(resp.read(65536))
+    run_id = str(run_data.get("run_id") or run_data.get("id") or "").strip()
+    if not run_id:
+        raise ValueError(f"Gateway runs API returned no run_id: {run_data!r}")
+    return run_id
+
+
+def _gateway_run_headers(session_id, api_key) -> dict:
+    headers = {
+        "Content-Type": "application/json",
+        "Accept": "application/json",
+        "X-Hermes-Session-Id": session_id,
+    }
+    if api_key:
+        headers["Authorization"] = f"Bearer {api_key}"
+        headers["X-Hermes-Session-Key"] = f"webui:{session_id}"
+    return headers
+
+
 def _run_gateway_runs_api_streaming(
     session_id, msg_text, model, workspace, stream_id,
     base_url, api_key, prefill_messages, body_extras,
     *, put_gateway_event, cancel_event,
     attachments=None, cfg=None, session=None,
     active_provider: str = "",
+    on_run_id=None,
 ):
     """Submit via POST /v1/runs and relay SSE events including approval."""
     try:
         url_runs = f"{base_url.rstrip('/')}/v1/runs"
-        headers = {
-            "Content-Type": "application/json",
-            "Accept": "application/json",
-            "X-Hermes-Session-Id": session_id,
-        }
-        if api_key:
-            headers["Authorization"] = f"Bearer {api_key}"
-            headers["X-Hermes-Session-Key"] = f"webui:{session_id}"
+        headers = _gateway_run_headers(session_id, api_key)
         message_content: Any = str(msg_text or "")
         if attachments:
             try:
@@ -635,24 +692,19 @@ def _run_gateway_runs_api_streaming(
             run_body["instructions"] = "\n\n".join(part for part in instructions_parts if part)
         if conversation_history:
             run_body["conversation_history"] = conversation_history
-        req = urllib.request.Request(
-            url_runs,
-            data=json.dumps(run_body).encode("utf-8"),
-            headers=headers,
-            method="POST",
-        )
         update_active_run(stream_id, phase="gateway-request")
-        with urllib.request.urlopen(req, timeout=30) as resp:
-            run_data = json.loads(resp.read(65536))
-        run_id = str(run_data.get("run_id") or run_data.get("id") or "").strip()
-        if not run_id:
-            raise ValueError(f"Gateway runs API returned no run_id: {run_data!r}")
+        # Persist the exact body first: a restart before the run id is saved replays this admission.
+        if on_run_id is not None:
+            on_run_id("", request=run_body)
+        run_id = _admit_gateway_run(url_runs, headers, run_body, stream_id)
     except Exception:
         _finish_gateway_run_starting(stream_id)
         raise
 
     usage: dict = {}
     _publish_gateway_run_id(stream_id, run_id)
+    if on_run_id is not None:
+        on_run_id(run_id)
 
     url_events = f"{base_url.rstrip('/')}/v1/runs/{run_id}/events"
     headers_sse = dict(headers)
@@ -683,22 +735,10 @@ def _run_gateway_runs_api_streaming(
                 continue
             payload_event = str(payload.get("event") or payload.get("type") or sse_event).strip() or "message"
             if payload_event == "approval.request":
-                approval_data = _gateway_runs_approval_event(payload)
-                if approval_data:
-                    approval_data["run_id"] = run_id
-                    from api.config import gateway_supports_approval_identity_v1
-                    identity_v1 = bool(approval_data.get("_gateway_raw_approval_id_present")) and gateway_supports_approval_identity_v1(base_url, api_key)
-                    approval_data["_gateway_agent_identity_v1"] = identity_v1
-                    auto_approved, head, total = _settle_gateway_run_approval(
-                        session_id,
-                        approval_data,
-                        base_url,
-                        api_key,
-                    )
-                    if auto_approved:
-                        sse_event = "message"
-                        continue
-                    put_gateway_event("approval", {**(head or approval_data), "pending_count": total})
+                _relay_gateway_run_approval(
+                    session_id, run_id, payload, base_url, api_key,
+                    put_gateway_event=put_gateway_event,
+                )
                 sse_event = "message"
                 continue
             if payload_event in {"tool.started", "tool.completed", "reasoning.available"}:
@@ -795,11 +835,7 @@ def stop_gateway_run(run_id: str) -> bool:
     run_id = str(run_id or "").strip()
     if not run_id:
         return False
-    from api.config import get_config
-
-    cfg = get_config()
-    base_url = _gateway_base_url(cfg)
-    api_key = _gateway_api_key()
+    base_url, api_key = gateway_run_endpoint(run_id)
     headers = {"Accept": "application/json", "Content-Type": "application/json"}
     if api_key:
         headers["Authorization"] = f"Bearer {api_key}"
@@ -824,6 +860,180 @@ def stop_gateway_run(run_id: str) -> bool:
         return False
 
 
+_GATEWAY_RUN_TERMINAL_STATUSES = frozenset({"completed", "failed", "cancelled", "interrupted"})
+GATEWAY_REATTACH_POLL_INTERVAL = 2.0
+# Consecutive unreachable polls tolerated before the reattached turn is failed (~5 min at 2s).
+GATEWAY_REATTACH_MAX_POLL_FAILURES = 150
+_REATTACH_SCAN_HEAD_BYTES = 16 * 1024
+
+
+def _record_gateway_run(session_id: str, stream_id: str, run_id: str, request=None, **extra) -> None:
+    """Persist the run id (or, before admission, the exact request) so a restarted WebUI can reattach."""
+    try:
+        with _get_session_agent_lock(session_id):
+            session = get_session(session_id)
+            if not _stream_writeback_is_current(session, stream_id):
+                return
+            session.gateway_run = {"run_id": run_id, "stream_id": stream_id, **extra}
+            if not run_id:
+                session.gateway_run["request"] = request
+            session.save(touch_updated_at=False)
+    except Exception:
+        logger.warning("Failed to persist gateway run %s for session %s", run_id, session_id, exc_info=True)
+        if not run_id:
+            raise  # never admit a run a restart could not recover
+
+
+def _get_gateway_run_status(base_url: str, api_key: str, run_id: str) -> dict:
+    headers = {"Accept": "application/json"}
+    if api_key:
+        headers["Authorization"] = f"Bearer {api_key}"
+    req = urllib.request.Request(
+        f"{base_url.rstrip('/')}/v1/runs/{urllib.parse.quote(run_id, safe='')}",
+        headers=headers,
+        method="GET",
+    )
+    with urllib.request.urlopen(req, timeout=30) as resp:
+        payload = json.loads(resp.read(1024 * 1024) or b"{}")
+    return payload if isinstance(payload, dict) else {}
+
+
+def _await_gateway_run_result(
+    session_id, stream_id, run_id, base_url, api_key,
+    *, put_gateway_event, cancel_event,
+):
+    """Poll a run started by a previous WebUI process; same return contract as _run_gateway_runs_api_streaming."""
+    from api.route_approvals import settle_gateway_pending_run
+
+    _publish_gateway_run_id(stream_id, run_id)
+    update_active_run(stream_id, phase="gateway-reattached")
+    surfaced_approval_ids: set[str] = set()
+    failures = 0
+    while True:
+        if cancel_event.is_set():
+            put_gateway_event("cancel", {"message": "Cancelled by user"})
+            return None, {}
+        try:
+            status = _get_gateway_run_status(base_url, api_key, run_id)
+            failures = 0
+        except (urllib.error.URLError, OSError, ValueError) as exc:
+            code = getattr(exc, "code", None)
+            if code == 404:
+                raise RuntimeError("Gateway no longer has this run; its result could not be recovered after the WebUI restart.") from exc
+            if code in (401, 403):
+                raise RuntimeError(f"Gateway rejected the WebUI credentials (HTTP {code}) while reattaching to a run after the restart.") from exc
+            failures += 1
+            if failures >= GATEWAY_REATTACH_MAX_POLL_FAILURES:
+                raise RuntimeError("Gateway became unreachable while waiting for a run that outlived the WebUI restart.") from exc
+            cancel_event.wait(GATEWAY_REATTACH_POLL_INTERVAL)
+            continue
+        state = str(status.get("status") or "").strip().lower()
+        approval = status.get("approval")
+        if state == "waiting_for_approval" and isinstance(approval, dict):
+            approval_key = str(approval.get("approval_id") or approval.get("id") or approval.get("timestamp") or "")
+            if approval_key not in surfaced_approval_ids:
+                surfaced_approval_ids.add(approval_key)
+                _relay_gateway_run_approval(
+                    session_id, run_id, approval, base_url, api_key, put_gateway_event=put_gateway_event,
+                )
+        if state in _GATEWAY_RUN_TERMINAL_STATUSES:
+            settle_gateway_pending_run(session_id, run_id, reason=f"Gateway run {state} before approval resolution")
+            if state == "cancelled":
+                put_gateway_event("cancel", {"message": "Cancelled by gateway"})
+                return None, {}
+            if state != "completed":
+                raise RuntimeError(str(status.get("error") or f"Gateway run {state}"))
+            output = str(status.get("output") or "")
+            if output and stream_id in STREAM_PARTIAL_TEXT:
+                STREAM_PARTIAL_TEXT[stream_id] = output
+            return output, {k: v for k, v in _gateway_stream_usage(status).items() if v}
+        cancel_event.wait(GATEWAY_REATTACH_POLL_INTERVAL)
+
+
+def resume_gateway_runs_after_restart() -> list[str]:
+    """Reattach pending turns whose gateway run outlived the previous WebUI process; never raises.
+
+    Call before serving so stale-pending repair does not mark these turns interrupted.
+    """
+    from api import models as _models
+
+    try:
+        candidates = _sidecars_with_active_stream(_models.SESSION_DIR)
+    except Exception:
+        logger.warning("gateway reattach: could not scan session sidecars", exc_info=True)
+        return []
+    resumed: list[str] = []
+    for sid in candidates:
+        try:
+            if _resume_gateway_run_for_session(_models.Session.load_metadata_only(sid)):
+                resumed.append(sid)
+        except Exception:
+            logger.warning("gateway reattach failed for session %s", sid, exc_info=True)
+    if resumed:
+        logger.info("Reattached %d gateway run(s) that outlived the previous WebUI process", len(resumed))
+    return resumed
+
+
+def _sidecars_with_active_stream(session_dir) -> list[str]:
+    """Ids of sidecars (not the index, which can lag a save) whose header shows an active stream."""
+    ids = []
+    for path in sorted(session_dir.glob("*.json")):
+        if path.name.startswith("_"):
+            continue
+        try:
+            with path.open("rb") as fp:
+                head = fp.read(_REATTACH_SCAN_HEAD_BYTES)
+        except OSError:
+            continue
+        if b'"active_stream_id": null' not in head:
+            ids.append(path.stem)
+    return ids
+
+
+def _gateway_endpoint_for_profile(profile_name) -> tuple[str, str]:
+    """URL and key of the session's own profile, root included, never the process-active profile."""
+    from api import profiles as _profiles
+    from api.config import get_config_for_profile_home
+
+    home = _profiles.get_hermes_home_for_profile(str(profile_name or "").strip())
+    environ = {k: v for k, v in os.environ.items() if k not in _profiles._loaded_profile_env_keys}
+    environ.update(_profiles.filter_runtime_env_for_gateway_parity(_profiles.get_profile_runtime_env(home)))
+    return _gateway_base_url(get_config_for_profile_home(home), environ), _gateway_api_key(environ)
+
+
+def _resume_gateway_run_for_session(session) -> bool:
+    from api.config import create_stream_channel, register_session_writeback_owner, register_stream_owner
+
+    run = (session.gateway_run if session is not None else None) or {}
+    stream_id = str(run.get("stream_id") or "")
+    # A persisted run is the authority whatever the process default backend is.
+    if not stream_id or not (run.get("run_id") or run.get("request")) or session.active_stream_id != stream_id:
+        return False
+    sid = session.session_id
+    endpoint = _gateway_endpoint_for_profile(session.profile)
+    with STREAMS_LOCK:
+        if stream_id in STREAMS:
+            return False
+        STREAMS[stream_id] = create_stream_channel()
+    register_stream_owner(stream_id, sid)
+    register_session_writeback_owner(sid, stream_id)
+    _mark_gateway_run_starting(stream_id)
+    threading.Thread(
+        target=_run_gateway_chat_streaming,
+        args=(sid, session.pending_user_message or "", session.model, session.workspace,
+              stream_id, list(session.pending_attachments or [])),
+        kwargs={
+            "model_provider": session.model_provider,
+            "goal_related": bool(run.get("goal_related")),
+            "regeneration": bool(run.get("regeneration")),
+            "reattach_run": run,
+            "reattach_endpoint": endpoint,
+        },
+        name=f"gateway-reattach-{stream_id[:12]}",
+        daemon=True,
+    ).start()
+    return True
+
 def _settle_gateway_terminal_error(session_id, stream_id, workspace, model, model_provider, terminal_error):
     from api.streaming import (
         _classify_provider_error,
@@ -847,6 +1057,7 @@ def _settle_gateway_terminal_error(session_id, stream_id, workspace, model, mode
         turn_duration = _terminal_turn_duration(session)
         _materialize_pending_user_turn_before_error(session)
         session.active_stream_id = None
+        session.gateway_run = None
         session.pending_user_message = None
         session.pending_attachments = []
         session.pending_started_at = None
@@ -890,14 +1101,32 @@ def _settle_gateway_terminal_error(session_id, stream_id, workspace, model, mode
         return error_payload
 
 
+def _settle_gateway_cancelled_turn(session_id, stream_id) -> None:
+    """Keep the prompt and a cancel marker when the run ended cancelled and Stop did not already settle it."""
+    from api.streaming import _persist_cancelled_turn
+
+    with _get_session_agent_lock(session_id):
+        session = get_session(session_id)
+        if not _stream_writeback_is_current(session, stream_id):
+            return
+        _persist_cancelled_turn(session, message="Cancelled by gateway")
+        session.gateway_run = None
+        session.save()
+
+
 def _stream_writeback_is_current(session: Any, stream_id: str) -> bool:
     return bool(stream_id and getattr(session, "active_stream_id", None) == stream_id)
 
 
 def _clear_gateway_pending_state(session: Any, stream_id: str) -> None:
     if not _stream_writeback_is_current(session, stream_id):
+        # Cancel clears active_stream_id eagerly; still drop this stream's run record.
+        if session is not None and (session.gateway_run or {}).get("stream_id") == stream_id:
+            session.gateway_run = None
+            session.save(touch_updated_at=False)
         return
     session.active_stream_id = None
+    session.gateway_run = None
     session.pending_user_message = None
     session.pending_attachments = None
     session.pending_started_at = None
@@ -927,6 +1156,8 @@ def _run_gateway_chat_streaming(
     model_provider=None,
     goal_related=False,
     regeneration=False,
+    reattach_run=None,
+    reattach_endpoint=None,
 ):
     """Bridge a WebUI chat turn through Hermes Gateway's API server.
 
@@ -1013,8 +1244,9 @@ def _run_gateway_chat_streaming(
             model=model,
             model_provider=model_provider,
         )
-        base_url = _gateway_base_url(cfg)
-        api_key = _gateway_api_key()
+        base_url, api_key = reattach_endpoint or (_gateway_base_url(cfg), _gateway_api_key())
+        with _STREAM_RUN_STARTING_CONDITION:
+            _STREAM_ENDPOINTS[stream_id] = (base_url, api_key)
         try:
             from api.config import _main_model_request_overrides
             _gw_overrides = _main_model_request_overrides(
@@ -1025,7 +1257,7 @@ def _run_gateway_chat_streaming(
         except Exception:
             _gw_overrides = {}
         _runs_api_enabled = _gateway_use_runs_api_enabled(cfg)
-        _use_runs_api = _runs_api_enabled and gateway_supports_approval(base_url, api_key)
+        _use_runs_api = bool(reattach_run) or (_runs_api_enabled and gateway_supports_approval(base_url, api_key))
         if not _use_runs_api and runs_api_pending_marked:
             _finish_gateway_run_starting(stream_id, result="fallback")
             runs_api_pending_marked = False
@@ -1075,17 +1307,37 @@ def _run_gateway_chat_streaming(
                 body_extras["reasoning_effort"] = reasoning_effort
             if _gw_overrides.get("service_tier"):
                 body_extras["service_tier"] = _gw_overrides["service_tier"]
+            record_run = lambda run_id, request=None: _record_gateway_run(
+                session_id, stream_id, run_id, request=request,
+                regeneration=bool(regeneration), goal_related=bool(goal_related),
+            )
             try:
-                final_text, usage = _run_gateway_runs_api_streaming(
-                    session_id, msg_text, model, workspace, stream_id,
-                    base_url, api_key, prefill_messages, body_extras,
-                    put_gateway_event=put_gateway_event,
-                    cancel_event=cancel_event,
-                    attachments=attachments,
-                    cfg=cfg,
-                    session=s,
-                    active_provider=(model_provider or ""),
-                )
+                if reattach_run:
+                    run_id = str(reattach_run.get("run_id") or "").strip()
+                    if not run_id:
+                        # Crashed between admission and saving the id: the same key returns the original run.
+                        run_id = _admit_gateway_run(
+                            f"{base_url.rstrip('/')}/v1/runs", _gateway_run_headers(session_id, api_key),
+                            reattach_run["request"], stream_id,
+                        )
+                        record_run(run_id)
+                    final_text, usage = _await_gateway_run_result(
+                        session_id, stream_id, run_id, base_url, api_key,
+                        put_gateway_event=put_gateway_event,
+                        cancel_event=cancel_event,
+                    )
+                else:
+                    final_text, usage = _run_gateway_runs_api_streaming(
+                        session_id, msg_text, model, workspace, stream_id,
+                        base_url, api_key, prefill_messages, body_extras,
+                        put_gateway_event=put_gateway_event,
+                        cancel_event=cancel_event,
+                        attachments=attachments,
+                        cfg=cfg,
+                        session=s,
+                        active_provider=(model_provider or ""),
+                        on_run_id=record_run,
+                    )
             except Exception as exc:
                 error_payload = _settle_gateway_terminal_error(
                     session_id,
@@ -1100,6 +1352,7 @@ def _run_gateway_chat_streaming(
                 put_gateway_event("apperror", error_payload)
                 return
             if final_text is None:
+                _settle_gateway_cancelled_turn(session_id, stream_id)
                 return
         else:
             # Legacy gateway path: emit unsupported approval notice once per session,
@@ -1387,6 +1640,7 @@ def _run_gateway_chat_streaming(
                 if len(current_display_rows) == 1:
                     current_display_rows[0]["timestamp"] = user_msg["timestamp"]
             s.active_stream_id = None
+            s.gateway_run = None
             s.pending_user_message = None
             s.pending_attachments = None
             s.pending_started_at = None
@@ -1537,6 +1791,8 @@ def _run_gateway_chat_streaming(
         if runs_api_pending_marked and gateway_run_id_pending(stream_id):
             _finish_gateway_run_starting(stream_id)
         _clear_gateway_run_starting(stream_id)
+        with _STREAM_RUN_STARTING_CONDITION:
+            _STREAM_ENDPOINTS.pop(stream_id, None)
         unregister_stream_owner(stream_id)
         unregister_active_run(stream_id)
         # Release the writeback-owner entry the route layer registered for this

api/models.py

@@ -1422,6 +1422,7 @@ class Session:
                  process_wakeup_pause=None,
                  share_token=None,
                  share_created_at=None,
+                 gateway_run=None,
                  **kwargs):
         self.session_id = session_id or uuid.uuid4().hex[:12]
         self.title = title
@@ -1536,6 +1537,7 @@ class Session:
         self.process_wakeup_pause = process_wakeup_pause if isinstance(process_wakeup_pause, dict) else {}
         self.share_token = str(share_token).strip() if share_token else None
         self.share_created_at = share_created_at
+        self.gateway_run = gateway_run if isinstance(gateway_run, dict) else None
         # #5854: a compact fingerprint of anchor_activity_scenes ({scene_key:
         # updated_at}) persisted BEFORE the messages array so the sidebar-poll
         # freshness check can compare scene freshness without parsing the full
@@ -1616,6 +1618,7 @@ class Session:
             'enabled_toolsets', 'composer_draft',
             'process_wakeup_pause',
             'share_token', 'share_created_at',
+            'gateway_run',
         ]
         meta = {k: getattr(self, k, None) for k in METADATA_FIELDS}
         # #5854: message_count and a compact anchor-scene fingerprint go in the

api/routes.py

@@ -27098,8 +27098,8 @@ def _relay_gateway_run_approval(
     Both the approval-card endpoint and the ordinary session-YOLO endpoint use
     this chokepoint so one tab cannot retire another tab's parked remote run.
     """
-    from api.config import gateway_supports_approval_identity_v1, get_config as _get_config
-    from api.gateway_chat import _gateway_api_key, _gateway_base_url
+    from api.config import gateway_supports_approval_identity_v1
+    from api.gateway_chat import gateway_run_endpoint
     from api.runner_client import HttpRunnerClient, RunnerClientError
 
     run_id = str(mirror.get("run_id") or "").strip()
@@ -27141,8 +27141,7 @@ def _relay_gateway_run_approval(
                 enable_yolo=enable_yolo,
             )
 
-        base_url = _gateway_base_url(_get_config())
-        api_key = _gateway_api_key()
+        base_url, api_key = gateway_run_endpoint(run_id)
         identity_v1 = bool(current_mirror.get(_GATEWAY_AGENT_IDENTITY_V1)) and (
             gateway_supports_approval_identity_v1(base_url, api_key)
         )

docs/advanced-chat-setup.md

@@ -119,6 +119,8 @@ HERMES_WEBUI_GATEWAY_USE_RUNS_API=true \
 
 Use this when the connected gateway advertises approval support and you want tool approval cards to appear in WebUI. Without `HERMES_WEBUI_GATEWAY_USE_RUNS_API=true`, gateway chat stays on the legacy chat-completions transport and approval-capable commands can remain pending in the agent without a WebUI approval card.
 
+On the runs API path the turn is executed by the Gateway, so restarting WebUI does not stop it. WebUI stores the Gateway `run_id` on the pending turn (and submits it with an `Idempotency-Key` so the Gateway keeps a durable run record). On startup, WebUI reattaches to every such run by polling `GET /v1/runs/{run_id}` until it settles, then writes the real final answer into the session instead of a "Response interrupted" marker. Stop still cancels a reattached run, and a pending approval is shown again. Token-by-token output from before the restart is not replayed; the reattached turn shows only the final answer. If the Gateway no longer knows the run (for example, it restarted too and the run was interrupted), the turn ends with an error message instead. The legacy chat-completions transport cannot reattach: its turn ends when the WebUI process that holds the HTTP stream exits.
+
 When YOLO is enabled for a gateway-backed browser session, WebUI approves every
 approval already parked for that session: Runs API prompts are relayed by their
 exact `run_id` and mirror token, and local/no-run waiters are all released. It

server.py

@@ -669,6 +669,7 @@ def main() -> None:
 
     _abort_if_already_serving(HOST, PORT)
     httpd = QuietHTTPServer((HOST, PORT), Handler)
+    from api.gateway_chat import resume_gateway_runs_after_restart; resume_gateway_runs_after_restart()  # bound, not yet serving
 
     from api.config import TLS_ENABLED, TLS_CERT, TLS_KEY
     scheme = 'https' if TLS_ENABLED else 'http'

tests/test_gateway_run_reattach_after_restart.py

@@ -0,0 +1,530 @@
+"""A gateway-owned run must survive a WebUI restart instead of being marked interrupted."""
+from collections import OrderedDict
+import io
+import json
+import os
+import threading
+import urllib.error
+from email.message import Message
+
+import pytest
+
+import api.gateway_chat as gateway_chat
+import api.models as models
+import api.streaming as streaming
+from api.config import ACTIVE_RUNS, STREAMS, STREAMS_LOCK, create_stream_channel
+from api import profiles
+from api.models import new_session
+
+
+@pytest.fixture
+def isolated_sessions(tmp_path, monkeypatch):
+    session_dir = tmp_path / "sessions"
+    session_dir.mkdir()
+    monkeypatch.setattr(models, "SESSION_DIR", session_dir)
+    monkeypatch.setattr(models, "SESSION_INDEX_FILE", session_dir / "_index.json")
+    monkeypatch.setattr(models, "SESSIONS", OrderedDict())
+    monkeypatch.setenv("HERMES_WEBUI_CHAT_BACKEND", "gateway")
+    monkeypatch.setenv("HERMES_WEBUI_GATEWAY_USE_RUNS_API", "1")
+    monkeypatch.setenv("HERMES_WEBUI_GATEWAY_BASE_URL", "http://gateway.local")
+    monkeypatch.setattr(gateway_chat, "GATEWAY_REATTACH_POLL_INTERVAL", 0.01)
+    monkeypatch.setattr(gateway_chat, "_gateway_reasoning_effort_for_request", lambda *a, **k: None)
+    monkeypatch.setattr(streaming, "_load_webui_prefill_context", lambda cfg: {
+        "status": "not_configured", "source": "none", "label": "", "message_count": 0, "messages": [],
+    })
+    monkeypatch.setattr(streaming, "_prefill_messages_with_webui_context", lambda ctx, cfg: [])
+    return session_dir
+
+
+def _orphaned_gateway_turn(run_id="run_survivor", stream_id="stream-before-restart"):
+    """Persist the sidecar exactly as a WebUI process leaves it when killed mid-run."""
+    s = new_session()
+    s.messages = [
+        {"role": "user", "content": "earlier", "timestamp": 0.5},
+        {"role": "assistant", "content": "earlier reply", "timestamp": 0.6},
+    ]
+    s.active_stream_id = stream_id
+    s.pending_user_message = "long task"
+    s.pending_attachments = []
+    s.pending_started_at = 1.0
+    s.pending_user_source = "webui"
+    s.gateway_run = {"run_id": run_id, "stream_id": stream_id, "regeneration": False, "goal_related": False}
+    s.save()
+    # Simulate the new process: nothing of the old run is in memory.
+    models.SESSIONS.clear()
+    with STREAMS_LOCK:
+        STREAMS.pop(stream_id, None)
+    ACTIVE_RUNS.pop(stream_id, None)
+    return s.session_id, stream_id
+
+
+def _saved(session_id):
+    return json.loads((models.SESSION_DIR / f"{session_id}.json").read_text())
+
+
+def _wait_for_reattach_threads(timeout=10.0):
+    for thread in threading.enumerate():
+        if thread.name.startswith("gateway-reattach-"):
+            thread.join(timeout)
+            assert not thread.is_alive(), "reattach worker did not settle"
+
+
+def test_runs_api_start_sends_idempotency_key_and_persists_run_id(isolated_sessions, monkeypatch):
+    s = new_session()
+    stream_id = "stream-live"
+    s.active_stream_id = stream_id
+    s.pending_user_message = "hi"
+    s.pending_attachments = []
+    s.pending_started_at = 123.0
+    s.save()
+    captured = {}
+
+    def fake_urlopen(req, timeout=None):
+        if req.get_method() == "POST":
+            captured["post_headers"] = dict(req.header_items())
+            return io.BytesIO(b'{"run_id":"run_live"}')
+        # The run id is durable before the first event is relayed.
+        captured["persisted_at_events"] = _saved(s.session_id).get("gateway_run")
+        return io.BytesIO(
+            b'data: {"event":"message.delta","delta":"done"}\n'
+            b'data: {"event":"run.completed","output":"done"}\n'
+            b"data: [DONE]\n"
+        )
+
+    monkeypatch.setattr(gateway_chat, "gateway_supports_approval", lambda *a, **k: True)
+    monkeypatch.setattr(gateway_chat.urllib.request, "urlopen", fake_urlopen)
+    with STREAMS_LOCK:
+        STREAMS[stream_id] = create_stream_channel()
+
+    gateway_chat._run_gateway_chat_streaming(s.session_id, "hi", "test-model", "/tmp", stream_id, [])
+
+    assert captured["post_headers"]["Idempotency-key"] == f"webui-{stream_id}"
+    # Only ids and flags are persisted, never a credential.
+    assert captured["persisted_at_events"] == {
+        "run_id": "run_live", "stream_id": stream_id, "regeneration": False, "goal_related": False,
+    }
+    saved = _saved(s.session_id)
+    assert saved["gateway_run"] is None
+    assert saved["active_stream_id"] is None
+    assert saved["messages"][-1]["content"] == "done"
+
+
+def test_restart_reattaches_and_writes_back_real_answer(isolated_sessions, monkeypatch):
+    sid, stream_id = _orphaned_gateway_turn()
+    release = threading.Event()
+    polls = []
+
+    def fake_status(base_url, api_key, run_id):
+        polls.append(run_id)
+        if not release.is_set():
+            return {"run_id": run_id, "status": "running"}
+        return {
+            "run_id": run_id,
+            "status": "completed",
+            "output": "finished after the restart",
+            "usage": {"input_tokens": 7, "output_tokens": 3},
+        }
+
+    monkeypatch.setattr(gateway_chat, "_get_gateway_run_status", fake_status)
+
+    assert gateway_chat.resume_gateway_runs_after_restart() == [sid]
+    # While the run is still going, the turn is live: no stale-pending repair, reconnect works.
+    assert stream_id in STREAMS
+    assert gateway_chat.wait_for_gateway_run_id(stream_id, 5.0) == (True, "run_survivor")
+    live = models.get_session(sid)
+    assert live.active_stream_id == stream_id
+    assert not any(m.get("_error") for m in live.messages)
+
+    release.set()
+    _wait_for_reattach_threads()
+
+    saved = _saved(sid)
+    assert [m["role"] for m in saved["messages"]] == ["user", "assistant", "user", "assistant"]
+    assert saved["messages"][2]["content"] == "long task"
+    assert saved["messages"][3]["content"] == "finished after the restart"
+    assert not any(m.get("_error") for m in saved["messages"])
+    assert saved["active_stream_id"] is None
+    assert saved["pending_user_message"] is None
+    assert saved["gateway_run"] is None
+    assert set(polls) == {"run_survivor"}
+    assert stream_id not in STREAMS
+
+
+def test_reattach_surfaces_pending_approval_once(isolated_sessions, monkeypatch):
+    sid, stream_id = _orphaned_gateway_turn(run_id="run_parked")
+    relayed = []
+    calls = {"n": 0}
+
+    def fake_status(base_url, api_key, run_id):
+        calls["n"] += 1
+        if calls["n"] < 4:
+            return {"run_id": run_id, "status": "waiting_for_approval", "approval": {
+                "event": "approval.request", "approval_id": "appr-1", "command": "rm -rf build", "description": "delete",
+            }}
+        return {"run_id": run_id, "status": "completed", "output": "approved and done"}
+
+    monkeypatch.setattr(gateway_chat, "_get_gateway_run_status", fake_status)
+    monkeypatch.setattr(
+        gateway_chat, "_relay_gateway_run_approval",
+        lambda session_id, run_id, payload, *a, **k: relayed.append((session_id, run_id, payload["approval_id"])),
+    )
+    gateway_chat.resume_gateway_runs_after_restart()
+    _wait_for_reattach_threads()
+
+    assert relayed == [(sid, "run_parked", "appr-1")]
+    assert _saved(sid)["messages"][-1]["content"] == "approved and done"
+
+
+def test_nothing_to_reattach_is_left_to_stale_pending_repair(isolated_sessions, monkeypatch):
+    sid, stream_id = _orphaned_gateway_turn()
+    s = models.Session.load(sid)
+    s.gateway_run = None
+    s.save()
+    monkeypatch.setattr(gateway_chat, "_get_gateway_run_status", lambda *a, **k: pytest.fail("must not poll"))
+    assert gateway_chat.resume_gateway_runs_after_restart() == []
+    assert stream_id not in STREAMS
+
+
+def test_cancel_still_stops_a_reattached_run(isolated_sessions, monkeypatch):
+    sid, stream_id = _orphaned_gateway_turn(run_id="run_to_stop")
+    stopped = []
+    monkeypatch.setattr(
+        gateway_chat, "_get_gateway_run_status",
+        lambda base_url, api_key, run_id: {"run_id": run_id, "status": "running"},
+    )
+    monkeypatch.setattr(gateway_chat, "stop_gateway_run", lambda run_id: stopped.append(run_id) or True)
+    gateway_chat.resume_gateway_runs_after_restart()
+
+    # Same two steps /api/chat/cancel performs for a gateway-backed stream.
+    _structured, run_id = gateway_chat.wait_for_gateway_run_id(stream_id, 5.0)
+    assert gateway_chat.stop_gateway_run(run_id)
+    for _ in range(500):
+        if stream_id in gateway_chat.CANCEL_FLAGS:
+            break
+        threading.Event().wait(0.01)
+    assert streaming.cancel_stream(stream_id)
+    _wait_for_reattach_threads()
+
+    assert stopped == ["run_to_stop"]
+    saved = _saved(sid)
+    assert saved["active_stream_id"] is None
+    assert saved["gateway_run"] is None
+    assert not any(m.get("content") == "long task" and m.get("role") == "assistant" for m in saved["messages"])
+
+
+def _poll_until_completed(monkeypatch):
+    seen = []
+
+    def fake_status(base_url, api_key, run_id):
+        seen.append((base_url, api_key, run_id))
+        return {"run_id": run_id, "status": "completed", "output": "answer"}
+
+    monkeypatch.setattr(gateway_chat, "_get_gateway_run_status", fake_status)
+    return seen
+
+
+@pytest.mark.parametrize("index_state", ["missing", "stale", "corrupt"])
+def test_reattach_does_not_depend_on_the_session_index(isolated_sessions, monkeypatch, index_state):
+    sid, _stream_id = _orphaned_gateway_turn()
+    index = isolated_sessions / "_index.json"
+    if index_state == "missing":
+        index.unlink(missing_ok=True)
+    elif index_state == "stale":
+        # Sidecar saved, index update lost: the row still shows no active stream.
+        index.write_text(json.dumps([{"session_id": sid, "active_stream_id": None}]))
+    else:
+        index.write_text("{not json")
+    _poll_until_completed(monkeypatch)
+
+    assert gateway_chat.resume_gateway_runs_after_restart() == [sid]
+    _wait_for_reattach_threads()
+
+    saved = _saved(sid)
+    assert saved["messages"][-1]["content"] == "answer"
+    assert saved["active_stream_id"] is None
+
+
+def test_reattach_skips_idle_sidecars_without_parsing_them(isolated_sessions, monkeypatch):
+    idle = new_session()
+    idle.messages = [{"role": "user", "content": "x", "timestamp": 1.0}]
+    idle.save()
+    sid, _ = _orphaned_gateway_turn()
+    loaded = []
+    real_load = models.Session.load_metadata_only
+    monkeypatch.setattr(
+        models.Session, "load_metadata_only",
+        classmethod(lambda cls, s, **kw: loaded.append(s) or real_load.__func__(cls, s, **kw)),
+    )
+    _poll_until_completed(monkeypatch)
+
+    assert gateway_chat.resume_gateway_runs_after_restart() == [sid]
+    _wait_for_reattach_threads()
+    assert idle.session_id not in loaded
+    assert sid in loaded
+
+
+@pytest.mark.parametrize("default_backend", ["gateway", "local"])
+@pytest.mark.parametrize("session_profile, process_profile", [("default", "work"), ("work", "default")])
+def test_reattach_resolves_gateway_from_the_session_profile(
+    isolated_sessions, tmp_path, monkeypatch, default_backend, session_profile, process_profile,
+):
+    """Poll the session profile's own gateway, root included, whatever profile the process restarted under."""
+    root = tmp_path / "hermes"
+    for name, home in (("default", root), ("work", root / "profiles" / "work")):
+        home.mkdir(parents=True, exist_ok=True)
+        (home / ".env").write_text(
+            f"HERMES_WEBUI_GATEWAY_BASE_URL=http://{name}-gateway:8642\nHERMES_WEBUI_GATEWAY_API_KEY={name}-key\n"
+        )
+    monkeypatch.setattr(profiles, "_DEFAULT_HERMES_HOME", root)
+    monkeypatch.setattr(profiles, "_active_profile", process_profile)
+    monkeypatch.setattr(profiles, "_loaded_profile_env_keys", set())
+    monkeypatch.delenv("HERMES_WEBUI_GATEWAY_BASE_URL")
+    monkeypatch.delenv("HERMES_WEBUI_GATEWAY_API_KEY", raising=False)
+    profiles._reload_dotenv(root if process_profile == "default" else root / "profiles" / "work")
+    if default_backend == "local":
+        monkeypatch.delenv("HERMES_WEBUI_CHAT_BACKEND")
+        monkeypatch.delenv("HERMES_WEBUI_GATEWAY_USE_RUNS_API")
+    sid, stream_id = _orphaned_gateway_turn()
+    s = models.Session.load(sid)
+    s.profile = session_profile
+    s.save(touch_updated_at=False)
+    models.SESSIONS.clear()
+    seen = _poll_until_completed(monkeypatch)
+
+    assert os.environ["HERMES_WEBUI_GATEWAY_API_KEY"] == f"{process_profile}-key"
+    assert gateway_chat.resume_gateway_runs_after_restart() == [sid]
+    _wait_for_reattach_threads()
+
+    assert {(b, k, r) for b, k, r in seen} == {
+        (f"http://{session_profile}-gateway:8642", f"{session_profile}-key", "run_survivor"),
+    }
+    saved = _saved(sid)
+    assert saved["messages"][-1]["content"] == "answer"
+    assert saved["active_stream_id"] is None and saved["gateway_run"] is None
+
+
+def test_crash_after_admission_before_run_id_saved_replays_the_same_admission(isolated_sessions, monkeypatch):
+    s = new_session()
+    stream_id = "stream-admitted"
+    s.active_stream_id = stream_id
+    s.pending_user_message = "long task"
+    s.pending_attachments = []
+    s.pending_started_at = 1.0
+    s.save()
+    posts = []
+
+    class Crash(BaseException):
+        pass
+
+    def fake_urlopen(req, timeout=None):
+        assert req.get_method() == "POST"
+        posts.append((req.get_header("Idempotency-key"), req.data))
+        return io.BytesIO(b'{"run_id":"run_original","replayed":%s}' % (b"true" if len(posts) > 1 else b"false"))
+
+    real_record = gateway_chat._record_gateway_run
+
+    def record_then_die(session_id, stream_id, run_id, **kw):
+        if run_id:
+            raise Crash  # process killed after the POST response, before the id is saved
+        real_record(session_id, stream_id, run_id, **kw)
+
+    monkeypatch.setattr(gateway_chat, "gateway_supports_approval", lambda *a, **k: True)
+    monkeypatch.setattr(gateway_chat.urllib.request, "urlopen", fake_urlopen)
+    monkeypatch.setattr(gateway_chat, "_record_gateway_run", record_then_die)
+    with STREAMS_LOCK:
+        STREAMS[stream_id] = create_stream_channel()
+    with pytest.raises(Crash):
+        gateway_chat._run_gateway_runs_api_streaming(
+            s.session_id, "long task", "test-model", "/tmp", stream_id, "http://gateway.local", "", [], {},
+            put_gateway_event=lambda *a: None, cancel_event=threading.Event(), session=s,
+            on_run_id=lambda run_id, **kw: gateway_chat._record_gateway_run(s.session_id, stream_id, run_id, **kw),
+        )
+    monkeypatch.setattr(gateway_chat, "_record_gateway_run", real_record)
+    models.SESSIONS.clear()
+    with STREAMS_LOCK:
+        STREAMS.pop(stream_id, None)
+    ACTIVE_RUNS.pop(stream_id, None)
+    assert _saved(s.session_id)["gateway_run"]["run_id"] == ""
+    seen = _poll_until_completed(monkeypatch)
+
+    assert gateway_chat.resume_gateway_runs_after_restart() == [s.session_id]
+    _wait_for_reattach_threads()
+
+    # Replayed with the same key and byte-identical body, so the gateway returns the original run.
+    assert len(posts) == 2 and posts[0] == posts[1] == (f"webui-{stream_id}", posts[0][1])
+    assert {r for _b, _k, r in seen} == {"run_original"}
+    saved = _saved(s.session_id)
+    assert saved["messages"][-1]["content"] == "answer"
+    assert not any(m.get("_error") for m in saved["messages"])
+    assert saved["active_stream_id"] is None and saved["gateway_run"] is None
+
+
+@pytest.mark.parametrize("code, message", [
+    (404, "could not be recovered after the WebUI restart"),
+    (401, "HTTP 401"),
+    (403, "HTTP 403"),
+])
+def test_reattach_http_rejection_settles_the_turn_with_an_error(isolated_sessions, monkeypatch, code, message):
+    sid, _stream_id = _orphaned_gateway_turn()
+    calls = []
+
+    def rejected(base_url, api_key, run_id):
+        calls.append(run_id)
+        raise urllib.error.HTTPError("http://gateway.local/v1/runs/x", code, "denied", Message(), io.BytesIO(b""))
+
+    monkeypatch.setattr(gateway_chat, "_get_gateway_run_status", rejected)
+    assert gateway_chat.resume_gateway_runs_after_restart() == [sid]
+    _wait_for_reattach_threads()
+
+    assert len(calls) == 1
+    saved = _saved(sid)
+    assert saved["messages"][-2]["content"] == "long task"
+    assert saved["messages"][-1]["_error"] is True
+    assert message in saved["messages"][-1]["content"]
+    assert saved["active_stream_id"] is None and saved["gateway_run"] is None
+
+
+def test_failed_recovery_checkpoint_never_admits_the_run(isolated_sessions, monkeypatch):
+    s = new_session()
+    stream_id = "stream-unsaved"
+    s.active_stream_id = stream_id
+    s.pending_user_message = "hi"
+    s.pending_attachments = []
+    s.pending_started_at = 1.0
+    s.save()
+    monkeypatch.setattr(gateway_chat, "gateway_supports_approval", lambda *a, **k: True)
+    monkeypatch.setattr(gateway_chat.urllib.request, "urlopen", lambda *a, **k: pytest.fail("admitted without a checkpoint"))
+    monkeypatch.setattr(models.Session, "save", lambda self, **kw: (_ for _ in ()).throw(OSError("disk full")))
+    with STREAMS_LOCK:
+        STREAMS[stream_id] = create_stream_channel()
+
+    gateway_chat._run_gateway_chat_streaming(s.session_id, "hi", "test-model", "/tmp", stream_id, [])
+    assert stream_id not in STREAMS
+
+
+@pytest.mark.parametrize("stopped_by", ["user", "gateway"])
+def test_cancelled_replayed_admission_keeps_the_prompt_and_is_not_reattached(isolated_sessions, monkeypatch, stopped_by):
+    sid, stream_id = _orphaned_gateway_turn(run_id="")
+    s = models.Session.load(sid)
+    s.gateway_run["request"] = {"input": "long task"}
+    s.save(touch_updated_at=False)
+    models.SESSIONS.clear()
+    monkeypatch.setattr(gateway_chat, "_admit_gateway_run", lambda *a, **k: "run_replayed")
+    state = "running" if stopped_by == "user" else "cancelled"
+    monkeypatch.setattr(gateway_chat, "_get_gateway_run_status", lambda b, k, r: {"run_id": r, "status": state})
+    monkeypatch.setattr(gateway_chat, "stop_gateway_run", lambda run_id: True)
+
+    assert gateway_chat.resume_gateway_runs_after_restart() == [sid]
+    if stopped_by == "user":
+        assert gateway_chat.wait_for_gateway_run_id(stream_id, 5.0) == (True, "run_replayed")
+        assert streaming.cancel_stream(stream_id)
+    _wait_for_reattach_threads()
+
+    saved = _saved(sid)
+    assert [(m["role"], m["content"]) for m in saved["messages"][:3]] == [
+        ("user", "earlier"), ("assistant", "earlier reply"), ("user", "long task"),
+    ]
+    assert len(saved["messages"]) == 4 and saved["messages"][3]["_error"] is True
+    assert "Task cancelled" in saved["messages"][3]["content"]
+    assert saved["active_stream_id"] is None and saved["gateway_run"] is None
+    assert saved["pending_user_message"] is None
+    models.SESSIONS.clear()
+    assert gateway_chat.resume_gateway_runs_after_restart() == []
+
+
+def _two_profile_gateways(tmp_path, monkeypatch, process_profile):
+    root = tmp_path / "hermes"
+    for name, home in (("default", root), ("work", root / "profiles" / "work")):
+        home.mkdir(parents=True, exist_ok=True)
+        (home / ".env").write_text(
+            f"HERMES_WEBUI_GATEWAY_BASE_URL=http://{name}-gateway:8642\nHERMES_WEBUI_GATEWAY_API_KEY={name}-key\n"
+        )
+    monkeypatch.setattr(profiles, "_DEFAULT_HERMES_HOME", root)
+    monkeypatch.setattr(profiles, "_active_profile", process_profile)
+    monkeypatch.setattr(profiles, "_loaded_profile_env_keys", set())
+    monkeypatch.delenv("HERMES_WEBUI_GATEWAY_BASE_URL")
+    monkeypatch.delenv("HERMES_WEBUI_GATEWAY_API_KEY", raising=False)
+    profiles._reload_dotenv(root if process_profile == "default" else root / "profiles" / "work")
+
+
+def _reattach_under_profile(session_profile, run_id):
+    sid, stream_id = _orphaned_gateway_turn(run_id=run_id)
+    s = models.Session.load(sid)
+    s.profile = session_profile
+    s.save(touch_updated_at=False)
+    models.SESSIONS.clear()
+    assert gateway_chat.resume_gateway_runs_after_restart() == [sid]
+    assert gateway_chat.wait_for_gateway_run_id(stream_id, 5.0) == (True, run_id)
+    return sid, stream_id
+
+
+@pytest.mark.parametrize("session_profile, process_profile", [("default", "work"), ("work", "default")])
+def test_stop_reaches_the_gateway_that_owns_a_reattached_run(
+    isolated_sessions, tmp_path, monkeypatch, session_profile, process_profile,
+):
+    _two_profile_gateways(tmp_path, monkeypatch, process_profile)
+    monkeypatch.setattr(gateway_chat, "_get_gateway_run_status", lambda b, k, r: {"run_id": r, "status": "running"})
+    sent = []
+
+    class _Resp(io.BytesIO):
+        status = 200
+
+    class _Opener:
+        def open(self, req, timeout=None):
+            sent.append((req.full_url, req.get_header("Authorization")))
+            resp = _Resp(b"{}")
+            resp.geturl = lambda: req.full_url
+            return resp
+
+    monkeypatch.setattr(gateway_chat.urllib.request, "build_opener", lambda *a: _Opener())
+    _sid, stream_id = _reattach_under_profile(session_profile, "run_to_stop")
+
+    assert gateway_chat.stop_gateway_run("run_to_stop")
+    assert streaming.cancel_stream(stream_id)
+    _wait_for_reattach_threads()
+
+    assert sent == [
+        (f"http://{session_profile}-gateway:8642/v1/runs/run_to_stop/stop", f"Bearer {session_profile}-key"),
+    ]
+    assert stream_id not in gateway_chat._STREAM_ENDPOINTS
+
+
+@pytest.mark.parametrize("session_profile, process_profile", [("default", "work"), ("work", "default")])
+def test_approval_reply_reaches_the_gateway_that_owns_a_reattached_run(
+    isolated_sessions, tmp_path, monkeypatch, session_profile, process_profile,
+):
+    from unittest.mock import MagicMock
+
+    import api.route_approvals as approvals
+    import api.runner_client as runner_client
+    from api import config, routes
+
+    _two_profile_gateways(tmp_path, monkeypatch, process_profile)
+    monkeypatch.setattr(gateway_chat, "_get_gateway_run_status", lambda b, k, r: {
+        "run_id": r, "status": "waiting_for_approval",
+        "approval": {"event": "approval.request", "approval_id": "appr-1", "command": "rm -rf build", "description": "d"},
+    })
+    monkeypatch.setattr(config, "gateway_supports_approval_identity_v1", lambda *a, **k: True)
+    replies = []
+
+    def fake_respond(self, run_id, approval_id, choice):
+        replies.append((self.base_url, self.api_key, run_id, approval_id, choice))
+        return {"ok": True}
+
+    monkeypatch.setattr(runner_client.HttpRunnerClient, "respond_approval", fake_respond)
+    sid, stream_id = _reattach_under_profile(session_profile, "run_parked")
+    try:
+        for _ in range(500):
+            if approvals.gateway_pending_mirror(sid, approval_id="appr-1", run_id="run_parked"):
+                break
+            threading.Event().wait(0.01)
+        handler = MagicMock()
+        handler.wfile = io.BytesIO()
+        routes._handle_approval_respond(handler, {"session_id": sid, "choice": "once", "approval_id": "appr-1"})
+
+        handler.send_response.assert_called_with(200)
+        assert replies == [(f"http://{session_profile}-gateway:8642", f"{session_profile}-key", "run_parked", "appr-1", "once")]
+    finally:
+        streaming.cancel_stream(stream_id)
+        _wait_for_reattach_threads()
+        approvals._pending.pop(sid, None)