happy5318
6af1a7e718
The suppression predicate _transcript_already_advanced_past_pending() required a run-journal terminal 'completed' state, but its caller in _apply_core_sync_or_error_marker already returns early on that same condition. The predicate branch was therefore unreachable dead code: it could never suppress anything, and stale-cancel recovery could still append a duplicate recovered user row, a journal clone and an interruption marker for a turn that actually finished. Derive the completion evidence from the crash-safe turn journal instead: _turn_journal_records_completion() reports True when the exact-stream 'completed' event is the latest terminal state for the pending stream id. The run journal's terminal write is not durable evidence for this path (it is consumed by the caller's early return), so the turn journal is the only still-observable source. A cancelled/crashed turn (interrupted) keeps recovery active, and completion for another stream of the same session never settles the pending one. Also record the same completion event on the Gateway success writeback path, so Gateway-backed runs have the equivalent completion persistence instead of leaving the run journal's lost terminal write unrecoverable. Regression coverage: - test_turn_journal_helper_reports_completed_for_matching_stream - test_turn_journal_helper_rejects_interrupted_turn - test_advances_past_pending_from_turn_journal_without_run_journal_terminal - test_full_recovery_suppresses_duplicates_on_turn_journal_completion - test_full_recovery_appends_rows_on_turn_journal_completion_loss - test_gateway_chat_worker_records_turn_journal_completion Reverting the predicate gate turns the two new CORE regressions red; reverting the Gateway write turns the Gateway regression red.
1513 lines
68 KiB
Python
1513 lines
68 KiB
Python
"""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 uuid
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
from typing import Any
|
|
|
|
from api.config import (
|
|
AGENT_INSTANCES,
|
|
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,
|
|
_parse_provider_qualified_model_id,
|
|
clear_session_writeback_owner_if_owned,
|
|
coerce_reasoning_effort_for_model,
|
|
gateway_approval_unavailable_reason,
|
|
gateway_supports_approval,
|
|
peek_stream,
|
|
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, bound_run_journal_snapshot_args
|
|
from api.turn_journal import append_turn_journal_event_for_stream
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Maps stream_id -> gateway run_id for approval response relay.
|
|
_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
|
|
|
|
|
|
def _mark_gateway_run_starting(stream_id: str) -> None:
|
|
with _STREAM_RUN_STARTING_CONDITION:
|
|
_STREAM_RUN_IDS.pop(stream_id, None)
|
|
_STREAM_RUN_LIFECYCLE[stream_id] = {
|
|
"phase": "pending",
|
|
"run_id": "",
|
|
"waiters": 0,
|
|
"owner_done": False,
|
|
}
|
|
|
|
|
|
def _publish_gateway_run_id(stream_id: str, run_id: str) -> None:
|
|
with _STREAM_RUN_STARTING_CONDITION:
|
|
_STREAM_RUN_IDS[stream_id] = run_id
|
|
state = _STREAM_RUN_LIFECYCLE.get(stream_id) or {}
|
|
_STREAM_RUN_LIFECYCLE[stream_id] = {
|
|
"phase": "ready",
|
|
"run_id": run_id,
|
|
"waiters": int(state.get("waiters") or 0),
|
|
"owner_done": bool(state.get("owner_done")),
|
|
}
|
|
_STREAM_RUN_STARTING_CONDITION.notify_all()
|
|
|
|
|
|
def _finish_gateway_run_starting(stream_id: str, *, result: str = "failed") -> None:
|
|
with _STREAM_RUN_STARTING_CONDITION:
|
|
state = _STREAM_RUN_LIFECYCLE.get(stream_id) or {}
|
|
if str(state.get("phase") or "").strip().lower() == "ready":
|
|
return
|
|
_STREAM_RUN_IDS.pop(stream_id, None)
|
|
_STREAM_RUN_LIFECYCLE[stream_id] = {
|
|
"phase": "fallback" if result == "fallback" else "failed",
|
|
"run_id": "",
|
|
"waiters": int(state.get("waiters") or 0),
|
|
"owner_done": bool(state.get("owner_done")),
|
|
}
|
|
_STREAM_RUN_STARTING_CONDITION.notify_all()
|
|
|
|
|
|
def _retire_gateway_run_starting_if_done(stream_id: str) -> bool:
|
|
state = _STREAM_RUN_LIFECYCLE.get(stream_id)
|
|
if not state:
|
|
return False
|
|
if int(state.get("waiters") or 0) > 0:
|
|
return False
|
|
if not bool(state.get("owner_done")):
|
|
return False
|
|
_STREAM_RUN_LIFECYCLE.pop(stream_id, None)
|
|
_STREAM_RUN_IDS.pop(stream_id, None)
|
|
return True
|
|
|
|
|
|
def _clear_gateway_run_starting(stream_id: str) -> None:
|
|
with _STREAM_RUN_STARTING_CONDITION:
|
|
state = _STREAM_RUN_LIFECYCLE.get(stream_id)
|
|
if state:
|
|
state["owner_done"] = True
|
|
_retire_gateway_run_starting_if_done(stream_id)
|
|
_STREAM_RUN_STARTING_CONDITION.notify_all()
|
|
|
|
|
|
def gateway_run_id_pending(stream_id: str) -> bool:
|
|
with _STREAM_RUN_STARTING_CONDITION:
|
|
return str((_STREAM_RUN_LIFECYCLE.get(stream_id) or {}).get("phase") or "").strip().lower() == "pending"
|
|
|
|
|
|
def wait_for_gateway_run_id(stream_id: str, timeout: float) -> tuple[bool, str | None]:
|
|
deadline = time.monotonic() + max(0.0, float(timeout))
|
|
with _STREAM_RUN_STARTING_CONDITION:
|
|
state = _STREAM_RUN_LIFECYCLE.get(stream_id)
|
|
if state:
|
|
state["waiters"] = int(state.get("waiters") or 0) + 1
|
|
try:
|
|
while True:
|
|
state = _STREAM_RUN_LIFECYCLE.get(stream_id)
|
|
phase = str((state or {}).get("phase") or "").strip().lower()
|
|
if phase == "fallback":
|
|
return False, None
|
|
if phase == "failed":
|
|
return True, None
|
|
run_id = str(_STREAM_RUN_IDS.get(stream_id) or "").strip()
|
|
if phase == "ready":
|
|
stored_run_id = str((state or {}).get("run_id") or "").strip()
|
|
return True, run_id or stored_run_id or None
|
|
if run_id:
|
|
return True, run_id
|
|
if not state:
|
|
return False, None
|
|
remaining = deadline - time.monotonic()
|
|
if remaining <= 0:
|
|
return True, None
|
|
_STREAM_RUN_STARTING_CONDITION.wait(timeout=remaining)
|
|
finally:
|
|
state = _STREAM_RUN_LIFECYCLE.get(stream_id)
|
|
if state:
|
|
waiters = max(0, int(state.get("waiters") or 0) - 1)
|
|
state["waiters"] = waiters
|
|
if _retire_gateway_run_starting_if_done(stream_id):
|
|
_STREAM_RUN_STARTING_CONDITION.notify_all()
|
|
|
|
_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"}
|
|
|
|
|
|
def _gateway_model_field(model: str | None) -> str:
|
|
"""Return the bare model name to put in a gateway request body.
|
|
|
|
The picker and ``_resolve_compatible_session_model_state`` intentionally
|
|
keep the full ``@provider:model`` string for internal routing (#1253), but
|
|
the gateway expects a bare model name and carries the provider separately.
|
|
Sent verbatim, ``@ollama-cloud:minimax-m2.7`` is forwarded to the upstream
|
|
provider API, which 404s on the ``@``-prefixed string (#6722).
|
|
|
|
Parsing is delegated to ``config._parse_provider_qualified_model_id()`` so
|
|
a multi-segment custom provider ID (``@custom:backup:model-a``) yields the
|
|
real model (``model-a``) instead of a positional-split fragment.
|
|
"""
|
|
if not model:
|
|
return ""
|
|
value = str(model).strip()
|
|
parsed = _parse_provider_qualified_model_id(value)
|
|
if parsed:
|
|
return str(parsed[0] or "").strip()
|
|
return value
|
|
|
|
|
|
# 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_session_yolo_enabled(session_id: str) -> bool:
|
|
"""Return the WebUI-owned, in-memory YOLO state for a browser session."""
|
|
try:
|
|
from tools.approval import is_session_yolo_enabled
|
|
|
|
return bool(is_session_yolo_enabled(str(session_id or "")))
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
def _settle_gateway_run_approval(
|
|
session_id: str,
|
|
approval_data: dict,
|
|
base_url: str,
|
|
api_key: str,
|
|
) -> tuple[bool, dict | None, int]:
|
|
"""Auto-approve or mirror one run approval at a session-linearized point."""
|
|
from api.route_approvals import gateway_yolo_handoff, submit_gateway_pending_mirror
|
|
|
|
run_id = str(approval_data.get("run_id") or "").strip()
|
|
identity_v1 = bool(approval_data.get("_gateway_agent_identity_v1"))
|
|
with gateway_yolo_handoff(session_id):
|
|
if _gateway_session_yolo_enabled(session_id):
|
|
try:
|
|
_auto_approve_gateway_run(
|
|
base_url,
|
|
api_key,
|
|
run_id,
|
|
approval_data["approval_id"] if identity_v1 else "",
|
|
)
|
|
return True, None, 0
|
|
except Exception:
|
|
# Fail closed: if remote approval fails, surface the real card
|
|
# before allowing a same-session toggle to pass the handoff.
|
|
logger.warning(
|
|
"WebUI YOLO could not auto-approve run %s; showing approval card",
|
|
run_id,
|
|
exc_info=True,
|
|
)
|
|
head, total = submit_gateway_pending_mirror(session_id, approval_data)
|
|
return False, head, total
|
|
|
|
|
|
def _auto_approve_gateway_run(
|
|
base_url: str,
|
|
api_key: str,
|
|
run_id: str,
|
|
approval_id: str,
|
|
) -> None:
|
|
"""Resolve one Runs API prompt using only the shipped approval contract.
|
|
|
|
This is a WebUI-owned compatibility path: the Runs API does not yet expose
|
|
session YOLO, so WebUI answers each approval request while its own session
|
|
flag is enabled. Native Agent-side YOLO would be preferable because it can
|
|
bypass gates before they pause and also covers Agent-owned computer-use
|
|
policy; https://github.com/NousResearch/hermes-agent/pull/61946 tracks that
|
|
API capability. Until then, do not send speculative fields to the Agent.
|
|
"""
|
|
from api.runner_client import HttpRunnerClient
|
|
|
|
HttpRunnerClient(base_url=base_url, api_key=api_key).respond_approval(
|
|
run_id,
|
|
approval_id,
|
|
"once",
|
|
)
|
|
|
|
|
|
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": bound_run_journal_snapshot_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()
|
|
raw_approval_id = str(payload.get("approval_id") or payload.get("id") or "").strip()
|
|
approval_id = raw_approval_id
|
|
if not approval_id:
|
|
approval_id = uuid.uuid4().hex
|
|
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,
|
|
"_gateway_raw_approval_id_present": bool(raw_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,
|
|
active_provider: str = "",
|
|
):
|
|
"""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}"
|
|
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, active_provider=active_provider, active_model=(model or ""), requested_provider=active_provider, profile=getattr(session, "profile", None))
|
|
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": _gateway_model_field(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}")
|
|
except Exception:
|
|
_finish_gateway_run_starting(stream_id)
|
|
raise
|
|
|
|
usage: dict = {}
|
|
_publish_gateway_run_id(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 = ""
|
|
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
|
|
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})
|
|
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":
|
|
from api.route_approvals import retire_gateway_pending_mirror
|
|
retire_gateway_pending_mirror(session_id, run_id=run_id)
|
|
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":
|
|
from api.route_approvals import retire_gateway_pending_mirror
|
|
retire_gateway_pending_mirror(session_id, run_id=run_id)
|
|
raise RuntimeError(str(payload.get("error") or "Gateway run failed"))
|
|
if payload_event == "run.cancelled":
|
|
from api.route_approvals import retire_gateway_pending_mirror
|
|
retire_gateway_pending_mirror(session_id, run_id=run_id)
|
|
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 stop_gateway_run(run_id: str) -> bool:
|
|
"""Request gateway interruption and report whether it was acknowledged."""
|
|
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()
|
|
headers = {"Accept": "application/json", "Content-Type": "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='')}/stop",
|
|
data=b"{}",
|
|
headers=headers,
|
|
method="POST",
|
|
)
|
|
class _NoRedirect(urllib.request.HTTPRedirectHandler):
|
|
def redirect_request(self, req, fp, code, msg, headers, newurl):
|
|
return None
|
|
|
|
try:
|
|
opener = urllib.request.build_opener(_NoRedirect)
|
|
with opener.open(req, timeout=10) as response:
|
|
final_url = str(getattr(response, "geturl", lambda: req.full_url)() or "")
|
|
status = int(getattr(response, "status", getattr(response, "code", 0)) or 0)
|
|
return 200 <= status < 300 and final_url == req.full_url
|
|
except (urllib.error.HTTPError, urllib.error.URLError, OSError, ValueError):
|
|
logger.debug("Gateway stop failed for run %s", run_id, exc_info=True)
|
|
return False
|
|
|
|
|
|
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,
|
|
_terminal_turn_duration,
|
|
)
|
|
|
|
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", ""),
|
|
)
|
|
turn_duration = _terminal_turn_duration(session)
|
|
_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 turn_duration is not None:
|
|
error_message["_turnDuration"] = turn_duration
|
|
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
|
|
terminal_session_persisted = False
|
|
try:
|
|
session.save()
|
|
terminal_session_persisted = True
|
|
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
|
|
error_payload["terminal_session_persisted"] = terminal_session_persisted
|
|
if terminal_session_persisted:
|
|
error_payload["terminal_session_persisted_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 (
|
|
retire_gateway_pending_mirror,
|
|
)
|
|
|
|
retire_gateway_pending_mirror(session_id)
|
|
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,
|
|
regeneration=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 = peek_stream(stream_id)
|
|
if q is None:
|
|
_finish_gateway_run_starting(stream_id, result="fallback")
|
|
_clear_gateway_run_starting(stream_id)
|
|
# 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)
|
|
# Also release the writeback-owner entry the route layer registered, so
|
|
# SESSION_WRITEBACK_OWNERS does not leak on this pre-start cancellation
|
|
# path (the teardown finally below never runs when we early-return here).
|
|
clear_session_writeback_owner_if_owned(session_id, 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
|
|
runs_api_pending_marked = True
|
|
|
|
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,
|
|
)
|
|
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 = {}
|
|
_runs_api_enabled = _gateway_use_runs_api_enabled(cfg)
|
|
_use_runs_api = _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
|
|
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 = []
|
|
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,
|
|
active_provider=(model_provider or ""),
|
|
)
|
|
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, active_provider=(model_provider or ""), active_model=(model or ""), requested_provider=(model_provider or ""), profile=getattr(s, "profile", None))
|
|
except Exception:
|
|
logger.debug("Failed to build gateway multimodal attachment payload", exc_info=True)
|
|
message_content = str(msg_text or "")
|
|
body = {
|
|
"model": _gateway_model_field(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
|
|
try:
|
|
from api.route_approvals import submit_gateway_pending_mirror
|
|
head, total = submit_gateway_pending_mirror(session_id, approval_data)
|
|
approval_data = {**(head or approval_data), "pending_count": total}
|
|
except Exception:
|
|
logger.debug("submit_gateway_pending_mirror failed", exc_info=True)
|
|
put_gateway_event("approval", approval_data)
|
|
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
|
|
pending_source = getattr(s, "pending_user_source", None) or "webui"
|
|
from api.streaming import _active_turn_authority, _materialize_active_turn_user
|
|
|
|
active_turn_identity = _active_turn_authority(s, stream_id, msg_text)
|
|
user_msg = _materialize_active_turn_user(
|
|
active_turn_identity,
|
|
str(msg_text or ""),
|
|
pending_source,
|
|
)
|
|
user_msg["timestamp"] = float(
|
|
active_turn_identity.get("timestamp") or now
|
|
)
|
|
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 [])
|
|
stored_context = getattr(s, "context_messages", None)
|
|
previous_context = list(
|
|
stored_context
|
|
if isinstance(stored_context, list) and (regeneration or stored_context)
|
|
else 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
|
|
# #6366 re-gate: record the durable same-stream completion
|
|
# event in the crash-safe turn journal. The run journal's
|
|
# terminal state is only reached on its own ``stream_end``
|
|
# write path, so a Gateway run whose terminal write is lost
|
|
# would otherwise leave no completion evidence at all and
|
|
# stale-cancel recovery would re-append a duplicate
|
|
# recovered row after the valid final answer.
|
|
try:
|
|
append_turn_journal_event_for_stream(
|
|
session_id,
|
|
stream_id,
|
|
{"event": "completed", "created_at": time.time()},
|
|
)
|
|
except Exception:
|
|
logger.debug("Failed to append completed turn journal event", exc_info=True)
|
|
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:
|
|
mapped_run_id = str(_STREAM_RUN_IDS.get(stream_id) or "").strip()
|
|
if mapped_run_id:
|
|
try:
|
|
from api.route_approvals import retire_gateway_pending_mirror
|
|
retire_gateway_pending_mirror(session_id, run_id=mapped_run_id)
|
|
except Exception:
|
|
logger.debug("Failed to retire gateway pending mirrors during teardown", exc_info=True)
|
|
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:
|
|
AGENT_INSTANCES.pop(stream_id, None)
|
|
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)
|
|
if runs_api_pending_marked and gateway_run_id_pending(stream_id):
|
|
_finish_gateway_run_starting(stream_id)
|
|
_clear_gateway_run_starting(stream_id)
|
|
unregister_stream_owner(stream_id)
|
|
unregister_active_run(stream_id)
|
|
# Release the writeback-owner entry the route layer registered for this
|
|
# Gateway run so SESSION_WRITEBACK_OWNERS does not grow unbounded across
|
|
# the process lifetime (compare-and-clear: only clears if still owned by
|
|
# this stream, mirroring the local streaming teardown).
|
|
clear_session_writeback_owner_if_owned(session_id, stream_id)
|