"""Default-off Hermes Gateway bridge for browser-originated chat turns.""" from __future__ import annotations import json import logging import os import threading import time import urllib.error import urllib.request from typing import Any from api.config import ( CANCEL_FLAGS, PENDING_GOAL_CONTINUATION, STREAM_GOAL_RELATED, STREAMS, STREAMS_LOCK, STREAM_LAST_EVENT_ID, STREAM_LIVE_TOOL_CALLS, STREAM_PARTIAL_TEXT, STREAM_REASONING_TEXT, _get_session_agent_lock, coerce_reasoning_effort_for_model, gateway_approval_unavailable_reason, gateway_supports_approval, register_active_run, unregister_active_run, unregister_stream_owner, update_active_run, ) from api.helpers import _redact_text, redact_session_data from api.models import clear_process_wakeup_pause, get_session, merge_session_messages_append_only from api.run_journal import RunJournalWriter logger = logging.getLogger(__name__) # Maps stream_id -> gateway run_id for approval response relay. _STREAM_RUN_IDS: dict[str, str] = {} _WEBUI_CHAT_BACKEND_ENV = "HERMES_WEBUI_CHAT_BACKEND" _WEBUI_GATEWAY_BASE_URL_ENV = "HERMES_WEBUI_GATEWAY_BASE_URL" _WEBUI_GATEWAY_API_KEY_ENV = "HERMES_WEBUI_GATEWAY_API_KEY" _WEBUI_GATEWAY_USE_RUNS_API_ENV = "HERMES_WEBUI_GATEWAY_USE_RUNS_API" _GATEWAY_CHAT_BACKENDS = {"gateway", "api_server", "api-server"} # Total byte-silence budget (seconds) for the gateway SSE socket, applied via # ``urlopen(timeout=...)``. A stream that emits ANY byte within the window never # trips it, so a genuinely alive (if slow) token stream is untouched; only more # than this much *total* byte-silence is treated as a dead/stalled gateway. # # This is a TERMINAL budget, not a per-read grace: CPython's ``socket.makefile`` # latches ``_timeout_occurred`` on the first ``socket.timeout``, after which every # further read raises a bare ``OSError`` — the connection cannot be resumed. So a # read timeout ends the turn (surfacing Stop if pressed). It replaces the old flat # 600s timeout, under which a half-open gateway (TCP open, zero bytes) pinned the # worker for the full 10 minutes and ignored Stop (cancel is only re-checked # between SSE lines). We KEEP the 600s default budget: there is no Gateway # protocol heartbeat guaranteeing sub-600s progress bytes, so a legitimately # long/fully-silent server-side tool call must not be terminated early — reducing # the default below 600s would kill currently-working turns (gate finding, #5789). # The win here is that a read timeout is now TERMINAL and Stop-honoring (the old # flat timeout ignored Stop on a half-open gateway); the budget itself stays 600s # for backward compatibility. Deployments that want a tighter dead-gateway cap can # lower ``HERMES_WEBUI_GATEWAY_READ_TIMEOUT``. _GATEWAY_READ_TIMEOUT_ENV = "HERMES_WEBUI_GATEWAY_READ_TIMEOUT" _GATEWAY_READ_TIMEOUT_DEFAULT = 600.0 def _gateway_read_timeout_secs() -> float: """Total byte-silence budget for gateway SSE reads (default 600s, env-tunable).""" raw = os.environ.get(_GATEWAY_READ_TIMEOUT_ENV) if raw: try: val = float(raw) if val > 0: return val except (TypeError, ValueError): pass return _GATEWAY_READ_TIMEOUT_DEFAULT def _iter_sse_lines_cancellable(resp, cancel_event): """Yield raw SSE lines from ``resp``, unblocking cleanly on a read timeout. ``resp``'s socket carries a read timeout (``urlopen(timeout=...)``). A read that blocks past it raises ``socket.timeout``, and that timeout is TERMINAL: CPython's ``socket.makefile`` latches ``_timeout_occurred`` on the first timeout, so every subsequent read raises a bare ``OSError`` ("cannot read from timed out object") — there is no multi-read grace to reclaim. So on a read timeout (or the poisoned-socket ``OSError``/any read error) this either surfaces the user's Stop or tears the stalled turn down: - cancel set -> yield ``b""`` (the caller's ``if cancel_event.is_set()`` branch emits its cancel event), then stop. This is why the old flat 600s pin — where a stalled gateway ignored Stop until it eventually errored — is gone: Stop is honored within one timeout window. - otherwise -> re-raise, so the caller's error handling reports the stall. A stream that keeps emitting bytes within the timeout window never trips it, so a genuinely alive (if slow) token stream is untouched. Emitting ``b""`` is safe: the SSE loops decode it to an empty line and ``continue`` (same as a real blank line). Iterates ``resp`` via the iterator protocol so a real ``HTTPResponse`` and the test fakes (which implement ``__iter__``) behave identically. """ resp_iter = iter(resp) while True: try: raw_line = next(resp_iter) except StopIteration: return # EOF except OSError: # socket.timeout / TimeoutError are OSError subclasses, as is the # post-timeout poisoned-socket "cannot read" error. All are terminal # for this connection. if cancel_event.is_set(): yield b"" # let the caller emit its cancel event return raise yield raw_line def webui_chat_backend_mode(config_data=None, environ: dict[str, str] | None = None) -> str: """Return the explicitly selected browser chat backend. The default remains the in-process WebUI runtime. Only explicit gateway values opt browser chat into the Hermes API server bridge; generic truthy strings are deliberately ignored so deployments do not change execution ownership by accident. """ source = os.environ if environ is None else environ cfg = config_data if isinstance(config_data, dict) else {} raw = str( source.get(_WEBUI_CHAT_BACKEND_ENV) or cfg.get("webui_chat_backend") or "" ).strip().lower() if raw in _GATEWAY_CHAT_BACKENDS: return "gateway" return "legacy" def webui_gateway_chat_enabled(config_data=None, environ: dict[str, str] | None = None) -> bool: return webui_chat_backend_mode(config_data, environ) == "gateway" def _gateway_base_url(config_data=None, environ: dict[str, str] | None = None) -> str: source = os.environ if environ is None else environ cfg = config_data if isinstance(config_data, dict) else {} raw = str( source.get(_WEBUI_GATEWAY_BASE_URL_ENV) or cfg.get("webui_gateway_base_url") or "http://127.0.0.1:8642" ).strip() return raw.rstrip("/") or "http://127.0.0.1:8642" def _gateway_api_key(environ: dict[str, str] | None = None) -> str: source = os.environ if environ is None else environ return str( source.get(_WEBUI_GATEWAY_API_KEY_ENV) or source.get("API_SERVER_KEY") or "" ).strip() def _gateway_use_runs_api_enabled(config_data=None, environ: dict[str, str] | None = None) -> bool: """Return True only when the operator has explicitly opted into the runs API path.""" source = os.environ if environ is None else environ cfg = config_data if isinstance(config_data, dict) else {} raw = str( source.get(_WEBUI_GATEWAY_USE_RUNS_API_ENV) or cfg.get("webui_gateway_use_runs_api") or "" ).strip().lower() return raw in ("1", "true", "yes", "on") def _gateway_reasoning_effort_for_request(cfg, *, model=None, model_provider=None): """Read and coerce user-configured reasoning effort for a gateway request.""" try: cfg_data = cfg if isinstance(cfg, dict) else {} effort_cfg = cfg_data.get("agent", {}) if isinstance(cfg_data, dict) else {} effort_raw = effort_cfg.get("reasoning_effort") if isinstance(effort_cfg, dict) else None coerced = coerce_reasoning_effort_for_model( effort_raw, model, provider_id=model_provider, ) # Preserve explicit "none" while still omitting absent or invalid effort. return None if not coerced else str(coerced) except Exception: return None def gateway_chat_config_status(config_data=None, environ: dict[str, str] | None = None) -> dict: """Return redacted Gateway-backed chat configuration status.""" mode = webui_chat_backend_mode(config_data, environ) base_url = _gateway_base_url(config_data, environ) return { "enabled": mode == "gateway", "backend": mode, "base_url_configured": bool(base_url), "api_key_configured": bool(_gateway_api_key(environ)), } def _gateway_http_error_event(exc: urllib.error.HTTPError, err_body: str, *, api_key_configured: bool) -> dict: safe = _redact_text(err_body or str(exc))[:500] if exc.code == 401: return { "label": "Gateway authentication failed", "type": "gateway_auth_error", "message": "Gateway rejected the WebUI API key (HTTP 401).", "hint": ( "Set HERMES_WEBUI_GATEWAY_API_KEY to the same value as the Hermes Gateway " "API_SERVER_KEY, or disable HERMES_WEBUI_CHAT_BACKEND=gateway." if not api_key_configured else "Check that HERMES_WEBUI_GATEWAY_API_KEY matches the Hermes Gateway API_SERVER_KEY." ), } return { "label": "Gateway request failed", "type": "gateway_http_error", "message": f"Gateway returned HTTP {exc.code}.", "hint": safe or "Check the configured Gateway API server.", } def _gateway_sse_delta(payload: dict) -> str: """Extract assistant text from an OpenAI-compatible streaming chunk.""" try: choices = payload.get("choices") or [] if not choices: return "" choice = choices[0] or {} delta = choice.get("delta") or {} content = delta.get("content") if isinstance(content, str): return content message = choice.get("message") or {} content = message.get("content") return content if isinstance(content, str) else "" except Exception: return "" def _gateway_sse_reasoning_delta(payload: dict) -> str: """Extract reasoning text from OpenAI-compatible streaming chunks.""" try: choices = payload.get("choices") or [] if not choices: return "" choice = choices[0] or {} delta = choice.get("delta") or {} reasoning = delta.get("reasoning_content") if isinstance(reasoning, str) and reasoning.strip(): return reasoning message = choice.get("message") or {} reasoning = message.get("reasoning_content") return reasoning if isinstance(reasoning, str) and reasoning.strip() else "" except Exception: return "" def _gateway_stream_usage(payload: dict) -> dict: usage = payload.get("usage") if isinstance(payload, dict) else None if not isinstance(usage, dict): return {} return { "input_tokens": int(usage.get("prompt_tokens") or usage.get("input_tokens") or 0), "output_tokens": int(usage.get("completion_tokens") or usage.get("output_tokens") or 0), "estimated_cost": usage.get("estimated_cost") or usage.get("estimated_cost_usd") or 0, } def _gateway_reasoning_delta(payload: dict) -> str: if not isinstance(payload, dict): return "" for key in ("text", "preview", "delta", "content"): value = payload.get(key) if isinstance(value, str) and value.strip(): return value return "" def _gateway_tool_progress_event(payload: dict) -> tuple[str, dict] | None: """Translate Hermes Gateway tool-progress SSE payloads to WebUI events.""" if not isinstance(payload, dict): return None event_type = str(payload.get("event") or "").strip().lower() if event_type == "reasoning.available": reason_delta = _gateway_reasoning_delta(payload) if not reason_delta: return None return "reasoning", {"text": reason_delta} name = str(payload.get("tool") or payload.get("name") or payload.get("function_name") or "").strip() if not name: return None if name == "_thinking": reason_delta = _gateway_reasoning_delta(payload) if not reason_delta: return None return "reasoning", {"text": reason_delta} if name.startswith("_"): return None status = str(payload.get("status") or "running").strip().lower() tid = payload.get("toolCallId") or payload.get("tool_call_id") or payload.get("id") is_complete = event_type == "tool.completed" or status in {"completed", "complete", "success", "error", "failed"} event_payload = { "event_type": "tool.completed" if is_complete else "tool.started", "name": name, "preview": payload.get("label") or payload.get("preview"), "args": payload.get("args") if isinstance(payload.get("args"), dict) else {}, "is_error": bool(payload.get("error")) or status in {"error", "failed"}, } if tid: event_payload["tid"] = str(tid) return ("tool_complete" if is_complete else "tool"), event_payload def _gateway_runs_approval_event(payload: dict) -> dict | None: """Map a runs-API approval.request payload to the WebUI approval contract.""" if not isinstance(payload, dict): return None tool = str(payload.get("tool") or payload.get("function_name") or payload.get("pattern_key") or "").strip() command = str(payload.get("command") or "").strip() description = str(payload.get("description") or "").strip() pattern_keys = payload.get("pattern_keys") if isinstance(payload.get("pattern_keys"), list) else [] pattern_key = str(payload.get("pattern_key") or "").strip() args = payload.get("args") if isinstance(payload.get("args"), (list, dict)) else [] run_id = str(payload.get("run_id") or "").strip() approval_id = str(payload.get("approval_id") or payload.get("id") or "").strip() risk = str(payload.get("risk_level") or "high").strip() choices = payload.get("choices") if isinstance(payload.get("choices"), list) else [] allow_permanent = payload.get("allow_permanent") if allow_permanent is None: allow_permanent = "always" in choices if not (tool or command or description): return None return { "tool": tool, "command": command, "description": description, "pattern_key": pattern_key, "pattern_keys": pattern_keys or ([pattern_key] if pattern_key else []), "args": args, "risk_level": risk, "run_id": run_id, "approval_id": approval_id, "choices": choices, "allow_permanent": bool(allow_permanent), } 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, ): """Submit via POST /v1/runs and relay SSE events including approval.""" 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}" message_content: Any = str(msg_text or "") if attachments: try: from api.streaming import _build_native_multimodal_message message_content = _build_native_multimodal_message("", str(msg_text or ""), attachments, str(workspace), cfg=cfg) except Exception: logger.debug("Failed to build runs-API multimodal attachment payload", exc_info=True) message_content = str(msg_text or "") from api.streaming import _strip_oob_blocks instructions_parts = [] conversation_history = [] for entry in getattr(session, "context_messages", None) or []: if not isinstance(entry, dict): continue role = str(entry.get("role") or "").strip().lower() if role not in {"user", "assistant"}: continue content = entry.get("content") if content is not None: content = _strip_oob_blocks(content) conversation_history.append({"role": role, "content": content}) for entry in prefill_messages or []: if not isinstance(entry, dict): continue role = str(entry.get("role") or "").strip().lower() content = entry.get("content") if role == "system": if isinstance(content, str) and content.strip(): instructions_parts.append(content) elif content is not None: instructions_parts.append(str(content)) continue if role not in {"user", "assistant"}: continue if content is not None: content = _strip_oob_blocks(content) conversation_history.append({"role": role, "content": content}) run_input = message_content if isinstance(run_input, list): run_input = [{"role": "user", "content": run_input}] run_body = { "model": model or "default", "input": run_input, **body_extras, "session_id": session_id, } if instructions_parts: run_body["instructions"] = "\n\n".join(part for part in instructions_parts if part) if conversation_history: run_body["conversation_history"] = conversation_history req = urllib.request.Request( url_runs, data=json.dumps(run_body).encode("utf-8"), headers=headers, method="POST", ) update_active_run(stream_id, phase="gateway-request") with urllib.request.urlopen(req, timeout=30) as resp: run_data = json.loads(resp.read(65536)) run_id = str(run_data.get("run_id") or run_data.get("id") or "").strip() if not run_id: raise ValueError(f"Gateway runs API returned no run_id: {run_data!r}") _STREAM_RUN_IDS[stream_id] = run_id url_events = f"{base_url.rstrip('/')}/v1/runs/{run_id}/events" headers_sse = dict(headers) headers_sse["Accept"] = "text/event-stream" req_events = urllib.request.Request(url_events, headers=headers_sse, method="GET") final_text = "" usage: dict = {} sse_event = "message" with urllib.request.urlopen(req_events, timeout=_gateway_read_timeout_secs()) as resp: for raw_line in _iter_sse_lines_cancellable(resp, cancel_event): if cancel_event.is_set(): put_gateway_event("cancel", {"message": "Cancelled by user"}) return None, usage line = raw_line.decode("utf-8", errors="replace").strip() if not line: sse_event = "message" continue if line.startswith("event:"): sse_event = line[6:].strip() or "message" continue if not line.startswith("data:"): continue data = line[5:].strip() if data == "[DONE]": break try: payload = json.loads(data) except json.JSONDecodeError: 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 put_gateway_event("approval", approval_data) sse_event = "message" continue if payload_event in {"tool.started", "tool.completed", "reasoning.available"}: translated = _gateway_tool_progress_event(payload) if translated: event_name, event_payload = translated if event_name == "reasoning": reason_delta = event_payload.get("text") if reason_delta and stream_id in STREAM_REASONING_TEXT: STREAM_REASONING_TEXT[stream_id] += reason_delta elif stream_id in STREAM_LIVE_TOOL_CALLS: if event_name == "tool": STREAM_LIVE_TOOL_CALLS[stream_id].append({ "name": event_payload.get("name"), "args": event_payload.get("args") or {}, "done": False, **({"tid": event_payload.get("tid")} if event_payload.get("tid") else {}), }) elif event_name == "tool_complete": for shared_tc in reversed(STREAM_LIVE_TOOL_CALLS[stream_id]): if shared_tc.get("done"): continue if ( event_payload.get("tid") and shared_tc.get("tid") == event_payload.get("tid") ) or shared_tc.get("name") == event_payload.get("name"): shared_tc["done"] = True shared_tc["is_error"] = bool(event_payload.get("is_error")) break put_gateway_event(event_name, event_payload) if event_name != "reasoning": update_active_run(stream_id, phase="gateway-tool", latest_tool=event_payload.get("name")) sse_event = "message" continue if payload_event == "message.delta": delta = str(payload.get("delta") or "") if delta: final_text += delta if stream_id in STREAM_PARTIAL_TEXT: STREAM_PARTIAL_TEXT[stream_id] += delta put_gateway_event("token", {"text": delta}) sse_event = "message" continue if payload_event == "run.completed": if payload.get("error"): raise RuntimeError(str(payload["error"])) output = str(payload.get("output") or "") if output and not final_text: final_text = output if stream_id in STREAM_PARTIAL_TEXT: STREAM_PARTIAL_TEXT[stream_id] = output usage.update({k: v for k, v in _gateway_stream_usage(payload).items() if v}) sse_event = "message" continue if payload_event == "run.failed": raise RuntimeError(str(payload.get("error") or "Gateway run failed")) if payload_event == "run.cancelled": put_gateway_event("cancel", {"message": "Cancelled by gateway"}) return None, usage reasoning_delta = _gateway_sse_reasoning_delta(payload) if reasoning_delta: if stream_id in STREAM_REASONING_TEXT: STREAM_REASONING_TEXT[stream_id] += reasoning_delta put_gateway_event("reasoning", {"text": reasoning_delta}) delta = _gateway_sse_delta(payload) if delta: final_text += delta if stream_id in STREAM_PARTIAL_TEXT: STREAM_PARTIAL_TEXT[stream_id] += delta put_gateway_event("token", {"text": delta}) usage.update({k: v for k, v in _gateway_stream_usage(payload).items() if v}) return final_text, usage def _settle_gateway_terminal_error(session_id, stream_id, workspace, model, model_provider, terminal_error): from api.streaming import ( _classify_provider_error, _materialize_pending_user_turn_before_error, _provider_error_payload, _session_payload_with_full_messages, _snapshot_and_append_partial_on_error, ) with _get_session_agent_lock(session_id): session = get_session(session_id) if not _stream_writeback_is_current(session, stream_id): return None error_classification = _classify_provider_error(terminal_error) error_payload = _provider_error_payload( terminal_error, error_classification["type"], error_classification.get("hint", ""), ) _materialize_pending_user_turn_before_error(session) session.active_stream_id = None session.pending_user_message = None session.pending_attachments = [] session.pending_started_at = None session.pending_user_source = None try: _snapshot_and_append_partial_on_error(session, stream_id) except Exception: logger.debug("Failed to snapshot gateway partials on terminal error", exc_info=True) error_message = { "role": "assistant", "content": ( f"**{error_classification['label']}:** " f"{error_payload.get('message') or error_classification['label']}" ) + (f"\n\n*{error_payload['hint']}*" if error_payload.get("hint") else ""), "timestamp": int(time.time()), "_error": True, } if error_payload.get("details"): error_message["provider_details"] = error_payload["details"] if not isinstance(session.messages, list): session.messages = [] session.messages.append(error_message) session.workspace = str(workspace) session.model = model session.model_provider = model_provider try: session.save() except Exception: logger.debug("Failed to persist gateway terminal error settlement", exc_info=True) error_payload["session"] = redact_session_data( _session_payload_with_full_messages(session, tool_calls=[]) ) error_payload["session_id"] = session.session_id return error_payload def _stream_writeback_is_current(session: Any, stream_id: str) -> bool: return bool(stream_id and getattr(session, "active_stream_id", None) == stream_id) def _clear_gateway_pending_state(session: Any, stream_id: str) -> None: if not _stream_writeback_is_current(session, stream_id): return session.active_stream_id = None session.pending_user_message = None session.pending_attachments = None session.pending_started_at = None session.pending_user_source = None session.save() def _cleanup_gateway_pending_mirror(session_id: str) -> None: try: from api.route_approvals import ( _approval_sse_notify_locked, _lock as _approval_lock, reconcile_gateway_pending_mirror_locked, ) with _approval_lock: head, total, _ = reconcile_gateway_pending_mirror_locked(session_id) _approval_sse_notify_locked(session_id, head, total) except Exception: logger.debug("Failed to reconcile gateway pending mirror during teardown", exc_info=True) def _run_gateway_chat_streaming( session_id, msg_text, model, workspace, stream_id, attachments=None, *, model_provider=None, goal_related=False, ): """Bridge a WebUI chat turn through Hermes Gateway's API server. This default-off path keeps the browser contract unchanged: /api/chat/start still returns a local stream_id and /api/chat/stream still receives WebUI SSE event names. The worker translates OpenAI-compatible streaming chunks from the configured Gateway API server into those local events and persists the final user/assistant turn back into the WebUI session. """ q = STREAMS.get(stream_id) if q is None: # Cancelled before the worker started; release the owner entry the route # layer registered so STREAM_SESSION_OWNERS does not leak (no teardown finally runs). unregister_stream_owner(stream_id) return register_active_run( stream_id, session_id=session_id, started_at=time.time(), phase="gateway-starting", workspace=str(workspace), model=model, provider=model_provider, backend="gateway", ) try: run_journal = RunJournalWriter(session_id, stream_id) except Exception: run_journal = None logger.debug("Failed to initialize gateway run journal for stream %s", stream_id, exc_info=True) cancel_event = threading.Event() with STREAMS_LOCK: CANCEL_FLAGS[stream_id] = cancel_event STREAM_PARTIAL_TEXT[stream_id] = "" STREAM_REASONING_TEXT[stream_id] = "" STREAM_LIVE_TOOL_CALLS[stream_id] = [] success_writeback_committed = False def put_gateway_event(event, data): if cancel_event.is_set() and not success_writeback_committed and event not in ("cancel", "error", "apperror"): return if event == "apperror" and isinstance(data, dict): data = data.copy() data.setdefault("session_id", session_id) event_id = None if run_journal is not None: try: journaled = run_journal.append_sse_event(event, data) event_id = (journaled or {}).get("event_id") if isinstance(journaled, dict) else None if event_id: STREAM_LAST_EVENT_ID[stream_id] = event_id except Exception: logger.debug("Failed to append gateway event %s for stream %s", event, stream_id, exc_info=True) if event_id and hasattr(q, "note_last_event_id"): try: q.note_last_event_id(event_id) except Exception: logger.debug("Failed to note gateway event_id %s for stream %s", event_id, stream_id, exc_info=True) try: queue_item = (event, data, event_id) if event_id and hasattr(q, "subscribe_with_snapshot") else (event, data) q.put_nowait(queue_item) except Exception: logger.debug("Failed to put gateway event to queue") s = None final_text = "" terminal_error = "" usage = {"input_tokens": 0, "output_tokens": 0, "estimated_cost": 0} try: s = get_session(session_id) from api.config import get_config # imported lazily to avoid config-cycle churn cfg = get_config() reasoning_effort = _gateway_reasoning_effort_for_request( cfg, model=model, model_provider=model_provider, ) try: from api.streaming import ( _load_webui_prefill_context, _prefill_messages_with_webui_context, _normalize_prefill_messages_before_user_turn, _public_prefill_context_status, _webui_ephemeral_system_prompt, ) prefill_context = _load_webui_prefill_context(cfg) # #3324: the WebUI session/delivery context (connected platforms, # home channels, delivery hints, session framing) is now carried in # the ephemeral system prompt rather than a prefill `user` message. # The gateway-backed path must build the SAME system prompt so that # context is not silently dropped on Gateway-routed WebUI chats. _gateway_system_prompt = _webui_ephemeral_system_prompt( None, surface_context={ "source": "webui", "session_id": session_id, "profile": getattr(s, "profile", None), "workspace": s.workspace if s is not None else str(workspace), }, config_data=cfg, ) prefill_messages = _prefill_messages_with_webui_context(prefill_context, cfg) prefill_messages = _normalize_prefill_messages_before_user_turn(prefill_messages) prefill_messages = [ {"role": "system", "content": _gateway_system_prompt}, *prefill_messages, ] put_gateway_event("context_status", { "session_id": session_id, "prefill": _public_prefill_context_status(prefill_context), }) except Exception: logger.debug("Failed to load WebUI gateway prefill context", exc_info=True) prefill_messages = [] 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( cfg, effective_model=model, effective_provider=model_provider, ) except Exception: _gw_overrides = {} # Capability gate: use runs API when gateway advertises approval support. _use_runs_api = _gateway_use_runs_api_enabled(cfg) and gateway_supports_approval(base_url, api_key) if _use_runs_api: body_extras = {} if model_provider: body_extras["provider"] = model_provider if reasoning_effort is not None: body_extras["reasoning_effort"] = reasoning_effort 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, ) except Exception as exc: error_payload = _settle_gateway_terminal_error( session_id, stream_id, workspace, model, model_provider, str(exc), ) if error_payload is None: return put_gateway_event("apperror", error_payload) return if final_text is None: return else: # Legacy gateway path: emit unsupported approval notice once per session, # but only when the gateway genuinely lacks approval capability. approval_reason = gateway_approval_unavailable_reason(base_url, api_key) if approval_reason is not None: if not hasattr(s, "_approval_notice_emitted"): s._approval_notice_emitted = False if not s._approval_notice_emitted: approval_message = "Approvals require a newer gateway. Upgrade the connected Hermes gateway to enable this." approval_type = "approval_gateway_unsupported" if approval_reason == "unreachable": approval_type = "approval_gateway_offline" approval_message = "Gateway connection failed. Check that the connected Hermes gateway is running and reachable." put_gateway_event("warning", { "type": approval_type, "message": approval_message, }) s._approval_notice_emitted = True url = f"{base_url}/v1/chat/completions" headers = { "Content-Type": "application/json", "Accept": "text/event-stream", "X-Hermes-Session-Id": session_id, } if api_key: headers["Authorization"] = f"Bearer {api_key}" # Scope Gateway long-term continuity to this WebUI conversation # without exposing the browser's auth cookie or CSRF material. headers["X-Hermes-Session-Key"] = f"webui:{session_id}" message_content: Any = str(msg_text or "") if attachments: try: from api.streaming import _build_native_multimodal_message message_content = _build_native_multimodal_message("", str(msg_text or ""), attachments, str(workspace), cfg=cfg) except Exception: logger.debug("Failed to build gateway multimodal attachment payload", exc_info=True) message_content = str(msg_text or "") body = { "model": model or "default", "stream": True, "messages": [*prefill_messages, {"role": "user", "content": message_content}], } if model_provider: body["provider"] = model_provider if reasoning_effort is not None: body["reasoning_effort"] = reasoning_effort if _gw_overrides.get("service_tier"): body["service_tier"] = _gw_overrides["service_tier"] req = urllib.request.Request( url, data=json.dumps(body).encode("utf-8"), headers=headers, method="POST", ) update_active_run(stream_id, phase="gateway-request") last_payload = {} sse_event = "message" with urllib.request.urlopen(req, timeout=_gateway_read_timeout_secs()) as resp: for raw_line in _iter_sse_lines_cancellable(resp, cancel_event): if cancel_event.is_set(): put_gateway_event("cancel", {"message": "Cancelled by user"}) return line = raw_line.decode("utf-8", errors="replace").strip() if not line: sse_event = "message" continue if line.startswith("event:"): sse_event = line[6:].strip() or "message" continue if not line.startswith("data:"): continue data = line[5:].strip() if data == "[DONE]": break try: payload = json.loads(data) except json.JSONDecodeError: continue _payload_event = str(payload.get("event") or payload.get("type") or sse_event).strip() if _payload_event in {"hermes.approval.request", "approval.request"}: approval_data = _gateway_runs_approval_event(payload) if approval_data: # Record the gateway run_id so /api/approval/respond # can relay the choice back and resume the parked run # (legacy path never creates a local run; without this # the card renders but approve/deny returns ok:false). # No-op when the payload omits run_id. _approval_run_id = str(approval_data.get("run_id") or "").strip() if _approval_run_id: _STREAM_RUN_IDS[stream_id] = _approval_run_id put_gateway_event("approval", approval_data) try: from api.route_approvals import submit_gateway_pending_mirror submit_gateway_pending_mirror(session_id, approval_data) except Exception: logger.debug("submit_gateway_pending_mirror failed", exc_info=True) else: logger.debug("Ignoring malformed gateway approval payload") sse_event = "message" continue if sse_event == "hermes.tool.progress": translated = _gateway_tool_progress_event(payload) if translated: event_name, event_payload = translated if event_name == "reasoning": reason_delta = event_payload.get("text") if reason_delta and stream_id in STREAM_REASONING_TEXT: STREAM_REASONING_TEXT[stream_id] += reason_delta elif stream_id in STREAM_LIVE_TOOL_CALLS: if event_name == "tool": STREAM_LIVE_TOOL_CALLS[stream_id].append({ "name": event_payload.get("name"), "args": event_payload.get("args") or {}, "done": False, **({"tid": event_payload.get("tid")} if event_payload.get("tid") else {}), }) else: for shared_tc in reversed(STREAM_LIVE_TOOL_CALLS[stream_id]): if shared_tc.get("done"): continue if ( event_payload.get("tid") and shared_tc.get("tid") == event_payload.get("tid") ) or shared_tc.get("name") == event_payload.get("name"): shared_tc["done"] = True shared_tc["is_error"] = bool(event_payload.get("is_error")) break put_gateway_event(event_name, event_payload) if event_name != "reasoning": update_active_run(stream_id, phase="gateway-tool", latest_tool=event_payload.get("name")) sse_event = "message" continue if sse_event == "reasoning.available": reason_delta = _gateway_reasoning_delta(payload) if reason_delta: if stream_id in STREAM_REASONING_TEXT: STREAM_REASONING_TEXT[stream_id] += reason_delta put_gateway_event("reasoning", {"text": reason_delta}) sse_event = "message" continue last_payload = payload if payload.get("error"): terminal_error = str(payload["error"]) reasoning_delta = _gateway_sse_reasoning_delta(payload) if reasoning_delta: if stream_id in STREAM_REASONING_TEXT: STREAM_REASONING_TEXT[stream_id] += reasoning_delta put_gateway_event("reasoning", {"text": reasoning_delta}) delta = _gateway_sse_delta(payload) if delta: final_text += delta if stream_id in STREAM_PARTIAL_TEXT: STREAM_PARTIAL_TEXT[stream_id] += delta put_gateway_event("token", {"text": delta}) usage.update({k: v for k, v in _gateway_stream_usage(payload).items() if v}) usage.update({k: v for k, v in _gateway_stream_usage(last_payload).items() if v}) assistant_text = final_text.strip() if terminal_error: error_payload = _settle_gateway_terminal_error( session_id, stream_id, workspace, model, model_provider, terminal_error, ) if error_payload is None: return put_gateway_event("apperror", error_payload) return if not assistant_text: put_gateway_event("apperror", { "label": "Gateway returned no response", "type": "gateway_empty_response", "message": "Gateway returned no assistant message for this turn.", "hint": "Check that Hermes Gateway API server is running and reachable.", }) return with _get_session_agent_lock(session_id): s = get_session(session_id) if not _stream_writeback_is_current(s, stream_id): return # A late Stop can land after Gateway has yielded a full answer but # before success writeback. Treat it as cancellation so any # credential-exhausted process-wakeup pause stays in place. if cancel_event.is_set(): put_gateway_event("cancel", {"message": "Cancelled by user"}) return now = time.time() # Preserve subsecond ordering for gateway-backed turns. Using an # integer seconds timestamp gives the user and assistant rows the # same sort key; later transcript merges can then fall back to # role/content ordering instead of turn order. assistant_ts = now + 0.000001 user_msg = {"role": "user", "content": str(msg_text or ""), "timestamp": now} pending_source = getattr(s, "pending_user_source", None) or "webui" if pending_source != "webui": user_msg["_source"] = pending_source if attachments: user_msg["attachments"] = list(attachments) assistant_msg = {"role": "assistant", "content": assistant_text, "timestamp": assistant_ts} saved_reasoning = STREAM_REASONING_TEXT.get(stream_id, "") if saved_reasoning: assistant_msg["reasoning"] = saved_reasoning previous_messages = list(getattr(s, "messages", None) or []) previous_context = list(getattr(s, "context_messages", None) or getattr(s, "messages", None) or []) previous_process_wakeup_pause = dict(getattr(s, "process_wakeup_pause", {}) or {}) # Stamp stable ids on the two new rows (shared with the display merge # below) so display and model-context copies share an id for the # fork/truncate aligner (#context-message-stable-id). try: from api.streaming import _assign_stable_message_ids _assign_stable_message_ids( [user_msg, assistant_msg], previous_context, list(getattr(s, "messages", None) or []), ) except Exception: logger.debug("Failed to stamp stable ids on gateway turn rows", exc_info=True) s.context_messages = previous_context + [user_msg, assistant_msg] try: from api.streaming import _is_context_compression_marker display_context = [ msg for msg in previous_context if not _is_context_compression_marker(msg) ] except Exception: logger.debug("Failed to filter gateway display context markers", exc_info=True) display_context = previous_context display = merge_session_messages_append_only( previous_messages, display_context, ) try: from api.streaming import _merge_display_messages_after_agent_result s.messages = _merge_display_messages_after_agent_result( display, previous_context, s.context_messages, str(msg_text or ""), source=pending_source, ) except Exception: logger.debug("Failed to merge gateway display transcript", exc_info=True) # Avoid duplicating the eager-save checkpointed user message. if display: latest = display[-1] if isinstance(latest, dict) and latest.get("role") == "user": latest_text = " ".join(str(latest.get("content") or "").split()) msg_norm = " ".join(str(msg_text or "").split()) if latest_text == msg_norm: display = display[:-1] s.messages = display + [user_msg, assistant_msg] s.active_stream_id = None s.pending_user_message = None s.pending_attachments = None s.pending_started_at = None s.pending_user_source = None s.workspace = str(workspace) s.model = model s.model_provider = model_provider def _restore_cancelled_success_writeback(): if pending_source == "process_wakeup": s.context_messages = previous_context s.messages = previous_messages s.process_wakeup_pause = dict(previous_process_wakeup_pause) elif previous_process_wakeup_pause: s.process_wakeup_pause = dict(previous_process_wakeup_pause) else: clear_process_wakeup_pause(s, reason="run_completed") s.save() put_gateway_event("cancel", {"message": "Cancelled by user"}) # Recheck immediately before clearing the pause; Stop can arrive # while the success transcript is being assembled. if cancel_event.is_set(): _restore_cancelled_success_writeback() return clear_process_wakeup_pause(s, reason="run_completed") if cancel_event.is_set(): _restore_cancelled_success_writeback() return s.save() if cancel_event.is_set(): _restore_cancelled_success_writeback() return success_writeback_committed = True try: from api.goals import evaluate_goal_after_turn, has_active_goal from api.profiles import get_hermes_home_for_profile profile_home = get_hermes_home_for_profile(getattr(s, "profile", None)) if goal_related and has_active_goal(session_id, profile_home=profile_home): put_gateway_event("goal", { "session_id": session_id, "state": "evaluating", "message": "Evaluating goal progress…", "message_key": "goal_evaluating_progress", }) decision = evaluate_goal_after_turn( session_id, assistant_text, user_initiated=True, profile_home=profile_home, ) or {} goal_message = str(decision.get("message") or "").strip() if goal_message: put_gateway_event("goal", { "session_id": session_id, "state": "continuing" if decision.get("should_continue") else "idle", "message": goal_message, "message_key": decision.get("message_key") or ( "goal_continuing" if goal_message else "" ), "message_args": decision.get("message_args") or [], "decision": decision, }) if decision.get("should_continue"): continuation_prompt = str(decision.get("continuation_prompt") or "").strip() if continuation_prompt: PENDING_GOAL_CONTINUATION.add(session_id) put_gateway_event("goal_continue", { "session_id": session_id, "continuation_prompt": continuation_prompt, "text": continuation_prompt, "message": goal_message, "message_key": decision.get("message_key") or "goal_continuing", "message_args": decision.get("message_args") or [], "decision": decision, }) except Exception as goal_exc: logger.debug( "Gateway goal continuation hook failed for session %s: %s", session_id, goal_exc, ) from api.streaming import _session_payload_with_full_messages gateway_session_payload = _session_payload_with_full_messages(s, tool_calls=[]) put_gateway_event("done", {"session": redact_session_data(gateway_session_payload), "usage": usage}) put_gateway_event("stream_end", {"session_id": session_id}) except urllib.error.HTTPError as exc: try: err_body = exc.read(2048).decode("utf-8", errors="replace") except Exception: err_body = "" put_gateway_event( "apperror", _gateway_http_error_event(exc, err_body, api_key_configured=bool(_gateway_api_key())), ) except Exception as exc: safe = _redact_text(str(exc))[:500] put_gateway_event("apperror", { "label": "Gateway request failed", "type": "gateway_error", "message": safe or "Gateway request failed.", "hint": "Check HERMES_WEBUI_GATEWAY_BASE_URL and Gateway API server health.", }) finally: if s is not None: try: with _get_session_agent_lock(session_id): _clear_gateway_pending_state(get_session(session_id), stream_id) except Exception: logger.debug("Failed to clear gateway stream state", exc_info=True) _cleanup_gateway_pending_mirror(session_id) with STREAMS_LOCK: CANCEL_FLAGS.pop(stream_id, None) STREAM_GOAL_RELATED.pop(stream_id, None) STREAM_PARTIAL_TEXT.pop(stream_id, None) STREAM_REASONING_TEXT.pop(stream_id, None) STREAM_LIVE_TOOL_CALLS.pop(stream_id, None) STREAM_LAST_EVENT_ID.pop(stream_id, None) STREAMS.pop(stream_id, None) _STREAM_RUN_IDS.pop(stream_id, None) unregister_active_run(stream_id)