From 7ba246bfc0bdeddc275bf6b593868027312f0c3b Mon Sep 17 00:00:00 2001 From: carlotestor Date: Wed, 23 Sep 2026 16:30:21 +0000 Subject: [PATCH 1/9] fix(gateway): reattach runs-API turns after a WebUI restart instead of marking them interrupted A gateway-backed turn on the runs API executes in the Hermes Gateway, but the run id only lived in process memory, so a WebUI restart orphaned the turn and stale-pending repair appended 'Response interrupted' while the run finished on the gateway. - persist session.gateway_run {run_id, stream_id} as soon as POST /v1/runs returns - send Idempotency-Key webui- so the gateway keeps a durable run record - on startup (after bind, before serving) re-register the stream and follow GET /v1/runs/{id} until terminal; write back the real output, surface a parked approval once, keep Stop working, clear the record on every settlement --- api/gateway_chat.py | 256 ++++++++++++++++-- api/models.py | 4 + docs/advanced-chat-setup.md | 12 + server.py | 1 + ...test_gateway_run_reattach_after_restart.py | 255 +++++++++++++++++ 5 files changed, 500 insertions(+), 28 deletions(-) create mode 100644 tests/test_gateway_run_reattach_after_restart.py diff --git a/api/gateway_chat.py b/api/gateway_chat.py index c66175d5..d8470b08 100644 --- a/api/gateway_chat.py +++ b/api/gateway_chat.py @@ -564,12 +564,32 @@ 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 _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: @@ -582,6 +602,8 @@ def _run_gateway_runs_api_streaming( if api_key: headers["Authorization"] = f"Bearer {api_key}" headers["X-Hermes-Session-Key"] = f"webui:{session_id}" + # Durable run record on the gateway: GET /v1/runs/{id} keeps answering after either side restarts. + post_headers = {**headers, "Idempotency-Key": f"webui-{stream_id}"} message_content: Any = str(msg_text or "") if attachments: try: @@ -637,7 +659,7 @@ def _run_gateway_runs_api_streaming( req = urllib.request.Request( url_runs, data=json.dumps(run_body).encode("utf-8"), - headers=headers, + headers=post_headers, method="POST", ) update_active_run(stream_id, phase="gateway-request") @@ -652,6 +674,8 @@ def _run_gateway_runs_api_streaming( 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) @@ -682,22 +706,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"}: @@ -823,6 +835,172 @@ 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 + + +def _record_gateway_run(session_id: str, stream_id: str, run_id: str, **extra) -> None: + """Persist the gateway run id on the pending turn 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} + 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) + + +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, +): + """Follow a gateway run started by a previous WebUI process until it settles. + + Returns the same ``(final_text | None, usage)`` contract as + ``_run_gateway_runs_api_streaming``; token/tool events emitted before the + restart are not replayed, only the terminal result is written back. + """ + 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.HTTPError as exc: + if exc.code == 404: + raise RuntimeError("Gateway no longer has this run; its result could not be recovered after the WebUI restart.") from exc + failures += 1 + status = None + except (urllib.error.URLError, OSError, ValueError): + failures += 1 + status = None + if status is None: + if failures >= GATEWAY_REATTACH_MAX_POLL_FAILURES: + raise RuntimeError("Gateway became unreachable while waiting for a run that outlived the WebUI restart.") + else: + state = str(status.get("status") or "").strip().lower() + if state == "waiting_for_approval" and isinstance(status.get("approval"), dict): + approval = status["approval"] + 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}")) + usage = {k: v for k, v in _gateway_stream_usage(status).items() if v} + output = str(status.get("output") or "") + if output and stream_id in STREAM_PARTIAL_TEXT: + STREAM_PARTIAL_TEXT[stream_id] = output + return output, usage + 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. + + Must run before the server accepts requests: registering the stream first + keeps stale-pending repair from marking these still-running turns interrupted. + Never raises; startup must not depend on the gateway being reachable. + """ + from api import models as _models + from api.config import get_config + + try: + if not webui_gateway_chat_enabled(get_config()): + return [] + entries = json.loads(_models.SESSION_INDEX_FILE.read_bytes()) + except Exception: + logger.debug("gateway reattach: no readable session index", exc_info=True) + return [] + resumed: list[str] = [] + for entry in entries if isinstance(entries, list) else []: + if not isinstance(entry, dict) or not entry.get("active_stream_id"): + continue + sid = str(entry.get("session_id") or "") + 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 _resume_gateway_run_for_session(session) -> bool: + from api.config import create_stream_channel, register_session_writeback_owner, register_stream_owner + + run = getattr(session, "gateway_run", None) if session is not None else None + stream_id = str(getattr(session, "active_stream_id", None) or "") + run_id = str((run or {}).get("run_id") or "").strip() + if not stream_id or not run_id or run.get("stream_id") != stream_id: + return False + sid = session.session_id + 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_id": run_id, + }, + 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, @@ -846,6 +1024,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 @@ -895,8 +1074,13 @@ def _stream_writeback_is_current(session: Any, stream_id: str) -> bool: 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 (getattr(session, "gateway_run", None) 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 @@ -926,6 +1110,7 @@ def _run_gateway_chat_streaming( model_provider=None, goal_related=False, regeneration=False, + reattach_run_id=None, ): """Bridge a WebUI chat turn through Hermes Gateway's API server. @@ -1024,7 +1209,9 @@ 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_id) 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,16 +1262,28 @@ def _run_gateway_chat_streaming( if _gw_overrides.get("service_tier"): body_extras["service_tier"] = _gw_overrides["service_tier"] 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_id: + final_text, usage = _await_gateway_run_result( + session_id, stream_id, reattach_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=lambda run_id: _record_gateway_run( + session_id, stream_id, run_id, + regeneration=bool(regeneration), + goal_related=bool(goal_related), + ), + ) except Exception as exc: error_payload = _settle_gateway_terminal_error( session_id, @@ -1371,6 +1570,7 @@ def _run_gateway_chat_streaming( display = display[:-1] s.messages = display + [user_msg, assistant_msg] s.active_stream_id = None + s.gateway_run = None s.pending_user_message = None s.pending_attachments = None s.pending_started_at = None diff --git a/api/models.py b/api/models.py index 0322a97b..3947a665 100644 --- a/api/models.py +++ b/api/models.py @@ -1319,6 +1319,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 @@ -1424,6 +1425,8 @@ 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 + # Gateway /v1/runs id for the turn named by stream_id; lets a restarted WebUI reattach. + 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 @@ -1496,6 +1499,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 diff --git a/docs/advanced-chat-setup.md b/docs/advanced-chat-setup.md index 8fadd447..7e434401 100644 --- a/docs/advanced-chat-setup.md +++ b/docs/advanced-chat-setup.md @@ -119,6 +119,18 @@ 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 diff --git a/server.py b/server.py index 06fc9a21..6ce78685 100644 --- a/server.py +++ b/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' diff --git a/tests/test_gateway_run_reattach_after_restart.py b/tests/test_gateway_run_reattach_after_restart.py new file mode 100644 index 00000000..2d459104 --- /dev/null +++ b/tests/test_gateway_run_reattach_after_restart.py @@ -0,0 +1,255 @@ +"""A gateway-owned run must survive a WebUI restart instead of being marked interrupted.""" +from collections import OrderedDict +import io +import json +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.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", started_at=1.0): + """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 = started_at + 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 _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 = {} + + class _Resp: + def __init__(self, body=None, lines=None): + self._body, self._lines = body, lines or [] + + def __enter__(self): + return self + + def __exit__(self, *a): + return False + + def read(self, *_a): + return self._body + + def __iter__(self): + return iter(self._lines) + + def fake_urlopen(req, timeout=None): + if req.get_method() == "POST": + captured["post_headers"] = dict(req.header_items()) + return _Resp(body=b'{"run_id":"run_live"}') + # The run id is durable before the first event is relayed. + captured["persisted_at_events"] = json.loads( + (isolated_sessions / f"{s.session_id}.json").read_text() + ).get("gateway_run") + return _Resp(lines=[ + 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}" + assert captured["persisted_at_events"]["run_id"] == "run_live" + assert captured["persisted_at_events"]["stream_id"] == stream_id + saved = json.loads((isolated_sessions / f"{s.session_id}.json").read_text()) + 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 = json.loads((isolated_sessions / f"{sid}.json").read_text()) + 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_reports_run_the_gateway_no_longer_knows(isolated_sessions, monkeypatch): + sid, _stream_id = _orphaned_gateway_turn(run_id="run_gone") + + def fake_status(base_url, api_key, run_id): + raise urllib.error.HTTPError("http://gateway.local", 404, "not found", Message(), io.BytesIO(b"{}")) + + monkeypatch.setattr(gateway_chat, "_get_gateway_run_status", fake_status) + assert gateway_chat.resume_gateway_runs_after_restart() == [sid] + _wait_for_reattach_threads() + + saved = json.loads((isolated_sessions / f"{sid}.json").read_text()) + assert saved["messages"][-2]["content"] == "long task" + assert saved["messages"][-1]["_error"] is True + assert "could not be recovered after the WebUI restart" in saved["messages"][-1]["content"] + assert saved["active_stream_id"] is None + assert saved["gateway_run"] is None + + +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")] + saved = json.loads((isolated_sessions / f"{sid}.json").read_text()) + assert saved["messages"][-1]["content"] == "approved and done" + + +def test_reattach_is_default_off_without_gateway_backend(isolated_sessions, monkeypatch): + _orphaned_gateway_turn() + monkeypatch.delenv("HERMES_WEBUI_CHAT_BACKEND") + monkeypatch.setattr( + gateway_chat, "_get_gateway_run_status", + lambda *a, **k: pytest.fail("legacy backend must not poll the gateway"), + ) + assert gateway_chat.resume_gateway_runs_after_restart() == [] + + +def test_sidecar_without_gateway_run_is_left_to_stale_pending_repair(isolated_sessions, monkeypatch): + sid, stream_id = _orphaned_gateway_turn() + s = models.Session.load(sid) + assert s is not None + s.gateway_run = None + s.save() + monkeypatch.setattr( + gateway_chat, "_get_gateway_run_status", + lambda *a, **k: pytest.fail("no recorded run id: nothing to reattach"), + ) + 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 = json.loads((isolated_sessions / f"{sid}.json").read_text()) + 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"]) From 65e336fb16e87a52dd56850ea7701df988f60625 Mon Sep 17 00:00:00 2001 From: carlotestor Date: Wed, 23 Sep 2026 17:00:38 +0000 Subject: [PATCH 2/9] fix(gateway): reattach through the session's profile and scan sidecars, not the index - resolve the reattach gateway URL/key inside profile_scope_for_detached_worker(session.profile) and prefer the base_url persisted when the run was accepted - discover pending turns from the sidecars (head read skips idle files) instead of the derived _index.json, which can be missing, corrupt, or lag a save --- api/gateway_chat.py | 55 +++++++-- ...test_gateway_run_reattach_after_restart.py | 108 ++++++++++++++++++ 2 files changed, 155 insertions(+), 8 deletions(-) diff --git a/api/gateway_chat.py b/api/gateway_chat.py index d8470b08..f18ee584 100644 --- a/api/gateway_chat.py +++ b/api/gateway_chat.py @@ -839,6 +839,7 @@ _GATEWAY_RUN_TERMINAL_STATUSES = frozenset({"completed", "failed", "cancelled", 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, **extra) -> None: @@ -945,15 +946,12 @@ def resume_gateway_runs_after_restart() -> list[str]: try: if not webui_gateway_chat_enabled(get_config()): return [] - entries = json.loads(_models.SESSION_INDEX_FILE.read_bytes()) + candidates = _sidecars_with_active_stream(_models.SESSION_DIR) except Exception: - logger.debug("gateway reattach: no readable session index", exc_info=True) + logger.warning("gateway reattach: could not scan session sidecars", exc_info=True) return [] resumed: list[str] = [] - for entry in entries if isinstance(entries, list) else []: - if not isinstance(entry, dict) or not entry.get("active_stream_id"): - continue - sid = str(entry.get("session_id") or "") + for sid in candidates: try: if _resume_gateway_run_for_session(_models.Session.load_metadata_only(sid)): resumed.append(sid) @@ -964,6 +962,40 @@ def resume_gateway_runs_after_restart() -> list[str]: return resumed +def _sidecars_with_active_stream(session_dir) -> list[str]: + """Session ids whose sidecar may still carry a pending turn. + + Scans the sidecars rather than the derived index, which can lag a save. + A head read skips the common idle case without parsing the file. + """ + 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_reattach(session, run: dict) -> tuple[str, str]: + """Resolve the gateway that owns ``run`` using the session's profile, not the process default.""" + from api import profiles as _profiles + from api.config import get_config + + with _profiles.profile_scope_for_detached_worker( + getattr(session, "profile", None), "gateway reattach", logger_override=logger, + ): + base_url = _gateway_base_url(get_config()) + api_key = _gateway_api_key() + # The run lives on the gateway that accepted it, even if config changed since. + return str(run.get("base_url") or base_url).rstrip("/"), api_key + + def _resume_gateway_run_for_session(session) -> bool: from api.config import create_stream_channel, register_session_writeback_owner, register_stream_owner @@ -973,6 +1005,7 @@ def _resume_gateway_run_for_session(session) -> bool: if not stream_id or not run_id or run.get("stream_id") != stream_id: return False sid = session.session_id + endpoint = _gateway_endpoint_for_reattach(session, run) with STREAMS_LOCK: if stream_id in STREAMS: return False @@ -995,6 +1028,7 @@ def _resume_gateway_run_for_session(session) -> bool: "goal_related": bool(run.get("goal_related")), "regeneration": bool(run.get("regeneration")), "reattach_run_id": run_id, + "reattach_endpoint": endpoint, }, name=f"gateway-reattach-{stream_id[:12]}", daemon=True, @@ -1111,6 +1145,7 @@ def _run_gateway_chat_streaming( goal_related=False, regeneration=False, reattach_run_id=None, + reattach_endpoint=None, ): """Bridge a WebUI chat turn through Hermes Gateway's API server. @@ -1197,8 +1232,11 @@ def _run_gateway_chat_streaming( model=model, model_provider=model_provider, ) - base_url = _gateway_base_url(cfg) - api_key = _gateway_api_key() + if reattach_endpoint: + base_url, api_key = reattach_endpoint + else: + base_url = _gateway_base_url(cfg) + api_key = _gateway_api_key() try: from api.config import _main_model_request_overrides _gw_overrides = _main_model_request_overrides( @@ -1280,6 +1318,7 @@ def _run_gateway_chat_streaming( active_provider=(model_provider or ""), on_run_id=lambda run_id: _record_gateway_run( session_id, stream_id, run_id, + base_url=base_url, regeneration=bool(regeneration), goal_related=bool(goal_related), ), diff --git a/tests/test_gateway_run_reattach_after_restart.py b/tests/test_gateway_run_reattach_after_restart.py index 2d459104..0ddf872a 100644 --- a/tests/test_gateway_run_reattach_after_restart.py +++ b/tests/test_gateway_run_reattach_after_restart.py @@ -113,6 +113,7 @@ def test_runs_api_start_sends_idempotency_key_and_persists_run_id(isolated_sessi assert captured["post_headers"]["Idempotency-key"] == f"webui-{stream_id}" assert captured["persisted_at_events"]["run_id"] == "run_live" assert captured["persisted_at_events"]["stream_id"] == stream_id + assert captured["persisted_at_events"]["base_url"] == "http://gateway.local" saved = json.loads((isolated_sessions / f"{s.session_id}.json").read_text()) assert saved["gateway_run"] is None assert saved["active_stream_id"] is None @@ -253,3 +254,110 @@ def test_cancel_still_stops_a_reattached_run(isolated_sessions, monkeypatch): 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) + + +@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") + seen = [] + _poll_until_completed(monkeypatch, seen) + + assert gateway_chat.resume_gateway_runs_after_restart() == [sid] + _wait_for_reattach_threads() + + saved = json.loads((isolated_sessions / f"{sid}.json").read_text()) + 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 + + +def test_reattach_resolves_gateway_from_the_session_profile(isolated_sessions, monkeypatch): + """The process may restart under another default profile; poll the session profile's gateway.""" + import contextlib + import os + from api import profiles + + sid, stream_id = _orphaned_gateway_turn() + s = models.Session.load(sid) + s.profile = "work" + s.save(touch_updated_at=False) + models.SESSIONS.clear() + scopes = [] + + @contextlib.contextmanager + def fake_scope(profile_name, purpose="", logger_override=None): + scopes.append(profile_name) + env = {"work": ("http://work-gateway:8642", "work-key")}.get(profile_name) + old = {k: os.environ.get(k) for k in ("HERMES_WEBUI_GATEWAY_BASE_URL", "HERMES_WEBUI_GATEWAY_API_KEY")} + if env: + os.environ["HERMES_WEBUI_GATEWAY_BASE_URL"], os.environ["HERMES_WEBUI_GATEWAY_API_KEY"] = env + try: + yield + finally: + for k, v in old.items(): + if v is None: + os.environ.pop(k, None) + else: + os.environ[k] = v + + monkeypatch.setattr(profiles, "profile_scope_for_detached_worker", fake_scope) + monkeypatch.setenv("HERMES_WEBUI_GATEWAY_API_KEY", "default-key") + seen = [] + _poll_until_completed(monkeypatch, seen) + + assert gateway_chat.resume_gateway_runs_after_restart() == [sid] + _wait_for_reattach_threads() + + assert scopes == ["work"] + assert {(b, k) for b, k, _ in seen} == {("http://work-gateway:8642", "work-key")} + saved = json.loads((isolated_sessions / f"{sid}.json").read_text()) + assert saved["messages"][-1]["content"] == "answer" + + +def test_reattach_polls_the_gateway_that_accepted_the_run(isolated_sessions, monkeypatch): + sid, stream_id = _orphaned_gateway_turn() + s = models.Session.load(sid) + s.gateway_run = {**s.gateway_run, "base_url": "http://accepted-gateway:8642/"} + s.save(touch_updated_at=False) + models.SESSIONS.clear() + seen = [] + _poll_until_completed(monkeypatch, seen) + + gateway_chat.resume_gateway_runs_after_restart() + _wait_for_reattach_threads() + + assert {b for b, _k, _r in seen} == {"http://accepted-gateway:8642"} From 94b07588dda8e1306a9bd82af67e1e311ea1d66a Mon Sep 17 00:00:00 2001 From: carlotestor Date: Wed, 23 Sep 2026 17:08:31 +0000 Subject: [PATCH 3/9] fix(gateway): resolve reattach URL and key as one pair; fail fast on 401/403 Drop the persisted base_url: pairing a saved URL with the profile's current key could poll the old gateway with a new credential. Both now come from the session profile's current config, and an auth rejection settles the turn at once with an explicit error instead of 150 retries. --- api/gateway_chat.py | 14 +++++------ ...test_gateway_run_reattach_after_restart.py | 24 +++++++++++-------- 2 files changed, 20 insertions(+), 18 deletions(-) diff --git a/api/gateway_chat.py b/api/gateway_chat.py index f18ee584..0ae1d5c1 100644 --- a/api/gateway_chat.py +++ b/api/gateway_chat.py @@ -895,6 +895,8 @@ def _await_gateway_run_result( except urllib.error.HTTPError as exc: if exc.code == 404: raise RuntimeError("Gateway no longer has this run; its result could not be recovered after the WebUI restart.") from exc + if exc.code in (401, 403): + raise RuntimeError(f"Gateway rejected the WebUI credentials (HTTP {exc.code}) while reattaching to a run after the restart.") from exc failures += 1 status = None except (urllib.error.URLError, OSError, ValueError): @@ -982,18 +984,15 @@ def _sidecars_with_active_stream(session_dir) -> list[str]: return ids -def _gateway_endpoint_for_reattach(session, run: dict) -> tuple[str, str]: - """Resolve the gateway that owns ``run`` using the session's profile, not the process default.""" +def _gateway_endpoint_for_reattach(session) -> tuple[str, str]: + """Resolve gateway URL and key together from the session's profile, never the process default.""" from api import profiles as _profiles from api.config import get_config with _profiles.profile_scope_for_detached_worker( getattr(session, "profile", None), "gateway reattach", logger_override=logger, ): - base_url = _gateway_base_url(get_config()) - api_key = _gateway_api_key() - # The run lives on the gateway that accepted it, even if config changed since. - return str(run.get("base_url") or base_url).rstrip("/"), api_key + return _gateway_base_url(get_config()), _gateway_api_key() def _resume_gateway_run_for_session(session) -> bool: @@ -1005,7 +1004,7 @@ def _resume_gateway_run_for_session(session) -> bool: if not stream_id or not run_id or run.get("stream_id") != stream_id: return False sid = session.session_id - endpoint = _gateway_endpoint_for_reattach(session, run) + endpoint = _gateway_endpoint_for_reattach(session) with STREAMS_LOCK: if stream_id in STREAMS: return False @@ -1318,7 +1317,6 @@ def _run_gateway_chat_streaming( active_provider=(model_provider or ""), on_run_id=lambda run_id: _record_gateway_run( session_id, stream_id, run_id, - base_url=base_url, regeneration=bool(regeneration), goal_related=bool(goal_related), ), diff --git a/tests/test_gateway_run_reattach_after_restart.py b/tests/test_gateway_run_reattach_after_restart.py index 0ddf872a..9597ce46 100644 --- a/tests/test_gateway_run_reattach_after_restart.py +++ b/tests/test_gateway_run_reattach_after_restart.py @@ -113,7 +113,7 @@ def test_runs_api_start_sends_idempotency_key_and_persists_run_id(isolated_sessi assert captured["post_headers"]["Idempotency-key"] == f"webui-{stream_id}" assert captured["persisted_at_events"]["run_id"] == "run_live" assert captured["persisted_at_events"]["stream_id"] == stream_id - assert captured["persisted_at_events"]["base_url"] == "http://gateway.local" + assert "api_key" not in captured["persisted_at_events"] saved = json.loads((isolated_sessions / f"{s.session_id}.json").read_text()) assert saved["gateway_run"] is None assert saved["active_stream_id"] is None @@ -348,16 +348,20 @@ def test_reattach_resolves_gateway_from_the_session_profile(isolated_sessions, m assert saved["messages"][-1]["content"] == "answer" -def test_reattach_polls_the_gateway_that_accepted_the_run(isolated_sessions, monkeypatch): - sid, stream_id = _orphaned_gateway_turn() - s = models.Session.load(sid) - s.gateway_run = {**s.gateway_run, "base_url": "http://accepted-gateway:8642/"} - s.save(touch_updated_at=False) - models.SESSIONS.clear() - seen = [] - _poll_until_completed(monkeypatch, seen) +@pytest.mark.parametrize("code", [401, 403]) +def test_reattach_auth_rejection_fails_fast_with_an_auth_error(isolated_sessions, monkeypatch, code): + 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) gateway_chat.resume_gateway_runs_after_restart() _wait_for_reattach_threads() - assert {b for b, _k, _r in seen} == {"http://accepted-gateway:8642"} + assert len(calls) == 1 + saved = json.loads((isolated_sessions / f"{sid}.json").read_text()) + assert saved["active_stream_id"] is None and saved["gateway_run"] is None + assert f"HTTP {code}" in json.dumps(saved["messages"][-1]) From 5d590bdc7fdc5732b578c78142eaf49043741484 Mon Sep 17 00:00:00 2001 From: carlotestor Date: Wed, 23 Sep 2026 19:12:46 +0000 Subject: [PATCH 4/9] refactor: shrink gateway reattach diff without behaviour change Inline endpoint resolution, fold HTTP error branches, merge duplicate tests into parametrized cases, trim comments and docstrings. --- api/gateway_chat.py | 135 ++++++--------- api/models.py | 1 - docs/advanced-chat-setup.md | 12 +- ...test_gateway_run_reattach_after_restart.py | 159 +++++++----------- 4 files changed, 106 insertions(+), 201 deletions(-) diff --git a/api/gateway_chat.py b/api/gateway_chat.py index 0ae1d5c1..0086e055 100644 --- a/api/gateway_chat.py +++ b/api/gateway_chat.py @@ -602,8 +602,6 @@ def _run_gateway_runs_api_streaming( if api_key: headers["Authorization"] = f"Bearer {api_key}" headers["X-Hermes-Session-Key"] = f"webui:{session_id}" - # Durable run record on the gateway: GET /v1/runs/{id} keeps answering after either side restarts. - post_headers = {**headers, "Idempotency-Key": f"webui-{stream_id}"} message_content: Any = str(msg_text or "") if attachments: try: @@ -659,7 +657,8 @@ def _run_gateway_runs_api_streaming( req = urllib.request.Request( url_runs, data=json.dumps(run_body).encode("utf-8"), - headers=post_headers, + # Durable run record on the gateway: GET /v1/runs/{id} survives either side restarting. + headers={**headers, "Idempotency-Key": f"webui-{stream_id}"}, method="POST", ) update_active_run(stream_id, phase="gateway-request") @@ -873,12 +872,7 @@ def _await_gateway_run_result( session_id, stream_id, run_id, base_url, api_key, *, put_gateway_event, cancel_event, ): - """Follow a gateway run started by a previous WebUI process until it settles. - - Returns the same ``(final_text | None, usage)`` contract as - ``_run_gateway_runs_api_streaming``; token/tool events emitted before the - restart are not replayed, only the terminal result is written back. - """ + """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) @@ -892,55 +886,44 @@ def _await_gateway_run_result( try: status = _get_gateway_run_status(base_url, api_key, run_id) failures = 0 - except urllib.error.HTTPError as exc: - if exc.code == 404: + 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 exc.code in (401, 403): - raise RuntimeError(f"Gateway rejected the WebUI credentials (HTTP {exc.code}) while reattaching to a run after the 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 - status = None - except (urllib.error.URLError, OSError, ValueError): - failures += 1 - status = None - if status is None: if failures >= GATEWAY_REATTACH_MAX_POLL_FAILURES: - raise RuntimeError("Gateway became unreachable while waiting for a run that outlived the WebUI restart.") - else: - state = str(status.get("status") or "").strip().lower() - if state == "waiting_for_approval" and isinstance(status.get("approval"), dict): - approval = status["approval"] - 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", + 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 == "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}")) - usage = {k: v for k, v in _gateway_stream_usage(status).items() if v} - output = str(status.get("output") or "") - if output and stream_id in STREAM_PARTIAL_TEXT: - STREAM_PARTIAL_TEXT[stream_id] = output - return output, usage + 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. + """Reattach pending turns whose gateway run outlived the previous WebUI process; never raises. - Must run before the server accepts requests: registering the stream first - keeps stale-pending repair from marking these still-running turns interrupted. - Never raises; startup must not depend on the gateway being reachable. + Call before serving so stale-pending repair does not mark these turns interrupted. """ from api import models as _models from api.config import get_config @@ -965,11 +948,7 @@ def resume_gateway_runs_after_restart() -> list[str]: def _sidecars_with_active_stream(session_dir) -> list[str]: - """Session ids whose sidecar may still carry a pending turn. - - Scans the sidecars rather than the derived index, which can lag a save. - A head read skips the common idle case without parsing the file. - """ + """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("_"): @@ -984,27 +963,19 @@ def _sidecars_with_active_stream(session_dir) -> list[str]: return ids -def _gateway_endpoint_for_reattach(session) -> tuple[str, str]: - """Resolve gateway URL and key together from the session's profile, never the process default.""" - from api import profiles as _profiles - from api.config import get_config - - with _profiles.profile_scope_for_detached_worker( - getattr(session, "profile", None), "gateway reattach", logger_override=logger, - ): - return _gateway_base_url(get_config()), _gateway_api_key() - - def _resume_gateway_run_for_session(session) -> bool: - from api.config import create_stream_channel, register_session_writeback_owner, register_stream_owner + from api import profiles as _profiles + from api.config import create_stream_channel, get_config, register_session_writeback_owner, register_stream_owner - run = getattr(session, "gateway_run", None) if session is not None else None - stream_id = str(getattr(session, "active_stream_id", None) or "") - run_id = str((run or {}).get("run_id") or "").strip() - if not stream_id or not run_id or run.get("stream_id") != stream_id: + run = (session.gateway_run if session is not None else None) or {} + stream_id = str(run.get("stream_id") or "") + run_id = str(run.get("run_id") or "").strip() + if not stream_id or not run_id or session.active_stream_id != stream_id: return False sid = session.session_id - endpoint = _gateway_endpoint_for_reattach(session) + # URL and key come together from the session's profile, never the process default. + with _profiles.profile_scope_for_detached_worker(session.profile, "gateway reattach", logger_override=logger): + endpoint = (_gateway_base_url(get_config()), _gateway_api_key()) with STREAMS_LOCK: if stream_id in STREAMS: return False @@ -1014,14 +985,8 @@ def _resume_gateway_run_for_session(session) -> bool: _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 []), - ), + 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")), @@ -1108,7 +1073,7 @@ def _stream_writeback_is_current(session: Any, stream_id: str) -> bool: 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 (getattr(session, "gateway_run", None) or {}).get("stream_id") == stream_id: + 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 @@ -1231,11 +1196,7 @@ def _run_gateway_chat_streaming( model=model, model_provider=model_provider, ) - if reattach_endpoint: - base_url, api_key = reattach_endpoint - else: - 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()) try: from api.config import _main_model_request_overrides _gw_overrides = _main_model_request_overrides( @@ -1246,9 +1207,7 @@ def _run_gateway_chat_streaming( except Exception: _gw_overrides = {} _runs_api_enabled = _gateway_use_runs_api_enabled(cfg) - _use_runs_api = bool(reattach_run_id) or ( - _runs_api_enabled and gateway_supports_approval(base_url, api_key) - ) + _use_runs_api = bool(reattach_run_id) 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 diff --git a/api/models.py b/api/models.py index 3947a665..3f3c2853 100644 --- a/api/models.py +++ b/api/models.py @@ -1425,7 +1425,6 @@ 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 - # Gateway /v1/runs id for the turn named by stream_id; lets a restarted WebUI reattach. 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 diff --git a/docs/advanced-chat-setup.md b/docs/advanced-chat-setup.md index 7e434401..54e33ebd 100644 --- a/docs/advanced-chat-setup.md +++ b/docs/advanced-chat-setup.md @@ -119,17 +119,7 @@ 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. +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 diff --git a/tests/test_gateway_run_reattach_after_restart.py b/tests/test_gateway_run_reattach_after_restart.py index 9597ce46..a6a38cc6 100644 --- a/tests/test_gateway_run_reattach_after_restart.py +++ b/tests/test_gateway_run_reattach_after_restart.py @@ -1,10 +1,13 @@ """A gateway-owned run must survive a WebUI restart instead of being marked interrupted.""" +import contextlib from collections import OrderedDict import io import json +import os import threading import urllib.error from email.message import Message +from unittest import mock import pytest @@ -12,6 +15,7 @@ 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 @@ -34,7 +38,7 @@ def isolated_sessions(tmp_path, monkeypatch): return session_dir -def _orphaned_gateway_turn(run_id="run_survivor", stream_id="stream-before-restart", started_at=1.0): +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 = [ @@ -44,7 +48,7 @@ def _orphaned_gateway_turn(run_id="run_survivor", stream_id="stream-before-resta s.active_stream_id = stream_id s.pending_user_message = "long task" s.pending_attachments = [] - s.pending_started_at = started_at + 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() @@ -56,6 +60,10 @@ def _orphaned_gateway_turn(run_id="run_survivor", stream_id="stream-before-resta 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-"): @@ -73,35 +81,17 @@ def test_runs_api_start_sends_idempotency_key_and_persists_run_id(isolated_sessi s.save() captured = {} - class _Resp: - def __init__(self, body=None, lines=None): - self._body, self._lines = body, lines or [] - - def __enter__(self): - return self - - def __exit__(self, *a): - return False - - def read(self, *_a): - return self._body - - def __iter__(self): - return iter(self._lines) - def fake_urlopen(req, timeout=None): if req.get_method() == "POST": captured["post_headers"] = dict(req.header_items()) - return _Resp(body=b'{"run_id":"run_live"}') + return io.BytesIO(b'{"run_id":"run_live"}') # The run id is durable before the first event is relayed. - captured["persisted_at_events"] = json.loads( - (isolated_sessions / f"{s.session_id}.json").read_text() - ).get("gateway_run") - return _Resp(lines=[ - b'data: {"event":"message.delta","delta":"done"}\n', - b'data: {"event":"run.completed","output":"done"}\n', - b"data: [DONE]\n", - ]) + 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) @@ -111,10 +101,11 @@ def test_runs_api_start_sends_idempotency_key_and_persists_run_id(isolated_sessi 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}" - assert captured["persisted_at_events"]["run_id"] == "run_live" - assert captured["persisted_at_events"]["stream_id"] == stream_id - assert "api_key" not in captured["persisted_at_events"] - saved = json.loads((isolated_sessions / f"{s.session_id}.json").read_text()) + # 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" @@ -149,7 +140,7 @@ def test_restart_reattaches_and_writes_back_real_answer(isolated_sessions, monke release.set() _wait_for_reattach_threads() - saved = json.loads((isolated_sessions / f"{sid}.json").read_text()) + 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" @@ -161,24 +152,6 @@ def test_restart_reattaches_and_writes_back_real_answer(isolated_sessions, monke assert stream_id not in STREAMS -def test_reattach_reports_run_the_gateway_no_longer_knows(isolated_sessions, monkeypatch): - sid, _stream_id = _orphaned_gateway_turn(run_id="run_gone") - - def fake_status(base_url, api_key, run_id): - raise urllib.error.HTTPError("http://gateway.local", 404, "not found", Message(), io.BytesIO(b"{}")) - - monkeypatch.setattr(gateway_chat, "_get_gateway_run_status", fake_status) - assert gateway_chat.resume_gateway_runs_after_restart() == [sid] - _wait_for_reattach_threads() - - saved = json.loads((isolated_sessions / f"{sid}.json").read_text()) - assert saved["messages"][-2]["content"] == "long task" - assert saved["messages"][-1]["_error"] is True - assert "could not be recovered after the WebUI restart" in saved["messages"][-1]["content"] - assert saved["active_stream_id"] is None - assert saved["gateway_run"] is None - - def test_reattach_surfaces_pending_approval_once(isolated_sessions, monkeypatch): sid, stream_id = _orphaned_gateway_turn(run_id="run_parked") relayed = [] @@ -201,30 +174,19 @@ def test_reattach_surfaces_pending_approval_once(isolated_sessions, monkeypatch) _wait_for_reattach_threads() assert relayed == [(sid, "run_parked", "appr-1")] - saved = json.loads((isolated_sessions / f"{sid}.json").read_text()) - assert saved["messages"][-1]["content"] == "approved and done" + assert _saved(sid)["messages"][-1]["content"] == "approved and done" -def test_reattach_is_default_off_without_gateway_backend(isolated_sessions, monkeypatch): - _orphaned_gateway_turn() - monkeypatch.delenv("HERMES_WEBUI_CHAT_BACKEND") - monkeypatch.setattr( - gateway_chat, "_get_gateway_run_status", - lambda *a, **k: pytest.fail("legacy backend must not poll the gateway"), - ) - assert gateway_chat.resume_gateway_runs_after_restart() == [] - - -def test_sidecar_without_gateway_run_is_left_to_stale_pending_repair(isolated_sessions, monkeypatch): +@pytest.mark.parametrize("case", ["legacy_backend", "no_gateway_run"]) +def test_nothing_to_reattach_is_left_to_stale_pending_repair(isolated_sessions, monkeypatch, case): sid, stream_id = _orphaned_gateway_turn() - s = models.Session.load(sid) - assert s is not None - s.gateway_run = None - s.save() - monkeypatch.setattr( - gateway_chat, "_get_gateway_run_status", - lambda *a, **k: pytest.fail("no recorded run id: nothing to reattach"), - ) + if case == "legacy_backend": + monkeypatch.delenv("HERMES_WEBUI_CHAT_BACKEND") + else: + 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 @@ -250,18 +212,21 @@ def test_cancel_still_stops_a_reattached_run(isolated_sessions, monkeypatch): _wait_for_reattach_threads() assert stopped == ["run_to_stop"] - saved = json.loads((isolated_sessions / f"{sid}.json").read_text()) + 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 _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"]) @@ -275,13 +240,12 @@ def test_reattach_does_not_depend_on_the_session_index(isolated_sessions, monkey index.write_text(json.dumps([{"session_id": sid, "active_stream_id": None}])) else: index.write_text("{not json") - seen = [] - _poll_until_completed(monkeypatch, seen) + _poll_until_completed(monkeypatch) assert gateway_chat.resume_gateway_runs_after_restart() == [sid] _wait_for_reattach_threads() - saved = json.loads((isolated_sessions / f"{sid}.json").read_text()) + saved = _saved(sid) assert saved["messages"][-1]["content"] == "answer" assert saved["active_stream_id"] is None @@ -297,7 +261,7 @@ def test_reattach_skips_idle_sidecars_without_parsing_them(isolated_sessions, mo models.Session, "load_metadata_only", classmethod(lambda cls, s, **kw: loaded.append(s) or real_load.__func__(cls, s, **kw)), ) - _poll_until_completed(monkeypatch, []) + _poll_until_completed(monkeypatch) assert gateway_chat.resume_gateway_runs_after_restart() == [sid] _wait_for_reattach_threads() @@ -307,10 +271,6 @@ def test_reattach_skips_idle_sidecars_without_parsing_them(isolated_sessions, mo def test_reattach_resolves_gateway_from_the_session_profile(isolated_sessions, monkeypatch): """The process may restart under another default profile; poll the session profile's gateway.""" - import contextlib - import os - from api import profiles - sid, stream_id = _orphaned_gateway_turn() s = models.Session.load(sid) s.profile = "work" @@ -321,35 +281,30 @@ def test_reattach_resolves_gateway_from_the_session_profile(isolated_sessions, m @contextlib.contextmanager def fake_scope(profile_name, purpose="", logger_override=None): scopes.append(profile_name) - env = {"work": ("http://work-gateway:8642", "work-key")}.get(profile_name) - old = {k: os.environ.get(k) for k in ("HERMES_WEBUI_GATEWAY_BASE_URL", "HERMES_WEBUI_GATEWAY_API_KEY")} - if env: - os.environ["HERMES_WEBUI_GATEWAY_BASE_URL"], os.environ["HERMES_WEBUI_GATEWAY_API_KEY"] = env - try: + with mock.patch.dict(os.environ, { + "HERMES_WEBUI_GATEWAY_BASE_URL": f"http://{profile_name}-gateway:8642", + "HERMES_WEBUI_GATEWAY_API_KEY": f"{profile_name}-key", + }): yield - finally: - for k, v in old.items(): - if v is None: - os.environ.pop(k, None) - else: - os.environ[k] = v monkeypatch.setattr(profiles, "profile_scope_for_detached_worker", fake_scope) monkeypatch.setenv("HERMES_WEBUI_GATEWAY_API_KEY", "default-key") - seen = [] - _poll_until_completed(monkeypatch, seen) + seen = _poll_until_completed(monkeypatch) assert gateway_chat.resume_gateway_runs_after_restart() == [sid] _wait_for_reattach_threads() assert scopes == ["work"] assert {(b, k) for b, k, _ in seen} == {("http://work-gateway:8642", "work-key")} - saved = json.loads((isolated_sessions / f"{sid}.json").read_text()) - assert saved["messages"][-1]["content"] == "answer" + assert _saved(sid)["messages"][-1]["content"] == "answer" -@pytest.mark.parametrize("code", [401, 403]) -def test_reattach_auth_rejection_fails_fast_with_an_auth_error(isolated_sessions, monkeypatch, code): +@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 = [] @@ -358,10 +313,12 @@ def test_reattach_auth_rejection_fails_fast_with_an_auth_error(isolated_sessions 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) - gateway_chat.resume_gateway_runs_after_restart() + assert gateway_chat.resume_gateway_runs_after_restart() == [sid] _wait_for_reattach_threads() assert len(calls) == 1 - saved = json.loads((isolated_sessions / f"{sid}.json").read_text()) + 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 - assert f"HTTP {code}" in json.dumps(saved["messages"][-1]) From b9fb2f7065e69b28353bddbd423b45e1baa02fa9 Mon Sep 17 00:00:00 2001 From: carlotestor Date: Wed, 23 Sep 2026 21:56:14 +0000 Subject: [PATCH 5/9] fix(gateway): reattach under the session profile and recover runs admitted before the id was saved --- api/gateway_chat.py | 95 +++++++++++-------- ...test_gateway_run_reattach_after_restart.py | 82 +++++++++++++--- 2 files changed, 128 insertions(+), 49 deletions(-) diff --git a/api/gateway_chat.py b/api/gateway_chat.py index 0086e055..22c4ea0d 100644 --- a/api/gateway_chat.py +++ b/api/gateway_chat.py @@ -583,6 +583,35 @@ def _relay_gateway_run_approval(session_id, run_id, payload, base_url, api_key, 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, @@ -594,14 +623,7 @@ def _run_gateway_runs_api_streaming( """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: @@ -654,19 +676,11 @@ 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"), - # Durable run record on the gateway: GET /v1/runs/{id} survives either side restarting. - headers={**headers, "Idempotency-Key": f"webui-{stream_id}"}, - 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 @@ -841,14 +855,16 @@ 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, **extra) -> None: - """Persist the gateway run id on the pending turn so a restarted WebUI can reattach.""" +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) @@ -926,11 +942,8 @@ def resume_gateway_runs_after_restart() -> list[str]: Call before serving so stale-pending repair does not mark these turns interrupted. """ from api import models as _models - from api.config import get_config try: - if not webui_gateway_chat_enabled(get_config()): - return [] candidates = _sidecars_with_active_stream(_models.SESSION_DIR) except Exception: logger.warning("gateway reattach: could not scan session sidecars", exc_info=True) @@ -969,8 +982,8 @@ def _resume_gateway_run_for_session(session) -> bool: run = (session.gateway_run if session is not None else None) or {} stream_id = str(run.get("stream_id") or "") - run_id = str(run.get("run_id") or "").strip() - if not stream_id or not run_id or session.active_stream_id != stream_id: + # 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 # URL and key come together from the session's profile, never the process default. @@ -991,7 +1004,7 @@ def _resume_gateway_run_for_session(session) -> bool: "model_provider": session.model_provider, "goal_related": bool(run.get("goal_related")), "regeneration": bool(run.get("regeneration")), - "reattach_run_id": run_id, + "reattach_run": run, "reattach_endpoint": endpoint, }, name=f"gateway-reattach-{stream_id[:12]}", @@ -1108,7 +1121,7 @@ def _run_gateway_chat_streaming( model_provider=None, goal_related=False, regeneration=False, - reattach_run_id=None, + reattach_run=None, reattach_endpoint=None, ): """Bridge a WebUI chat turn through Hermes Gateway's API server. @@ -1207,7 +1220,7 @@ def _run_gateway_chat_streaming( except Exception: _gw_overrides = {} _runs_api_enabled = _gateway_use_runs_api_enabled(cfg) - _use_runs_api = bool(reattach_run_id) or (_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 @@ -1257,10 +1270,22 @@ 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: - if reattach_run_id: + 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, reattach_run_id, base_url, api_key, + session_id, stream_id, run_id, base_url, api_key, put_gateway_event=put_gateway_event, cancel_event=cancel_event, ) @@ -1274,11 +1299,7 @@ def _run_gateway_chat_streaming( cfg=cfg, session=s, active_provider=(model_provider or ""), - on_run_id=lambda run_id: _record_gateway_run( - session_id, stream_id, run_id, - regeneration=bool(regeneration), - goal_related=bool(goal_related), - ), + on_run_id=record_run, ) except Exception as exc: error_payload = _settle_gateway_terminal_error( diff --git a/tests/test_gateway_run_reattach_after_restart.py b/tests/test_gateway_run_reattach_after_restart.py index a6a38cc6..5ffb9426 100644 --- a/tests/test_gateway_run_reattach_after_restart.py +++ b/tests/test_gateway_run_reattach_after_restart.py @@ -177,15 +177,11 @@ def test_reattach_surfaces_pending_approval_once(isolated_sessions, monkeypatch) assert _saved(sid)["messages"][-1]["content"] == "approved and done" -@pytest.mark.parametrize("case", ["legacy_backend", "no_gateway_run"]) -def test_nothing_to_reattach_is_left_to_stale_pending_repair(isolated_sessions, monkeypatch, case): +def test_nothing_to_reattach_is_left_to_stale_pending_repair(isolated_sessions, monkeypatch): sid, stream_id = _orphaned_gateway_turn() - if case == "legacy_backend": - monkeypatch.delenv("HERMES_WEBUI_CHAT_BACKEND") - else: - s = models.Session.load(sid) - s.gateway_run = None - s.save() + 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 @@ -269,8 +265,12 @@ def test_reattach_skips_idle_sidecars_without_parsing_them(isolated_sessions, mo assert sid in loaded -def test_reattach_resolves_gateway_from_the_session_profile(isolated_sessions, monkeypatch): - """The process may restart under another default profile; poll the session profile's gateway.""" +@pytest.mark.parametrize("default_backend", ["gateway", "local"]) +def test_reattach_resolves_gateway_from_the_session_profile(isolated_sessions, monkeypatch, default_backend): + """The process may restart under another default profile, even a local one; poll the session profile's gateway.""" + 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 = "work" @@ -295,8 +295,66 @@ def test_reattach_resolves_gateway_from_the_session_profile(isolated_sessions, m _wait_for_reattach_threads() assert scopes == ["work"] - assert {(b, k) for b, k, _ in seen} == {("http://work-gateway:8642", "work-key")} - assert _saved(sid)["messages"][-1]["content"] == "answer" + assert {(b, k, r) for b, k, r in seen} == {("http://work-gateway:8642", "work-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", [ From 5fdc881799b9667fd8d620b602c19fb248990f24 Mon Sep 17 00:00:00 2001 From: carlotestor Date: Wed, 23 Sep 2026 22:05:09 +0000 Subject: [PATCH 6/9] fix(gateway): do not admit a run whose recovery checkpoint failed to save --- api/gateway_chat.py | 2 ++ .../test_gateway_run_reattach_after_restart.py | 18 ++++++++++++++++++ 2 files changed, 20 insertions(+) diff --git a/api/gateway_chat.py b/api/gateway_chat.py index 22c4ea0d..1ce74e9a 100644 --- a/api/gateway_chat.py +++ b/api/gateway_chat.py @@ -868,6 +868,8 @@ def _record_gateway_run(session_id: str, stream_id: str, run_id: str, request=No 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: diff --git a/tests/test_gateway_run_reattach_after_restart.py b/tests/test_gateway_run_reattach_after_restart.py index 5ffb9426..7845ab53 100644 --- a/tests/test_gateway_run_reattach_after_restart.py +++ b/tests/test_gateway_run_reattach_after_restart.py @@ -380,3 +380,21 @@ def test_reattach_http_rejection_settles_the_turn_with_an_error(isolated_session 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 From 00973abe0c2404e7df2cf65ec1f07f53c8e8b05a Mon Sep 17 00:00:00 2001 From: carlotestor Date: Thu, 24 Sep 2026 15:18:53 +0000 Subject: [PATCH 7/9] fix(gateway): reattach with the session profile's own endpoint (root too); keep the prompt when a replayed run ends cancelled --- api/gateway_chat.py | 32 +++++++-- ...test_gateway_run_reattach_after_restart.py | 72 +++++++++++++------ 2 files changed, 79 insertions(+), 25 deletions(-) diff --git a/api/gateway_chat.py b/api/gateway_chat.py index 1daa497f..a846704d 100644 --- a/api/gateway_chat.py +++ b/api/gateway_chat.py @@ -979,9 +979,19 @@ def _sidecars_with_active_stream(session_dir) -> list[str]: return ids -def _resume_gateway_run_for_session(session) -> bool: +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 create_stream_channel, get_config, register_session_writeback_owner, register_stream_owner + 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 "") @@ -989,9 +999,7 @@ def _resume_gateway_run_for_session(session) -> bool: 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 - # URL and key come together from the session's profile, never the process default. - with _profiles.profile_scope_for_detached_worker(session.profile, "gateway reattach", logger_override=logger): - endpoint = (_gateway_base_url(get_config()), _gateway_api_key()) + endpoint = _gateway_endpoint_for_profile(session.profile) with STREAMS_LOCK: if stream_id in STREAMS: return False @@ -1082,6 +1090,19 @@ 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) @@ -1318,6 +1339,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, diff --git a/tests/test_gateway_run_reattach_after_restart.py b/tests/test_gateway_run_reattach_after_restart.py index 7845ab53..4fef0461 100644 --- a/tests/test_gateway_run_reattach_after_restart.py +++ b/tests/test_gateway_run_reattach_after_restart.py @@ -1,5 +1,4 @@ """A gateway-owned run must survive a WebUI restart instead of being marked interrupted.""" -import contextlib from collections import OrderedDict import io import json @@ -7,7 +6,6 @@ import os import threading import urllib.error from email.message import Message -from unittest import mock import pytest @@ -266,36 +264,40 @@ def test_reattach_skips_idle_sidecars_without_parsing_them(isolated_sessions, mo @pytest.mark.parametrize("default_backend", ["gateway", "local"]) -def test_reattach_resolves_gateway_from_the_session_profile(isolated_sessions, monkeypatch, default_backend): - """The process may restart under another default profile, even a local one; poll the session profile's gateway.""" +@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 = "work" + s.profile = session_profile s.save(touch_updated_at=False) models.SESSIONS.clear() - scopes = [] - - @contextlib.contextmanager - def fake_scope(profile_name, purpose="", logger_override=None): - scopes.append(profile_name) - with mock.patch.dict(os.environ, { - "HERMES_WEBUI_GATEWAY_BASE_URL": f"http://{profile_name}-gateway:8642", - "HERMES_WEBUI_GATEWAY_API_KEY": f"{profile_name}-key", - }): - yield - - monkeypatch.setattr(profiles, "profile_scope_for_detached_worker", fake_scope) - monkeypatch.setenv("HERMES_WEBUI_GATEWAY_API_KEY", "default-key") 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 scopes == ["work"] - assert {(b, k, r) for b, k, r in seen} == {("http://work-gateway:8642", "work-key", "run_survivor")} + 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 @@ -398,3 +400,33 @@ def test_failed_recovery_checkpoint_never_admits_the_run(isolated_sessions, monk 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() == [] From cfe13b5fc28eb85ead96088f53b29a8592e6327b Mon Sep 17 00:00:00 2001 From: carlotestor Date: Thu, 24 Sep 2026 16:56:58 +0000 Subject: [PATCH 8/9] fix(gateway): send Stop and approval replies to the Gateway that owns a reattached run stop_gateway_run() and the /api/approval/respond relay resolved the process-active profile's Gateway, while a reattached run is polled on its session profile's Gateway. Record the endpoint per stream when the worker resolves it and look it up by run id; fall back to the process Gateway only when no live stream holds the run. --- api/gateway_chat.py | 25 ++++- api/routes.py | 7 +- ...test_gateway_run_reattach_after_restart.py | 98 +++++++++++++++++++ 3 files changed, 121 insertions(+), 9 deletions(-) diff --git a/api/gateway_chat.py b/api/gateway_chat.py index a846704d..9f5934e1 100644 --- a/api/gateway_chat.py +++ b/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: @@ -820,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}" @@ -1234,6 +1245,8 @@ def _run_gateway_chat_streaming( model_provider=model_provider, ) 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( @@ -1778,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 diff --git a/api/routes.py b/api/routes.py index a81660f8..c8bc62d6 100644 --- a/api/routes.py +++ b/api/routes.py @@ -26943,8 +26943,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() @@ -26986,8 +26986,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) ) diff --git a/tests/test_gateway_run_reattach_after_restart.py b/tests/test_gateway_run_reattach_after_restart.py index 4fef0461..267c51cb 100644 --- a/tests/test_gateway_run_reattach_after_restart.py +++ b/tests/test_gateway_run_reattach_after_restart.py @@ -430,3 +430,101 @@ def test_cancelled_replayed_admission_keeps_the_prompt_and_is_not_reattached(iso 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) From ae05896eb5c0836f34d0477aecd5273c0d22f42c Mon Sep 17 00:00:00 2001 From: nesquena-hermes Date: Thu, 24 Sep 2026 19:15:19 +0000 Subject: [PATCH 9/9] docs(changelog): #7785 gateway turns survive a WebUI restart --- CHANGELOG.md | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 671eb877..4fc39cd9 100644 --- a/CHANGELOG.md +++ b/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