yilsem/aiturk-hermes-ide
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 3 Wiki Activity
Files
aiturk/main
aiturk-hermes-ide/gateway/platforms/api_server.py
T

8093 lines
361 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""
OpenAI-compatible API server platform adapter.
Exposes an HTTP server with endpoints:
- POST /v1/chat/completions — OpenAI Chat Completions format (stateless; opt-in session continuity via X-Hermes-Session-Id header; opt-in long-term memory scoping via X-Hermes-Session-Key header)
- POST /v1/responses — OpenAI Responses API format (stateful via previous_response_id; X-Hermes-Session-Key supported)
- GET /v1/responses/{response_id} — Retrieve a stored response
- DELETE /v1/responses/{response_id} — Delete a stored response
- GET /v1/models — lists hermes-agent and any configured model_routes aliases
- GET /v1/capabilities — machine-readable API capabilities for external UIs
- GET /api/sessions — list client-visible Hermes sessions
- POST /api/sessions — create an empty Hermes session
- GET/PATCH/DELETE /api/sessions/{session_id} — read/update/delete a session
- GET /api/sessions/{session_id}/messages — read session message history
- POST /api/sessions/{session_id}/fork — branch a session using SessionDB lineage
- POST /api/sessions/{session_id}/chat[/stream] — chat with a persisted session
- POST /v1/runs — start a run, returns run_id immediately (202)
- GET /v1/runs/{run_id} — retrieve current run status
- GET /v1/runs/{run_id}/events — SSE stream of structured lifecycle events
- POST /v1/runs/{run_id}/approval — resolve a pending run approval
- POST /v1/runs/{run_id}/steer — inject guidance into a running agent
- POST /v1/runs/{run_id}/stop — interrupt a running agent
- GET /health — health check
- GET /health/detailed — rich status for cross-container dashboard probing
Any OpenAI-compatible frontend (Open WebUI, LobeChat, LibreChat,
AnythingLLM, NextChat, ChatBox, etc.) can connect to hermes-agent
through this adapter by pointing at http://localhost:8642/v1 and
authenticating with API_SERVER_KEY.
When ``gateway.multiplex_profiles`` is on, the default profile owns this
listener and secondary profiles are reached via a URL prefix — same contract
as the webhook adapter:
GET /p/<profile>/v1/models
POST /p/<profile>/v1/chat/completions
...
Requires:
- aiohttp (already available in the gateway)
"""
import asyncio
import concurrent.futures
import errno
import hashlib
import hmac
import itertools
import json
from contextlib import contextmanager, nullcontext, suppress
from contextvars import ContextVar
from functools import wraps
import logging
import os
import re
import sqlite3
import sys
import threading
import time
import uuid
from pathlib import Path
from typing import Any, Dict, List, Optional
# Sentinel returned by _resolve_request_profile when a /p/<profile>/ prefix
# names a profile this gateway does not serve (→ 404). Distinct from None
# (no prefix / multiplexing off → handle as the default profile).
_PROFILE_REJECTED = object()
def _prefix_names_served_profile(profile: str) -> bool:
"""True when a /p/<profile>/ prefix names the profile this gateway serves.
Single-profile (non-multiplex) gateways historically ignored the prefix
and answered every /p/<x>/ request from their own profile's config —
which silently served the gateway owner's toolsets/capabilities under
another profile's URL (#91583 defect 2). Only a self-referential prefix
may fall through; anything else must be rejected. Fail closed.
"""
try:
from hermes_cli.profiles import profile_matches_home
return profile_matches_home(profile)
except Exception:
return False
# Profile selected by the /p/<profile>/ URL prefix for the current request.
# Set by the profile-prefix middleware; read by handlers / _run_agent.
_api_request_profile: ContextVar[Optional[str]] = ContextVar(
"api_server_request_profile", default=None
)
_api_request_browser_control_principal: ContextVar[str] = ContextVar(
"api_server_browser_control_principal", default=""
)
_api_request_browser_control_transport_family: ContextVar[str] = ContextVar(
"api_server_browser_control_transport_family", default=""
)
#: Minimal scope shape accepted by :func:`gateway.browser_control_artifacts
#: .artifact_scope_key`: principal + session + transport family. The API
#: server authenticates the caller itself, so the facade carries only the
#: server-derived principal and the loopback/remote family.
class _ArtifactScopeFacade:
__slots__ = ("principal_id", "session_id", "transport_family")
def __init__(self, principal_id: str, *, session_id: str = "", transport_family: str = ""):
self.principal_id = principal_id
self.session_id = session_id
self.transport_family = transport_family
def __repr__(self) -> str: # pragma: no cover - debugging aid
return f"_ArtifactScopeFacade(principal={self.principal_id!r})"
#: Browser-extension control protocol version advertised in capabilities and
#: echoed in registration responses. Strict validation is centralized in the
#: broker's ``browser_control_protocol_supported`` helper.
_BROWSER_CONTROL_PROTOCOL_VERSION = 1
_BROWSER_CONTROL_WS_PROTOCOL = "hermes-browser-control-v1"
_BROWSER_CONTROL_TICKET_PROTOCOL_PREFIX = "hermes-browser-control-ticket."
def _approval_event_choices(
*, smart_denied: bool, allow_session: bool, allow_permanent: bool
) -> list[str]:
if smart_denied or not allow_session:
return ["once", "deny"]
return (
["once", "session", "always", "deny"]
if allow_permanent
else ["once", "session", "deny"]
)
try:
from aiohttp import web
AIOHTTP_AVAILABLE = True
except ImportError:
AIOHTTP_AVAILABLE = False
web = None # type: ignore[assignment]
from gateway.config import Platform, PlatformConfig
from gateway.platforms import api_server_room_dispatch as _room_dispatch
from gateway.platforms import api_server_room_grants as _room_grants
from gateway.platforms import api_server_runs as _api_runs
from gateway.platforms.base import (
MEDIA_TAG_CLEANUP_RE,
BasePlatformAdapter,
SendResult,
is_network_accessible,
validate_media_delivery_path,
)
# Re-exported here for existing imports and constructor monkeypatches.
from gateway.platforms.api_server_run_idempotency import RunIdempotencyStore
from agent.redact import redact_sensitive_text
from agent.interrupt_compat import request_hard_interrupt
from gateway.readiness import collect_runtime_readiness
from gateway.browser_control_artifacts import (
ArtifactError,
ArtifactRateLimiter,
ArtifactStore,
ArtifactTooLarge,
DEFAULT_ALLOWED_MIME_TYPES,
DEFAULT_MAX_ARTIFACT_BYTES,
DEFAULT_ARTIFACT_TTL_SECONDS,
)
from gateway.browser_control_broker import (
BROWSER_CONTROL_ARTIFACT_CAPABILITIES,
BROWSER_CONTROL_CAPABILITIES,
BROWSER_CONTROL_DEVELOPER_CAPABILITIES,
ControllerScope,
ControllerTicketInvalid,
browser_control_developer_mode,
browser_control_protocol_supported,
filter_browser_control_capabilities,
get_browser_control_broker,
)
from agent.secret_scope import UnscopedSecretError as _UnscopedSecretError
from agent.secret_scope import get_secret as _scoped_get_secret
def _get_scoped_secret(name, default=None):
"""Scope-aware credential read with the default-profile startup fallback.
Secondary profiles construct their adapters under a profile secret
scope -- the scope is authoritative and a scoped miss returns ``default``
(no cross-profile borrow from ``os.environ``, which may hold another
profile's value). The DEFAULT profile's adapter constructs and sends
*unscoped* under multiplexing, where a bare ``get_secret`` would raise
``UnscopedSecretError`` and crash this path; there ``os.environ`` is that
profile's own value, so fall back to it. Same pattern as the Slack
``SLACK_APP_TOKEN`` read (#59739) and
``gateway/platforms/whatsapp_common.py::_get_wsecret``.
"""
try:
val = _scoped_get_secret(name, default)
except _UnscopedSecretError:
val = os.getenv(name)
return val if val is not None else default
logger = logging.getLogger(__name__)
def _browser_controller_ws_sender(ws, loop, *, wait_timeout: float = 10.0):
"""Return a loop-aware broker sender for one aiohttp controller socket.
A wait timeout means the coroutine is still in flight on a live loop, not
that the frame was rejected. Keep the broker command pending and let its
own deadline/cancel path decide; a real send exception still propagates.
"""
def send(frame: dict) -> None:
if ws.closed:
raise ConnectionError("browser-control websocket is closed")
try:
on_loop = asyncio.get_running_loop() is loop
except RuntimeError:
on_loop = False
if on_loop:
loop.create_task(ws.send_json(frame))
return
future = asyncio.run_coroutine_threadsafe(ws.send_json(frame), loop)
try:
future.result(timeout=wait_timeout)
except concurrent.futures.TimeoutError:
if future.done():
raise
def observe_late_send(completed):
try:
completed.result()
except Exception:
logger.exception("browser-controller websocket send failed after wait timeout")
future.add_done_callback(observe_late_send)
return send
def _hermes_version() -> str:
"""Return the canonical Hermes Agent version string.
``hermes_cli.__version__`` is the runtime source of truth used by the CLI,
dashboard, portal tags, and release script. Prefer it over installed
distribution metadata because editable/source checkouts can retain stale
``hermes_agent-*.dist-info`` after a source update until the environment is
reinstalled. Never raises — a version probe must not be able to break the
health endpoint.
"""
try:
from hermes_cli import __version__
return __version__
except Exception:
pass
try:
from importlib.metadata import version
return version("hermes-agent")
except Exception:
return "dev"
# Default settings
DEFAULT_HOST = "127.0.0.1"
DEFAULT_PORT = 8642
MAX_STORED_RESPONSES = 100
MAX_REQUEST_BYTES = 10_000_000 # 10 MB — accommodates long agent conversations with tool calls
CHAT_COMPLETIONS_SSE_KEEPALIVE_SECONDS = 30.0
MAX_NORMALIZED_TEXT_LENGTH = 65_536 # 64 KB cap for normalized content parts
MAX_CONTENT_LIST_SIZE = 1_000 # Max items when content is an array
RESPONSES_AUTO_TRUNCATION_HISTORY_LIMIT = 100
_COMPRESSED_SUMMARY_METADATA_KEY = "_compressed_summary"
class ThreadSafeAsyncQueue(asyncio.Queue):
"""An ``asyncio.Queue`` that a non-loop thread can push into safely.
The SSE writers' streaming loops used to bridge a plain ``queue.Queue``
into the event loop via ``await loop.run_in_executor(None, lambda:
stream_q.get(timeout=0.5))`` inside a ``while True`` poll — a thread-pool
round trip on every 0.5s tick even when idle, plus up to 500ms of tail
latency between a delta landing in the queue and it reaching the
response. ``run_conversation`` itself runs on a worker thread (via
``loop.run_in_executor``), so its ``stream_delta_callback`` closures
(``_on_delta`` etc.) call ``put_threadsafe`` from off the loop thread;
the consumer side just does a plain ``await queue.get()``/
``asyncio.wait_for(queue.get(), timeout=...)``, woken immediately by
``call_soon_threadsafe`` instead of polling.
"""
def put_threadsafe(self, item, *, loop: asyncio.AbstractEventLoop = None) -> None:
(loop or self._loop_ref).call_soon_threadsafe(self.put_nowait, item)
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
# Always constructed inside a running async handler (the SSE
# request handlers below), so get_running_loop() is safe here.
self._loop_ref = asyncio.get_running_loop()
def _sse_frame(data: Any, *, event: str = None, ensure_ascii: bool = True) -> bytes:
"""Encode one SSE frame: optional ``event:`` line, then ``data: <json>\n\n``.
The single source of truth for SSE frame serialization across every
streaming writer in this module — ``_write_sse_chat_completion`` (the
five call sites it was first extracted from), ``_write_sse_responses``'s
inner ``_write_event`` closure, and the ``/v1/runs`` event stream. All
three used the identical ``json.dumps(data)`` / ``json.dumps(...,
ensure_ascii=False)`` + ``"\\ndata: ...\\n\\n"`` shape; routing them all
through here keeps the on-the-wire format in exactly one place.
``ensure_ascii`` defaults to ``True``, byte-identical to a bare
``json.dumps(data)``. Callers that must preserve raw non-ASCII bytes on
the wire (the Responses-API writer historically used
``ensure_ascii=False``) pass ``ensure_ascii=False`` explicitly — the
option exists so every writer shares one helper without changing any
existing byte stream.
"""
prefix = f"event: {event}\n" if event else ""
return f"{prefix}data: {json.dumps(data, ensure_ascii=ensure_ascii)}\n\n".encode()
def _coerce_port(value: Any, default: int = DEFAULT_PORT) -> int:
"""Parse a listen port without letting malformed env/config values crash startup."""
try:
return int(value)
except (TypeError, ValueError):
return default
_TRUE_REQUEST_BOOL_STRINGS = frozenset({"1", "true", "yes", "on"})
_FALSE_REQUEST_BOOL_STRINGS = frozenset({"0", "false", "no", "off"})
def _coerce_request_bool(value: Any, default: bool = False) -> bool:
"""Normalize boolean-like API payload values.
External clients should send real JSON booleans, but some OpenAI-compatible
frontends and middleware serialize flags like ``stream`` as strings. Using
Python truthiness on those values misroutes requests because ``"false"`` is
still truthy. Treat only explicit bool-ish scalars as booleans; everything
else falls back to the caller's default.
"""
if isinstance(value, bool):
return value
if value is None:
return default
if isinstance(value, str):
normalized = value.strip().lower()
if normalized in _TRUE_REQUEST_BOOL_STRINGS:
return True
if normalized in _FALSE_REQUEST_BOOL_STRINGS:
return False
return default
if isinstance(value, (int, float)):
return bool(value)
return default
_REQUEST_OPTION_MISSING = object()
# Full internal ladder + "none": the API server accepts what /reasoning and
# config.yaml accept (hermes_constants.VALID_REASONING_EFFORTS); wire-level
# clamping to each provider's vocabulary happens downstream in the
# transports/profiles via agent.reasoning_effort. Rejecting "max"/"ultra"
# here made API/browser clients second-class citizens of the ladder
# (#78216's api_server observation).
_REASONING_EFFORTS = frozenset(
{"none", "minimal", "low", "medium", "high", "xhigh", "max", "ultra"}
)
_RUNTIME_AGENT_OVERRIDE_KEYS = (
"api_key",
"base_url",
"provider",
"api_mode",
"command",
"args",
"credential_pool",
"max_tokens",
)
def _clean_request_string(value: Any) -> Optional[str]:
"""Return a stripped request string, or None for absent/non-string values."""
if not isinstance(value, str):
return None
cleaned = value.strip()
return cleaned or None
def _request_reasoning_config(model_options: Any) -> Optional[Dict[str, Any]]:
"""Translate browser/API model_options into AIAgent reasoning_config.
The browser extension sends both a structured ``reasoning`` object and a
compatibility ``reasoning_effort`` scalar. Keep this parser permissive so
older clients can send either shape, but ignore unknown effort values rather
than raising on a chat request.
"""
if not isinstance(model_options, dict):
return None
reasoning = model_options.get("reasoning")
enabled: Any = None
effort: Any = model_options.get("reasoning_effort")
if isinstance(reasoning, dict):
enabled = reasoning.get("enabled")
effort = reasoning.get("effort", effort)
effort_norm = str(effort).strip().lower() if effort is not None else ""
if enabled is False or effort_norm == "none":
return {"enabled": False}
if effort_norm in _REASONING_EFFORTS and effort_norm != "none":
return {"enabled": True, "effort": effort_norm}
if enabled is True:
return {"enabled": True}
return None
def _request_service_tier(model_options: Any) -> Any:
"""Return a per-request service_tier override or _REQUEST_OPTION_MISSING."""
if not isinstance(model_options, dict):
return _REQUEST_OPTION_MISSING
if "service_tier" in model_options:
raw_tier = model_options.get("service_tier")
if raw_tier is None:
return None
if isinstance(raw_tier, str):
return raw_tier.strip() or None
return raw_tier
if "fast" in model_options:
return "priority" if _coerce_request_bool(model_options.get("fast"), default=False) else None
return _REQUEST_OPTION_MISSING
def _apply_runtime_agent_overrides(
runtime_kwargs: Dict[str, Any], overrides: Optional[Dict[str, Any]]
) -> Dict[str, Any]:
"""Merge resolved provider/runtime fields into ``runtime_kwargs`` in place."""
if not isinstance(overrides, dict):
return runtime_kwargs
for key in _RUNTIME_AGENT_OVERRIDE_KEYS:
if key not in overrides:
continue
value = overrides.get(key)
if value is None:
continue
runtime_kwargs[key] = list(value) if key == "args" and isinstance(value, (list, tuple)) else value
return runtime_kwargs
def _resolve_request_runtime_agent_kwargs(provider: str, target_model: Optional[str] = None) -> Dict[str, Any]:
"""Resolve runtime kwargs for a one-request provider override.
This mirrors gateway.run._resolve_runtime_agent_kwargs(), but accepts an
explicit provider/model so an API caller can use the same authenticated
provider catalog as the TUI without mutating config.yaml.
"""
from hermes_cli.runtime_provider import resolve_runtime_provider, format_runtime_provider_error, _get_model_config
try:
runtime = resolve_runtime_provider(requested=provider, target_model=target_model)
except Exception as exc:
raise RuntimeError(format_runtime_provider_error(exc)) from exc
model_cfg = _get_model_config()
max_tokens = None
env_max_tokens = os.environ.get("HERMES_MAX_TOKENS")
if env_max_tokens:
try:
max_tokens = int(env_max_tokens)
except (ValueError, TypeError):
max_tokens = None
elif isinstance(model_cfg, dict):
cfg_max_tokens = model_cfg.get("max_tokens")
if isinstance(cfg_max_tokens, int):
max_tokens = cfg_max_tokens
if max_tokens is None:
runtime_max_tokens = runtime.get("max_output_tokens")
if isinstance(runtime_max_tokens, int) and runtime_max_tokens > 0:
max_tokens = runtime_max_tokens
return {
"api_key": runtime.get("api_key"),
"base_url": runtime.get("base_url"),
"provider": runtime.get("provider"),
"api_mode": runtime.get("api_mode"),
"command": runtime.get("command"),
"args": list(runtime.get("args") or []),
"credential_pool": runtime.get("credential_pool"),
"max_tokens": max_tokens,
}
def _request_agent_overrides(
body: Any,
*,
virtual_model: Optional[str] = None,
allow_bare_model: bool = True,
) -> Dict[str, Any]:
"""Extract per-request model/provider/options for _run_agent.
``/v1/models`` advertises a stable virtual model (usually ``hermes-agent``)
for OpenAI-compatible clients. Treat that alias as "use the gateway
default"; real model picker selections from the browser extension send the
raw provider model id plus a provider slug and should override this turn.
``allow_bare_model`` controls whether a ``model`` value WITHOUT an
accompanying ``provider`` is honored. Generic OpenAI clients routinely
hardcode model names ("gpt-4o", ...), and existing deployments rely on
those falling back to the gateway default on the OpenAI-compatible
surfaces — so those handlers pass the opt-in
``direct_model_requests`` config value here, while Hermes-native
endpoints (session chat, /v1/runs) always allow it. A request that
sends an explicit ``provider`` is unambiguously Hermes-aware and is
always honored.
"""
if not isinstance(body, dict):
return {}
overrides: Dict[str, Any] = {}
provider = _clean_request_string(body.get("provider"))
if provider:
overrides["requested_provider"] = provider
model = _clean_request_string(body.get("model"))
if model and model != virtual_model and (provider or allow_bare_model):
overrides["requested_model"] = model
model_options = body.get("model_options")
if isinstance(model_options, dict):
overrides["model_options"] = dict(model_options)
return overrides
def _is_compressed_summary_message(message: Any) -> bool:
"""Recognize every model-side compaction carrier shape.
SessionDB does not persist the in-process metadata marker, so client
projections must share the compressor's content classifier rather than a
prefix-only approximation that misses merge-into-tail carriers.
"""
if not isinstance(message, dict):
return False
from agent.context_compressor import is_compaction_summary_message
return is_compaction_summary_message(message)
def _project_client_message(message: Dict[str, Any]) -> Dict[str, Any]:
"""Remove model-only compaction scaffolding from a client message.
Standalone handoffs have no transcript content and remain as hidden empty
rows so clients can reconcile stable message identities. Merged handoffs
preserve only the real prior-tail content that precedes the internal
summary delimiter. Tool calls are dropped from both shapes because a
carrier's inherited calls are historical context, not live client output.
"""
from agent.compaction_display import (
_COMPACTION_INTERNAL_FIELDS,
project_compaction_message_for_display,
)
projected = project_compaction_message_for_display(message)
if projected is None:
projected = message.copy()
for internal_key in _COMPACTION_INTERNAL_FIELDS:
projected.pop(internal_key, None)
projected["content"] = ""
projected["display_kind"] = "hidden"
return projected
def _auto_truncate_response_history(
conversation_history: List[Dict[str, Any]],
*,
limit: int = RESPONSES_AUTO_TRUNCATION_HISTORY_LIMIT,
) -> List[Dict[str, Any]]:
"""Keep recent Responses history without dropping the compaction handoff.
Compaction summaries are preserved wherever they sit in the history —
the gateway /compress path can leave them after a retained system head
(see ``context_compressor`` force-user-leading handling), so a
leading-block-only scan would silently drop them.
"""
if limit <= 0 or len(conversation_history) <= limit:
return conversation_history
summary_indices = [
index
for index, message in enumerate(conversation_history)
if _is_compressed_summary_message(message)
]
if not summary_indices:
return conversation_history[-limit:]
kept_indices = set(summary_indices[:limit])
remaining = limit - len(kept_indices)
if remaining > 0:
summary_index_set = set(summary_indices)
for index in range(len(conversation_history) - 1, -1, -1):
if index in summary_index_set:
continue
kept_indices.add(index)
remaining -= 1
if remaining <= 0:
break
return [conversation_history[index] for index in sorted(kept_indices)]
def _normalize_chat_content(
content: Any, *, _max_depth: int = 10, _depth: int = 0,
) -> str:
"""Normalize OpenAI chat message content into a plain text string.
Some clients (Open WebUI, LobeChat, etc.) send content as an array of
typed parts instead of a plain string::
[{"type": "text", "text": "hello"}, {"type": "input_text", "text": "..."}]
This function flattens those into a single string so the agent pipeline
(which expects strings) doesn't choke.
Defensive limits prevent abuse: recursion depth, list size, and output
length are all bounded.
"""
if _depth > _max_depth:
return ""
if content is None:
return ""
if isinstance(content, str):
return content[:MAX_NORMALIZED_TEXT_LENGTH] if len(content) > MAX_NORMALIZED_TEXT_LENGTH else content
if isinstance(content, list):
parts: List[str] = []
total_len = 0
items = content[:MAX_CONTENT_LIST_SIZE] if len(content) > MAX_CONTENT_LIST_SIZE else content
for item in items:
if isinstance(item, str):
if item:
part = item[:MAX_NORMALIZED_TEXT_LENGTH]
parts.append(part)
total_len += len(part)
elif isinstance(item, dict):
item_type = str(item.get("type") or "").strip().lower()
if item_type in {"text", "input_text", "output_text"}:
text = item.get("text", "")
if text:
try:
part = str(text)[:MAX_NORMALIZED_TEXT_LENGTH]
parts.append(part)
total_len += len(part)
except Exception:
pass
# Silently skip image_url / other non-text parts
elif isinstance(item, list):
nested = _normalize_chat_content(item, _max_depth=_max_depth, _depth=_depth + 1)
if nested:
parts.append(nested)
total_len += len(nested)
# Check accumulated size
if total_len >= MAX_NORMALIZED_TEXT_LENGTH:
break
result = "\n".join(parts)
return result[:MAX_NORMALIZED_TEXT_LENGTH] if len(result) > MAX_NORMALIZED_TEXT_LENGTH else result
# Fallback for unexpected types (int, float, bool, etc.)
try:
result = str(content)
return result[:MAX_NORMALIZED_TEXT_LENGTH] if len(result) > MAX_NORMALIZED_TEXT_LENGTH else result
except Exception:
return ""
# Content part type aliases used by the OpenAI Chat Completions and Responses
# APIs. We accept both spellings on input and emit a single canonical internal
# shape (``{"type": "text", ...}`` / ``{"type": "image_url", ...}``) that the
# rest of the agent pipeline already understands.
_TEXT_PART_TYPES = frozenset({"text", "input_text", "output_text"})
_IMAGE_PART_TYPES = frozenset({"image_url", "input_image"})
_FILE_PART_TYPES = frozenset({"file", "input_file"})
def _normalize_multimodal_content(content: Any) -> Any:
"""Validate and normalize multimodal content for the API server.
Returns a plain string when the content is text-only, or a list of
``{"type": "text"|"image_url", ...}`` parts when images are present.
The output shape is the native OpenAI Chat Completions vision format,
which the agent pipeline accepts verbatim (OpenAI-wire providers) or
converts (``_preprocess_anthropic_content`` for Anthropic).
Raises ``ValueError`` with an OpenAI-style code on invalid input:
* ``unsupported_content_type`` — file/input_file/file_id parts, or
non-image ``data:`` URLs.
* ``invalid_image_url`` — missing URL or unsupported scheme.
* ``invalid_content_part`` — malformed text/image objects.
Callers translate the ValueError into a 400 response.
"""
# Scalar passthrough mirrors ``_normalize_chat_content``.
if content is None:
return ""
if isinstance(content, str):
return content[:MAX_NORMALIZED_TEXT_LENGTH] if len(content) > MAX_NORMALIZED_TEXT_LENGTH else content
if not isinstance(content, list):
# Mirror the legacy text-normalizer's fallback so callers that
# pre-existed image support still get a string back.
return _normalize_chat_content(content)
items = content[:MAX_CONTENT_LIST_SIZE] if len(content) > MAX_CONTENT_LIST_SIZE else content
normalized_parts: List[Dict[str, Any]] = []
text_accum_len = 0
for part in items:
if isinstance(part, str):
if part:
trimmed = part[:MAX_NORMALIZED_TEXT_LENGTH]
normalized_parts.append({"type": "text", "text": trimmed})
text_accum_len += len(trimmed)
continue
if not isinstance(part, dict):
# Ignore unknown scalars for forward compatibility with future
# Responses API additions (e.g. ``refusal``). The same policy
# the text normalizer applies.
continue
raw_type = part.get("type")
part_type = str(raw_type or "").strip().lower()
if part_type in _TEXT_PART_TYPES:
text = part.get("text")
if text is None:
continue
if not isinstance(text, str):
text = str(text)
if text:
trimmed = text[:MAX_NORMALIZED_TEXT_LENGTH]
normalized_parts.append({"type": "text", "text": trimmed})
text_accum_len += len(trimmed)
continue
if part_type in _IMAGE_PART_TYPES:
detail = part.get("detail")
image_ref = part.get("image_url")
# OpenAI Responses sends ``input_image`` with a top-level
# ``image_url`` string; Chat Completions sends ``image_url`` as
# ``{"url": "...", "detail": "..."}``. Support both.
if isinstance(image_ref, dict):
url_value = image_ref.get("url")
detail = image_ref.get("detail", detail)
else:
url_value = image_ref
if not isinstance(url_value, str) or not url_value.strip():
raise ValueError("invalid_image_url:Image parts must include a non-empty image URL.")
url_value = url_value.strip()
lowered = url_value.lower()
if lowered.startswith("data:"):
if not lowered.startswith("data:image/") or "," not in url_value:
raise ValueError(
"unsupported_content_type:Only image data URLs are supported. "
"Non-image data payloads are not supported."
)
elif not (lowered.startswith("http://") or lowered.startswith("https://")):
raise ValueError(
"invalid_image_url:Image inputs must use http(s) URLs or data:image/... URLs."
)
image_part: Dict[str, Any] = {"type": "image_url", "image_url": {"url": url_value}}
if detail is not None:
if not isinstance(detail, str) or not detail.strip():
raise ValueError("invalid_content_part:Image detail must be a non-empty string when provided.")
image_part["image_url"]["detail"] = detail.strip()
normalized_parts.append(image_part)
continue
if part_type in _FILE_PART_TYPES:
raise ValueError(
"unsupported_content_type:Inline image inputs are supported, "
"but uploaded files and document inputs are not supported on this endpoint."
)
# Unknown part type — reject explicitly so clients get a clear error
# instead of a silently dropped turn.
raise ValueError(
f"unsupported_content_type:Unsupported content part type {raw_type!r}. "
"Only text and image_url/input_image parts are supported."
)
if not normalized_parts:
return ""
# Text-only: collapse to a plain string so downstream logging/trajectory
# code sees the native shape and prompt caching on text-only turns is
# unaffected.
if all(p.get("type") == "text" for p in normalized_parts):
return "\n".join(p["text"] for p in normalized_parts if p.get("text"))
return normalized_parts
def _content_has_visible_payload(content: Any) -> bool:
"""True when content has any text or image attachment. Used to reject empty turns."""
if isinstance(content, str):
return bool(content.strip())
if isinstance(content, list):
for part in content:
if isinstance(part, dict):
ptype = str(part.get("type") or "").strip().lower()
if ptype in _TEXT_PART_TYPES and str(part.get("text") or "").strip():
return True
if ptype in _IMAGE_PART_TYPES:
return True
return False
def _multimodal_validation_error(exc: ValueError, *, param: str) -> "web.Response":
"""Translate a ``_normalize_multimodal_content`` ValueError into a 400 response."""
raw = str(exc)
code, _, message = raw.partition(":")
if not message:
code, message = "invalid_content_part", raw
return web.json_response(
_openai_error(message, code=code, param=param),
status=400,
)
def _reap_disconnected_agent_processes(
agent: Any, *, source: str = "api_server_sse_disconnect"
) -> None:
"""Reap background processes an abandoned API-server turn created.
Mirrors the gateway-turn cleanup in ``gateway/run.py`` (#76115) for this
API-server surface, which runs its own agent lifecycle via ``_run_agent``
and never passes through ``TurnRunner`` — so it needs its own trigger for
the same baseline-diff reap. Fire-and-forget on a daemon thread so the
SSE handler's own cleanup isn't blocked on process-tree teardown.
Reaping is epoch-gated: client-provided session IDs are conversation
scopes, and multiple concurrent runs can intentionally share one (see
``_handle_runs``). Without the gate, run A disconnecting could kill a
process a still-live run B (same task_id) spawned after A's baseline
snapshot — the same stale-reaper bug class the gateway path gates via
``run_generation``. The epoch closure skips the reap when a newer run
has since claimed the task_id; that newer run's own baseline covers its
eventual cleanup.
"""
process_task_id = getattr(agent, "_gateway_turn_process_task_id", "")
process_baseline = getattr(agent, "_gateway_turn_process_baseline", None)
if not process_task_id or process_baseline is None:
return
epoch = getattr(agent, "_gateway_turn_process_epoch", None)
is_still_current: Optional[Any] = None
if epoch is not None:
def _epoch_still_current(_task_id=process_task_id, _epoch=epoch):
# Skip only when a NEWER run has claimed this task_id. A missing
# entry means the abandoned run's own clear pruned it (worker
# returned after the interrupt) — no newer claimant exists, so
# the reap must still proceed or the leak survives. This matches
# the gateway gate's semantics: worker completion does not bump
# run_generation either.
with _TURN_PROCESS_EPOCH_LOCK:
current = _TURN_PROCESS_EPOCHS.get(_task_id)
return current is None or current == _epoch
is_still_current = _epoch_still_current
from gateway.run import _reap_gateway_turn_processes
threading.Thread(
target=_reap_gateway_turn_processes,
args=(process_task_id, process_baseline),
kwargs={"source": source, "is_still_current": is_still_current},
name=f"api-turn-reaper-{process_task_id[:12]}",
daemon=True,
).start()
# Per-task-id run epochs for the reap gate above. task_id is a conversation
# scope shared by concurrent API runs, so each run that claims it bumps the
# epoch; a reaper holding a stale epoch declines to kill. Epochs come from a
# single monotonic counter (never reused), so pruning an entry and later
# re-claiming the task_id can never resurrect a stale reaper's claim.
# Entries are pruned on clear when still current, bounding the dict to
# in-flight runs.
_TURN_PROCESS_EPOCHS: Dict[str, int] = {}
_TURN_PROCESS_EPOCH_LOCK = threading.Lock()
_TURN_PROCESS_EPOCH_COUNTER = itertools.count(1)
def _publish_turn_process_ownership(agent: Any, task_id: str) -> None:
"""Snapshot the process baseline and claim the task_id's current epoch.
Single place all API-server agent lifecycles (chat/responses ``_run_agent``
and ``/v1/runs``) record turn ownership, so the marker attribute names and
epoch bookkeeping cannot drift between surfaces.
"""
from tools.process_registry import process_registry
with _TURN_PROCESS_EPOCH_LOCK:
epoch = next(_TURN_PROCESS_EPOCH_COUNTER)
_TURN_PROCESS_EPOCHS[task_id] = epoch
agent._gateway_turn_process_task_id = task_id
agent._gateway_turn_process_baseline = process_registry.snapshot_running_ids(
task_id
)
agent._gateway_turn_process_epoch = epoch
def _clear_turn_process_ownership(agent: Any) -> None:
"""Clear turn ownership the moment the turn finishes (success or crash).
A disconnect/cancel landing after this point must not reap background
work the turn deliberately left running — mirrors the same race-window
guard in ``gateway/run.py``'s ``_run_sync_with_timeout_lifecycle``.
"""
task_id = getattr(agent, "_gateway_turn_process_task_id", "")
epoch = getattr(agent, "_gateway_turn_process_epoch", None)
if task_id and epoch is not None:
with _TURN_PROCESS_EPOCH_LOCK:
# Prune only when this run is still the current claimant; a
# newer concurrent run owns the entry otherwise.
if _TURN_PROCESS_EPOCHS.get(task_id) == epoch:
del _TURN_PROCESS_EPOCHS[task_id]
agent._gateway_turn_process_task_id = ""
agent._gateway_turn_process_baseline = frozenset()
agent._gateway_turn_process_epoch = None
def _session_chat_user_message(body: Dict[str, Any], *, param: str = "message") -> tuple[Any, Optional["web.Response"]]:
"""Parse and normalize session chat ``message`` / ``input`` like chat completions."""
user_message = body.get("message") or body.get("input")
if not _content_has_visible_payload(user_message):
return None, web.json_response(
_openai_error("Missing 'message' field", code="missing_message"),
status=400,
)
try:
return _normalize_multimodal_content(user_message), None
except ValueError as exc:
return None, _multimodal_validation_error(exc, param=param)
def check_api_server_requirements() -> bool:
"""Check if API server dependencies are available."""
return AIOHTTP_AVAILABLE
class ResponseStore:
"""
SQLite-backed LRU store for Responses API state.
Each stored response includes the full internal conversation history
(with tool calls and results) so it can be reconstructed on subsequent
requests via previous_response_id.
Persists across gateway restarts. Falls back to in-memory SQLite
if the on-disk path is unavailable.
"""
def __init__(self, max_size: int = MAX_STORED_RESPONSES, db_path: str = None):
self._max_size = max_size
if db_path is None:
try:
from hermes_cli.config import get_hermes_home
db_path = str(get_hermes_home() / "response_store.db")
except Exception:
db_path = ":memory:"
self._db_path: Optional[str] = db_path if db_path != ":memory:" else None
try:
self._conn = sqlite3.connect(db_path, check_same_thread=False)
except Exception:
self._conn = sqlite3.connect(":memory:", check_same_thread=False)
self._db_path = None
# Use shared WAL-fallback helper so response_store.db degrades
# gracefully on NFS/SMB/FUSE-mounted HERMES_HOME (same filesystem
# issue addressed for state.db/kanban.db — see
# hermes_state._WAL_INCOMPAT_MARKERS).
from hermes_state import apply_wal_with_fallback
apply_wal_with_fallback(self._conn, db_label="response_store.db")
self._conn.execute(
"""CREATE TABLE IF NOT EXISTS responses (
response_id TEXT PRIMARY KEY,
data TEXT NOT NULL,
accessed_at REAL NOT NULL
)"""
)
self._conn.execute(
"""CREATE TABLE IF NOT EXISTS conversations (
name TEXT PRIMARY KEY,
response_id TEXT NOT NULL
)"""
)
self._conn.commit()
# response_store.db contains conversation history (tool payloads,
# prompts, results). Tighten to owner-only after creation so other
# local users on a shared box can't read it. Run once at __init__
# rather than after every commit — chmod-on-every-write is wasted
# syscalls on a hot path.
self._tighten_file_permissions()
def _tighten_file_permissions(self) -> None:
"""Force owner-only permissions on the DB and SQLite sidecars."""
if not self._db_path:
return
for candidate in (
Path(self._db_path),
Path(f"{self._db_path}-wal"),
Path(f"{self._db_path}-shm"),
):
try:
if candidate.exists():
candidate.chmod(0o600)
except OSError:
logger.debug(
"Failed to restrict response store permissions for %s",
candidate,
exc_info=True,
)
def get(self, response_id: str) -> Optional[Dict[str, Any]]:
"""Retrieve a stored response by ID (updates access time for LRU)."""
row = self._conn.execute(
"SELECT data FROM responses WHERE response_id = ?", (response_id,)
).fetchone()
if row is None:
return None
self._conn.execute(
"UPDATE responses SET accessed_at = ? WHERE response_id = ?",
(time.time(), response_id),
)
self._conn.commit()
try:
return json.loads(row[0])
except (json.JSONDecodeError, TypeError):
logger.warning(
"Corrupted JSON in response store for id=%s, evicting entry",
response_id,
)
self._conn.execute(
"DELETE FROM responses WHERE response_id = ?",
(response_id,),
)
self._conn.commit()
return None
def put(self, response_id: str, data: Dict[str, Any]) -> None:
"""Store a response, evicting the oldest if at capacity."""
self._conn.execute(
"INSERT OR REPLACE INTO responses (response_id, data, accessed_at) VALUES (?, ?, ?)",
(response_id, json.dumps(data, default=str), time.time()),
)
# Evict oldest entries beyond max_size
count = self._conn.execute("SELECT COUNT(*) FROM responses").fetchone()[0]
if count > self._max_size:
# Collect IDs that will be evicted
evict_ids = [
row[0]
for row in self._conn.execute(
"SELECT response_id FROM responses ORDER BY accessed_at ASC LIMIT ?",
(count - self._max_size,),
).fetchall()
]
if evict_ids:
placeholders = ",".join("?" for _ in evict_ids)
# Clear conversation mappings pointing to evicted responses
self._conn.execute(
f"DELETE FROM conversations WHERE response_id IN ({placeholders})",
evict_ids,
)
# Delete evicted responses
self._conn.execute(
f"DELETE FROM responses WHERE response_id IN ({placeholders})",
evict_ids,
)
self._conn.commit()
def delete(self, response_id: str) -> bool:
"""Remove a response from the store. Returns True if found and deleted."""
# Clear conversation mappings pointing to this response
self._conn.execute(
"DELETE FROM conversations WHERE response_id = ?", (response_id,)
)
cursor = self._conn.execute(
"DELETE FROM responses WHERE response_id = ?", (response_id,)
)
self._conn.commit()
return cursor.rowcount > 0
def get_conversation(self, name: str) -> Optional[str]:
"""Get the latest response_id for a conversation name."""
row = self._conn.execute(
"SELECT response_id FROM conversations WHERE name = ?", (name,)
).fetchone()
return row[0] if row else None
def set_conversation(self, name: str, response_id: str) -> None:
"""Map a conversation name to its latest response_id."""
self._conn.execute(
"INSERT OR REPLACE INTO conversations (name, response_id) VALUES (?, ?)",
(name, response_id),
)
self._conn.commit()
def close(self) -> None:
"""Close the database connection."""
try:
self._conn.close()
except Exception:
pass
def __len__(self) -> int:
row = self._conn.execute("SELECT COUNT(*) FROM responses").fetchone()
return row[0] if row else 0
# ---------------------------------------------------------------------------
# CORS middleware
# ---------------------------------------------------------------------------
_CORS_HEADERS = {
"Access-Control-Allow-Methods": "GET, POST, DELETE, OPTIONS",
"Access-Control-Allow-Headers": "Authorization, Content-Type, Idempotency-Key",
}
if AIOHTTP_AVAILABLE:
@web.middleware
async def cors_middleware(request, handler):
"""Add CORS headers for explicitly allowed origins; handle OPTIONS preflight."""
adapter = request.app.get("api_server_adapter")
origin = request.headers.get("Origin", "")
cors_headers = None
if adapter is not None:
if not adapter._origin_allowed(origin):
return web.Response(status=403)
cors_headers = adapter._cors_headers_for_origin(origin)
if request.method == "OPTIONS":
if cors_headers is None:
return web.Response(status=403)
return web.Response(status=200, headers=cors_headers)
response = await handler(request)
if cors_headers is not None:
response.headers.update(cors_headers)
return response
else:
cors_middleware = None # type: ignore[assignment]
_MEDIA_IMG_EXT = {".png", ".jpg", ".jpeg", ".gif", ".webp", ".bmp"}
_MEDIA_MIME = {
".png": "image/png",
".jpg": "image/jpeg",
".jpeg": "image/jpeg",
".gif": "image/gif",
".webp": "image/webp",
".bmp": "image/bmp",
}
_MEDIA_DATA_URL_MAX_BYTES = 5 * 1024 * 1024 # skip images larger than 5MB
def _resolve_media_to_data_urls(text: str) -> str:
"""Replace ``MEDIA:<path>`` image tags with inline base64 data URLs.
Remote OpenAI-compatible frontends can't read local file paths, so
``MEDIA:`` tags referencing images on the server are useless to them.
Inline small local images as markdown data URLs; non-image or unreadable
paths are left untouched.
Uses the same anchored ``MEDIA_TAG_CLEANUP_RE`` matcher and
``validate_media_delivery_path`` safety check every other platform
adapter's media delivery already goes through (gateway/platforms/base.py)
— an absolute-path anchor plus a known-extension requirement, and a
resolved-path check against the credential/system-path denylist. The
prior pattern here matched any bare token after ``MEDIA:`` (including a
relative/traversal path like ``../../etc/passwd.png``) and read the file
directly with no denylist, so any image-suffixed, readable file the
process could see was base64-exfiltrated to the API caller if its path
merely appeared in the model's own final reply text.
"""
if not text or "MEDIA:" not in text:
return text
import base64
def _to_data_url(path_str: str) -> Optional[str]:
# validate_media_delivery_path() strips wrapping quotes/backticks
# and trailing punctuation internally, same as MEDIA_TAG_CLEANUP_RE's
# other callers (extract_media / _strip_media_tag_directives) rely on.
safe_path = validate_media_delivery_path(path_str)
if not safe_path:
return None
p = Path(safe_path)
suffix = p.suffix.lower()
if suffix not in _MEDIA_IMG_EXT:
return None
try:
if p.stat().st_size > _MEDIA_DATA_URL_MAX_BYTES:
return None
b64 = base64.b64encode(p.read_bytes()).decode()
except OSError:
return None
return f"![image](data:{_MEDIA_MIME[suffix]};base64,{b64})"
def _repl(m: "re.Match[str]") -> str:
return _to_data_url(m.group("path")) or m.group(0)
try:
return MEDIA_TAG_CLEANUP_RE.sub(_repl, text)
except Exception:
return text
def _redact_api_error_text(value: Any, *, limit: int | None = None) -> str:
"""Redact API-bound error text before it crosses the HTTP boundary."""
redacted = redact_sensitive_text(str(value), force=True)
if limit is not None:
return redacted[:limit]
return redacted
def _openai_error(message: str, err_type: str = "invalid_request_error", param: str = None, code: str = None) -> Dict[str, Any]:
"""OpenAI-style error envelope."""
return {
"error": {
"message": _redact_api_error_text(message),
"type": err_type,
"param": param,
"code": code,
}
}
_api_agent_request_reservation: ContextVar[Optional[dict[str, bool]]] = ContextVar(
"api_agent_request_reservation", default=None
)
def _admit_api_agent_request(handler):
"""Reserve an authenticated API turn before its handler first awaits.
Gateway shutdown and aiohttp requests share an event loop. Keeping the
drain check and reservation in one non-awaiting block prevents a request
admitted immediately before shutdown from becoming invisible while it is
still parsing its body or resolving session state. The mutable reservation
is intentionally shared with child tasks so agent/task bookkeeping releases
this one slot exactly once.
"""
@wraps(handler)
async def _wrapped(self, request, *args, **kwargs):
auth_err = (
self._check_run_auth(request, permission="dispatch")
if _api_runs._uses_room_run_auth(self, request)
else self._check_auth(request)
)
if auth_err:
return auth_err
draining = self._draining_response()
if draining is not None:
return draining
reservation = {"active": True}
token = _api_agent_request_reservation.set(reservation)
self._pending_agent_requests += 1
try:
return await handler(self, request, *args, **kwargs)
finally:
if reservation["active"]:
reservation["active"] = False
self._pending_agent_requests = max(0, self._pending_agent_requests - 1)
_api_agent_request_reservation.reset(token)
return _wrapped
def _release_pending_api_work(adapter, reservation: dict[str, bool]) -> None:
"""Release a pending-work reservation exactly once."""
if reservation["active"]:
reservation["active"] = False
adapter._pending_agent_requests = max(0, adapter._pending_agent_requests - 1)
@contextmanager
def _reserve_pending_api_work(adapter):
"""Keep externally-triggered background work visible across awaits.
A handler can detach the reservation to an asyncio task; its done callback
then owns release so shutdown cannot miss the handoff to background work.
"""
reservation = {"active": True, "detached": False}
adapter._pending_agent_requests += 1
try:
yield reservation
finally:
if not reservation["detached"]:
_release_pending_api_work(adapter, reservation)
if AIOHTTP_AVAILABLE:
@web.middleware
async def body_limit_middleware(request, handler):
"""Reject overly large request bodies early based on Content-Length."""
if request.method in {"POST", "PUT", "PATCH"}:
cl = request.headers.get("Content-Length")
if cl is not None:
try:
if int(cl) > MAX_REQUEST_BYTES:
return web.json_response(_openai_error("Request body too large.", code="body_too_large"), status=413)
except ValueError:
return web.json_response(_openai_error("Invalid Content-Length header.", code="invalid_content_length"), status=400)
try:
return await handler(request)
except web.HTTPRequestEntityTooLarge:
# aiohttp's client_max_size tripped mid-read (chunked bodies carry
# no Content-Length) — return a proper 413 instead of letting the
# handler's broad JSON except turn it into 400 "Invalid JSON".
return web.json_response(
_openai_error("Request body too large.", code="body_too_large"),
status=413,
)
else:
body_limit_middleware = None # type: ignore[assignment]
_SECURITY_HEADERS = {
"Content-Security-Policy": "default-src 'none'; frame-ancestors 'none'",
"Permissions-Policy": "camera=(), microphone=(), geolocation=()",
"Strict-Transport-Security": "max-age=31536000; includeSubDomains",
"X-Content-Type-Options": "nosniff",
"X-Frame-Options": "DENY",
"X-XSS-Protection": "0",
"Referrer-Policy": "no-referrer",
}
if AIOHTTP_AVAILABLE:
@web.middleware
async def security_headers_middleware(request, handler):
"""Add security headers to all responses (including errors)."""
response = await handler(request)
for k, v in _SECURITY_HEADERS.items():
response.headers.setdefault(k, v)
return response
else:
security_headers_middleware = None # type: ignore[assignment]
class _IdempotencyCache:
"""In-memory idempotency cache with TTL and basic LRU semantics."""
def __init__(self, max_items: int = 1000, ttl_seconds: int = 300):
from collections import OrderedDict
self._store = OrderedDict()
self._inflight: Dict[tuple[str, str], "asyncio.Task[Any]"] = {}
self._ttl = ttl_seconds
self._max = max_items
def _purge(self):
now = time.time()
expired = [k for k, v in self._store.items() if now - v["ts"] > self._ttl]
for k in expired:
self._store.pop(k, None)
while len(self._store) > self._max:
self._store.popitem(last=False)
async def get_or_set(self, key: str, fingerprint: str, compute_coro):
self._purge()
item = self._store.get(key)
if item and item["fp"] == fingerprint:
return item["resp"]
inflight_key = (key, fingerprint)
task = self._inflight.get(inflight_key)
if task is None:
async def _compute_and_store():
resp = await compute_coro()
import time as _t
self._store[key] = {"resp": resp, "fp": fingerprint, "ts": _t.time()}
self._purge()
return resp
task = asyncio.create_task(_compute_and_store())
self._inflight[inflight_key] = task
def _clear_inflight(done_task: "asyncio.Task[Any]") -> None:
if self._inflight.get(inflight_key) is done_task:
self._inflight.pop(inflight_key, None)
task.add_done_callback(_clear_inflight)
return await asyncio.shield(task)
_idem_cache = _IdempotencyCache()
def _make_request_fingerprint(body: Dict[str, Any], keys: List[str]) -> str:
from hashlib import sha256
subset = {k: body.get(k) for k in keys}
return sha256(repr(subset).encode("utf-8")).hexdigest()
def _derive_chat_session_id(
system_prompt: Optional[str],
first_user_message: str,
) -> str:
"""Derive a stable session ID from the conversation's first user message.
OpenAI-compatible frontends (Open WebUI, LibreChat, etc.) send the full
conversation history with every request. The system prompt and first user
message are constant across all turns of the same conversation, so hashing
them produces a deterministic session ID that lets the API server reuse
the same Hermes session (and therefore the same Docker container sandbox
directory) across turns.
"""
seed = f"{system_prompt or ''}\n{first_user_message}"
digest = hashlib.sha256(seed.encode("utf-8")).hexdigest()[:16]
return f"api-{digest}"
_CRON_AVAILABLE = False
try:
from cron.jobs import (
list_jobs as _cron_list,
get_job as _cron_get,
update_job as _cron_update,
remove_job as _cron_remove,
pause_job as _cron_pause,
resume_job as _cron_resume,
trigger_job as _cron_trigger,
)
from cron.scheduler import (
CronSchedulerRegistrationError as _CronSchedulerRegistrationError,
create_job_with_scheduler_registration as _cron_create,
)
_CRON_AVAILABLE = True
except ImportError:
_cron_list = None
_cron_get = None
_cron_create = None
_cron_update = None
_cron_remove = None
_cron_pause = None
_cron_resume = None
_cron_trigger = None
class _CronSchedulerRegistrationError(RuntimeError):
pass
def _notify_cron_provider_jobs_changed() -> None:
"""Tell the active cron scheduler provider the job set changed after a REST
mutation (no-op for the built-in). Best-effort — never breaks the handler."""
try:
from cron.scheduler import _notify_provider_jobs_changed
_notify_provider_jobs_changed()
except Exception:
pass
# Defense-in-depth: mirror the agent-facing cronjob tool, which scans the
# user-supplied prompt for exfiltration/injection payloads at create/update
# time (tools/cronjob_tools.py). The REST cron endpoints are authenticated
# (every handler runs _check_auth, and connect() refuses to start without
# API_SERVER_KEY), so this is not the trust boundary — it's parity with the
# tool path so a malicious prompt is rejected the same way regardless of
# which surface created the job. Imported defensively: a missing scanner
# must not disable the cron REST API.
try:
from tools.cronjob_tools import _scan_cron_prompt as _scan_cron_prompt
except Exception: # pragma: no cover - scanner is optional hardening
_scan_cron_prompt = None
class _ProviderAuthResolutionError(RuntimeError):
"""Raised only when gateway.run._resolve_runtime_agent_kwargs() fails
to resolve provider credentials.
That function is the sole raiser of RuntimeError(format_runtime_
provider_error(...)) anywhere in _create_agent()'s call graph.
Re-raising it as this dedicated subclass -- instead of catching bare
RuntimeError around the much wider _create_agent()+run_conversation()
span -- lets callers distinguish "provider auth/credential failure"
from any other RuntimeError a provider adapter or run_conversation()
might legitimately raise (e.g. run_agent.py's "Failed to recreate
closed OpenAI client"), which a bare `except RuntimeError` there would
otherwise mislabel as an auth failure.
"""
class APIServerAdapter(BasePlatformAdapter):
"""
OpenAI-compatible HTTP API server adapter.
Runs an aiohttp web server that accepts OpenAI-format requests
and routes them through hermes-agent's AIAgent.
"""
# Stateless request/response: every route (the OpenAI-spec
# /v1/chat/completions and /v1/responses, and the proprietary /v1/runs SSE
# stream) tears down its channel when the turn ends. There is no persistent
# outbound channel to push a background completion to a client that already
# received its response, and ``send()`` is a no-op stub. So async-delivery
# tools (terminal notify_on_complete / watch_patterns, delegate_task
# background=True) must NOT promise delivery on this path — see
# ``async_delivery_supported()``.
supports_async_delivery: bool = False
# Same statelessness applies to the startup auto-resume prompt: no client
# is waiting to answer "session restored — what next?", so a resumed turn
# should complete the interrupted work rather than acknowledge (#57056).
interactive_resume: bool = False
def __init__(self, config: PlatformConfig):
super().__init__(config, Platform.API_SERVER)
extra = config.extra or {}
self._host: str = extra.get("host", os.getenv("API_SERVER_HOST", DEFAULT_HOST))
raw_port = extra.get("port")
if raw_port is None:
raw_port = os.getenv("API_SERVER_PORT", str(DEFAULT_PORT))
self._port: int = _coerce_port(raw_port, DEFAULT_PORT)
self._api_key: str = extra.get("key", _get_scoped_secret("API_SERVER_KEY", ""))
self._cors_origins: tuple[str, ...] = self._parse_cors_origins(
extra.get("cors_origins", os.getenv("API_SERVER_CORS_ORIGINS", "")),
)
self._model_name: str = self._resolve_model_name(
extra.get("model_name", os.getenv("API_SERVER_MODEL_NAME", "")),
)
# model_routes: maps incoming ``model`` field values to specific
# provider/model configs so one API server instance can serve
# multiple clients on different backends.
#
# Config format (platforms.api_server.extra in the gateway config):
# model_routes:
# minimax-m2: # alias the client sends as the "model" field
# model: "minimax/minimax-m1"
# provider: "openrouter" # optional — resolved via the provider
# # credential chain when set
# api_key: "sk-…" # optional — per-route UPSTREAM provider
# # key override (NOT caller auth; never logged)
# base_url: "https://…" # optional — per-route base URL override
self._model_routes: Dict[str, Dict[str, Any]] = self._parse_model_routes(
extra.get("model_routes"),
)
# direct_model_requests: opt-in passthrough for a bare ``model`` value
# (no ``provider``) on the OpenAI-compatible surfaces
# (/v1/chat/completions, /v1/responses). Off by default: generic
# OpenAI clients routinely hardcode model names ("gpt-4o", ...), and
# existing deployments rely on those falling back to the gateway
# default rather than switching the executing model. Requests that
# send an explicit ``provider`` — and the Hermes-native session-chat
# and /v1/runs endpoints — are always honored regardless of this flag.
# (Idea credit: PR #22825 by @mssteuer.)
self._direct_model_requests: bool = _coerce_request_bool(
extra.get("direct_model_requests"), default=False
)
self._app: Optional["web.Application"] = None
self._runner: Optional["web.AppRunner"] = None
self._site: Optional["web.TCPSite"] = None
self._response_store = ResponseStore()
_api_runs._initialize_run_state(
self,
store_factory=RunIdempotencyStore,
)
self._session_db: Optional[Any] = None # Lazy-init SessionDB for session continuity
self._session_dbs: Dict[str, Any] = {}
self._session_db_cache_lock = threading.Lock()
self._session_db_cache_closed = False
# Last-known-good resolved model per session (keyed by gateway_session_key
# ONLY — never session_id, which rotates/is ephemeral for one-off API
# server requests; "*" is the process-wide fallback), mirroring
# GatewayRunner._last_resolved_model in run.py — recovers from a
# transient empty model resolution (#35314) instead of building an
# agent with model="" that 400s every call until manual retry.
self._last_resolved_model: Dict[str, str] = {}
self._session_db_lock: Optional[asyncio.Lock] = None # Single-flight for lazy init
# Concurrency cap shared across all agent-serving endpoints
# (/v1/chat/completions, /v1/responses, /v1/runs). Read from
# config.yaml gateway.api_server.max_concurrent_runs; 0 disables
# the cap. Bounds CPU / memory / upstream-LLM-quota exhaustion
# from a request flood (#7483).
self._max_concurrent_runs: int = self._resolve_max_concurrent_runs()
# Number of in-flight runs on the non-streaming chat/responses paths
# (the /v1/runs path tracks its own in-flight set via
# _active_run_tasks).
self._inflight_agent_runs: int = 0
# Every agent currently inside _run_agent(), i.e. exactly the turns
# counted by _inflight_agent_runs above. Shutdown needs the whole
# adapter-owned set, so this is deliberately NOT _active_run_agents:
# that one is run_id-keyed and scoped to the public /v1/runs stop API,
# and only /v1/runs has a run_id at all. Keyed by id() because the
# other six agent-entry paths have no stable identifier of their own;
# the dict holds a strong reference for the life of the turn, so an
# id() can never be recycled while it is still registered.
self._shutdown_interruptible_agents: Dict[int, Any] = {}
# Back-reference to the owning GatewayRunner (set by gateway/run.py)
# so /api/platforms/{platform}/events can resolve sibling adapters.
# BasePlatformAdapter declares the class-level default of None.
self.gateway_runner: Optional[Any] = None
# Requests admitted before their handler reaches agent bookkeeping.
# Shutdown counts this reservation so the request cannot slip through
# the drain between its first await and _run_agent()/task registration.
self._pending_agent_requests: int = 0
# Browser-control broker core: transport-neutral ticket, controller,
# and command lifecycle shared with the dashboard Gateway transport. This adapter only maps HTTP registration and the
# controller WebSocket onto the broker; it owns no broker state.
self._browser_control_broker = get_browser_control_broker()
# One-shot artifact transport (Phase 8 Task 29). Lazy per-profile
# stores + limiter are created on first authenticated artifact use;
# tests inject their own store/limiter via
# _inject_browser_control_artifacts().
self._browser_control_artifacts: Dict[str, ArtifactStore] = {}
self._browser_control_artifact_limiter: Optional[ArtifactRateLimiter] = None
def active_agent_work_count(self) -> int:
"""Return all live agent work owned by this API adapter.
``/v1/runs`` registers an asyncio task before it constructs and stores
its agent, so ``_active_run_agents`` has a real queued-before-agent gap.
Reuse the task-based accounting used by the concurrent-run limit: it
covers that gap and excludes completed tasks retained until cleanup.
"""
try:
return (
int(getattr(self, "_pending_agent_requests", 0))
+ int(self._inflight_agent_runs)
+ sum(not task.done() for task in self._active_run_tasks.values())
)
except Exception:
return 0
def interrupt_active_runs(self, reason: str) -> int:
"""Cooperatively interrupt every adapter-owned agent during shutdown.
The gateway drain accounts for API-server work through
``active_agent_work_count()``, but those agents are owned by this
adapter rather than ``GatewayRunner._running_agents``, so
``GatewayRunner._interrupt_running_agents()`` never reaches them: the
turn runs to the drain timeout with no cooperative interrupt and is
then amputated by the post-interrupt tool-subprocess kill.
Cover the same set the drain waits on, so accounting and interrupt
agree:
* ``_active_run_agents`` — the ``/v1/runs`` agents counted through
``_active_run_tasks``.
* ``_shutdown_interruptible_agents`` — every ``_run_agent()`` turn
counted through ``_inflight_agent_runs``, i.e. both session-chat
routes, ``/v1/chat/completions`` and ``/v1/responses`` in their
streaming and non-streaming forms.
``_pending_agent_requests`` is intentionally not covered: it counts
admitted requests that have not constructed an agent yet, so there is
no object to interrupt.
Returns the number of agents that accepted an interrupt.
"""
agents: Dict[int, Any] = {}
for agent in list(self._active_run_agents.values()):
if agent is not None:
agents[id(agent)] = agent
for agent in list(self._shutdown_interruptible_agents.values()):
if agent is not None:
# Dedupe by object identity — the two registries are disjoint
# today (/v1/runs runs its own lifecycle, not _run_agent), but
# an agent published to both must still be interrupted once.
agents[id(agent)] = agent
interrupted = 0
for agent in agents.values():
try:
if request_hard_interrupt(agent, reason):
interrupted += 1
except Exception as exc:
logger.debug("[api_server] failed interrupting active agent: %s", exc)
return interrupted
@staticmethod
def _gateway_is_draining() -> bool:
"""Whether the owning gateway currently refuses new agent turns."""
try:
from gateway.run import _gateway_runner_ref
runner = _gateway_runner_ref()
return bool(
runner
and (
getattr(runner, "_draining", False)
or getattr(runner, "_external_drain_active", False)
)
)
except Exception:
return False
def _draining_response(self) -> Optional["web.Response"]:
"""Return a retryable response while the gateway drains existing work."""
if not self._gateway_is_draining():
return None
return web.json_response(
_openai_error(
"Gateway is draining existing work; retry shortly.",
code="gateway_draining",
),
status=503,
headers={"Retry-After": "1"},
)
def _activate_admitted_request(self) -> None:
"""Transfer this request's drain reservation to agent bookkeeping."""
reservation = _api_agent_request_reservation.get()
if reservation and reservation["active"]:
reservation["active"] = False
self._pending_agent_requests = max(0, self._pending_agent_requests - 1)
def _readiness_work_counts(self) -> tuple[int, int, int]:
"""Return bounded work counts from each subsystem's public state."""
active_api_runs = sum(
1
for status in self._run_statuses.values()
# "stopping" (set by _handle_stop_run) is not terminal: the run
# stays in this state, doing real executor-thread work, until the
# agent actually notices the interrupt and the task settles to
# "cancelled" — an unbounded window, not the old ~5s hard-timeout
# wait. Excluding it here undercounts active_api_runs for the
# whole duration of a cooperative stop.
if status.get("status") in {"queued", "running", "waiting_for_approval", "stopping"}
)
process_depth = 0
active_delegations = 0
try:
from tools.process_registry import process_registry
process_depth = process_registry.completion_queue.qsize()
except Exception:
pass
try:
from tools.async_delegation import active_count
active_delegations = active_count()
except Exception:
pass
return active_api_runs, process_depth, active_delegations
@staticmethod
def _parse_cors_origins(value: Any) -> tuple[str, ...]:
"""Normalize configured CORS origins into a stable tuple."""
if not value:
return ()
if isinstance(value, str):
items = value.split(",")
elif isinstance(value, (list, tuple, set)):
items = value
else:
items = [str(value)]
return tuple(str(item).strip() for item in items if str(item).strip())
@staticmethod
def _resolve_max_concurrent_runs() -> int:
"""Read the concurrent-run cap from config.yaml (0 disables).
gateway.api_server.max_concurrent_runs. Falls back to the historical
default of 10 when unset or malformed. Negative values are clamped
to 0 (disabled).
"""
default = 10
try:
from hermes_cli.config import cfg_get, load_config
raw = cfg_get(
load_config(),
"gateway",
"api_server",
"max_concurrent_runs",
default=default,
)
value = int(raw)
except Exception:
return default
return max(0, value)
@staticmethod
def _resolve_model_name(explicit: str) -> str:
"""Derive the advertised model name for /v1/models.
Priority:
1. Explicit override (config extra or API_SERVER_MODEL_NAME env var)
2. Active profile name (so each profile advertises a distinct model)
3. Fallback: "hermes-agent"
Delegates the tiered fallthrough to
:func:`hermes_cli.model_switch.resolve_effective_model` (the shared
override > mid-tier > default precedence owner).
"""
from hermes_cli.model_switch import resolve_effective_model
profile_name = ""
try:
from hermes_cli.profiles import get_active_profile_name
profile = get_active_profile_name()
if profile and profile not in {"default", "custom"}:
profile_name = profile
except Exception:
pass
return resolve_effective_model(explicit, profile_name, "hermes-agent")
def _cors_headers_for_origin(self, origin: str) -> Optional[Dict[str, str]]:
"""Return CORS headers for an allowed browser origin."""
if not origin or not self._cors_origins:
return None
if "*" in self._cors_origins:
headers = dict(_CORS_HEADERS)
headers["Access-Control-Allow-Origin"] = "*"
headers["Access-Control-Max-Age"] = "600"
return headers
if origin not in self._cors_origins:
return None
headers = dict(_CORS_HEADERS)
headers["Access-Control-Allow-Origin"] = origin
headers["Vary"] = "Origin"
headers["Access-Control-Max-Age"] = "600"
return headers
def _origin_allowed(self, origin: str) -> bool:
"""Allow non-browser clients and explicitly configured browser origins."""
if not origin:
return True
if not self._cors_origins:
return False
return "*" in self._cors_origins or origin in self._cors_origins
@staticmethod
def _clean_log_value(value: Any, *, max_len: int = 200) -> str:
"""Sanitize request metadata before it reaches security logs."""
if value is None:
return ""
text = str(value).replace("\r", " ").replace("\n", " ").strip()
return text[:max_len]
def _request_audit_context(self, request: "web.Request") -> Dict[str, str]:
"""Return non-secret source metadata for security/audit warnings."""
peer_ip = ""
try:
peer = request.transport.get_extra_info("peername") if request.transport else None
if isinstance(peer, (tuple, list)) and peer:
peer_ip = str(peer[0])
except Exception:
peer_ip = ""
return {
"remote": self._clean_log_value(getattr(request, "remote", "") or peer_ip),
"peer_ip": self._clean_log_value(peer_ip),
"forwarded_for": self._clean_log_value(request.headers.get("X-Forwarded-For", "")),
"real_ip": self._clean_log_value(request.headers.get("X-Real-IP", "")),
"method": self._clean_log_value(request.method, max_len=16),
"path": self._clean_log_value(request.path_qs, max_len=500),
"user_agent": self._clean_log_value(request.headers.get("User-Agent", ""), max_len=300),
}
def _request_audit_log_suffix(self, request: "web.Request") -> str:
ctx = self._request_audit_context(request)
fields = [f"{key}={value!r}" for key, value in ctx.items() if value]
return " ".join(fields) if fields else "source='unknown'"
def _cron_origin_from_request(self, request: "web.Request") -> Dict[str, str]:
"""Persist safe API source metadata on cron jobs created over HTTP."""
ctx = self._request_audit_context(request)
origin = {
"platform": "api_server",
"chat_id": "api",
}
if ctx.get("remote"):
origin["source_ip"] = ctx["remote"]
if ctx.get("peer_ip"):
origin["peer_ip"] = ctx["peer_ip"]
if ctx.get("forwarded_for"):
origin["forwarded_for"] = ctx["forwarded_for"]
if ctx.get("real_ip"):
origin["real_ip"] = ctx["real_ip"]
if ctx.get("user_agent"):
origin["user_agent"] = ctx["user_agent"]
return origin
# ------------------------------------------------------------------
# Auth helper
# ------------------------------------------------------------------
def _expected_api_key(self) -> str:
"""Return the API key authorized for the URL-selected profile."""
profile = _api_request_profile.get()
if not profile or profile == "default":
return self._api_key
try:
from agent.secret_scope import get_secret
from hermes_cli.auth import has_usable_secret
key = get_secret("API_SERVER_KEY", "") or ""
if not has_usable_secret(key, min_length=16):
return ""
return key
except Exception as exc:
# Fail closed if the profile scope or strength guard cannot resolve
# the credential. Do not log the key or exception text.
logger.warning(
"Failed to resolve a usable profile-scoped API_SERVER_KEY for %r: %s",
profile,
type(exc).__name__,
)
return ""
def _check_auth(self, request: "web.Request") -> Optional["web.Response"]:
"""
Validate Bearer token from Authorization header.
Returns None if auth is OK, or a 401 web.Response on failure.
connect() refuses to start the API server without API_SERVER_KEY, so
the no-key branch only exists for tests or unsupported manual wiring.
"""
profile = _api_request_profile.get()
is_named_profile = bool(profile and profile != "default")
expected_key = self._expected_api_key()
if not expected_key:
# Preserve the historical no-key test/manual-wiring behavior only
# for the default listener. Named profiles must fail closed rather
# than inherit the listener owner's key.
if not is_named_profile:
return None
logger.warning(
"API server rejected request for profile %r: no profile-scoped "
"API_SERVER_KEY is configured; %s",
profile,
self._request_audit_log_suffix(request),
)
return web.json_response(
{
"error": {
"message": "Invalid gateway API key (API_SERVER_KEY)",
"type": "gateway_auth_error",
"code": "gateway_auth_failed",
}
},
status=401,
)
auth_header = request.headers.get("Authorization", "")
if auth_header.startswith("Bearer "):
token = auth_header[7:].strip()
# Compare as bytes: ``hmac.compare_digest`` raises TypeError on a
# str containing non-ASCII characters, and ``token`` is the raw
# client-supplied header. A stray non-ASCII byte in the key would
# otherwise crash this handler (500) instead of returning a clean
# 401. Encoding both sides keeps the timing-safe comparison and
# matches web_server.py's dashboard-token check.
if hmac.compare_digest(token.encode(), expected_key.encode()):
return None # Auth OK
logger.warning(
"API server rejected invalid API key: %s",
self._request_audit_log_suffix(request),
)
return web.json_response(
{"error": {"message": "Invalid gateway API key (API_SERVER_KEY)", "type": "gateway_auth_error", "code": "gateway_auth_failed"}},
status=401,
)
@staticmethod
def _normalize_callback_platform(value: str) -> str:
normalized = (value or "").strip().lower().replace("-", "_")
if not re.fullmatch(r"[a-z0-9_]+", normalized):
return ""
return normalized
def _get_platform_callback_adapter(
self,
request: "web.Request",
platform_name: str,
) -> Optional[Any]:
injected = request.app.get("platform_event_adapters")
if isinstance(injected, dict):
adapter = injected.get(platform_name)
if adapter is not None:
return adapter
adapter = request.app.get(f"{platform_name}_adapter")
if adapter is not None:
return adapter
runner = self.gateway_runner or request.app.get("gateway_runner")
adapters = getattr(runner, "adapters", None)
if not adapters:
return None
try:
from gateway.config import Platform as _Platform
return adapters.get(_Platform(platform_name))
except Exception:
for platform, candidate in adapters.items():
if getattr(platform, "value", platform) == platform_name:
return candidate
return None
async def _handle_platform_event_callback(self, request: "web.Request") -> "web.Response":
platform_name = self._normalize_callback_platform(
request.match_info.get("platform", "")
)
if not platform_name:
return web.json_response(
_openai_error(
"Invalid platform name",
code="invalid_platform",
),
status=400,
)
adapter = self._get_platform_callback_adapter(request, platform_name)
if adapter is None:
return web.json_response(
_openai_error(
"Platform adapter is not connected",
code="platform_unavailable",
),
status=503,
)
verifier = getattr(adapter, "verify_http_event_request", None)
dispatcher = getattr(adapter, "dispatch_http_event", None)
if verifier is None or dispatcher is None:
return web.json_response(
_openai_error(
"Platform adapter does not support HTTP events",
code="platform_http_events_unsupported",
),
status=503,
)
auth_header = request.headers.get("Authorization", "")
try:
if asyncio.iscoroutinefunction(verifier):
ok, code = await verifier(auth_header)
else:
# Platform verifiers may do blocking network I/O (e.g. Google
# signing-cert fetches) — keep that off the event loop.
ok, code = await asyncio.to_thread(verifier, auth_header)
except Exception:
# Fail closed: a crashing verifier must never admit the event.
logger.exception(
"Platform HTTP event verifier failed for %s", platform_name
)
ok, code = False, "platform_event_verifier_error"
if not ok:
return web.json_response(
_openai_error(
"Invalid platform event authorization",
code=code or "invalid_platform_event_authorization",
),
status=401,
)
try:
payload = await request.json()
except Exception:
return web.json_response(
_openai_error("Invalid JSON in platform event", code="invalid_json"),
status=400,
)
if not isinstance(payload, dict):
return web.json_response(
_openai_error(
"Platform event must be a JSON object",
code="invalid_request",
),
status=400,
)
try:
result = await dispatcher(payload)
except Exception:
logger.exception("Platform HTTP event dispatch failed for %s", platform_name)
return web.json_response(
_openai_error(
"Platform event dispatch failed",
err_type="server_error",
code="platform_event_dispatch_failed",
),
status=500,
)
return web.json_response(result if isinstance(result, dict) else {})
# ------------------------------------------------------------------
# Multi-profile multiplexing (/p/<profile>/…)
# ------------------------------------------------------------------
def _resolve_request_profile(self, request: "web.Request"):
"""Resolve + validate the /p/<profile>/ URL prefix on an API request.
Returns:
- ``None`` when no profile prefix is present, or when multiplexing
is off and the prefix names this gateway's own profile (the
request is handled as the serving profile).
- the profile name (str) when present, multiplexing is on, and the
profile is one this gateway serves.
- ``_PROFILE_REJECTED`` when a prefix is present but the profile is
unknown/unconfigured, or names a profile this single-profile
gateway does not serve (handler/middleware returns 404).
"""
profile = (request.match_info.get("profile") or "").strip()
if not profile:
return None
runner = getattr(self, "gateway_runner", None)
cfg = getattr(runner, "config", None)
if not getattr(cfg, "multiplex_profiles", False):
# Prefix supplied but multiplexing is off. Only a self-referential
# prefix (naming the profile this gateway already serves) may fall
# through to the bare route. Silently ignoring ANY prefix served
# the gateway owner's config/toolsets/capabilities under another
# profile's URL — cross-profile capability leakage (#91583
# defect 2) and silently misdelivered peer DMs (observed live:
# `hermes peer dm mini/researcher` answered by the mini's default
# agent) — so anything else fails closed as unknown.
return (
None
if _prefix_names_served_profile(profile)
else _PROFILE_REJECTED
)
try:
from hermes_cli.profiles import profiles_to_serve
served = {
name
for name, _ in profiles_to_serve(
multiplex=True,
profile_allowlist=getattr(
cfg, "multiplex_profile_allowlist", None
),
)
}
except Exception:
return _PROFILE_REJECTED
if profile not in served:
return _PROFILE_REJECTED
return profile
@staticmethod
def _profile_scope(profile: Optional[str]):
"""Enter the multiplex profile runtime scope, or a no-op when unset.
When no ``/p/<profile>/`` prefix was given AND multiplexing is active,
enter the DEFAULT profile's scope instead of a no-op: api_server is a
port-binding platform that lives on the default profile, and with
multiplex fail-closed ``get_secret`` active, an unscoped agent run
raises ``UnscopedSecretError`` on its first credential read (#61276).
Single-profile gateways keep the no-op — ``get_secret`` falls through
to ``os.environ`` there, unchanged.
"""
if not profile:
try:
from agent.secret_scope import is_multiplex_active
if is_multiplex_active():
from gateway.run import _profile_runtime_scope
from hermes_constants import get_hermes_home
return _profile_runtime_scope(get_hermes_home())
except Exception:
pass
return nullcontext()
from gateway.run import _profile_runtime_scope
from hermes_cli.profiles import get_profile_dir
return _profile_runtime_scope(get_profile_dir(profile))
def _make_profile_prefix_middleware(self):
"""Reject unknown /p/<profile>/ prefixes and scope the request home."""
@web.middleware
async def profile_prefix_middleware(request: "web.Request", handler):
profile = self._resolve_request_profile(request)
if profile is _PROFILE_REJECTED:
return web.json_response(
{"error": "Unknown or unconfigured profile"},
status=404,
)
token = _api_request_profile.set(profile)
try:
with self._profile_scope(profile):
resolved_profile = profile or "default"
principal_token = _api_request_browser_control_principal.set(
self._derive_browser_control_principal(resolved_profile)
)
family_token = _api_request_browser_control_transport_family.set(
self._browser_control_transport_family(request)
)
try:
return await handler(request)
finally:
_api_request_browser_control_transport_family.reset(family_token)
_api_request_browser_control_principal.reset(principal_token)
finally:
_api_request_profile.reset(token)
return profile_prefix_middleware
def _http_route_table(self) -> List[tuple]:
"""Return (method, path, handler) rows registered by ``connect()``.
Kept as a method so multiplex tests can assert the /p/<profile>/
mirrors without starting a real aiohttp listener.
"""
routes: List[tuple] = [
("GET", "/health", self._handle_health),
("GET", "/health/detailed", self._handle_health_detailed),
("GET", "/v1/health", self._handle_health),
("GET", "/v1/models", self._handle_models),
("GET", "/api/model/options", self._handle_model_options),
("GET", "/v1/capabilities", self._handle_capabilities),
# Authenticated browser-control surface: POST registration
# mints a short-lived ticket; the controller then opens the WS with
# that ticket. Both are gated on browser.extension_control.enabled
# and API-key auth (see the handlers for the exact status ladder).
("POST", "/v1/browser-control/register", self._handle_browser_control_register),
("GET", "/v1/browser-control/ws", self._handle_browser_control_ws),
# One-shot artifact transport (Phase 8 Task 29): bounded, SHA-256
# validated HTTPS upload/download bound to a browser-control
# scope. Gated identically to registration (feature flag + API
# key) plus per-principal rate limits.
("POST", "/v1/artifacts/upload", self._handle_artifact_upload),
("GET", "/v1/artifacts/download/{artifact_id}", self._handle_artifact_download),
("GET", "/v1/skills", self._handle_skills),
("GET", "/v1/toolsets", self._handle_toolsets),
("GET", "/api/sessions", self._handle_list_sessions),
("POST", "/api/sessions", self._handle_create_session),
("GET", "/api/sessions/{session_id}", self._handle_get_session),
("PATCH", "/api/sessions/{session_id}", self._handle_patch_session),
("DELETE", "/api/sessions/{session_id}", self._handle_delete_session),
("GET", "/api/sessions/{session_id}/messages", self._handle_session_messages),
("POST", "/api/sessions/{session_id}/fork", self._handle_fork_session),
("POST", "/api/sessions/{session_id}/chat", self._handle_session_chat),
("POST", "/api/sessions/{session_id}/chat/stream", self._handle_session_chat_stream),
("POST", "/api/sessions/{session_id}/model", self._handle_session_model_lock),
("POST", "/v1/chat/completions", self._handle_chat_completions),
("POST", "/v1/responses", self._handle_responses),
("GET", "/v1/responses/{response_id}", self._handle_get_response),
("DELETE", "/v1/responses/{response_id}", self._handle_delete_response),
# Generic platform HTTP event callback ingress. Authenticated by
# the target adapter's own verifier (platform-signed bearer), NOT
# API_SERVER_KEY — external platforms hold no API server key.
("POST", "/api/platforms/{platform}/events", self._handle_platform_event_callback),
("GET", "/api/jobs", self._handle_list_jobs),
("POST", "/api/jobs", self._handle_create_job),
("GET", "/api/jobs/{job_id}", self._handle_get_job),
("PATCH", "/api/jobs/{job_id}", self._handle_update_job),
("DELETE", "/api/jobs/{job_id}", self._handle_delete_job),
("POST", "/api/jobs/{job_id}/pause", self._handle_pause_job),
("POST", "/api/jobs/{job_id}/resume", self._handle_resume_job),
("POST", "/api/jobs/{job_id}/run", self._handle_run_job),
]
routes.extend(_room_grants._http_routes(self))
routes.extend(_api_runs._http_routes(self))
if _CRON_AVAILABLE:
# Chronos managed-cron fire webhook (NAS → agent). Authenticated
# by a NAS-minted JWT (NOT API_SERVER_KEY).
routes.append(("POST", "/api/cron/fire", self._handle_cron_fire))
return routes
# ------------------------------------------------------------------
# Session header helpers
# ------------------------------------------------------------------
# Soft length cap for session identifiers. Headers are bounded in
# aggregate by aiohttp (``client_max_size`` / default 8 KiB per
# header), but we impose a tighter limit on the session headers so a
# caller can't burn memory by passing a multi-kilobyte "session key".
# 256 chars is well above any realistic stable channel identifier
# (e.g. ``agent:main:webui:dm:user-42``) while staying small enough
# that the sanitized form is safe to pass into Honcho / state.db.
_MAX_SESSION_HEADER_LEN = 256
# Source stamped on every session row this platform owns. Hardwired in
# both places that create one — ``_bind_api_server_session`` (session
# ContextVars) and ``_create_agent`` (``platform="api_server"``) — so the
# peer lookup below can filter on it without guessing.
_SESSION_SOURCE = "api_server"
def _declared_conversation_session(
self, gateway_session_key: Optional[str]
) -> Optional[str]:
"""Resolve the live session a client declared with ``X-Hermes-Session-Key``.
The key names the *conversation*; ``session_id`` names the transcript
that conversation is currently on. A client that manages its own
history has no ``previous_response_id`` chain to carry the transcript
forward, so the handlers used to mint a fresh id per request — and
every conversation-affinity hint Hermes sends off that id
(``prompt_cache_key`` on both OpenAI-wire transports, the
OpenRouter/Nous sticky ``session_id``, and xAI's ``x-grok-conv-id``)
re-keyed on every single reply (#96811).
This is the same reset-fenced recovery every native gateway platform
already uses (``SessionStore._recover_session_for_peer``): rows ended
at a conversation boundary — ``session_reset`` (/new),
``session_switch``, ``idle``, ``daily``, ``suspended`` — are fenced
out, so a new conversation still gets a new id and a cold affinity
scope. The generation that must rotate is durable independently of
this lookup, in the ``conversation_generations`` counter advanced
inside each boundary's own transaction
(``SessionDB._bump_conversation_generation``); all this has to resolve
is the live transcript.
Two first requests arriving concurrently on one declared key can each
miss this lookup, mint their own row and both bind — the mismatch
guard in :meth:`_bind_declared_conversation` does not fire, because
each row is still unkeyed at bind time. That converges instead of
crossing: both rows carry the same key under the same source, so this
lookup returns the later one for every subsequent reply and the
earlier row is an abandoned transcript, never another conversation's
identity.
Returns ``None`` when nothing was declared, when no live row is
recorded for the declared key, or on any DB error — every one of
those leaves the caller's per-request id exactly as it is today.
"""
key = (gateway_session_key or "").strip()
if not key:
return None
db = self._ensure_session_db()
if db is None:
return None
try:
row = db.find_latest_gateway_session_for_peer(
source=self._SESSION_SOURCE, session_key=key
)
except Exception:
logger.debug(
"[%s] declared-conversation lookup failed", self.name, exc_info=True
)
return None
return str(row["id"]) if row and row.get("id") else None
def _bind_declared_conversation(
self, session_id: Optional[str], gateway_session_key: Optional[str]
) -> None:
"""Record the declared conversation key on the session row.
Counterpart to :meth:`_declared_conversation_session`. Without it the
row is written unkeyed by ``AIAgent._ensure_db_session`` (which knows
the key but does not persist it), and the reset-fenced lookup can
never see it — the mapping the next reply needs would not exist.
``include_compression_ancestors`` carries the key up a mid-turn
compression rotation so the pre- and post-rotation rows of one
conversation share it, while that same walk deliberately stops at
``/branch``, delegate and tool children (#79161). The statement is an
UPDATE, so it is a harmless no-op on a turn that failed before the row
was created.
"""
key = (gateway_session_key or "").strip()
sid = str(session_id or "").strip()
if not key or not sid:
return
db = self._ensure_session_db()
if db is None:
return
try:
# Defence in depth behind the callers' precedence gate: never
# rewrite a row that already belongs to a different conversation.
# record_gateway_session_peer does SET session_key = ?, so a
# mistaken bind would strand the original conversation and hand its
# session to this request's key.
existing = db.get_session(sid) or {}
current = str(existing.get("session_key") or "").strip()
if current and current != key:
logger.debug(
"[%s] refusing to rebind session %s from a different "
"declared conversation",
self.name,
sid,
)
return
db.record_gateway_session_peer(
sid,
source=self._SESSION_SOURCE,
session_key=key,
include_compression_ancestors=True,
)
except Exception:
logger.debug(
"[%s] declared-conversation bind failed for %s",
self.name,
sid,
exc_info=True,
)
def _parse_session_key_header(
self, request: "web.Request"
) -> tuple[Optional[str], Optional["web.Response"]]:
"""Extract and validate the ``X-Hermes-Session-Key`` header.
The session key is a stable per-channel identifier that scopes
long-term memory (e.g. Honcho sessions) across transcripts. It
is independent of ``X-Hermes-Session-Id``: callers may send
either, both, or neither.
Returns ``(session_key, None)`` on success (with an empty/absent
header yielding ``None`` for the key), or ``(None, error_response)``
on validation failure.
Security: like session continuation, accepting a caller-supplied
memory scope requires API-key authentication so that an
unauthenticated client on a local-only server can't inject itself
into another user's long-term memory scope by guessing a key.
"""
raw = request.headers.get("X-Hermes-Session-Key", "").strip()
if not raw:
return None, None
if not self._api_key:
logger.warning(
"X-Hermes-Session-Key rejected: no API key configured. "
"Set API_SERVER_KEY to enable long-term memory scoping."
)
return None, web.json_response(
_openai_error(
"X-Hermes-Session-Key requires API key authentication. "
"Configure API_SERVER_KEY to enable this feature."
),
status=403,
)
# Reject control characters that could enable header injection on
# the echo path.
if re.search(r'[\r\n\x00]', raw):
return None, web.json_response(
{"error": {"message": "Invalid session key", "type": "invalid_request_error"}},
status=400,
)
if len(raw) > self._MAX_SESSION_HEADER_LEN:
return None, web.json_response(
{"error": {"message": "Session key too long", "type": "invalid_request_error"}},
status=400,
)
return raw, None
# ------------------------------------------------------------------
# Session DB helper
# ------------------------------------------------------------------
def _open_and_cache_session_db(self, home) -> Optional[Any]:
"""Sync core: return the cached SessionDB for ``home``, opening it once.
Shared by the sync (``_ensure_session_db``) and async
(``_ensure_session_db_async``) entry points so both honor the same
per-profile cache. Deliberately does NOT write into ``self._session_db``
— that stays reserved for an explicit test/manual override, so the first
profile served can't pin every later request to its DB.
"""
from hermes_state import SessionDB
key = str(home)
with self._session_db_cache_lock:
if self._session_db_cache_closed:
return None
db = self._session_dbs.get(key)
if db is None:
db = SessionDB(db_path=home / "state.db")
self._session_dbs[key] = db
return db
def _close_cached_session_dbs(self) -> None:
"""Close SessionDB handles owned by this adapter's profile cache."""
with self._session_db_cache_lock:
self._session_db_cache_closed = True
cached = list(self._session_dbs.values())
self._session_dbs.clear()
shared_db = getattr(self, "_session_db", None)
for db in cached:
if db is shared_db:
continue
try:
db.close()
except Exception:
logger.debug("Failed to close API-server SessionDB", exc_info=True)
def _ensure_session_db(self):
"""Lazily initialise and return the SessionDB for the active profile home.
Sessions are persisted to ``state.db`` so that ``hermes sessions list``
shows API-server conversations alongside CLI and gateway ones.
Under multiplex ``/p/<profile>/`` requests the profile runtime scope
redirects ``get_hermes_home()``, so each profile gets its own DB —
never the default profile's file. Synchronous: used by ``_create_agent``
(itself sync, and run in both loop and worker contexts). Request
handlers use ``_ensure_session_db_async`` to keep the SQLite open off
the event loop.
"""
# Explicit override (tests / manual wiring) wins.
if self._session_db is not None:
return self._session_db
try:
from hermes_constants import get_hermes_home
return self._open_and_cache_session_db(get_hermes_home())
except Exception as e:
logger.debug("SessionDB unavailable for API server: %s", e)
return None
async def _ensure_session_db_async(self):
"""Async variant for request handlers: offload the SQLite open/schema
init off the single aiohttp event-loop thread.
The active profile home is captured on the loop thread (its runtime
scope is not visible inside ``asyncio.to_thread``); only the blocking
construction runs in the worker. A single-flight lock prevents duplicate
concurrent construction for the same home.
"""
if self._session_db is not None:
return self._session_db
try:
from hermes_constants import get_hermes_home
home = get_hermes_home()
key = str(home)
with self._session_db_cache_lock:
cached = self._session_dbs.get(key)
if cached is not None:
return cached
if self._session_db_lock is None:
self._session_db_lock = asyncio.Lock()
async with self._session_db_lock:
with self._session_db_cache_lock:
cached = self._session_dbs.get(key)
if cached is not None:
return cached
return await asyncio.to_thread(self._open_and_cache_session_db, home)
except Exception as e:
logger.debug("SessionDB unavailable for API server: %s", e)
return None
# ------------------------------------------------------------------
# Agent creation helper
# ------------------------------------------------------------------
@staticmethod
def _parse_model_routes(raw: Any) -> Dict[str, Dict[str, Any]]:
"""Validate and normalize the ``model_routes`` config block.
Accepts a mapping of ``alias -> {model, provider?, api_key?, base_url?}``.
Invalid shapes are dropped (never raised) so a config typo can't take
the whole API server down. Route values are coerced to strings.
Security: per-route ``api_key`` values are UPSTREAM provider
credentials (used to call the routed model's backend), not caller
authentication — callers still authenticate with the global
API_SERVER_KEY bearer token via ``_check_auth``. Route api_keys must
never be logged; only alias names and non-secret fields may appear in
logs.
"""
if not isinstance(raw, dict):
if raw:
logger.warning(
"api_server model_routes ignored: expected a mapping, got %s",
type(raw).__name__,
)
return {}
allowed_keys = ("model", "provider", "api_key", "base_url")
routes: Dict[str, Dict[str, Any]] = {}
for alias, cfg in raw.items():
alias_str = str(alias).strip()
if not alias_str or not isinstance(cfg, dict):
logger.warning(
"api_server model_routes: dropping invalid route entry %r", alias_str or alias
)
continue
route = {
key: str(cfg[key]).strip()
for key in allowed_keys
if cfg.get(key) is not None and str(cfg[key]).strip()
}
if not route.get("model"):
logger.warning(
"api_server model_routes: route %r has no 'model'; dropping", alias_str
)
continue
routes[alias_str] = route
return routes
def _resolve_route(self, model_alias: Any) -> Optional[Dict[str, Any]]:
"""Return the model_routes entry for *model_alias*, or None."""
if not self._model_routes or not isinstance(model_alias, str):
return None
return self._model_routes.get(model_alias)
def _stored_session_model(self, session: Any) -> Optional[str]:
"""The model persisted on a session row, minus the virtual alias.
The advertised virtual model (usually ``hermes-agent``) means "use
the gateway default". Session creation persists it when the client
sent no model, and replaying it upstream as a raw provider model id
400s ("hermes-agent is not a valid model ID") — the same filter
``_request_agent_overrides`` applies to per-request bodies. One
resolver for both session-chat sites (sync + stream).
"""
stored = session.get("model") if isinstance(session, dict) else None
if not stored or stored == self._model_name:
return None
return stored
@staticmethod
def _clean_runtime_id(value: Any, *, max_len: int = 200) -> str:
if value is None:
return ""
text = str(value).strip()
if not text or len(text) > max_len:
return ""
if re.search(r"[\r\n\x00]", text):
return ""
return text
@classmethod
def _split_provider_prefixed_model(cls, model: str) -> tuple[str, str]:
text = cls._clean_runtime_id(model)
if "::" in text:
provider, raw = text.split("::", 1)
if re.match(r"^[a-zA-Z0-9_.-]{2,64}$", provider) and raw.strip():
return provider, raw.strip()
return "", text
@classmethod
def _runtime_options_from_model_options(cls, model_options: Any) -> Dict[str, Any]:
if not isinstance(model_options, dict):
return {}
runtime_options: Dict[str, Any] = {}
reasoning = model_options.get("reasoning")
if isinstance(reasoning, dict):
enabled = reasoning.get("enabled")
effort = cls._clean_runtime_id(reasoning.get("effort"), max_len=32)
if enabled is False:
runtime_options["reasoning_config"] = {"enabled": False}
elif effort:
runtime_options["reasoning_config"] = {"enabled": True, "effort": effort}
elif enabled is True:
runtime_options["reasoning_config"] = {"enabled": True}
service_tier = cls._clean_runtime_id(model_options.get("service_tier"), max_len=32)
if service_tier:
runtime_options["service_tier"] = service_tier
elif _coerce_request_bool(model_options.get("fast"), default=False):
runtime_options["service_tier"] = "priority"
return runtime_options
def _session_runtime_request_from_body(self, body: Dict[str, Any]) -> Dict[str, Any]:
raw_model = self._clean_runtime_id(body.get("model") or body.get("model_id"))
raw_provider = self._clean_runtime_id(body.get("provider") or body.get("provider_id"), max_len=80)
prefixed_provider, split_model = self._split_provider_prefixed_model(raw_model)
provider = raw_provider or prefixed_provider
model = split_model or raw_model
alias_route = self._resolve_route(raw_model) or self._resolve_route(model)
route = dict(alias_route) if isinstance(alias_route, dict) else None
# The virtual model alias (self._model_name, e.g. "hermes-agent") is
# not a real provider model id — it's the id /v1/models advertises
# for "use the gateway default". A client that echoes it back
# (explicitly or via a generic model picker) means "no real request",
# same as omitting model entirely. Null it out here, upstream of
# both the route-building below and every caller's "requested"
# dict, so it never gets persisted as a session's model or
# misread later as a raw session_model override (#session-model-
# alias-leak — see _handle_create_session).
if model == self._model_name:
model = None
route_source = "model_routes" if route else "global"
if not route and model:
route = {"model": model}
if provider:
route["provider"] = provider
route_source = "raw_request"
elif not route and provider and model:
route = {"model": model, "provider": provider}
route_source = "raw_request"
runtime_options = self._runtime_options_from_model_options(body.get("model_options"))
requested = {"provider": provider, "model": model, "raw_model": raw_model}
return {
"requested": requested,
"route": route,
"route_source": route_source,
"runtime_options": runtime_options,
"require_model_lock": _coerce_request_bool(body.get("require_model_lock"), default=False),
"model_options": body.get("model_options") if isinstance(body.get("model_options"), dict) else {},
}
def _runtime_lock_error(self, runtime_request: Dict[str, Any]) -> Optional["web.Response"]:
if not runtime_request.get("require_model_lock"):
return None
requested = runtime_request.get("requested") or {}
model = self._clean_runtime_id(requested.get("model"))
provider = self._clean_runtime_id(requested.get("provider"), max_len=80)
route = runtime_request.get("route")
if not model and not provider:
return web.json_response(
_openai_error("require_model_lock was set but no model/provider was provided", code="missing_model"),
status=400,
)
if not route or runtime_request.get("route_source") == "global":
return web.json_response(
_openai_error("Requested Browser model lock cannot be routed; refusing silent global fallback", code="model_lock_unavailable"),
status=409,
)
return None
def _persist_session_runtime_lock(self, session_id: str, runtime_request: Dict[str, Any]) -> bool:
# Persist only a newly confirmed lock. Reusing a stored lock should not
# rewrite its timestamp/prompt state on every turn, and an ordinary
# one-off request override must not erase a previously confirmed lock.
if runtime_request.get("persisted_lock") or not runtime_request.get("require_model_lock"):
return True
requested = runtime_request.get("requested") or {}
model = self._clean_runtime_id(requested.get("model"))
provider = self._clean_runtime_id(requested.get("provider"), max_len=80)
if not model and not provider:
return False
db = self._ensure_session_db()
if db is None:
return False
try:
db.update_session_runtime_lock(
session_id,
model=model or None,
provider=provider or None,
model_options=runtime_request.get("model_options") or {},
route_source=runtime_request.get("route_source") or "",
confirmed=bool(runtime_request.get("require_model_lock")),
)
return True
except Exception:
logger.warning("[%s] failed to persist session runtime lock for %s", self.name, session_id, exc_info=True)
return False
@staticmethod
def _parse_session_model_config(raw: Any) -> Dict[str, Any]:
if isinstance(raw, dict):
return dict(raw)
if isinstance(raw, str) and raw.strip():
try:
parsed = json.loads(raw)
except Exception:
return {}
if isinstance(parsed, dict):
return parsed
return {}
def _runtime_request_from_persisted_session_lock(
self,
session: Optional[Dict[str, Any]],
body: Dict[str, Any],
) -> Optional[Dict[str, Any]]:
if not isinstance(session, dict):
return None
model_config = self._parse_session_model_config(session.get("model_config"))
lock = model_config.get("browser_model_lock")
if not isinstance(lock, dict) or not _coerce_request_bool(lock.get("confirmed"), default=False):
return None
model = self._clean_runtime_id(lock.get("model"))
provider = self._clean_runtime_id(lock.get("provider"), max_len=80)
if not model and not provider:
return None
persisted_route_source = self._clean_runtime_id(
lock.get("route_source"),
max_len=64,
).lower()
route: Optional[Dict[str, Any]] = None
if persisted_route_source == "model_routes":
route = self._resolve_route(model) if model else None
else:
route = {"model": model} if model else {}
if provider:
route["provider"] = provider
model_options = (
body.get("model_options")
if isinstance(body.get("model_options"), dict)
else lock.get("model_options")
)
return {
"requested": {
"provider": provider,
"model": model,
"raw_model": model,
},
"route": route or None,
"route_source": "session_model_lock",
"runtime_options": self._runtime_options_from_model_options(model_options),
"require_model_lock": True,
"model_options": model_options if isinstance(model_options, dict) else {},
"persisted_lock": True,
}
def _effective_session_runtime_request(
self,
*,
session: Optional[Dict[str, Any]],
body: Dict[str, Any],
) -> Dict[str, Any]:
runtime_request = self._session_runtime_request_from_body(body)
requested = runtime_request.get("requested") or {}
if requested.get("model") or requested.get("provider"):
return runtime_request
persisted = self._runtime_request_from_persisted_session_lock(session, body)
return persisted or runtime_request
@classmethod
def _sanitize_runtime_metadata(
cls,
*,
runtime: Optional[Dict[str, Any]] = None,
requested_runtime: Optional[Dict[str, Any]] = None,
route_source: str = "global",
model_lock: str = "",
) -> Dict[str, Any]:
payload = dict(runtime or {})
provider = cls._clean_runtime_id(
payload.get("provider") or payload.get("provider_id") or payload.get("effective_provider"),
max_len=80,
)
model = cls._clean_runtime_id(payload.get("model") or payload.get("model_id") or payload.get("effective_model"))
result: Dict[str, Any] = {
"provider": provider,
"model": model,
"route_source": cls._clean_runtime_id(payload.get("route_source") or route_source, max_len=64) or "global",
}
if requested_runtime or payload.get("requested"):
req = requested_runtime or payload.get("requested") or {}
result["requested"] = {
"provider": cls._clean_runtime_id(req.get("provider"), max_len=80),
"model": cls._clean_runtime_id(req.get("model")),
}
if model_lock or payload.get("model_lock"):
result["model_lock"] = cls._clean_runtime_id(model_lock or payload.get("model_lock"), max_len=32)
return result
@staticmethod
def _normalize_session_source(value: Any) -> str:
text = str(value or "").strip().lower()
allowed = {"api_server", "hermes_browser", "browser", "cli", "telegram", "discord", "slack", "desktop", "dashboard"}
if text in allowed:
return "hermes_browser" if text == "browser" else text
return "api_server"
def _session_model_override_for(self, session_key: Optional[str]) -> Optional[Dict[str, Any]]:
"""Return the gateway's session ``/model`` override for *session_key*, if any.
The gateway tracks per-session ``/model`` switches in
``GatewayRunner._session_model_overrides``. API-server requests that
share such a session key must keep honouring the explicit session
override even when the request's ``model`` field matches a configured
route — a user-issued ``/model`` always wins over static config.
"""
if not session_key:
return None
try:
from gateway.run import _gateway_runner_ref
runner = _gateway_runner_ref()
if runner is None:
return None
try:
rehydrate = getattr(runner, "_rehydrate_session_model_override", None)
if callable(rehydrate):
rehydrate(session_key)
except Exception:
logger.debug(
"api_server failed to rehydrate session /model override for %s",
session_key,
exc_info=True,
)
override = runner._session_model_overrides.get(session_key)
return dict(override) if isinstance(override, dict) else None
except Exception:
return None
def _request_route_conflict_error(
self,
*,
session_id: Optional[str],
gateway_session_key: Optional[str],
requested_model: Optional[str],
requested_provider: Optional[str],
route: Optional[Dict[str, Any]],
) -> Optional[str]:
"""Return a 400-worthy conflict string for ambiguous route/provider mixes."""
request_provider = _clean_request_string(requested_provider)
if not request_provider or not isinstance(route, dict):
return None
if self._session_model_override_for(gateway_session_key or session_id):
# Session /model wins over both the route and the request override, so
# there is no ambiguity to reject on this request path.
return None
route_provider = _clean_request_string(route.get("provider"))
route_api_key = _clean_request_string(route.get("api_key"))
route_base_url = _clean_request_string(route.get("base_url"))
route_alias = _clean_request_string(requested_model) or "requested model"
if route_provider and request_provider != route_provider:
return (
f"Model route '{route_alias}' is pinned to provider '{route_provider}'. "
f"Remove 'provider' or use '{route_provider}'."
)
if not route_provider and (route_api_key or route_base_url):
return (
f"Model route '{route_alias}' pins route credentials/base_url. "
"Do not combine it with an explicit 'provider'."
)
return None
def _create_agent(
self,
ephemeral_system_prompt: Optional[str] = None,
session_id: Optional[str] = None,
stream_delta_callback=None,
tool_progress_callback=None,
tool_start_callback=None,
tool_complete_callback=None,
gateway_session_key: Optional[str] = None,
requested_model: Optional[str] = None,
requested_provider: Optional[str] = None,
model_options: Optional[Dict[str, Any]] = None,
route: Optional[Dict[str, Any]] = None,
session_model: Optional[str] = None,
confirmed_runtime_lock: bool = False,
room_dispatch: Optional[Dict[str, Any]] = None,
room_execution_policy: Optional[Dict[str, Any]] = None,
) -> Any:
"""
Create an AIAgent instance using the gateway's runtime config.
Uses _resolve_runtime_agent_kwargs() to pick up model, api_key,
base_url, etc. from config.yaml / env vars. Toolsets are resolved
from config.yaml platform_toolsets.api_server (same as all other
gateway platforms), falling back to the hermes-api-server default.
``gateway_session_key`` is a stable per-channel identifier supplied
by the client (via ``X-Hermes-Session-Key``). Unlike ``session_id``
which scopes the short-term transcript and rotates on /new, this
key is meant to persist across transcripts so long-term memory
providers (e.g. Honcho) can scope their per-chat state correctly
— matching the semantics of the native gateway's ``session_key``.
``route`` is an optional ``model_routes`` entry (per-client model
routing). When set — and no session ``/model`` override exists for
this session — its model/provider/api_key/base_url override the
global defaults for this agent instance only.
``session_model`` is the raw model persisted on a native API session
row at creation time (``POST /api/sessions {"model": ...}``) when
that value does not resolve to a ``model_routes`` alias. Session-chat
handlers pass either ``route`` (alias hit) or ``session_model`` (raw
model), never both. Precedence: session ``/model`` override →
``session_model`` → route alias / per-request selection → global.
``confirmed_runtime_lock`` marks a backend-acknowledged Browser model
lock (POST /api/sessions/{id}/model). A confirmed lock beats the
session ``/model`` override, disables the global fallback model
chain, and fails closed if the locked provider's credentials cannot
be resolved.
"""
from run_agent import AIAgent
from gateway.run import (
_checkpoint_agent_kwargs,
_current_max_iterations,
_resolve_runtime_agent_kwargs,
_resolve_gateway_model,
_load_gateway_config,
GatewayRunner,
)
from hermes_cli.tools_config import _get_platform_tools
# Catch RuntimeError ONLY around this call, not the wider
# _create_agent()+run_conversation() span --
# _resolve_runtime_agent_kwargs() is the sole raiser of
# RuntimeError(format_runtime_provider_error(...)) for provider
# auth/credential failure. Re-raising as
# _ProviderAuthResolutionError lets _run_agent() (and
# _handle_runs()) distinguish this from an unrelated RuntimeError
# elsewhere in the call graph.
try:
runtime_kwargs = _resolve_runtime_agent_kwargs()
except RuntimeError as exc:
raise _ProviderAuthResolutionError(str(exc)) from exc
model = _resolve_gateway_model()
# When the primary provider's auth fails (expired token / 429 quota
# cap), _resolve_runtime_agent_kwargs() falls through to the fallback
# provider chain, whose runtime dict carries its own ``model`` key.
# Pop it and let it override the config model, mirroring the native
# gateway path (_resolve_session_agent_runtime in run.py). Otherwise
# the explicit ``model=model`` below collides with the ``**runtime_kwargs``
# spread → "got multiple values for keyword argument 'model'", 500ing
# every /v1/chat/completions request while a fallback is active.
runtime_model = runtime_kwargs.pop("model", None)
if runtime_model:
model = runtime_model
request_reasoning_config = _request_reasoning_config(model_options)
request_service_tier = _request_service_tier(model_options)
request_model = _clean_request_string(requested_model)
request_provider = _clean_request_string(requested_provider)
route_model = _clean_request_string(route.get("model")) if isinstance(route, dict) else None
route_provider = _clean_request_string(route.get("provider")) if isinstance(route, dict) else None
route_api_key = _clean_request_string(route.get("api_key")) if isinstance(route, dict) else None
route_base_url = _clean_request_string(route.get("base_url")) if isinstance(route, dict) else None
def _resolve_provider_runtime(
provider: Optional[str],
*,
target_model: Optional[str],
required: bool,
) -> Optional[Dict[str, Any]]:
provider_name = _clean_request_string(provider)
if not provider_name:
return None
try:
return _resolve_request_runtime_agent_kwargs(
provider_name,
target_model=target_model or None,
)
except Exception as exc:
try:
from gateway.run import _resolve_runtime_agent_kwargs_for_provider
return _resolve_runtime_agent_kwargs_for_provider(provider_name)
except Exception:
pass
if required:
# Surface as the typed provider-auth failure so
# _run_agent()/_handle_runs() return the controlled
# response shape instead of a raw 500.
raise _ProviderAuthResolutionError(str(exc)) from exc
logger.debug(
"api_server provider-runtime refresh failed for provider=%s model=%s",
provider_name,
target_model or "",
exc_info=True,
)
return None
# Final precedence mirrors the gateway contract:
# confirmed Browser model lock → session /model override →
# session-persisted model (POST /api/sessions {"model": ...}) →
# model_routes mapping selected by the request model alias → direct
# per-request provider/model → global defaults. model_options stay
# request-scoped regardless of which selection wins. A confirmed
# lock is an execution contract: it bypasses the session /model
# override and fails closed (never reuses global credentials) if
# its provider cannot be resolved.
session_key = gateway_session_key or session_id
session_row_model = _clean_request_string(session_model)
session_override = None
if not confirmed_runtime_lock:
session_override = self._session_model_override_for(session_key)
# Model-string precedence delegates to the shared owner
# hermes_cli.model_switch.resolve_effective_model (session /model
# override > session-persisted model > global) — the rule 7dd00bb47d
# had to re-fix here after it diverged from gateway/run.py.
from hermes_cli.model_switch import resolve_effective_model
if session_override:
override_model = resolve_effective_model(session_override, None, model)
session_provider = _clean_request_string(session_override.get("provider"))
current_provider = _clean_request_string(runtime_kwargs.get("provider"))
provider_runtime = _resolve_provider_runtime(
session_provider or current_provider,
target_model=override_model,
required=False,
)
if provider_runtime:
_apply_runtime_agent_overrides(runtime_kwargs, provider_runtime)
_apply_runtime_agent_overrides(runtime_kwargs, session_override)
model = override_model
if route or request_model or request_provider:
logger.debug(
"api_server request selection skipped: session /model override wins for %s",
session_key or "",
)
elif session_row_model and not confirmed_runtime_lock:
# Session-persisted model (raw string that resolved to no route
# alias). Pins this session's turns ahead of per-request body
# values — a session's chosen model is a standing selection,
# matching the native gateway's session-model semantics.
current_provider = _clean_request_string(runtime_kwargs.get("provider"))
provider_runtime = _resolve_provider_runtime(
current_provider,
target_model=session_row_model,
required=False,
)
if provider_runtime:
_apply_runtime_agent_overrides(runtime_kwargs, provider_runtime)
model = resolve_effective_model(None, session_row_model, model)
if request_model or request_provider:
logger.debug(
"api_server request selection skipped: session-persisted model wins for %s",
session_key or "",
)
else:
if route is not None:
# The request's ``model`` field selected this route, so its
# value is the route ALIAS — never usable as a model name.
# A route with no ``model`` key keeps the global default
# (pre-existing model_routes behavior).
effective_model = route_model or model
else:
effective_model = request_model or model
current_provider = _clean_request_string(runtime_kwargs.get("provider"))
effective_provider = request_provider or route_provider or current_provider
provider_runtime = None
if effective_provider and (
bool(request_provider or route_provider) or effective_model != model
):
provider_runtime = _resolve_provider_runtime(
effective_provider,
target_model=effective_model,
# A confirmed Browser lock fails closed: if the locked
# provider cannot be resolved, never fall through to
# the previous global provider's credentials.
required=bool(request_provider) or confirmed_runtime_lock,
)
if provider_runtime:
_apply_runtime_agent_overrides(runtime_kwargs, provider_runtime)
elif effective_provider and effective_provider != current_provider:
runtime_kwargs["provider"] = effective_provider
model = effective_model
# Per-route explicit transport secrets/base URLs win within the
# route contract after provider resolution.
if route_api_key:
runtime_kwargs["api_key"] = route_api_key
if route_base_url:
runtime_kwargs["base_url"] = route_base_url
if route:
logger.debug(
"api_server request selection applied: model=%s provider=%s route_provider=%s request_provider=%s",
model,
runtime_kwargs.get("provider"),
route_provider or "",
request_provider or "",
)
# When the config has no model.default but a provider was resolved
# (e.g. user ran `hermes auth add openai-codex` without `hermes model`),
# fall back to the provider's first catalog model so the API call
# doesn't fail with "model must be a non-empty string". Mirrors
# run.py::_resolve_session_agent_runtime. Runs after the selection
# block above so a route/session/request override that already
# resolved a model is never treated as "empty" here.
if not model and runtime_kwargs.get("provider"):
try:
from hermes_cli.models import get_default_model_for_provider
model = get_default_model_for_provider(runtime_kwargs["provider"])
if model:
logger.info(
"No model configured — defaulting to %s for provider %s",
model, runtime_kwargs["provider"],
)
except Exception:
pass
# Final safety net (#35314): if resolution still produced an empty
# model — e.g. a transient config-cache miss — reuse the last model
# successfully resolved for this session (or, failing that, the most
# recent one resolved process-wide). Building an agent with model=""
# makes every API call fail HTTP 400 until a manual retry. Mirrors
# run.py::_resolve_session_agent_runtime.
#
# Cache key is gateway_session_key ONLY, never session_id — unlike
# run.py's native gateway (stable, long-lived chat scopes), the API
# server hands out a fresh UUID session_id per one-off request
# (/v1/responses, /v1/runs when no explicit session is supplied).
# Keying on session_id would leave one permanent dict entry per
# stateless request, growing unbounded for the life of the process.
_resolved_key = gateway_session_key or ""
if not model:
_recovered = (self._last_resolved_model.get(_resolved_key)
or self._last_resolved_model.get("*"))
if _recovered and _recovered != self._model_name:
logger.warning(
"Empty model resolved for session=%s — recovering "
"last-known-good model %s (config read likely returned "
"empty; see #35314)",
_resolved_key, _recovered,
)
model = _recovered
elif model:
if model != self._model_name:
if _resolved_key:
self._last_resolved_model[_resolved_key] = model
self._last_resolved_model["*"] = model
user_config = _load_gateway_config()
enabled_toolsets = sorted(_get_platform_tools(user_config, "api_server"))
max_iterations = _current_max_iterations()
if room_dispatch is not None:
from gateway.hosted_room_execution_policy import RoomExecutionPolicy
policy = RoomExecutionPolicy.from_mapping(room_execution_policy or {})
enabled_toolsets = list(policy.enabled_toolsets)
max_iterations = policy.max_iterations
# Load fallback provider chain so the API server platform has the
# same fallback behaviour as Telegram/Discord/Slack (fixes #4954).
fallback_model = (
None
if confirmed_runtime_lock
else GatewayRunner._load_fallback_model()
)
# Resolve reasoning against the model this request will actually
# run. Per-model ``agent.reasoning_overrides`` key off that model,
# and it is only settled after the precedence chain above (browser
# lock -> session /model -> session row -> route -> per-request ->
# defaults). Resolving at function entry keyed them off
# ``model.default`` instead — the defect e81d18dfb removed from the
# native gateway paths. An explicit per-request reasoning parameter
# still wins over config.
reasoning_config = (
request_reasoning_config
if request_reasoning_config is not None
else GatewayRunner._load_reasoning_config(model)
)
agent_kwargs = {
"model": model,
**runtime_kwargs,
**_checkpoint_agent_kwargs(user_config),
"max_iterations": max_iterations,
"quiet_mode": True,
"verbose_logging": False,
"ephemeral_system_prompt": ephemeral_system_prompt or None,
"enabled_toolsets": enabled_toolsets,
"session_id": session_id,
"platform": "api_server",
"stream_delta_callback": stream_delta_callback,
"tool_progress_callback": tool_progress_callback,
"tool_start_callback": tool_start_callback,
"tool_complete_callback": tool_complete_callback,
"session_db": self._ensure_session_db(),
"fallback_model": fallback_model,
"reasoning_config": reasoning_config,
"gateway_session_key": gateway_session_key,
}
if request_service_tier is not _REQUEST_OPTION_MISSING:
agent_kwargs["service_tier"] = request_service_tier
agent = AIAgent(**agent_kwargs)
agent._hermes_api_runtime = {
"provider": runtime_kwargs.get("provider") or getattr(agent, "provider", "") or "",
"model": getattr(agent, "model", None) or model,
"route_source": (
"session_model_lock"
if confirmed_runtime_lock
else "session_model_override"
if session_override
else "raw_request"
if route or request_model or request_provider
else "global"
),
}
return agent
# ------------------------------------------------------------------
# HTTP Handlers
# ------------------------------------------------------------------
async def _handle_health(self, request: "web.Request") -> "web.Response":
"""GET /health — simple health check."""
return web.json_response(
{"status": "ok", "platform": "hermes-agent", "version": _hermes_version()}
)
async def _handle_health_detailed(self, request: "web.Request") -> "web.Response":
"""GET /health/detailed — rich status for cross-container dashboard probing.
Returns gateway state, connected platforms, PID, and uptime so the
dashboard can display full status without needing a shared PID file or
/proc access. Requires the same Bearer auth as other API routes.
"""
auth_err = self._check_auth(request)
if auth_err:
return auth_err
from gateway.status import (
derive_gateway_busy,
derive_gateway_drainable,
normalize_updated_at,
parse_active_agents,
read_runtime_status,
)
runtime = read_runtime_status() or {}
gw_state = runtime.get("gateway_state")
gw_active = parse_active_agents(runtime.get("active_agents", 0))
# This endpoint is served BY the gateway process, so it is by definition
# alive — gateway_running is True. Derive busy/drainable from the same
# shared contract /api/status uses so the two surfaces never disagree.
active_api_runs, process_depth, active_delegations = self._readiness_work_counts()
from gateway.run import _resolve_gateway_model
readiness = collect_runtime_readiness(
configured_model=_resolve_gateway_model(),
runtime_status=runtime,
active_api_runs=active_api_runs,
process_completion_queue_depth=process_depth,
active_delegations=active_delegations,
)
return web.json_response({
"status": readiness["status"],
"readiness": readiness,
"platform": "hermes-agent",
"version": _hermes_version(),
"gateway_state": gw_state,
"platforms": runtime.get("platforms", {}),
"active_agents": gw_active,
"gateway_busy": derive_gateway_busy(
gateway_running=True,
gateway_state=gw_state,
active_agents=gw_active,
),
"gateway_drainable": derive_gateway_drainable(
gateway_running=True,
gateway_state=gw_state,
),
"exit_reason": runtime.get("exit_reason"),
# Contract: updated_at is RFC3339 string | null, never a number —
# the state file may carry legacy epoch floats or hand-edited junk.
"updated_at": normalize_updated_at(runtime.get("updated_at")),
"pid": os.getpid(),
})
async def _handle_models(self, request: "web.Request") -> "web.Response":
"""GET /v1/models — list hermes-agent and any configured model_routes aliases.
Under ``/p/<profile>/v1/models`` (multiplex on) the advertised primary
model id follows that profile's name/config, not the default adapter's
cached ``_model_name``.
"""
auth_err = self._check_auth(request)
if auth_err:
return auth_err
now = int(time.time())
# Middleware already entered the profile runtime scope when a /p/
# prefix was present, so get_active_profile_name() resolves correctly.
model_name = (
self._resolve_model_name("")
if _api_request_profile.get()
else self._model_name
)
models = [
{
"id": model_name,
"object": "model",
"created": now,
"owned_by": "hermes",
"permission": [],
"root": model_name,
"parent": None,
}
]
# Expose configured model route aliases so clients can discover them.
# Only the alias and resolved model name are exposed — never provider
# credentials.
for alias, route_cfg in self._model_routes.items():
if alias == model_name:
continue # already listed above
models.append({
"id": alias,
"object": "model",
"created": now,
"owned_by": "hermes",
"permission": [],
"root": route_cfg.get("model", alias),
"parent": model_name,
})
return web.json_response({"object": "list", "data": models})
async def _handle_model_options(self, request: "web.Request") -> "web.Response":
"""GET /api/model/options — return Hermes provider/model inventory.
This mirrors the dashboard/TUI model picker inventory endpoint so
external clients using the API server can sync to the user's configured
Hermes provider catalog instead of scraping the single OpenAI-compatible
`/v1/models` alias.
"""
auth_err = self._check_auth(request)
if auth_err:
return auth_err
refresh = _coerce_request_bool(request.query.get("refresh"), default=False)
try:
from hermes_cli.inventory import build_model_options_payload, load_picker_context
def _build_payload() -> Dict[str, Any]:
return build_model_options_payload(
load_picker_context(),
include_unconfigured=True,
refresh=refresh,
)
# Inventory enrichment can fetch pricing and provider catalogs.
# Keep all synchronous picker work off aiohttp's event loop.
payload = await asyncio.to_thread(_build_payload)
return web.json_response(payload)
except Exception:
logger.exception("[%s] GET /api/model/options failed", self.name)
return web.json_response(
_openai_error(
"Failed to list model options.",
code="model_options_failed",
),
status=500,
)
async def _handle_capabilities(self, request: "web.Request") -> "web.Response":
"""GET /v1/capabilities — advertise the stable API surface.
External UIs and orchestrators use this endpoint to discover the API
server's plugin-safe contract without scraping docs or assuming that
every Hermes version exposes the same endpoints.
"""
auth_err = self._check_auth(request)
if auth_err:
return auth_err
return web.json_response({
"object": "hermes.api_server.capabilities",
"platform": "hermes-agent",
"model": self._model_name,
"auth": {
"type": "bearer",
"required": bool(self._api_key),
},
"runtime": {
"mode": "server_agent",
"tool_execution": "server",
"split_runtime": False,
"description": (
"The API server creates a server-side Hermes AIAgent; "
"tools execute on the API-server host unless a future "
"explicit split-runtime mode is enabled."
),
},
"features": {
"chat_completions": True,
"chat_completions_streaming": True,
"responses_api": True,
"responses_streaming": True,
"run_submission": True,
"runs_idempotency": _api_runs._idempotency_capabilities(
self,
store_type=RunIdempotencyStore,
),
"run_status": True,
"run_events_sse": True,
"run_stop": True,
"run_steer": True,
"run_approval_response": True,
"tool_progress_events": True,
"approval_events": True,
"session_resources": True,
"model_options": True,
"session_chat": True,
"session_chat_streaming": True,
"session_fork": True,
"session_model_lock": True,
"admin_config_rw": False,
"jobs_admin": False,
"memory_write_api": False,
"skills_api": True,
"audio_api": False,
"realtime_voice": False,
"session_continuity_header": "X-Hermes-Session-Id",
"session_key_header": "X-Hermes-Session-Key",
"cors": bool(self._cors_origins),
# Browser-extension control is always advertised so clients
# can feature-detect it, but remains disabled until
# browser.extension_control.enabled is explicitly set.
"browser_extension_control": {
"enabled": self._browser_control_enabled(),
"protocol_version": _BROWSER_CONTROL_PROTOCOL_VERSION,
"capabilities": sorted(BROWSER_CONTROL_CAPABILITIES),
"artifact_capabilities": sorted(BROWSER_CONTROL_ARTIFACT_CAPABILITIES),
"developer_capabilities": sorted(BROWSER_CONTROL_DEVELOPER_CAPABILITIES),
"developer_mode": self._browser_control_developer_mode(),
"artifact_transport": {
"upload": {"method": "POST", "path": "/v1/artifacts/upload"},
"download": {
"method": "GET",
"path": "/v1/artifacts/download/{artifact_id}",
},
"max_bytes": DEFAULT_MAX_ARTIFACT_BYTES,
"ttl_seconds": DEFAULT_ARTIFACT_TTL_SECONDS,
"allowed_mime_types": sorted(DEFAULT_ALLOWED_MIME_TYPES),
},
"real_browser_actions": True,
"transports": {
"local_vps": "websocket-subprotocol-ticket",
"cloud": "authenticated-gateway-rpc",
},
},
},
"endpoints": {
"health": {"method": "GET", "path": "/health"},
"health_detailed": {"method": "GET", "path": "/health/detailed"},
"models": {"method": "GET", "path": "/v1/models"},
"model_options": {"method": "GET", "path": "/api/model/options"},
"chat_completions": {"method": "POST", "path": "/v1/chat/completions"},
"responses": {"method": "POST", "path": "/v1/responses"},
"runs": {"method": "POST", "path": "/v1/runs"},
"run_status": {"method": "GET", "path": "/v1/runs/{run_id}"},
"run_events": {"method": "GET", "path": "/v1/runs/{run_id}/events"},
"run_approval": {"method": "POST", "path": "/v1/runs/{run_id}/approval"},
"run_steer": {"method": "POST", "path": "/v1/runs/{run_id}/steer"},
"run_stop": {"method": "POST", "path": "/v1/runs/{run_id}/stop"},
"skills": {"method": "GET", "path": "/v1/skills"},
"toolsets": {"method": "GET", "path": "/v1/toolsets"},
"sessions": {"method": "GET", "path": "/api/sessions"},
"session_create": {"method": "POST", "path": "/api/sessions"},
"session": {"method": "GET", "path": "/api/sessions/{session_id}"},
"session_update": {"method": "PATCH", "path": "/api/sessions/{session_id}"},
"session_delete": {"method": "DELETE", "path": "/api/sessions/{session_id}"},
"session_messages": {"method": "GET", "path": "/api/sessions/{session_id}/messages"},
"session_fork": {"method": "POST", "path": "/api/sessions/{session_id}/fork"},
"session_chat": {"method": "POST", "path": "/api/sessions/{session_id}/chat"},
"session_chat_stream": {"method": "POST", "path": "/api/sessions/{session_id}/chat/stream"},
"session_model_lock": {"method": "POST", "path": "/api/sessions/{session_id}/model"},
"browser_control_register": {"method": "POST", "path": "/v1/browser-control/register"},
"browser_control_ws": {"method": "GET", "path": "/v1/browser-control/ws"},
"artifact_upload": {"method": "POST", "path": "/v1/artifacts/upload"},
"artifact_download": {
"method": "GET",
"path": "/v1/artifacts/download/{artifact_id}",
},
},
})
# ------------------------------------------------------------------
# Browser-extension control (authenticated local/VPS API)
# ------------------------------------------------------------------
async def _handle_browser_control_register(self, request: "web.Request") -> "web.Response":
"""POST /v1/browser-control/register — mint a controller ticket.
The extension controller proves itself with the same Bearer API key
every other API-server client uses, then receives a short-lived,
single-use ticket to open the controller WebSocket. Identity is NOT
taken from the request body: the scope principal is derived
server-side from the authenticated key/profile as a non-reversible
digest, and the capability set is filtered to the shared browser
action allowlist, so a spoofed
``principal_id`` or inflated capability list in the payload is
ignored rather than honored. The named session must already exist in
the active profile's server-owned SessionDB before a ticket is minted.
Status ladder: 404 when the feature is disabled, 403 when no API key
is configured at all (registration can never be authenticated), 401
for a missing/invalid Bearer token, 201 on success.
"""
if not self._browser_control_enabled():
return web.json_response(
_openai_error(
"Browser control is not enabled on this server.",
code="browser_control_disabled",
),
status=404,
)
if not self._api_key:
logger.warning(
"browser-control registration rejected: no API key configured; "
"set API_SERVER_KEY to enable authenticated browser control."
)
return web.json_response(
_openai_error(
"Browser control registration requires a configured API key.",
err_type="gateway_auth_error",
code="browser_control_auth_required",
),
status=403,
)
auth_err = self._check_auth(request)
if auth_err:
return auth_err
try:
payload = await request.json()
except Exception:
return web.json_response(
_openai_error("Request body must be valid JSON."), status=400
)
if not isinstance(payload, dict):
return web.json_response(
_openai_error("Request body must be a JSON object."), status=400
)
if not browser_control_protocol_supported(payload.get("protocol_version")):
return web.json_response(
_openai_error(
"Unsupported browser-control protocol version.",
code="browser_control_protocol_unsupported",
),
status=400,
)
controller_id = str(payload.get("controller_id") or "").strip()
browser_profile_id = str(payload.get("browser_profile_id") or "").strip()
session_id = str(payload.get("session_id") or "").strip()
if not controller_id or not browser_profile_id or not session_id:
return web.json_response(
_openai_error(
"controller_id, browser_profile_id, and session_id are required.",
code="browser_control_invalid_registration",
),
status=400,
)
db = await self._ensure_session_db_async()
if db is None:
return web.json_response(
_openai_error(
"Session database unavailable.",
code="session_db_unavailable",
),
status=503,
)
session = await asyncio.to_thread(db.get_session, session_id)
if not session:
return web.json_response(
_openai_error(
"Browser control may register only for an existing server session.",
err_type="gateway_auth_error",
code="browser_control_session_forbidden",
),
status=403,
)
profile = _api_request_profile.get() or "default"
capabilities = filter_browser_control_capabilities(
payload.get("capabilities"),
developer_mode=self._browser_control_developer_mode(),
)
if not capabilities:
return web.json_response(
_openai_error(
"At least one permitted browser-control capability is required.",
code="browser_control_no_capabilities",
),
status=400,
)
# Developer capabilities may only be negotiated while the broker
# itself runs in Developer Mode (fail closed even if a registration
# somehow slipped through the filter).
if (
capabilities & BROWSER_CONTROL_DEVELOPER_CAPABILITIES
and not self._browser_control_developer_mode()
):
return web.json_response(
_openai_error(
"Developer Mode is required for browser_evaluate and raw CDP.",
code="browser_control_developer_mode_required",
),
status=403,
)
scope = ControllerScope(
principal_id=self._derive_browser_control_principal(profile),
profile_id=profile,
session_id=session_id or None,
controller_id=controller_id,
browser_profile_id=browser_profile_id,
transport_family=self._browser_control_transport_family(request),
capabilities=capabilities,
)
ticket = self._browser_control_broker.mint_ticket(scope)
ticket_ttl = self._browser_control_broker.ticket_ttl_seconds
return web.json_response(
{
"protocol_version": _BROWSER_CONTROL_PROTOCOL_VERSION,
"ticket": ticket.value,
# Best-effort wall-clock projection for clients; the broker
# enforces expiry on its monotonic clock, so after an NTP
# step trust ticket_expires_in_seconds, not this absolute.
"ticket_expires_at": time.time() + ticket_ttl,
"ticket_expires_in_seconds": ticket_ttl,
"ws_path": "/v1/browser-control/ws",
"scope": {
"principal_id": scope.principal_id,
"profile_id": scope.profile_id,
"session_id": scope.session_id,
"controller_id": scope.controller_id,
"browser_profile_id": scope.browser_profile_id,
"transport_family": scope.transport_family,
"capabilities": sorted(scope.capabilities),
},
},
status=201,
)
async def _handle_browser_control_ws(self, request: "web.Request") -> "web.WebSocketResponse":
"""GET /v1/browser-control/ws — controller WebSocket (one-shot ticket).
A ticket-bearing ``Sec-WebSocket-Protocol`` token is exchanged exactly
once for the identity scope minted at registration; query-string,
unknown, already-consumed, or expired tickets are rejected with 401
before upgrade. The socket then attaches to the shared broker under
that scope, forwards broker command/cancel frames onto the aiohttp loop
thread-safely, and accepts controller result/cancel frames. Completion
is exact-scope checked, and owner-aware teardown cannot detach a newer
replacement controller generation.
"""
# Re-check at upgrade time so disabling the feature immediately closes
# the admission gate without consuming still-live one-shot tickets.
if not self._browser_control_enabled():
raise web.HTTPNotFound()
# Credentials in the request target are liable to appear in access
# logs. Accept the one-shot ticket only as a WebSocket subprotocol;
# reject the former query-string shape without consuming it.
if request.query.get("ticket"):
raise web.HTTPUnauthorized()
requested_protocols = [
value.strip()
for value in request.headers.get("Sec-WebSocket-Protocol", "").split(",")
if value.strip()
]
ticket_protocols = [
value
for value in requested_protocols
if value.startswith(_BROWSER_CONTROL_TICKET_PROTOCOL_PREFIX)
]
if (
_BROWSER_CONTROL_WS_PROTOCOL not in requested_protocols
or len(ticket_protocols) != 1
):
raise web.HTTPUnauthorized()
ticket_value = ticket_protocols[0][len(_BROWSER_CONTROL_TICKET_PROTOCOL_PREFIX) :]
if not ticket_value:
raise web.HTTPUnauthorized()
try:
scope = self._browser_control_broker.consume_ticket(ticket_value)
except ControllerTicketInvalid:
raise web.HTTPUnauthorized() from None
except Exception:
logger.exception("browser-control WS ticket consumption failed")
raise web.HTTPUnauthorized() from None
ws = web.WebSocketResponse(
heartbeat=30.0,
protocols=(_BROWSER_CONTROL_WS_PROTOCOL,),
)
await ws.prepare(request)
loop = asyncio.get_running_loop()
_send = _browser_controller_ws_sender(ws, loop)
# attach/disconnect/detach acquire the controller's send_lock, which a
# worker-thread dispatch may hold while blocking on THIS loop to
# transmit its frame (run_coroutine_threadsafe + result(timeout=10)).
# Offload them so a teardown/attach racing an in-flight send parks a
# worker thread, never the event loop.
await asyncio.to_thread(
self._browser_control_broker.attach, scope, _send, owner=ws
)
try:
async for msg in ws:
if msg.type == web.WSMsgType.TEXT:
try:
frame = msg.json()
except Exception:
continue
if isinstance(frame, dict):
reply = await asyncio.to_thread(
self._handle_browser_control_frame,
scope,
frame,
owner=ws,
)
if isinstance(reply, dict):
await ws.send_json(reply)
elif msg.type in (web.WSMsgType.CLOSE, web.WSMsgType.ERROR):
break
finally:
await asyncio.to_thread(
self._browser_control_broker.disconnect,
scope,
owner=ws,
)
return ws
def _handle_browser_control_frame(
self,
scope: "ControllerScope",
frame: dict,
*,
owner: Any = None,
) -> Optional[dict]:
"""Apply one controller→broker frame with exact-scope checks."""
method = frame.get("method")
params = frame.get("params")
if not isinstance(params, dict):
return
if owner is None or not self._browser_control_broker.is_owner(scope, owner):
return
if method == "browser.controller.heartbeat":
nonce = str(params.get("nonce") or "").strip()
if not nonce or len(nonce) > 128:
return
# Echo only the caller's opaque nonce on the already authenticated,
# exact-scope controller socket. This proves the socket path is live
# without granting a new capability or touching broker commands.
return {
"method": "browser.controller.heartbeat",
"params": {"nonce": nonce, "ok": True},
}
if method == "browser.controller.detach":
self._browser_control_broker.detach(
scope,
owner=owner,
notify_controller=False,
)
return {
"method": "browser.controller.detach",
"params": {"ok": True},
}
if method == "browser.controller.result":
command_id = params.get("command_id")
if isinstance(command_id, str) and command_id:
# Broker resolves only the pending command whose scope equals
# this socket's scope; a stranger's command id is a no-op.
ok = params.get("ok") is True
self._browser_control_broker.complete(
command_id,
scope=scope,
ok=ok,
result=params.get("result") if ok else params.get("error"),
)
elif method == "browser.controller.cancel":
tool_call_id = params.get("tool_call_id")
if isinstance(tool_call_id, str) and tool_call_id:
self._browser_control_broker.cancel(scope, tool_call_id=tool_call_id)
def _browser_control_enabled(self) -> bool:
"""Feature flag; False unless explicitly enabled.
Reads ``browser.extension_control.enabled`` from the global config
(defaults to False). Tests monkeypatch this method directly to force
the feature on/off without touching config.
"""
try:
from gateway.browser_control_broker import browser_control_enabled as _flag
return _flag()
except Exception:
return False
def _derive_browser_control_principal(self, profile: str) -> str:
"""Server-derived controller principal (non-reversible digest).
The principal is bound to the credential that authenticated the
registration request — the expected API key for the request's
profile — so a client cannot impersonate another controller by
echoing an id in the registration body.
"""
key = self._expected_api_key() or self._api_key or ""
digest = hashlib.sha256(f"{profile}\x00{key}".encode("utf-8")).hexdigest()
return f"principal:{profile}:{digest[:32]}"
def _browser_control_transport_family(self, request: "web.Request") -> str:
"""Local vs remote API family, decided by the loopback peer.
A controller speaking to a localhost listener is in the same trust
domain as the host and gets the ``local-api`` family; anything else
is ``remote-api``. The broker treats the family as part of exact
identity, so a remote controller can never satisfy a local-only
dispatch (and vice versa).
"""
host = None
try:
transport = request.transport
if transport is not None:
peer = transport.get_extra_info("peername")
if isinstance(peer, tuple) and peer:
host = peer[0]
elif isinstance(peer, str):
host = peer
except Exception:
host = None
if host in ("127.0.0.1", "::1", "localhost"):
return "local-api"
return "remote-api"
def _browser_control_developer_mode(self) -> bool:
"""Developer Mode flag; False unless explicitly enabled.
Mirrors the broker's gate for ``browser_evaluate`` and raw CDP.
Tests monkeypatch this method directly to force the gate on/off.
"""
try:
return browser_control_developer_mode()
except Exception:
return False
# ------------------------------------------------------------------
# One-shot artifact transport (Phase 8 Task 29)
# ------------------------------------------------------------------
def _artifact_store_for(self, profile: str) -> ArtifactStore:
"""Return the profile-scoped artifact store, creating it lazily.
The store root lives under the profile's data directory
(``<HERMES_HOME>/plugin-data/.../artifacts``-style controlled root),
so artifacts never escape the profile boundary. Stores are cached
BY RESOLVED PROFILE — on a multiplex listener, profile A touching
the artifact route first must never pin profile B to A's physical
root (same frozen-handle class as the per-profile session-storage
fix in #88734). The root itself is created on first use; TTL
cleanup runs on every store/load/prune.
"""
profile_key = str(profile or "default")
store = self._browser_control_artifacts.get(profile_key)
if store is not None:
return store
try:
from hermes_cli.profiles import get_profile_dir
profile_root = get_profile_dir(profile or "default")
root = Path(profile_root) / "artifacts" / "browser-control"
except Exception:
# Unscoped fallback used only when profile resolution is
# unavailable (tests/manual wiring): keep the controlled root
# under the Hermes home.
try:
from hermes_state import get_hermes_home
root = Path(get_hermes_home()) / "artifacts" / "browser-control"
except Exception:
raise ArtifactError("no artifact root is resolvable") from None
store = ArtifactStore(
root,
ttl_seconds=DEFAULT_ARTIFACT_TTL_SECONDS,
max_bytes=DEFAULT_MAX_ARTIFACT_BYTES,
allowed_mime_types=DEFAULT_ALLOWED_MIME_TYPES,
)
store.prune_expired()
self._browser_control_artifacts[profile_key] = store
# Share the store with the broker so artifact actions dispatched to a
# controller validate their artifact reference against the same
# profile's controlled root ("approved artifact id only").
try:
self._browser_control_broker.attach_artifact_store(
store, profile_id=profile_key
)
except Exception:
logger.debug("could not attach artifact store to broker", exc_info=True)
return store
def _artifact_limiter(self) -> ArtifactRateLimiter:
"""Return the per-principal artifact route limiter (lazy)."""
if self._browser_control_artifact_limiter is None:
self._browser_control_artifact_limiter = ArtifactRateLimiter(
window_seconds=60.0,
max_requests=30,
)
return self._browser_control_artifact_limiter
def _inject_browser_control_artifacts(
self,
store: Optional[ArtifactStore],
limiter: Optional[ArtifactRateLimiter] = None,
*,
profile: str = "default",
) -> None:
"""Inject a store/limiter (tests, diagnostics)."""
if store is None:
self._browser_control_artifacts.pop(profile, None)
else:
self._browser_control_artifacts[profile] = store
if limiter is not None:
self._browser_control_artifact_limiter = limiter
@staticmethod
def _artifact_auth_fail(request: "web.Request", status: int, code: str, message: str):
return web.json_response(
_openai_error(
message,
err_type="gateway_auth_error" if status == 401 else "invalid_request_error",
code=code,
),
status=status,
)
async def _handle_artifact_upload(self, request: "web.Request") -> "web.Response":
"""POST /v1/artifacts/upload — one-shot bounded artifact upload.
Authenticated with the same Bearer API key as every other API-server
route and gated on browser.extension_control.enabled. The body is
read as raw bytes with an exact size cap; ``Content-Type`` must name
an allowed MIME type and ``X-Artifact-Filename`` supplies the
display-only name. On success returns a provenance receipt carrying
the server-minted artifact id, SHA-256, size, TTL, and download path
— never a filesystem path.
Status ladder: 404 feature disabled, 403 no API key configured, 401
bad/missing Bearer, 429 rate limited, 413 too large, 415 MIME
rejected, 400 missing filename/scope, 201 success.
"""
if not self._browser_control_enabled():
return web.json_response(
_openai_error(
"Browser control is not enabled on this server.",
code="browser_control_disabled",
),
status=404,
)
if not self._api_key:
return web.json_response(
_openai_error(
"Artifact transport requires a configured API key.",
err_type="gateway_auth_error",
code="browser_control_auth_required",
),
status=403,
)
auth_err = self._check_auth(request)
if auth_err:
return auth_err
profile = _api_request_profile.get() or "default"
principal = self._derive_browser_control_principal(profile)
limiter = self._artifact_limiter()
if not limiter.allow(f"upload:{principal}"):
return web.json_response(
_openai_error(
"Artifact upload rate limit exceeded.",
err_type="rate_limit_error",
code="rate_limit_exceeded",
),
status=429,
headers={"Retry-After": "1"},
)
content_type = request.headers.get("Content-Type", "")
filename = request.headers.get("X-Artifact-Filename", "").strip()
if not filename:
return web.json_response(
_openai_error("X-Artifact-Filename header is required."),
status=400,
)
# Bounded read: cap at the store's byte cap + 1 so an oversize body
# is detected and rejected without buffering unbounded data.
try:
store = self._artifact_store_for(profile)
except ArtifactError as exc:
return web.json_response(_openai_error(str(exc), code="artifact_rejected"), status=500)
max_bytes = store.max_bytes
try:
data = await request.content.read(max_bytes + 1)
except Exception:
return web.json_response(_openai_error("Failed to read request body."), status=400)
if len(data) > max_bytes:
return web.json_response(
_openai_error(
f"Artifact exceeds the {max_bytes}-byte cap.",
code="artifact_too_large",
),
status=413,
)
if not data:
return web.json_response(_openai_error("Empty artifact body."), status=400)
try:
receipt = store.store(
data,
filename=filename,
content_type=content_type,
scope=_ArtifactScopeFacade(
principal,
transport_family=self._browser_control_transport_family(request),
),
)
except ArtifactTooLarge as exc:
return web.json_response(
_openai_error(str(exc), code="artifact_too_large"), status=413
)
except ArtifactError as exc:
code = "artifact_mime_rejected" if "allowlist" in str(exc) else "artifact_rejected"
status = 415 if "allowlist" in str(exc) else 400
return web.json_response(_openai_error(str(exc), code=code), status=status)
return web.json_response(
receipt.to_dict(download_path=f"/v1/artifacts/download/{receipt.artifact_id}"),
status=201,
)
async def _handle_artifact_download(self, request: "web.Request") -> "web.Response":
"""GET /v1/artifacts/download/{artifact_id} — one-shot download.
Authenticated and gated identically to upload. The artifact is
consumed on success: the second download of the same id returns 404.
The response streams the verified bytes with the recorded content
type and a ``X-Artifact-Sha256`` header for client-side validation.
Status ladder: 404 feature disabled / unknown artifact, 403 no API
key, 401 bad/missing Bearer, 429 rate limited, 410 expired, 400
invalid id / scope mismatch, 200 success.
"""
if not self._browser_control_enabled():
raise web.HTTPNotFound()
if not self._api_key:
return web.json_response(
_openai_error(
"Artifact transport requires a configured API key.",
err_type="gateway_auth_error",
code="browser_control_auth_required",
),
status=403,
)
auth_err = self._check_auth(request)
if auth_err:
return auth_err
profile = _api_request_profile.get() or "default"
principal = self._derive_browser_control_principal(profile)
limiter = self._artifact_limiter()
if not limiter.allow(f"download:{principal}"):
return web.json_response(
_openai_error(
"Artifact download rate limit exceeded.",
err_type="rate_limit_error",
code="rate_limit_exceeded",
),
status=429,
headers={"Retry-After": "1"},
)
artifact_id = request.match_info.get("artifact_id", "")
try:
store = self._artifact_store_for(profile)
data, receipt = store.load(
artifact_id,
scope=_ArtifactScopeFacade(
principal,
transport_family=self._browser_control_transport_family(request),
),
)
except ArtifactError as exc:
message = str(exc)
if "expired" in message:
return web.json_response(
_openai_error(message, code="artifact_expired"), status=410
)
status = 400 if "scope" in message or "invalid" in message else 404
return web.json_response(
_openai_error(message, code="artifact_not_found"), status=status
)
return web.Response(
body=data,
status=200,
content_type=receipt.content_type,
headers={
"X-Artifact-Sha256": receipt.sha256,
"X-Artifact-Id": receipt.artifact_id,
"Content-Disposition": f'attachment; filename="{receipt.filename}"',
},
)
async def _handle_skills(self, request: "web.Request") -> "web.Response":
"""GET /v1/skills — list installed skills visible to the API-server agent.
Read-only listing intended for external clients that need to know
which skills are available without sending a chat message and asking
the model. Mirrors what the gateway/CLI surfaces through
``/skills list``, but as a deterministic JSON payload.
Returns the same skill metadata (name, description, category) the
skills hub uses internally. Disabled skills are excluded so the
listing matches what the agent actually loads.
"""
auth_err = self._check_auth(request)
if auth_err:
return auth_err
try:
from tools.skills_tool import _find_all_skills, _sort_skills
skills = _sort_skills(_find_all_skills(skip_disabled=False))
except Exception:
logger.exception("GET /v1/skills failed")
return web.json_response(
_openai_error("Failed to enumerate skills", err_type="server_error"),
status=500,
)
return web.json_response({
"object": "list",
"data": skills,
})
async def _handle_toolsets(self, request: "web.Request") -> "web.Response":
"""GET /v1/toolsets — list toolsets and their resolved tools.
Returns the toolset surface the api_server platform actually exposes
to its agent: each toolset's enabled/configured state plus the
concrete tool names it expands to. This is the deterministic
equivalent of what a client would otherwise have to recover by
asking the model what tools it can call.
"""
auth_err = self._check_auth(request)
if auth_err:
return auth_err
try:
from hermes_cli.config import load_config
from hermes_cli.tools_config import (
_get_effective_configurable_toolsets,
_get_platform_tools,
_toolset_has_keys,
get_nous_subscription_features,
)
from toolsets import resolve_toolset
config = load_config()
enabled_toolsets = _get_platform_tools(
config,
"api_server",
include_default_mcp_servers=False,
)
features = get_nous_subscription_features(config)
data: List[Dict[str, Any]] = []
for name, label, desc in _get_effective_configurable_toolsets():
try:
tools = sorted(set(resolve_toolset(name)))
except Exception:
tools = []
is_enabled = name in enabled_toolsets
data.append({
"name": name,
"label": label,
"description": desc,
"enabled": is_enabled,
"configured": _toolset_has_keys(name, config, features=features),
"tools": tools,
})
except Exception:
logger.exception("GET /v1/toolsets failed")
return web.json_response(
_openai_error("Failed to enumerate toolsets", err_type="server_error"),
status=500,
)
return web.json_response({
"object": "list",
"platform": "api_server",
"data": data,
})
# ------------------------------------------------------------------
# /api/sessions — thin client/session resource API
# ------------------------------------------------------------------
@staticmethod
def _parse_nonnegative_int(value: Any, default: int, maximum: int) -> int:
try:
parsed = int(value)
except (TypeError, ValueError):
return default
if parsed < 0:
return default
return min(parsed, maximum)
@staticmethod
def _session_response(session: Dict[str, Any]) -> Dict[str, Any]:
"""Return a stable, client-safe session representation."""
safe_keys = (
"id", "source", "user_id", "model", "title", "started_at", "ended_at",
"end_reason", "message_count", "tool_call_count", "input_tokens",
"output_tokens", "cache_read_tokens", "cache_write_tokens",
"reasoning_tokens", "estimated_cost_usd", "actual_cost_usd",
"api_call_count", "parent_session_id", "last_active", "preview",
"_lineage_root_id", "pinned", "archived", "hidden",
)
payload = {key: session.get(key) for key in safe_keys if key in session}
# SQLite stores these as 0/1; clients reconcile against a real boolean.
for flag in ("pinned", "archived", "hidden"):
if flag in payload:
payload[flag] = bool(payload[flag])
# Avoid exposing full system prompts/model_config through the client API;
# callers only need to know whether those snapshots exist.
payload["has_system_prompt"] = bool(session.get("system_prompt"))
payload["has_model_config"] = bool(session.get("model_config"))
return payload
@staticmethod
def _message_response(message: Dict[str, Any]) -> Dict[str, Any]:
message = _project_client_message(message)
safe_keys = (
"id", "session_id", "role", "content", "tool_call_id", "tool_calls",
"tool_name", "timestamp", "token_count", "finish_reason", "reasoning",
"reasoning_content", "display_kind",
)
return {key: message.get(key) for key in safe_keys if key in message}
async def _read_json_body(self, request: "web.Request") -> tuple[Dict[str, Any], Optional["web.Response"]]:
try:
body = await request.json()
except Exception:
return {}, web.json_response(_openai_error("Invalid JSON in request body"), status=400)
if not isinstance(body, dict):
return {}, web.json_response(_openai_error("Request body must be a JSON object"), status=400)
return body, None
async def _get_existing_session_or_404(self, session_id: str) -> tuple[Optional[Dict[str, Any]], Optional["web.Response"]]:
db = await self._ensure_session_db_async()
if db is None:
return None, web.json_response(_openai_error("Session database unavailable", code="session_db_unavailable"), status=503)
# Offload the blocking SQLite read off the event loop (CWE/perf: the
# API server is single-threaded aiohttp; a sync SessionDB call here
# freezes every in-flight request, see PR discussion on event-loop
# blocking SQLite in the gateway surface).
session = await asyncio.to_thread(db.get_session, session_id)
if not session:
return None, web.json_response(_openai_error(f"Session not found: {session_id}", code="session_not_found"), status=404)
return session, None
async def _conversation_history_for_session(self, session_id: str) -> List[Dict[str, Any]]:
db = await self._ensure_session_db_async()
if db is None:
return []
try:
return await asyncio.to_thread(db.get_messages_as_conversation, session_id)
except Exception as exc:
logger.warning("Failed to load session history for %s: %s", session_id, exc)
return []
async def _handle_list_sessions(self, request: "web.Request") -> "web.Response":
"""GET /api/sessions — list persisted Hermes sessions."""
auth_err = self._check_auth(request)
if auth_err:
return auth_err
db = await self._ensure_session_db_async()
if db is None:
return web.json_response(_openai_error("Session database unavailable", code="session_db_unavailable"), status=503)
limit = self._parse_nonnegative_int(request.query.get("limit"), default=50, maximum=200)
offset = self._parse_nonnegative_int(request.query.get("offset"), default=0, maximum=1_000_000)
source = request.query.get("source") or None
include_children = _coerce_request_bool(request.query.get("include_children"), default=False)
# Exact-title lookup, used by `hermes peer dm` to resolve a peer's
# canonical "Bot Chat" session. ``include_hidden`` is honored ONLY
# alongside a title filter: Bot Mode hides canonical chats, so a
# title-scoped lookup must see them (issue #91583), but a blanket
# hidden listing stays off this client surface.
title_filter = (request.query.get("title") or "").strip() or None
include_hidden = bool(title_filter) and _coerce_request_bool(
request.query.get("include_hidden"), default=False
)
sessions = await asyncio.to_thread(db.list_sessions_rich,
source=source,
limit=limit,
offset=offset,
include_children=include_children,
order_by_last_active=True,
# A pin means "always reachable", so a pinned conversation that has
# aged past the recency window is back-filled rather than dropped.
include_pinned=True,
# Push the title needle into SQL so a hidden/old canonical row is
# found even when it falls outside the recency window, then apply
# the exact-match contract below (search_query is substring-based).
search_query=title_filter,
include_hidden=include_hidden,
)
if title_filter:
sessions = [s for s in sessions if (s.get("title") or "").strip() == title_filter]
if not sessions:
# Recoverable-archive resurrection (#92687): a canonical Bot
# Chat archived by the ws-orphan reaper / older agent cleanup
# is invisible to list_sessions_rich (include_archived=False),
# which would fail `hermes peer dm` resolution and mint
# transient sessions — same accident the tui_gateway lookups
# heal. Resurrect and re-list; deliberate archives stay put.
try:
from tools.bot_mode_probe import BOT_CHAT_TITLE
stale = db.get_session_by_title(title_filter) if title_filter == BOT_CHAT_TITLE else None
if stale and stale.get("archived") and db.unarchive_recoverable_session(stale["id"]):
sessions = await asyncio.to_thread(db.list_sessions_rich,
source=source,
limit=limit,
offset=offset,
include_children=include_children,
order_by_last_active=True,
include_pinned=True,
search_query=title_filter,
include_hidden=include_hidden,
)
sessions = [s for s in sessions if (s.get("title") or "").strip() == title_filter]
except Exception:
pass # resolution degrades to today's no-row behavior
# Back-filled pins arrive PAST the limit, so counting them would report
# another page that doesn't exist. Only the recency window decides.
windowed = sum(1 for s in sessions if not s.get("pinned"))
return web.json_response({
"object": "list",
"data": [self._session_response(s) for s in sessions],
"limit": limit,
"offset": offset,
"has_more": windowed >= limit,
})
async def _handle_create_session(self, request: "web.Request") -> "web.Response":
"""POST /api/sessions -- create an empty Hermes session row.
The existence check, insert, title handling, and invalid-title
rollback run as a single off-loop operation to avoid a TOCTOU
window between the duplicate check and the insert (concurrent
same-ID creates could otherwise both pass the check and both
return 201 via the ON CONFLICT enrichment upsert).
"""
auth_err = self._check_auth(request)
if auth_err:
return auth_err
body, err = await self._read_json_body(request)
if err:
return err
db = await self._ensure_session_db_async()
if db is None:
return web.json_response(_openai_error("Session database unavailable", code="session_db_unavailable"), status=503)
raw_id = body.get("id") or body.get("session_id")
session_id = str(raw_id).strip() if raw_id else f"api_{int(time.time())}_{uuid.uuid4().hex[:8]}"
from gateway.session import _is_path_unsafe
if not session_id or re.search(r'[\r\n\x00]', session_id) or _is_path_unsafe(session_id):
return web.json_response(_openai_error("Invalid session ID", code="invalid_session_id"), status=400)
if len(session_id) > self._MAX_SESSION_HEADER_LEN:
return web.json_response(_openai_error("Session ID too long", code="invalid_session_id"), status=400)
system_prompt = body.get("system_prompt")
if system_prompt is not None and not isinstance(system_prompt, str):
return web.json_response(_openai_error("system_prompt must be a string", code="invalid_system_prompt"), status=400)
source = self._normalize_session_source(body.get("source") or "api_server")
runtime_request = self._session_runtime_request_from_body(body)
lock_error = self._runtime_lock_error(runtime_request)
if lock_error is not None:
return lock_error
requested = runtime_request.get("requested") or {}
# requested["model"] is already normalized by
# _session_runtime_request_from_body: provider-prefixed values
# (e.g. "provider::hermes-agent") are split, and the virtual model
# alias (self._model_name, e.g. "hermes-agent") is nulled out
# there — a bare "hermes-agent" is not a model_routes alias, so a
# later chat on this session would otherwise fall into the raw
# session_model precedence branch in _handle_session_chat and get
# sent to the provider literally, failing with "invalid model
# identifier" (#session-model-alias-leak). Re-deriving straight
# from the raw body here would bypass that normalization and
# reintroduce the leak for the provider-prefixed case.
model_name = self._clean_runtime_id(requested.get("model")) or None
model_config = None
if requested.get("model") or requested.get("provider"):
model_config = {
"browser_model_lock": {
"provider": requested.get("provider") or "",
"model": requested.get("model") or "",
"model_options": runtime_request.get("model_options") or {},
"route_source": runtime_request.get("route_source") or "",
"confirmed": bool(runtime_request.get("require_model_lock")),
"updated_at": time.time(),
}
}
title = body.get("title")
# Run the entire check-insert-title sequence inside a single
# _execute_write call (BEGIN IMMEDIATE + commit) so the existence
# check and the insert are atomic at the SQLite level. Two
# concurrent requests for the same ID serialize here: the second
# one blocks on the write lock and sees the row the first inserted.
def _do_create():
def _atomic(conn):
row = conn.execute(
"SELECT id FROM sessions WHERE id = ?", (session_id,)
).fetchone()
if row:
return None, "exists"
import time as _time
conn.execute(
"""INSERT INTO sessions (
id, source, model, model_config, system_prompt, started_at
) VALUES (?, ?, ?, ?, ?, ?)""",
(
session_id,
source,
model_name,
json.dumps(model_config) if model_config else None,
system_prompt,
_time.time(),
),
)
if title is not None:
clean_title = db.sanitize_title(str(title))
if clean_title:
conflict = conn.execute(
"SELECT id FROM sessions WHERE title = ? AND id != ?",
(clean_title, session_id),
).fetchone()
if conflict:
conn.execute(
"DELETE FROM sessions WHERE id = ?", (session_id,)
)
return None, f"title:Title already in use by session {conflict['id']}"
conn.execute(
"UPDATE sessions SET title = ? WHERE id = ?",
(clean_title, session_id),
)
session_row = conn.execute(
"SELECT * FROM sessions WHERE id = ?", (session_id,)
).fetchone()
return (dict(session_row) if session_row else {
"id": session_id, "source": source,
"model": model_name, "title": title,
}), None
return db._execute_write(_atomic)
session, err = await asyncio.to_thread(_do_create)
if err == "exists":
return web.json_response(_openai_error(f"Session already exists: {session_id}", code="session_exists"), status=409)
if err and err.startswith("title:"):
return web.json_response(_openai_error(err[len("title:"):], code="invalid_title"), status=400)
return web.json_response({"object": "hermes.session", "session": self._session_response(session)}, status=201)
async def _handle_get_session(self, request: "web.Request") -> "web.Response":
"""GET /api/sessions/{session_id}."""
auth_err = self._check_auth(request)
if auth_err:
return auth_err
session, err = await self._get_existing_session_or_404(request.match_info["session_id"])
if err:
return err
return web.json_response({"object": "hermes.session", "session": self._session_response(session)})
async def _handle_patch_session(self, request: "web.Request") -> "web.Response":
"""PATCH /api/sessions/{session_id} — update client-safe session metadata."""
auth_err = self._check_auth(request)
if auth_err:
return auth_err
session_id = request.match_info["session_id"]
session, err = await self._get_existing_session_or_404(session_id)
if err:
return err
body, err = await self._read_json_body(request)
if err:
return err
# `pinned` and `archived` are durable per-session flags the desktop
# sidebar owns (the "keep" flag exempts a chat from the auto-archive
# sweep). Rejecting them here was silently 400ing every pin the desktop
# made, so pins only ever lived in that one app's localStorage.
# `unread` is the read-state watermark toggle (same desktop owner).
allowed = {"title", "end_reason", "pinned", "archived", "hidden", "unread"}
unknown = sorted(set(body) - allowed)
if unknown:
return web.json_response(_openai_error(f"Unsupported session fields: {', '.join(unknown)}", code="unsupported_session_field"), status=400)
for flag in ("pinned", "archived", "hidden", "unread"):
if flag in body and not isinstance(body[flag], bool):
return web.json_response(_openai_error(f"'{flag}' must be a boolean", code="invalid_session_field"), status=400)
db = await self._ensure_session_db_async()
if db is None:
return web.json_response(_openai_error("Session database unavailable", code="session_db_unavailable"), status=503)
if "title" in body:
try:
await asyncio.to_thread(db.set_session_title, session_id, "" if body["title"] is None else str(body["title"]))
except ValueError as exc:
return web.json_response(_openai_error(str(exc), code="invalid_title"), status=400)
if "pinned" in body:
await asyncio.to_thread(db.set_session_pinned, session_id, body["pinned"])
if "archived" in body:
await asyncio.to_thread(db.set_session_archived, session_id, body["archived"])
if "hidden" in body:
await asyncio.to_thread(db.set_session_hidden, session_id, body["hidden"])
if "unread" in body:
await asyncio.to_thread(db.set_session_read, session_id, read=not body["unread"])
if body.get("end_reason"):
await asyncio.to_thread(db.end_session, session_id, str(body["end_reason"]))
session = await asyncio.to_thread(db.get_session, session_id) or session
return web.json_response({"object": "hermes.session", "session": self._session_response(session)})
async def _handle_delete_session(self, request: "web.Request") -> "web.Response":
"""DELETE /api/sessions/{session_id}."""
auth_err = self._check_auth(request)
if auth_err:
return auth_err
session_id = request.match_info["session_id"]
session, err = await self._get_existing_session_or_404(session_id)
if err:
return err
db = await self._ensure_session_db_async()
deleted = await asyncio.to_thread(db.delete_session, session_id)
return web.json_response({"object": "hermes.session.deleted", "id": session_id, "deleted": bool(deleted)})
async def _handle_session_messages(self, request: "web.Request") -> "web.Response":
"""GET /api/sessions/{session_id}/messages."""
auth_err = self._check_auth(request)
if auth_err:
return auth_err
session_id = request.match_info["session_id"]
_, err = await self._get_existing_session_or_404(session_id)
if err:
return err
db = await self._ensure_session_db_async()
resolved_id = await asyncio.to_thread(db.resolve_resume_session_id, session_id)
raw_limit = request.query.get("limit")
raw_offset = request.query.get("offset", "0")
order = request.query.get("order")
if order not in (None, "oldest", "latest"):
return web.json_response(
_openai_error(
"order must be one of: oldest, latest",
code="invalid_pagination",
),
status=400,
)
try:
offset = int(raw_offset)
requested_limit = None if raw_limit is None else int(raw_limit)
except (TypeError, ValueError):
offset = -1
requested_limit = -1
if offset < 0 or (requested_limit is not None and requested_limit < 0):
return web.json_response(
_openai_error(
"limit and offset must be non-negative integers",
code="invalid_pagination",
),
status=400,
)
default_page = requested_limit is None
latest_page = order == "latest" or (order is None and default_page)
limit = 500 if default_page else min(requested_limit, 500)
messages = await asyncio.to_thread(
db.get_messages,
resolved_id,
limit=limit,
offset=offset,
latest=latest_page,
)
return web.json_response({
"object": "list",
"session_id": resolved_id,
"data": [self._message_response(m) for m in messages],
"pagination": {
"limit": limit,
"offset": offset,
"order": order or ("latest" if default_page else "oldest"),
"returned": len(messages),
},
})
async def _handle_fork_session(self, request: "web.Request") -> "web.Response":
"""POST /api/sessions/{session_id}/fork — branch via current SessionDB primitives."""
auth_err = self._check_auth(request)
if auth_err:
return auth_err
source_id = request.match_info["session_id"]
source, err = await self._get_existing_session_or_404(source_id)
if err:
return err
body, err = await self._read_json_body(request)
if err:
return err
db = await self._ensure_session_db_async()
fork_id = str(body.get("id") or body.get("session_id") or f"api_{int(time.time())}_{uuid.uuid4().hex[:8]}").strip()
if not fork_id or re.search(r'[\r\n\x00]', fork_id):
return web.json_response(_openai_error("Invalid session ID", code="invalid_session_id"), status=400)
if await asyncio.to_thread(db.get_session, fork_id):
return web.json_response(_openai_error(f"Session already exists: {fork_id}", code="session_exists"), status=409)
# Match the CLI /branch semantics: mark the original as branched, then
# create a child session that carries the transcript forward. This uses
# SessionDB's native parent_session_id/end_reason visibility model rather
# than inventing a parallel fork store.
await asyncio.to_thread(db.end_session, source_id, "branched")
await asyncio.to_thread(db.create_session,
fork_id,
"api_server",
model=source.get("model"),
system_prompt=source.get("system_prompt"),
parent_session_id=source_id,
)
messages = await asyncio.to_thread(db.get_messages, source_id)
await asyncio.to_thread(db.replace_messages, fork_id, messages)
title = body.get("title")
if title is None:
base = source.get("title") or "fork"
try:
title = await asyncio.to_thread(db.get_next_title_in_lineage, base)
except Exception:
title = f"{base} fork"
try:
await asyncio.to_thread(db.set_session_title, fork_id, str(title))
except ValueError as exc:
return web.json_response(_openai_error(str(exc), code="invalid_title"), status=400)
fork = await asyncio.to_thread(db.get_session, fork_id) or {"id": fork_id, "parent_session_id": source_id}
return web.json_response({"object": "hermes.session", "session": self._session_response(fork)}, status=201)
@_admit_api_agent_request
async def _handle_session_chat(self, request: "web.Request") -> "web.Response":
"""POST /api/sessions/{session_id}/chat — one synchronous agent turn."""
gateway_session_key, key_err = self._parse_session_key_header(request)
if key_err is not None:
return key_err
session_id = request.match_info["session_id"]
session, err = await self._get_existing_session_or_404(session_id)
if err:
return err
body, err = await self._read_json_body(request)
if err:
return err
user_message, err = _session_chat_user_message(body)
if err is not None:
return err
system_prompt = body.get("system_message") or body.get("instructions")
if system_prompt is not None and not isinstance(system_prompt, str):
return web.json_response(_openai_error("system_message must be a string", code="invalid_system_message"), status=400)
# Runtime selection. A backend-acknowledged Browser model lock
# (require_model_lock in the body, or a previously confirmed lock
# persisted on the session row) is an execution contract and wins.
# Otherwise: session-persisted model (POST /api/sessions
# {"model": ...}) — previously fetched and discarded here — routes
# through model_routes when it is an alias (route
# provider/credentials come along) or threads through as
# session_model when it is a raw string; per-request body values
# come after that.
runtime_request = self._effective_session_runtime_request(
session=session,
body=body,
)
lock_error = self._runtime_lock_error(runtime_request)
if lock_error is not None:
return lock_error
if not self._persist_session_runtime_lock(session_id, runtime_request):
return web.json_response(
_openai_error(
"Could not persist the requested session model lock",
code="model_lock_persistence_failed",
),
status=500,
)
lock_active = bool(runtime_request.get("require_model_lock"))
if lock_active:
route = runtime_request.get("route")
session_model = None
requested = runtime_request.get("requested") or {}
agent_overrides: Dict[str, Any] = {}
if requested.get("model"):
agent_overrides["requested_model"] = requested["model"]
if requested.get("provider"):
agent_overrides["requested_provider"] = requested["provider"]
if runtime_request.get("model_options"):
agent_overrides["model_options"] = runtime_request["model_options"]
else:
stored_model = self._stored_session_model(session)
stored_route = self._resolve_route(stored_model)
route = stored_route or self._resolve_route(body.get("model"))
session_model = stored_model if (stored_model and stored_route is None) else None
agent_overrides = _request_agent_overrides(body, virtual_model=self._model_name)
selection_error = self._request_route_conflict_error(
session_id=session_id,
gateway_session_key=gateway_session_key,
requested_model=agent_overrides.get("requested_model"),
requested_provider=agent_overrides.get("requested_provider"),
route=route,
)
if selection_error:
return web.json_response(_openai_error(selection_error), status=400)
history = await self._conversation_history_for_session(session_id)
result, usage = await self._run_agent(
user_message=user_message,
conversation_history=history,
ephemeral_system_prompt=system_prompt,
session_id=session_id,
gateway_session_key=gateway_session_key,
route=route,
session_model=session_model,
requested_runtime=runtime_request.get("requested") or {},
route_source=runtime_request.get("route_source") or "global",
confirmed_runtime_lock=lock_active,
**agent_overrides,
)
effective_session_id = result.get("session_id") if isinstance(result, dict) else session_id
final_response = _resolve_media_to_data_urls(result.get("final_response", "") if isinstance(result, dict) else "")
headers = {"X-Hermes-Session-Id": effective_session_id or session_id}
if gateway_session_key:
headers["X-Hermes-Session-Key"] = gateway_session_key
runtime = {}
if isinstance(result, dict):
runtime = result.get("runtime") or {}
if not runtime and isinstance(usage, dict):
runtime = usage.get("runtime") or {}
runtime = self._sanitize_runtime_metadata(
runtime=runtime,
requested_runtime=runtime_request.get("requested"),
route_source=runtime_request.get("route_source") or "global",
model_lock=(
"confirmed"
if runtime and runtime_request.get("require_model_lock")
else "accepted"
if runtime_request.get("require_model_lock")
else ""
),
)
return web.json_response(
{
"object": "hermes.session.chat.completion",
"session_id": effective_session_id or session_id,
"message": {"role": "assistant", "content": final_response},
"usage": usage,
"runtime": runtime,
},
headers=headers,
)
@_admit_api_agent_request
async def _handle_session_chat_stream(self, request: "web.Request") -> "web.StreamResponse":
"""POST /api/sessions/{session_id}/chat/stream — SSE wrapper over _run_agent."""
gateway_session_key, key_err = self._parse_session_key_header(request)
if key_err is not None:
return key_err
session_id = request.match_info["session_id"]
session, err = await self._get_existing_session_or_404(session_id)
if err:
return err
body, err = await self._read_json_body(request)
if err:
return err
user_message, err = _session_chat_user_message(body)
if err is not None:
return err
system_prompt = body.get("system_message") or body.get("instructions")
if system_prompt is not None and not isinstance(system_prompt, str):
return web.json_response(_openai_error("system_message must be a string", code="invalid_system_message"), status=400)
# Runtime selection — mirrors _handle_session_chat (lock wins,
# otherwise session-persisted model then per-request values).
runtime_request = self._effective_session_runtime_request(
session=session,
body=body,
)
lock_error = self._runtime_lock_error(runtime_request)
if lock_error is not None:
return lock_error
if not self._persist_session_runtime_lock(session_id, runtime_request):
return web.json_response(
_openai_error(
"Could not persist the requested session model lock",
code="model_lock_persistence_failed",
),
status=500,
)
lock_active = bool(runtime_request.get("require_model_lock"))
if lock_active:
route = runtime_request.get("route")
session_model = None
requested = runtime_request.get("requested") or {}
agent_overrides: Dict[str, Any] = {}
if requested.get("model"):
agent_overrides["requested_model"] = requested["model"]
if requested.get("provider"):
agent_overrides["requested_provider"] = requested["provider"]
if runtime_request.get("model_options"):
agent_overrides["model_options"] = runtime_request["model_options"]
else:
stored_model = self._stored_session_model(session)
stored_route = self._resolve_route(stored_model)
route = stored_route or self._resolve_route(body.get("model"))
session_model = stored_model if (stored_model and stored_route is None) else None
agent_overrides = _request_agent_overrides(body, virtual_model=self._model_name)
selection_error = self._request_route_conflict_error(
session_id=session_id,
gateway_session_key=gateway_session_key,
requested_model=agent_overrides.get("requested_model"),
requested_provider=agent_overrides.get("requested_provider"),
route=route,
)
if selection_error:
return web.json_response(_openai_error(selection_error), status=400)
runtime_meta = self._sanitize_runtime_metadata(
requested_runtime=runtime_request.get("requested"),
route_source=runtime_request.get("route_source") or "global",
model_lock=("accepted" if lock_active else ""),
)
loop = asyncio.get_running_loop()
queue: "asyncio.Queue[Optional[tuple[str, Dict[str, Any]]]]" = asyncio.Queue()
message_id = f"msg_{uuid.uuid4().hex}"
run_id = f"run_{uuid.uuid4().hex}"
# Claim ownership while still inside the request's profile scope,
# before any run-keyed state exists — the same rule as /v1/runs, so
# /v1/runs/{id}* control of this turn is confined to the profile
# that started it (#93689).
self._run_owners[run_id] = self._run_idempotency_scope(request)
self._set_run_status(
run_id,
"queued",
session_id=session_id,
model=body.get("model", self._model_name),
)
seq = 0
def _event_payload(name: str, payload: Dict[str, Any]) -> tuple[str, Dict[str, Any]]:
nonlocal seq
seq += 1
payload.setdefault("session_id", session_id)
payload.setdefault("run_id", run_id)
payload.setdefault("seq", seq)
payload.setdefault("ts", time.time())
return name, payload
def _enqueue(name: str, payload: Dict[str, Any]) -> None:
event = _event_payload(name, payload)
try:
running_loop = asyncio.get_running_loop()
except RuntimeError:
running_loop = None
try:
if running_loop is loop:
queue.put_nowait(event)
else:
loop.call_soon_threadsafe(queue.put_nowait, event)
except RuntimeError:
pass
def _delta(delta: str) -> None:
if delta:
_enqueue("assistant.delta", {"message_id": message_id, "delta": delta})
def _tool_progress(event_type: str, tool_name: str = None, preview: str = None, args=None, **kwargs) -> None:
if event_type == "reasoning.available":
_enqueue("tool.progress", {"message_id": message_id, "tool_name": tool_name or "_thinking", "delta": preview or ""})
elif event_type in {"tool.started", "tool.completed", "tool.failed"}:
event_name = event_type.replace("tool.", "tool.")
_enqueue(event_name, {"message_id": message_id, "tool_name": tool_name, "preview": preview, "args": args})
async def _run_and_signal() -> None:
try:
await queue.put(_event_payload("run.started", {
"user_message": {"role": "user", "content": user_message},
"runtime": runtime_meta,
}))
self._set_run_status(run_id, "running", last_event="run.started")
await queue.put(_event_payload("message.started", {"message": {"id": message_id, "role": "assistant"}}))
history = await self._conversation_history_for_session(session_id)
result, usage = await self._run_agent(
user_message=user_message,
conversation_history=history,
ephemeral_system_prompt=system_prompt,
session_id=session_id,
stream_delta_callback=_delta,
tool_progress_callback=_tool_progress,
active_run_id=run_id,
gateway_session_key=gateway_session_key,
route=route,
session_model=session_model,
requested_runtime=runtime_request.get("requested") or {},
route_source=runtime_request.get("route_source") or "global",
confirmed_runtime_lock=lock_active,
**agent_overrides,
)
final_response = _resolve_media_to_data_urls(result.get("final_response", "") if isinstance(result, dict) else "")
effective_session_id = result.get("session_id", session_id) if isinstance(result, dict) else session_id
turn_messages = self._turn_transcript_messages(history, user_message, result) if isinstance(result, dict) else []
effective_runtime = {}
if isinstance(result, dict):
effective_runtime = result.get("runtime") or {}
if not effective_runtime and isinstance(usage, dict):
effective_runtime = usage.get("runtime") or {}
effective_runtime = self._sanitize_runtime_metadata(
runtime=effective_runtime,
requested_runtime=runtime_request.get("requested"),
route_source=runtime_request.get("route_source") or "global",
model_lock=(
"confirmed"
if effective_runtime and runtime_request.get("require_model_lock")
else "accepted"
if runtime_request.get("require_model_lock")
else ""
),
)
is_partial = bool(result.get("partial")) if isinstance(result, dict) else False
await queue.put(_event_payload("assistant.completed", {
"session_id": effective_session_id,
"message_id": message_id,
"content": final_response,
"completed": True,
"partial": is_partial,
"interrupted": False,
"runtime": effective_runtime,
}))
# A steer accepted after the final assistant response is drained
# into result["pending_steer"] by the turn finalizer instead of
# being consumed; surface it so clients can replay it as the
# next user turn rather than silently losing it.
pending_steer = result.get("pending_steer") if isinstance(result, dict) else None
completed_payload = {
"session_id": effective_session_id,
"message_id": message_id,
"completed": True,
"messages": turn_messages,
"usage": usage,
"runtime": effective_runtime,
}
if pending_steer:
completed_payload["pending_steer"] = pending_steer
await queue.put(_event_payload("run.completed", completed_payload))
self._set_run_status(
run_id,
"completed",
session_id=effective_session_id,
usage=usage,
last_event="run.completed",
**({"pending_steer": pending_steer} if pending_steer else {}),
)
except asyncio.CancelledError:
self._set_run_status(run_id, "cancelled", last_event="run.cancelled")
raise
except Exception as exc:
logger.exception("[api_server] session chat stream failed")
self._set_run_status(
run_id,
"failed",
error=_redact_api_error_text(exc),
last_event="run.failed",
)
await queue.put(_event_payload("error", {"message": _redact_api_error_text(exc)}))
finally:
self._active_run_agents.pop(run_id, None)
self._release_run_owner_if_forgotten(run_id)
await queue.put(_event_payload("done", {}))
await queue.put(None)
# NOTE: deliberately NOT registered in _active_run_tasks — this turn
# is already counted by active_agent_work_count() via
# _inflight_agent_runs (_run_agent), and a second task-based entry
# would double-count it in the shutdown drain. Run-scoped control
# needs only the agent ref, registered by _run_agent(active_run_id).
task = asyncio.create_task(_run_and_signal())
try:
self._background_tasks.add(task)
except TypeError:
pass
if hasattr(task, "add_done_callback"):
task.add_done_callback(self._background_tasks.discard)
headers = {
"Content-Type": "text/event-stream",
"Cache-Control": "no-cache",
"X-Accel-Buffering": "no",
"X-Hermes-Session-Id": session_id,
}
if gateway_session_key:
headers["X-Hermes-Session-Key"] = gateway_session_key
response = web.StreamResponse(status=200, headers=headers)
await response.prepare(request)
try:
while True:
try:
item = await asyncio.wait_for(queue.get(), timeout=CHAT_COMPLETIONS_SSE_KEEPALIVE_SECONDS)
except asyncio.TimeoutError:
await response.write(b": keepalive\n\n")
continue
if item is None:
break
name, payload = item
await response.write(_sse_frame(payload, event=name, ensure_ascii=False))
except (ConnectionResetError, ConnectionAbortedError, BrokenPipeError, OSError):
await self._drain_session_stream_task_on_disconnect(
run_id, task, interrupt_message="SSE client disconnected", shield_wait=False
)
logger.info("Session SSE client disconnected; interrupted live run %s", run_id)
except asyncio.CancelledError:
await self._drain_session_stream_task_on_disconnect(
run_id, task, interrupt_message="SSE task cancelled", shield_wait=True
)
logger.info("Session SSE task cancelled; drained live run %s", run_id)
raise
except Exception as exc:
logger.debug("[api_server] session SSE stream error: %s", exc)
return response
async def _drain_session_stream_task_on_disconnect(
self,
run_id: str,
task: "asyncio.Task",
*,
interrupt_message: str,
shield_wait: bool,
) -> None:
"""Preserve live run control refs until the executor-backed turn actually exits."""
agent = self._active_run_agents.get(run_id)
if agent is None:
if not task.done():
task.cancel()
with suppress(Exception):
await task
return
with suppress(Exception):
agent.interrupt(interrupt_message)
if not task.done():
with suppress(Exception):
await (asyncio.shield(task) if shield_wait else task)
async def _handle_session_model_lock(self, request: "web.Request") -> "web.Response":
"""POST /api/sessions/{session_id}/model — backend-ack a Browser model lock."""
auth_err = self._check_auth(request)
if auth_err:
return auth_err
session_id = request.match_info["session_id"]
_, err = await self._get_existing_session_or_404(session_id)
if err:
return err
body, err = await self._read_json_body(request)
if err:
return err
runtime_request = self._session_runtime_request_from_body(body)
runtime_request["require_model_lock"] = True
lock_error = self._runtime_lock_error(runtime_request)
if lock_error is not None:
return lock_error
if not self._persist_session_runtime_lock(session_id, runtime_request):
return web.json_response(
_openai_error(
"Could not persist the requested session model lock",
code="model_lock_persistence_failed",
),
status=500,
)
requested = runtime_request.get("requested") or {}
route = runtime_request.get("route") or {}
runtime = self._sanitize_runtime_metadata(
runtime={
"provider": route.get("provider") or requested.get("provider") or "",
"model": route.get("model") or requested.get("model") or "",
"route_source": runtime_request.get("route_source") or "raw_request",
},
requested_runtime=requested,
route_source=runtime_request.get("route_source") or "raw_request",
model_lock="accepted",
)
return web.json_response({
"object": "hermes.session.model_lock",
"session_id": session_id,
"runtime": runtime,
})
@_admit_api_agent_request
async def _handle_chat_completions(self, request: "web.Request") -> "web.Response":
"""POST /v1/chat/completions — OpenAI Chat Completions format."""
# Bound total in-flight agent runs (configurable; #7483).
limited = self._concurrency_limited_response()
if limited is not None:
return limited
# Parse request body
try:
body = await request.json()
except (json.JSONDecodeError, Exception):
return web.json_response(_openai_error("Invalid JSON in request body"), status=400)
messages = body.get("messages")
if not messages or not isinstance(messages, list):
return web.json_response(
{"error": {"message": "Missing or invalid 'messages' field", "type": "invalid_request_error"}},
status=400,
)
stream = _coerce_request_bool(body.get("stream"), default=False)
# Extract system message (becomes ephemeral system prompt layered ON TOP of core)
system_prompt = None
conversation_messages: List[Dict[str, str]] = []
for idx, msg in enumerate(messages):
role = msg.get("role", "")
raw_content = msg.get("content", "")
if role == "system":
# System messages don't support images (Anthropic rejects, OpenAI
# text-model systems don't render them). Flatten to text.
content = _normalize_chat_content(raw_content)
if system_prompt is None:
system_prompt = content
else:
system_prompt = system_prompt + "\n" + content
elif role in {"user", "assistant"}:
try:
content = _normalize_multimodal_content(raw_content)
except ValueError as exc:
return _multimodal_validation_error(exc, param=f"messages[{idx}].content")
conversation_messages.append({"role": role, "content": content})
# Extract the last user message as the primary input
user_message: Any = ""
history = []
if conversation_messages:
user_message = conversation_messages[-1].get("content", "")
history = conversation_messages[:-1]
if not _content_has_visible_payload(user_message):
return web.json_response(
{"error": {"message": "No user message found in messages", "type": "invalid_request_error"}},
status=400,
)
# Allow caller to scope long-term memory (e.g. Honcho) with a
# stable per-channel identifier via X-Hermes-Session-Key. This
# is independent of X-Hermes-Session-Id: the key persists across
# transcripts while the id rotates when the caller starts a new
# transcript (i.e. /new semantics). See _parse_session_key_header.
gateway_session_key, key_err = self._parse_session_key_header(request)
if key_err is not None:
return key_err
# Allow caller to continue an existing session by passing X-Hermes-Session-Id.
# When provided, history is loaded from state.db instead of from the request body.
#
# Security: session continuation exposes conversation history, so it is
# only allowed when the API key is configured and the request is
# authenticated. Without this gate, any unauthenticated client could
# read arbitrary session history by guessing/enumerating session IDs.
provided_session_id = request.headers.get("X-Hermes-Session-Id", "").strip()
if provided_session_id:
if not self._api_key:
logger.warning(
"Session continuation via X-Hermes-Session-Id rejected: "
"no API key configured. Set API_SERVER_KEY to enable "
"session continuity."
)
return web.json_response(
_openai_error(
"Session continuation requires API key authentication. "
"Configure API_SERVER_KEY to enable this feature."
),
status=403,
)
# Sanitize: reject control characters that could enable header
# injection, and path-traversal-shaped IDs that would escape the
# sessions directory when interpolated into on-disk artifact
# filenames (session snapshots, request dumps). Mirrors the native
# gateway's entry-boundary guard (gateway.session._is_path_unsafe).
from gateway.session import _is_path_unsafe
if re.search(r'[\r\n\x00]', provided_session_id) or _is_path_unsafe(provided_session_id):
return web.json_response(
{"error": {"message": "Invalid session ID", "type": "invalid_request_error"}},
status=400,
)
if len(provided_session_id) > self._MAX_SESSION_HEADER_LEN:
return web.json_response(
{"error": {"message": "Session ID too long", "type": "invalid_request_error"}},
status=400,
)
session_id = provided_session_id
try:
db = await self._ensure_session_db_async()
if db is not None:
history = await asyncio.to_thread(db.get_messages_as_conversation, session_id)
except Exception as e:
logger.warning("Failed to load session history for %s: %s", session_id, e)
history = []
else:
# Derive a stable session ID from the conversation fingerprint so
# that consecutive messages from the same Open WebUI (or similar)
# conversation map to the same Hermes session. The first user
# message + system prompt are constant across all turns.
first_user = ""
for cm in conversation_messages:
if cm.get("role") == "user":
first_user = cm.get("content", "")
break
session_id = _derive_chat_session_id(system_prompt, first_user)
# history already set from request body above
completion_id = f"chatcmpl-{uuid.uuid4().hex[:29]}"
model_name = body.get("model", self._model_name)
created = int(time.time())
# Per-client model routing: if the requested model matches a
# configured model_routes alias, this request's agent is created
# with that route's model/provider instead of the global default.
route = self._resolve_route(model_name)
agent_overrides = _request_agent_overrides(
body,
virtual_model=self._model_name,
allow_bare_model=self._direct_model_requests,
)
selection_error = self._request_route_conflict_error(
session_id=session_id,
gateway_session_key=gateway_session_key,
requested_model=agent_overrides.get("requested_model"),
requested_provider=agent_overrides.get("requested_provider"),
route=route,
)
if selection_error:
return web.json_response(_openai_error(selection_error), status=400)
if stream:
_stream_q = ThreadSafeAsyncQueue()
def _on_delta(delta):
# Filter out None — the agent fires stream_delta_callback(None)
# to signal the CLI display to close its response box before
# tool execution, but the SSE writer uses None as end-of-stream
# sentinel. Forwarding it would prematurely close the HTTP
# response, causing Open WebUI (and similar frontends) to miss
# the final answer after tool calls. The SSE loop detects
# completion via agent_task.done() instead.
# Called from the worker thread running run_conversation —
# put_threadsafe (not put_nowait) is required here.
if delta is not None:
_stream_q.put_threadsafe(delta)
# Track which tool_call_ids we've emitted a "running" lifecycle
# event for, so a "completed" event without a matching "running"
# (e.g. internal/filtered tools) is silently dropped instead of
# producing an orphaned event clients can't correlate.
_started_tool_call_ids: set[str] = set()
def _on_tool_start(tool_call_id, function_name, function_args):
"""Emit ``hermes.tool.progress`` with ``status: running``.
Replaces the old ``tool_progress_callback("tool.started",
...)`` emit so SSE consumers receive a single event per
tool start, carrying both the legacy ``tool``/``emoji``/
``label`` payload (for #6972 frontends) and the new
``toolCallId``/``status`` correlation fields (#16588).
Skips tools whose names start with ``_`` so internal
events (``_thinking``, …) stay off the wire — matching
the prior ``_on_tool_progress`` filter exactly.
"""
if not tool_call_id or function_name.startswith("_"):
return
_started_tool_call_ids.add(tool_call_id)
from agent.display import build_tool_preview, get_tool_emoji
label = build_tool_preview(function_name, function_args) or function_name
_stream_q.put_threadsafe(("__tool_progress__", {
"tool": function_name,
"emoji": get_tool_emoji(function_name),
"label": label,
"toolCallId": tool_call_id,
"status": "running",
}))
def _on_tool_complete(tool_call_id, function_name, function_args, function_result):
"""Emit the matching ``status: completed`` event.
Dropped if the start was filtered (internal tool, missing
id, or never seen) so clients never get an orphaned
``completed`` they can't correlate to a prior ``running``.
"""
if not tool_call_id or tool_call_id not in _started_tool_call_ids:
return
_started_tool_call_ids.discard(tool_call_id)
_stream_q.put_threadsafe(("__tool_progress__", {
"tool": function_name,
"toolCallId": tool_call_id,
"status": "completed",
}))
# Start agent in background. agent_ref is a mutable container
# so the SSE writer can interrupt the agent on client disconnect.
#
# ``tool_progress_callback`` is intentionally not wired here:
# it would duplicate every emit because ``run_agent`` fires it
# side-by-side with ``tool_start_callback``/``tool_complete_callback``.
# The structured callbacks are strictly richer (they carry
# the tool_call id), so they own the chat-completions SSE channel.
agent_ref = [None]
agent_task = asyncio.ensure_future(self._run_agent(
user_message=user_message,
conversation_history=history,
ephemeral_system_prompt=system_prompt,
session_id=session_id,
stream_delta_callback=_on_delta,
tool_start_callback=_on_tool_start,
tool_complete_callback=_on_tool_complete,
agent_ref=agent_ref,
gateway_session_key=gateway_session_key,
**agent_overrides,
route=route,
))
# Ensure SSE drain loops can terminate without relying on polling
# agent_task.done(), which can race with queue timeout checks.
agent_task.add_done_callback(lambda _fut: _stream_q.put_nowait(None))
return await self._write_sse_chat_completion(
request, completion_id, model_name, created, _stream_q,
agent_task, agent_ref, session_id=session_id,
gateway_session_key=gateway_session_key,
)
# Non-streaming: run the agent (with optional Idempotency-Key)
async def _compute_completion():
return await self._run_agent(
user_message=user_message,
conversation_history=history,
ephemeral_system_prompt=system_prompt,
session_id=session_id,
gateway_session_key=gateway_session_key,
**agent_overrides,
route=route,
)
idempotency_key = request.headers.get("Idempotency-Key")
if idempotency_key:
fp = _make_request_fingerprint(
body,
keys=[
"model",
"provider",
"model_options",
"messages",
"tools",
"tool_choice",
"stream",
],
)
try:
result, usage = await _idem_cache.get_or_set(
idempotency_key, fp, _compute_completion
)
except Exception as e:
logger.error(
"Error running agent for chat completions: %s", e, exc_info=True
)
return web.json_response(
_openai_error(
f"Internal server error: {e}", err_type="server_error"
),
status=500,
)
else:
try:
result, usage = await _compute_completion()
except Exception as e:
logger.error(
"Error running agent for chat completions: %s", e, exc_info=True
)
return web.json_response(
_openai_error(
f"Internal server error: {e}", err_type="server_error"
),
status=500,
)
final_response = _resolve_media_to_data_urls(result.get("final_response") or "")
is_partial = bool(result.get("partial"))
is_failed = bool(result.get("failed"))
completed = bool(result.get("completed", True))
raw_err_msg = result.get("error")
err_msg = _redact_api_error_text(raw_err_msg) if raw_err_msg else raw_err_msg
# Decide finish_reason. OpenAI uses "length" for truncation, "stop"
# for normal completion, and downstream SDKs accept "error" / custom
# codes. See issue #22496.
if is_partial and err_msg and "truncat" in err_msg.lower():
finish_reason = "length"
elif is_failed or (not completed and err_msg):
finish_reason = "error"
else:
finish_reason = "stop"
response_headers = {
"X-Hermes-Session-Id": result.get("session_id", session_id),
}
if gateway_session_key:
response_headers["X-Hermes-Session-Key"] = gateway_session_key
# Hard-fail path: no usable assistant text AND a real failure → 5xx
# with OpenAI-style error envelope so SDK clients raise instead of
# silently rendering the internal failure string as message.content.
if not final_response and (is_failed or is_partial):
err_body = _openai_error(
err_msg or "Agent run did not produce a response.",
err_type="server_error",
code="agent_incomplete",
)
err_body["error"]["hermes"] = {
"completed": completed,
"partial": is_partial,
"failed": is_failed,
}
response_headers["X-Hermes-Completed"] = "false"
response_headers["X-Hermes-Partial"] = "true" if is_partial else "false"
return web.json_response(err_body, status=502, headers=response_headers)
# Soft-partial path: we have *some* text but the run did not complete
# (e.g. truncation with partial buffered output). Still 200 but signal
# truncation via finish_reason="length" + Hermes-specific extras.
response_data = {
"id": completion_id,
"object": "chat.completion",
"created": created,
"model": model_name,
"choices": [
{
"index": 0,
"message": {
"role": "assistant",
"content": final_response,
},
"finish_reason": finish_reason,
}
],
"usage": {
"prompt_tokens": usage.get("input_tokens", 0),
"completion_tokens": usage.get("output_tokens", 0),
"total_tokens": usage.get("total_tokens", 0),
},
}
if is_partial or is_failed or not completed:
response_data["hermes"] = {
"completed": completed,
"partial": is_partial,
"failed": is_failed,
"error": err_msg,
"error_code": "output_truncated" if finish_reason == "length" else "agent_error",
}
response_headers["X-Hermes-Completed"] = "false"
response_headers["X-Hermes-Partial"] = "true" if is_partial else "false"
if err_msg:
response_headers["X-Hermes-Error"] = _redact_api_error_text(err_msg, limit=200)
return web.json_response(response_data, headers=response_headers)
async def _write_sse_chat_completion(
self, request: "web.Request", completion_id: str, model: str,
created: int, stream_q, agent_task, agent_ref=None, session_id: str = None,
gateway_session_key: str = None,
) -> "web.StreamResponse":
"""Write real streaming SSE from agent's stream_delta_callback queue.
If the client disconnects mid-stream (network drop, browser tab close),
the agent is interrupted via ``agent.interrupt()`` so it stops making
LLM API calls, and the asyncio task wrapper is cancelled.
"""
sse_headers = {
"Content-Type": "text/event-stream",
"Cache-Control": "no-cache",
"X-Accel-Buffering": "no",
}
# CORS middleware can't inject headers into StreamResponse after
# prepare() flushes them, so resolve CORS headers up front.
origin = request.headers.get("Origin", "")
cors = self._cors_headers_for_origin(origin) if origin else None
if cors:
sse_headers.update(cors)
if session_id:
sse_headers["X-Hermes-Session-Id"] = session_id
if gateway_session_key:
sse_headers["X-Hermes-Session-Key"] = gateway_session_key
response = web.StreamResponse(status=200, headers=sse_headers)
await response.prepare(request)
try:
last_activity = time.monotonic()
# Role chunk
role_chunk = {
"id": completion_id, "object": "chat.completion.chunk",
"created": created, "model": model,
"choices": [{"index": 0, "delta": {"role": "assistant"}, "finish_reason": None}],
}
await response.write(_sse_frame(role_chunk))
last_activity = time.monotonic()
# Helper — route a queue item to the correct SSE event.
async def _emit(item):
"""Write a single queue item to the SSE stream.
Plain strings are sent as normal ``delta.content`` chunks.
Tagged tuples ``("__tool_progress__", payload)`` are sent
as a custom ``event: hermes.tool.progress`` SSE event so
frontends can display them without storing the markers in
conversation history. See #6972 for the original event,
#16588 for the ``toolCallId``/``status`` lifecycle fields.
"""
if isinstance(item, tuple) and len(item) == 2 and item[0] == "__tool_progress__":
await response.write(_sse_frame(item[1], event="hermes.tool.progress"))
else:
content_chunk = {
"id": completion_id, "object": "chat.completion.chunk",
"created": created, "model": model,
"choices": [{"index": 0, "delta": {"content": item}, "finish_reason": None}],
}
await response.write(_sse_frame(content_chunk))
return time.monotonic()
# Stream content chunks as they arrive from the agent. Woken
# directly by put_threadsafe's call_soon_threadsafe — no
# executor hop, no poll-interval latency (see
# ThreadSafeAsyncQueue's docstring).
while True:
try:
delta = await asyncio.wait_for(stream_q.get(), timeout=0.5)
except asyncio.TimeoutError:
if agent_task.done():
# Drain any remaining items
while True:
try:
delta = stream_q.get_nowait()
if delta is None:
break
last_activity = await _emit(delta)
except asyncio.QueueEmpty:
break
break
if time.monotonic() - last_activity >= CHAT_COMPLETIONS_SSE_KEEPALIVE_SECONDS:
await response.write(b": keepalive\n\n")
last_activity = time.monotonic()
continue
if delta is None: # End of stream sentinel
break
last_activity = await _emit(delta)
# Get usage from completed agent. The agent can fail two ways
# after the content queue terminates cleanly: (1) ``agent_task``
# raises, or (2) it returns a ``result`` dict flagged
# failed/partial/incomplete. Both previously fell through to a
# ``finish_reason: "stop"`` chunk, so OpenAI-compatible clients
# saw a fake success. Surface either as a non-"stop" finish so
# the failure is detectable — mirroring the non-streaming path's
# decision logic (see the finish_reason block above).
usage = {"input_tokens": 0, "output_tokens": 0, "total_tokens": 0}
result = None
agent_error = None
try:
result, agent_usage = await agent_task
usage = agent_usage or usage
except Exception as exc:
agent_error = exc
logger.error(
"Agent task %s failed during SSE streaming: %s", completion_id, exc
)
# Inspect the result dict for a flagged (non-exception) failure.
is_partial = bool(result.get("partial")) if isinstance(result, dict) else False
is_failed = bool(result.get("failed")) if isinstance(result, dict) else False
completed = bool(result.get("completed", True)) if isinstance(result, dict) else True
err_msg = result.get("error") if isinstance(result, dict) else None
if agent_error is not None:
is_failed = True
err_msg = err_msg or str(agent_error)
# Decide finish_reason, matching the non-streaming logic: "length"
# for truncation, "error" for failure, "stop" for normal completion.
if is_partial and err_msg and "truncat" in err_msg.lower():
finish_reason = "length"
elif agent_error is not None or is_failed or (not completed and err_msg):
finish_reason = "error"
else:
finish_reason = "stop"
# Finish chunk
finish_chunk = {
"id": completion_id, "object": "chat.completion.chunk",
"created": created, "model": model,
"choices": [{"index": 0, "delta": {}, "finish_reason": finish_reason}],
"usage": {
"prompt_tokens": usage.get("input_tokens", 0),
"completion_tokens": usage.get("output_tokens", 0),
"total_tokens": usage.get("total_tokens", 0),
},
}
if finish_reason != "stop":
finish_chunk["choices"][0]["delta"] = {}
if err_msg:
finish_chunk["error"] = {
"message": err_msg,
"type": type(agent_error).__name__ if agent_error else "agent_error",
}
finish_chunk["hermes"] = {
"completed": completed,
"partial": is_partial,
"failed": is_failed,
"error": err_msg,
"error_code": "output_truncated" if finish_reason == "length" else "agent_error",
}
await response.write(_sse_frame(finish_chunk))
await response.write(b"data: [DONE]\n\n")
except (ConnectionResetError, ConnectionAbortedError, BrokenPipeError, OSError):
# Client disconnected mid-stream. Interrupt the agent so it
# stops making LLM API calls at the next loop iteration, then
# cancel the asyncio task wrapper.
agent = agent_ref[0] if agent_ref else None
if agent is not None:
try:
request_hard_interrupt(agent, "SSE client disconnected")
except Exception:
pass
_reap_disconnected_agent_processes(agent)
if not agent_task.done():
agent_task.cancel()
try:
await agent_task
except (asyncio.CancelledError, Exception):
pass
logger.info("SSE client disconnected; interrupted agent task %s", completion_id)
except Exception as _exc:
# Agent crashed mid-stream. Try to emit an error chunk
# so the client gets a proper response instead of a
# TransferEncodingError from incomplete chunked encoding.
import traceback as _tb
logger.error("Agent crashed mid-stream for %s: %s", completion_id, _tb.format_exc()[:300])
try:
error_chunk = {
"id": completion_id, "object": "chat.completion.chunk",
"created": created, "model": model,
"choices": [{"index": 0, "delta": {}, "finish_reason": "error"}],
}
await response.write(_sse_frame(error_chunk))
await response.write(b"data: [DONE]\n\n")
except Exception:
pass
return response
async def _write_sse_responses(
self,
request: "web.Request",
response_id: str,
model: str,
created_at: int,
stream_q,
agent_task,
agent_ref,
conversation_history: List[Dict[str, str]],
user_message: str,
instructions: Optional[str],
conversation: Optional[str],
store: bool,
session_id: str,
gateway_session_key: Optional[str] = None,
) -> "web.StreamResponse":
"""Write an SSE stream for POST /v1/responses (OpenAI Responses API).
Emits spec-compliant event types as the agent runs:
- ``response.created`` — initial envelope (status=in_progress)
- ``response.output_text.delta`` / ``response.output_text.done`` —
streamed assistant text
- ``response.output_item.added`` / ``response.output_item.done``
with ``item.type == "function_call"`` — when the agent invokes a
tool (both events fire; the ``done`` event carries the finalized
``arguments`` string)
- ``response.output_item.added`` with
``item.type == "function_call_output"`` — tool result with
``{call_id, output, status}``
- ``response.completed`` — terminal event carrying the full
response object with all output items + usage (same payload
shape as the non-streaming path for parity)
- ``response.failed`` — terminal event on agent error
If the client disconnects mid-stream, ``agent.interrupt()`` is
called so the agent stops issuing upstream LLM calls, then the
asyncio task is cancelled. When ``store=True`` an initial
``in_progress`` snapshot is persisted immediately after
``response.created`` and disconnects update it to an
``incomplete`` snapshot so GET /v1/responses/{id} and
``previous_response_id`` chaining still have something to
recover from.
"""
sse_headers = {
"Content-Type": "text/event-stream",
"Cache-Control": "no-cache",
"X-Accel-Buffering": "no",
}
origin = request.headers.get("Origin", "")
cors = self._cors_headers_for_origin(origin) if origin else None
if cors:
sse_headers.update(cors)
if session_id:
sse_headers["X-Hermes-Session-Id"] = session_id
if gateway_session_key:
sse_headers["X-Hermes-Session-Key"] = gateway_session_key
response = web.StreamResponse(status=200, headers=sse_headers)
await response.prepare(request)
# State accumulated during the stream
final_text_parts: List[str] = []
# Track open function_call items by name so we can emit a matching
# ``done`` event when the tool completes. Order preserved.
pending_tool_calls: List[Dict[str, Any]] = []
# Output items we've emitted so far (used to build the terminal
# response.completed payload). Kept in the order they appeared.
emitted_items: List[Dict[str, Any]] = []
# Monotonic counter for output_index (spec requires it).
output_index = 0
# Monotonic counter for call_id generation if the agent doesn't
# provide one (it doesn't, from tool_progress_callback).
call_counter = 0
# Canonical Responses SSE events include a monotonically increasing
# sequence_number. Add it server-side for every emitted event so
# clients that validate the OpenAI event schema can parse our stream.
sequence_number = 0
# Track the assistant message item id + content index for text
# delta events — the spec ties deltas to a specific item.
message_item_id = f"msg_{uuid.uuid4().hex[:24]}"
message_output_index: Optional[int] = None
message_opened = False
async def _write_event(event_type: str, data: Dict[str, Any]) -> None:
nonlocal sequence_number
if "sequence_number" not in data:
data["sequence_number"] = sequence_number
sequence_number += 1
await response.write(_sse_frame(data, event=event_type))
def _envelope(status: str) -> Dict[str, Any]:
env: Dict[str, Any] = {
"id": response_id,
"object": "response",
"status": status,
"created_at": created_at,
"model": model,
}
return env
final_response_text = ""
agent_error: Optional[str] = None
usage: Dict[str, int] = {"input_tokens": 0, "output_tokens": 0, "total_tokens": 0}
terminal_snapshot_persisted = False
def _persist_response_snapshot(
response_env: Dict[str, Any],
*,
conversation_history_snapshot: Optional[List[Dict[str, Any]]] = None,
session_id_snapshot: Optional[str] = None,
) -> None:
if not store:
return
if conversation_history_snapshot is None:
conversation_history_snapshot = list(conversation_history)
conversation_history_snapshot.append({"role": "user", "content": user_message})
self._response_store.put(response_id, {
"response": response_env,
"conversation_history": conversation_history_snapshot,
"instructions": instructions,
"session_id": session_id_snapshot or session_id,
})
if conversation:
self._response_store.set_conversation(conversation, response_id)
def _persist_incomplete_if_needed() -> None:
"""Persist an ``incomplete`` snapshot if no terminal one was written.
Called from both the client-disconnect (``ConnectionResetError``)
and server-cancellation (``asyncio.CancelledError``) paths so
GET /v1/responses/{id} and ``previous_response_id`` chaining keep
working after abrupt stream termination.
"""
if not store or terminal_snapshot_persisted:
return
incomplete_text = "".join(final_text_parts) or final_response_text
incomplete_items: List[Dict[str, Any]] = list(emitted_items)
if incomplete_text:
incomplete_items.append({
"type": "message",
"role": "assistant",
"content": [{"type": "output_text", "text": incomplete_text}],
})
incomplete_env = _envelope("incomplete")
incomplete_env["output"] = incomplete_items
incomplete_env["usage"] = {
"input_tokens": usage.get("input_tokens", 0),
"output_tokens": usage.get("output_tokens", 0),
"total_tokens": usage.get("total_tokens", 0),
}
incomplete_history = list(conversation_history)
incomplete_history.append({"role": "user", "content": user_message})
if incomplete_text:
incomplete_history.append({"role": "assistant", "content": incomplete_text})
_persist_response_snapshot(
incomplete_env,
conversation_history_snapshot=incomplete_history,
)
try:
# response.created — initial envelope, status=in_progress
created_env = _envelope("in_progress")
created_env["output"] = []
await _write_event("response.created", {
"type": "response.created",
"response": created_env,
})
_persist_response_snapshot(created_env)
last_activity = time.monotonic()
async def _open_message_item() -> None:
"""Emit response.output_item.added for the assistant message
the first time any text delta arrives."""
nonlocal message_opened, message_output_index, output_index
if message_opened:
return
message_opened = True
message_output_index = output_index
output_index += 1
item = {
"id": message_item_id,
"type": "message",
"status": "in_progress",
"role": "assistant",
"content": [],
}
await _write_event("response.output_item.added", {
"type": "response.output_item.added",
"output_index": message_output_index,
"item": item,
})
async def _emit_text_delta(delta_text: str) -> None:
await _open_message_item()
final_text_parts.append(delta_text)
await _write_event("response.output_text.delta", {
"type": "response.output_text.delta",
"item_id": message_item_id,
"output_index": message_output_index,
"content_index": 0,
"delta": delta_text,
"logprobs": [],
})
async def _emit_tool_started(payload: Dict[str, Any]) -> str:
"""Emit response.output_item.added for a function_call.
Returns the call_id so the matching completion event can
reference it. Prefer the real ``tool_call_id`` from the
agent when available; fall back to a generated call id for
safety in tests or older code paths.
"""
nonlocal output_index, call_counter
call_counter += 1
call_id = payload.get("tool_call_id") or f"call_{response_id[5:]}_{call_counter}"
args = payload.get("arguments", {})
if isinstance(args, dict):
arguments_str = json.dumps(args)
else:
arguments_str = str(args)
item = {
"id": f"fc_{uuid.uuid4().hex[:24]}",
"type": "function_call",
"status": "in_progress",
"name": payload.get("name", ""),
"call_id": call_id,
"arguments": arguments_str,
}
idx = output_index
output_index += 1
pending_tool_calls.append({
"call_id": call_id,
"name": payload.get("name", ""),
"arguments": arguments_str,
"item_id": item["id"],
"output_index": idx,
})
emitted_items.append({
"type": "function_call",
"name": payload.get("name", ""),
"arguments": arguments_str,
"call_id": call_id,
})
await _write_event("response.output_item.added", {
"type": "response.output_item.added",
"output_index": idx,
"item": item,
})
return call_id
async def _emit_tool_completed(payload: Dict[str, Any]) -> None:
"""Emit response.output_item.done (function_call) followed
by response.output_item.added (function_call_output)."""
nonlocal output_index
call_id = payload.get("tool_call_id")
result = payload.get("result", "")
pending = None
if call_id:
for i, p in enumerate(pending_tool_calls):
if p["call_id"] == call_id:
pending = pending_tool_calls.pop(i)
break
if pending is None:
# Completion without a matching start — skip to avoid
# emitting orphaned done events.
return
# function_call done
done_item = {
"id": pending["item_id"],
"type": "function_call",
"status": "completed",
"name": pending["name"],
"call_id": pending["call_id"],
"arguments": pending["arguments"],
}
await _write_event("response.output_item.done", {
"type": "response.output_item.done",
"output_index": pending["output_index"],
"item": done_item,
})
# function_call_output added (result)
result_str = result if isinstance(result, str) else json.dumps(result)
output_parts = [{"type": "input_text", "text": result_str}]
output_item = {
"id": f"fco_{uuid.uuid4().hex[:24]}",
"type": "function_call_output",
"call_id": pending["call_id"],
"output": output_parts,
"status": "completed",
}
idx = output_index
output_index += 1
emitted_items.append({
"type": "function_call_output",
"call_id": pending["call_id"],
"output": output_parts,
})
await _write_event("response.output_item.added", {
"type": "response.output_item.added",
"output_index": idx,
"item": output_item,
})
await _write_event("response.output_item.done", {
"type": "response.output_item.done",
"output_index": idx,
"item": output_item,
})
# Main drain loop — thread-safe queue fed by agent callbacks.
async def _dispatch(it) -> None:
"""Route a queue item to the correct SSE emitter.
Plain strings are text deltas — they are batched (50ms)
to reduce Open WebUI re-render storms. Tagged tuples
with ``__tool_started__`` / ``__tool_completed__``
prefixes are tool lifecycle events and flush the buffer
before emitting.
"""
nonlocal _batch_timer
if isinstance(it, tuple) and len(it) == 2 and isinstance(it[0], str):
tag, payload = it
# Flush batched text before tool events
if _batch_buf:
await _flush_batch()
if tag == "__tool_started__":
await _emit_tool_started(payload)
elif tag == "__tool_completed__":
await _emit_tool_completed(payload)
elif isinstance(it, str):
# Batch text deltas — append to buffer, flush on timer
_batch_buf.append(it)
if _batch_timer is None:
_batch_timer = asyncio.create_task(_batch_flush_after(0.05))
# Other types are silently dropped.
# ── Batching state ──
_batch_buf: List[str] = []
_batch_timer: Optional[asyncio.Task] = None
_batch_lock = asyncio.Lock()
async def _batch_flush_after(delay: float) -> None:
"""Wait delay seconds, then flush accumulated text deltas."""
try:
await asyncio.sleep(delay)
except asyncio.CancelledError:
return
# Clear timer reference BEFORE flush so new deltas
# can start a fresh timer while we emit
nonlocal _batch_buf, _batch_timer
_batch_timer = None
await _flush_batch()
async def _flush_batch() -> None:
"""Emit a single SSE delta for all accumulated text."""
nonlocal _batch_buf
async with _batch_lock:
if _batch_buf:
combined = "".join(_batch_buf)
_batch_buf = []
await _emit_text_delta(combined)
while True:
try:
item = await asyncio.wait_for(stream_q.get(), timeout=0.5)
except asyncio.TimeoutError:
if agent_task.done():
# Drain remaining
while True:
try:
item = stream_q.get_nowait()
if item is None:
break
await _dispatch(item)
last_activity = time.monotonic()
except asyncio.QueueEmpty:
break
break
if time.monotonic() - last_activity >= CHAT_COMPLETIONS_SSE_KEEPALIVE_SECONDS:
await response.write(b": keepalive\n\n")
last_activity = time.monotonic()
continue
if item is None: # EOS sentinel
# Cancel pending timer and flush remaining batched text
if _batch_timer and not _batch_timer.done():
_batch_timer.cancel()
_batch_timer = None
if _batch_buf:
await _flush_batch()
break
await _dispatch(item)
last_activity = time.monotonic()
# Flush any final batched text before processing result
if _batch_buf:
await _flush_batch()
# Pick up agent result + usage from the completed task
try:
result, agent_usage = await agent_task
usage = agent_usage or usage
# If the agent produced a final_response but no text
# deltas were streamed (e.g. some providers only emit
# the full response at the end), emit a single fallback
# delta so Responses clients still receive a live text part.
agent_final = result.get("final_response", "") if isinstance(result, dict) else ""
if agent_final and not final_text_parts:
await _emit_text_delta(agent_final)
if agent_final and not final_response_text:
final_response_text = agent_final
if isinstance(result, dict) and result.get("error") and not final_response_text:
agent_error = _redact_api_error_text(result["error"])
except Exception as e: # noqa: BLE001
logger.error("Error running agent for streaming responses: %s", e, exc_info=True)
agent_error = _redact_api_error_text(e)
# Close the message item if it was opened
final_response_text = "".join(final_text_parts) or final_response_text
if message_opened:
await _write_event("response.output_text.done", {
"type": "response.output_text.done",
"item_id": message_item_id,
"output_index": message_output_index,
"content_index": 0,
"text": final_response_text,
"logprobs": [],
})
msg_done_item = {
"id": message_item_id,
"type": "message",
"status": "completed",
"role": "assistant",
"content": [
{"type": "output_text", "text": final_response_text}
],
}
await _write_event("response.output_item.done", {
"type": "response.output_item.done",
"output_index": message_output_index,
"item": msg_done_item,
})
# Always append a final message item in the completed
# response envelope so clients that only parse the terminal
# payload still see the assistant text. This mirrors the
# shape produced by _extract_output_items in the batch path.
final_items: List[Dict[str, Any]] = list(emitted_items)
# Trim large content from tool call arguments to keep the
# response.completed event under ~100KB. Clients already
# received full details via incremental events.
for _item in final_items:
if _item.get("type") == "function_call":
try:
_args = json.loads(_item.get("arguments", "{}")) if isinstance(_item.get("arguments"), str) else _item.get("arguments", {})
if isinstance(_args, dict):
for _k in ("content", "query", "pattern", "old_string", "new_string"):
if isinstance(_args.get(_k), str) and len(_args[_k]) > 500:
_args[_k] = "[" + str(len(_args[_k])) + " chars — truncated for response.completed]"
_item["arguments"] = json.dumps(_args)
except Exception:
pass
elif _item.get("type") == "function_call_output":
_output = _item.get("output", [])
if isinstance(_output, list) and _output:
_first = _output[0]
if isinstance(_first, dict) and _first.get("type") == "input_text":
_text = _first.get("text", "")
if len(_text) > 1000:
_first["text"] = _text[:500] + "...[" + str(len(_text) - 500) + " more chars]"
_item["output"] = [_first]
final_items.append({
"type": "message",
"role": "assistant",
"content": [
{"type": "output_text", "text": final_response_text or (_redact_api_error_text(agent_error) if agent_error else "")}
],
})
if agent_error:
failed_env = _envelope("failed")
failed_env["output"] = final_items
failed_env["error"] = {"message": _redact_api_error_text(agent_error), "type": "server_error"}
failed_env["usage"] = {
"input_tokens": usage.get("input_tokens", 0),
"output_tokens": usage.get("output_tokens", 0),
"total_tokens": usage.get("total_tokens", 0),
}
_failed_history = list(conversation_history)
_failed_history.append({"role": "user", "content": user_message})
if final_response_text or agent_error:
_failed_history.append({
"role": "assistant",
"content": final_response_text or _redact_api_error_text(agent_error),
})
_persist_response_snapshot(
failed_env,
conversation_history_snapshot=_failed_history,
)
terminal_snapshot_persisted = True
await _write_event("response.failed", {
"type": "response.failed",
"response": failed_env,
})
else:
completed_env = _envelope("completed")
completed_env["output"] = final_items
completed_env["usage"] = {
"input_tokens": usage.get("input_tokens", 0),
"output_tokens": usage.get("output_tokens", 0),
"total_tokens": usage.get("total_tokens", 0),
}
full_history = self._build_response_conversation_history(
conversation_history,
user_message,
result,
final_response_text,
)
# Compression-aware transcript substitution happens inside
# _build_response_conversation_history (result["_compressed"]);
# here we only propagate a compression-rotated session_id so
# previous_response_id chaining resumes the child session.
_result_sid = result.get("session_id") if isinstance(result, dict) else None
_persist_response_snapshot(
completed_env,
conversation_history_snapshot=full_history,
session_id_snapshot=_result_sid if isinstance(_result_sid, str) and _result_sid else None,
)
terminal_snapshot_persisted = True
await _write_event("response.completed", {
"type": "response.completed",
"response": completed_env,
})
except (ConnectionResetError, ConnectionAbortedError, BrokenPipeError, OSError):
_persist_incomplete_if_needed()
# Client disconnected — interrupt the agent so it stops
# making upstream LLM calls, then cancel the task.
agent = agent_ref[0] if agent_ref else None
if agent is not None:
try:
request_hard_interrupt(agent, "SSE client disconnected")
except Exception:
pass
_reap_disconnected_agent_processes(agent)
if not agent_task.done():
agent_task.cancel()
try:
await agent_task
except (asyncio.CancelledError, Exception):
pass
logger.info("SSE client disconnected; interrupted agent task %s", response_id)
except asyncio.CancelledError:
# Server-side cancellation (e.g. shutdown, request timeout) —
# persist an incomplete snapshot so GET /v1/responses/{id} and
# previous_response_id chaining still work, then re-raise so the
# runtime's cancellation semantics are respected.
_persist_incomplete_if_needed()
agent = agent_ref[0] if agent_ref else None
if agent is not None:
try:
request_hard_interrupt(agent, "SSE task cancelled")
except Exception:
pass
# Same abandonment as a client disconnect: the run will never
# be resumed, so reap the background processes it created
# (#76115). Epoch-gated; no-op when the turn already
# finished and cleared its markers.
_reap_disconnected_agent_processes(
agent, source="api_server_sse_cancelled"
)
if not agent_task.done():
agent_task.cancel()
logger.info("SSE task cancelled; persisted incomplete snapshot for %s", response_id)
raise
except Exception as _exc:
# Agent crashed with an unhandled error (e.g. model API error like
# BadRequestError, AuthenticationError). Emit a response.failed
# event and properly terminate the SSE stream so the client doesn't
# get a TransferEncodingError from incomplete chunked encoding.
import traceback as _tb
_persist_incomplete_if_needed()
agent_error = _redact_api_error_text(_tb.format_exc())
try:
failed_env = _envelope("failed")
failed_env["output"] = list(emitted_items)
failed_env["error"] = {"message": _redact_api_error_text(_exc, limit=500), "type": "server_error"}
failed_env["usage"] = {
"input_tokens": usage.get("input_tokens", 0),
"output_tokens": usage.get("output_tokens", 0),
"total_tokens": usage.get("total_tokens", 0),
}
await _write_event("response.failed", {
"type": "response.failed",
"response": failed_env,
})
except Exception:
pass
logger.error("Agent crashed mid-stream for %s: %s", response_id, str(agent_error)[:300])
return response
@_admit_api_agent_request
async def _handle_responses(self, request: "web.Request") -> "web.Response":
"""POST /v1/responses — OpenAI Responses API format."""
# Bound total in-flight agent runs (configurable; #7483).
limited = self._concurrency_limited_response()
if limited is not None:
return limited
# Long-term memory scope header (see chat_completions for details).
gateway_session_key, key_err = self._parse_session_key_header(request)
if key_err is not None:
return key_err
# Parse request body
try:
body = await request.json()
except (json.JSONDecodeError, Exception):
return web.json_response(
{"error": {"message": "Invalid JSON in request body", "type": "invalid_request_error"}},
status=400,
)
raw_input = body.get("input")
if raw_input is None:
return web.json_response(_openai_error("Missing 'input' field"), status=400)
instructions = body.get("instructions")
previous_response_id = body.get("previous_response_id")
conversation = body.get("conversation")
store = _coerce_request_bool(body.get("store"), default=True)
# conversation and previous_response_id are mutually exclusive
if conversation and previous_response_id:
return web.json_response(_openai_error("Cannot use both 'conversation' and 'previous_response_id'"), status=400)
# Resolve conversation name to latest response_id
if conversation:
previous_response_id = self._response_store.get_conversation(conversation)
# No error if conversation doesn't exist yet — it's a new conversation
# Normalize input to message list
input_messages: List[Dict[str, Any]] = []
if isinstance(raw_input, str):
input_messages = [{"role": "user", "content": raw_input}]
elif isinstance(raw_input, list):
for idx, item in enumerate(raw_input):
if isinstance(item, str):
input_messages.append({"role": "user", "content": item})
elif isinstance(item, dict):
role = item.get("role", "user")
try:
content = _normalize_multimodal_content(item.get("content", ""))
except ValueError as exc:
return _multimodal_validation_error(exc, param=f"input[{idx}].content")
input_messages.append({"role": role, "content": content})
else:
return web.json_response(_openai_error("'input' must be a string or array"), status=400)
# Accept explicit conversation_history from the request body.
# This lets stateless clients supply their own history instead of
# relying on server-side response chaining via previous_response_id.
# Precedence: explicit conversation_history > previous_response_id.
conversation_history: List[Dict[str, Any]] = []
raw_history = body.get("conversation_history")
if raw_history:
if not isinstance(raw_history, list):
return web.json_response(
_openai_error("'conversation_history' must be an array of message objects"),
status=400,
)
for i, entry in enumerate(raw_history):
if not isinstance(entry, dict) or "role" not in entry or "content" not in entry:
return web.json_response(
_openai_error(f"conversation_history[{i}] must have 'role' and 'content' fields"),
status=400,
)
try:
entry_content = _normalize_multimodal_content(entry["content"])
except ValueError as exc:
return _multimodal_validation_error(exc, param=f"conversation_history[{i}].content")
conversation_history.append({"role": str(entry["role"]), "content": entry_content})
if previous_response_id:
logger.debug("Both conversation_history and previous_response_id provided; using conversation_history")
stored_session_id = None
if not conversation_history and previous_response_id:
stored = self._response_store.get(previous_response_id)
if stored is None:
return web.json_response(_openai_error(f"Previous response not found: {previous_response_id}"), status=404)
conversation_history = list(stored.get("conversation_history", []))
stored_session_id = stored.get("session_id")
# If no instructions provided, carry forward from previous
if instructions is None:
instructions = stored.get("instructions")
# Append new input messages to history (all but the last become history)
for msg in input_messages[:-1]:
conversation_history.append(msg)
# Last input message is the user_message
user_message: Any = input_messages[-1].get("content", "") if input_messages else ""
if not _content_has_visible_payload(user_message):
return web.json_response(_openai_error("No user message found in input"), status=400)
# Truncation support
if body.get("truncation") == "auto":
conversation_history = _auto_truncate_response_history(conversation_history)
# Reuse session from previous_response_id chain so the dashboard
# groups the entire conversation under one session entry. A client
# that manages its own history has no chain to reuse, so fall back to
# the conversation it declared via ``X-Hermes-Session-Key`` before
# minting a throwaway id — otherwise every reply is a new conversation
# to every affinity surface (#96811).
# The response chain still outranks the declared key. Recording is
# gated on that same precedence: binding a session the chain selected
# would rewrite ITS routing key to this request's header
# (record_gateway_session_peer does SET session_key = ?), stranding the
# original conversation and letting the header key recover it instead.
_declared_selected = not stored_session_id and bool(gateway_session_key)
session_id = (
stored_session_id
or self._declared_conversation_session(gateway_session_key)
or str(uuid.uuid4())
)
stream = _coerce_request_bool(body.get("stream"), default=False)
route = self._resolve_route(body.get("model"))
agent_overrides = _request_agent_overrides(
body,
virtual_model=self._model_name,
allow_bare_model=self._direct_model_requests,
)
selection_error = self._request_route_conflict_error(
session_id=session_id,
gateway_session_key=gateway_session_key,
requested_model=agent_overrides.get("requested_model"),
requested_provider=agent_overrides.get("requested_provider"),
route=route,
)
if selection_error:
return web.json_response(_openai_error(selection_error), status=400)
if stream:
# Streaming branch — emit OpenAI Responses SSE events as the
# agent runs so frontends can render text deltas and tool
# calls in real time. See _write_sse_responses for details.
_stream_q = ThreadSafeAsyncQueue()
def _on_delta(delta):
# None from the agent is a CLI box-close signal, not EOS.
# Forwarding would kill the SSE stream prematurely; the
# SSE writer detects completion via agent_task.done().
# Called from the worker thread running run_conversation —
# put_threadsafe (not put_nowait) is required here.
if delta is not None:
_stream_q.put_threadsafe(delta)
def _on_tool_progress(event_type, name, preview, args, **kwargs):
"""Queue non-start tool progress events if needed in future.
The structured Responses stream uses ``tool_start_callback``
and ``tool_complete_callback`` for exact call-id correlation,
so progress events are currently ignored here.
"""
return
def _on_tool_start(tool_call_id, function_name, function_args):
"""Queue a started tool for live function_call streaming."""
_stream_q.put_threadsafe(("__tool_started__", {
"tool_call_id": tool_call_id,
"name": function_name,
"arguments": function_args or {},
}))
def _on_tool_complete(tool_call_id, function_name, function_args, function_result):
"""Queue a completed tool result for live function_call_output streaming."""
_stream_q.put_threadsafe(("__tool_completed__", {
"tool_call_id": tool_call_id,
"name": function_name,
"arguments": function_args or {},
"result": function_result,
}))
agent_ref = [None]
agent_task = asyncio.ensure_future(self._run_agent(
user_message=user_message,
conversation_history=conversation_history,
ephemeral_system_prompt=instructions,
session_id=session_id,
stream_delta_callback=_on_delta,
tool_progress_callback=_on_tool_progress,
tool_start_callback=_on_tool_start,
tool_complete_callback=_on_tool_complete,
agent_ref=agent_ref,
gateway_session_key=gateway_session_key,
bind_declared_conversation=_declared_selected,
**agent_overrides,
route=route,
))
# Ensure SSE drain loops can terminate without relying on polling
# agent_task.done(), which can race with queue timeout checks.
agent_task.add_done_callback(lambda _fut: _stream_q.put_nowait(None))
response_id = f"resp_{uuid.uuid4().hex[:28]}"
model_name = body.get("model", self._model_name)
created_at = int(time.time())
return await self._write_sse_responses(
request=request,
response_id=response_id,
model=model_name,
created_at=created_at,
stream_q=_stream_q,
agent_task=agent_task,
agent_ref=agent_ref,
conversation_history=conversation_history,
user_message=user_message,
instructions=instructions,
conversation=conversation,
store=store,
session_id=session_id,
gateway_session_key=gateway_session_key,
)
async def _compute_response():
return await self._run_agent(
user_message=user_message,
conversation_history=conversation_history,
ephemeral_system_prompt=instructions,
session_id=session_id,
gateway_session_key=gateway_session_key,
bind_declared_conversation=_declared_selected,
**agent_overrides,
route=route,
)
idempotency_key = request.headers.get("Idempotency-Key")
if idempotency_key:
fp = _make_request_fingerprint(
body,
keys=[
"input",
"instructions",
"previous_response_id",
"conversation",
"model",
"provider",
"model_options",
"tools",
],
)
try:
result, usage = await _idem_cache.get_or_set(
idempotency_key, fp, _compute_response
)
except Exception as e:
logger.error("Error running agent for responses: %s", e, exc_info=True)
return web.json_response(
_openai_error(
f"Internal server error: {e}", err_type="server_error"
),
status=500,
)
else:
try:
result, usage = await _compute_response()
except Exception as e:
logger.error("Error running agent for responses: %s", e, exc_info=True)
return web.json_response(
_openai_error(f"Internal server error: {e}", err_type="server_error"),
status=500,
)
final_response = _resolve_media_to_data_urls(result.get("final_response", ""))
if not final_response:
final_response = _redact_api_error_text(result.get("error", "(No response generated)"))
response_id = f"resp_{uuid.uuid4().hex[:28]}"
created_at = int(time.time())
# Build the full conversation history for storage
# (includes tool calls from the agent run)
full_history = self._build_response_conversation_history(
conversation_history,
user_message,
result,
final_response,
)
# Persist the effective session ID surfaced by _run_agent so that
# compression-triggered session rotations propagate to the stored
# response and the X-Hermes-Session-Id header. Without this,
# previous_response_id chaining keeps resuming the pre-rotation
# session and re-triggers compression on every subsequent request.
_effective_session_id = session_id
_result_sid = result.get("session_id") if isinstance(result, dict) else None
if isinstance(_result_sid, str) and _result_sid:
_effective_session_id = _result_sid
# Build output items from the current turn only. AIAgent returns a
# full transcript in result["messages"], while older/mocked paths may
# return only the current turn suffix.
output_start_index = self._response_messages_turn_start_index(
conversation_history,
user_message,
result,
)
output_items = self._extract_output_items(result, start_index=output_start_index)
response_data = {
"id": response_id,
"object": "response",
"status": "completed",
"created_at": created_at,
"model": body.get("model", self._model_name),
"output": output_items,
"usage": {
"input_tokens": usage.get("input_tokens", 0),
"output_tokens": usage.get("output_tokens", 0),
"total_tokens": usage.get("total_tokens", 0),
},
}
# Store the complete response object for future chaining / GET retrieval
if store:
self._response_store.put(response_id, {
"response": response_data,
"conversation_history": full_history,
"instructions": instructions,
"session_id": _effective_session_id,
})
# Update conversation mapping so the next request with the same
# conversation name automatically chains to this response
if conversation:
self._response_store.set_conversation(conversation, response_id)
response_headers = {"X-Hermes-Session-Id": _effective_session_id}
if gateway_session_key:
response_headers["X-Hermes-Session-Key"] = gateway_session_key
return web.json_response(response_data, headers=response_headers)
# ------------------------------------------------------------------
# GET / DELETE response endpoints
# ------------------------------------------------------------------
async def _handle_get_response(self, request: "web.Request") -> "web.Response":
"""GET /v1/responses/{response_id} — retrieve a stored response."""
auth_err = self._check_auth(request)
if auth_err:
return auth_err
response_id = request.match_info["response_id"]
stored = self._response_store.get(response_id)
if stored is None:
return web.json_response(_openai_error(f"Response not found: {response_id}"), status=404)
return web.json_response(stored["response"])
async def _handle_delete_response(self, request: "web.Request") -> "web.Response":
"""DELETE /v1/responses/{response_id} — delete a stored response."""
auth_err = self._check_auth(request)
if auth_err:
return auth_err
response_id = request.match_info["response_id"]
deleted = self._response_store.delete(response_id)
if not deleted:
return web.json_response(_openai_error(f"Response not found: {response_id}"), status=404)
return web.json_response({
"id": response_id,
"object": "response",
"deleted": True,
})
# ------------------------------------------------------------------
# Cron jobs API
# ------------------------------------------------------------------
_JOB_ID_RE = __import__("re").compile(r"[a-f0-9]{12}")
# Allowed fields for update — prevents clients injecting arbitrary keys
_UPDATE_ALLOWED_FIELDS = {"name", "schedule", "prompt", "deliver", "skills", "skill", "repeat", "enabled"}
_MAX_NAME_LENGTH = 200
_MAX_PROMPT_LENGTH = 5000
@staticmethod
def _check_jobs_available() -> Optional["web.Response"]:
"""Return error response if cron module isn't available."""
if not _CRON_AVAILABLE:
return web.json_response(
{"error": "Cron module not available"}, status=501,
)
return None
def _check_job_id(self, request: "web.Request") -> tuple:
"""Validate and extract job_id. Returns (job_id, error_response)."""
job_id = request.match_info["job_id"]
if not self._JOB_ID_RE.fullmatch(job_id):
logger.warning(
"Cron jobs API rejected invalid job_id %r: %s",
job_id,
self._request_audit_log_suffix(request),
)
return job_id, web.json_response(
{"error": "Invalid job ID format"}, status=400,
)
return job_id, None
async def _handle_list_jobs(self, request: "web.Request") -> "web.Response":
"""GET /api/jobs — list all cron jobs."""
auth_err = self._check_auth(request)
if auth_err:
return auth_err
cron_err = self._check_jobs_available()
if cron_err:
return cron_err
try:
include_disabled = request.query.get("include_disabled", "").lower() in {"true", "1"}
jobs = _cron_list(include_disabled=include_disabled)
return web.json_response({"jobs": jobs})
except Exception as e:
return web.json_response({"error": _redact_api_error_text(e)}, status=500)
async def _handle_create_job(self, request: "web.Request") -> "web.Response":
"""POST /api/jobs — create a new cron job."""
auth_err = self._check_auth(request)
if auth_err:
return auth_err
cron_err = self._check_jobs_available()
if cron_err:
return cron_err
try:
body = await request.json()
name = (body.get("name") or "").strip()
schedule = (body.get("schedule") or "").strip()
prompt = body.get("prompt", "")
deliver = body.get("deliver", "local")
skills = body.get("skills")
repeat = body.get("repeat")
if not name:
return web.json_response({"error": "Name is required"}, status=400)
if len(name) > self._MAX_NAME_LENGTH:
return web.json_response(
{"error": f"Name must be ≤ {self._MAX_NAME_LENGTH} characters"}, status=400,
)
if not schedule:
return web.json_response({"error": "Schedule is required"}, status=400)
if len(prompt) > self._MAX_PROMPT_LENGTH:
return web.json_response(
{"error": f"Prompt must be ≤ {self._MAX_PROMPT_LENGTH} characters"}, status=400,
)
if prompt and _scan_cron_prompt is not None:
scan_error = _scan_cron_prompt(prompt)
if scan_error:
return web.json_response({"error": scan_error}, status=400)
if repeat is not None and (not isinstance(repeat, int) or repeat < 1):
return web.json_response({"error": "Repeat must be a positive integer"}, status=400)
kwargs = {
"prompt": prompt,
"schedule": schedule,
"name": name,
"deliver": deliver,
"origin": self._cron_origin_from_request(request),
}
if skills:
kwargs["skills"] = skills
if repeat is not None:
kwargs["repeat"] = repeat
job = _cron_create(**kwargs)
return web.json_response({"job": job})
except _CronSchedulerRegistrationError as e:
return web.json_response(e.to_dict(), status=424)
except Exception as e:
return web.json_response({"error": _redact_api_error_text(e)}, status=500)
async def _handle_get_job(self, request: "web.Request") -> "web.Response":
"""GET /api/jobs/{job_id} — get a single cron job."""
auth_err = self._check_auth(request)
if auth_err:
return auth_err
cron_err = self._check_jobs_available()
if cron_err:
return cron_err
job_id, id_err = self._check_job_id(request)
if id_err:
return id_err
try:
job = _cron_get(job_id)
if not job:
return web.json_response({"error": "Job not found"}, status=404)
return web.json_response({"job": job})
except Exception as e:
return web.json_response({"error": _redact_api_error_text(e)}, status=500)
async def _handle_update_job(self, request: "web.Request") -> "web.Response":
"""PATCH /api/jobs/{job_id} — update a cron job."""
auth_err = self._check_auth(request)
if auth_err:
return auth_err
cron_err = self._check_jobs_available()
if cron_err:
return cron_err
job_id, id_err = self._check_job_id(request)
if id_err:
return id_err
try:
body = await request.json()
# Whitelist allowed fields to prevent arbitrary key injection
sanitized = {k: v for k, v in body.items() if k in self._UPDATE_ALLOWED_FIELDS}
if not sanitized:
return web.json_response({"error": "No valid fields to update"}, status=400)
# Validate lengths if present
if "name" in sanitized and len(sanitized["name"]) > self._MAX_NAME_LENGTH:
return web.json_response(
{"error": f"Name must be ≤ {self._MAX_NAME_LENGTH} characters"}, status=400,
)
if "prompt" in sanitized and len(sanitized["prompt"]) > self._MAX_PROMPT_LENGTH:
return web.json_response(
{"error": f"Prompt must be ≤ {self._MAX_PROMPT_LENGTH} characters"}, status=400,
)
if sanitized.get("prompt") and _scan_cron_prompt is not None:
scan_error = _scan_cron_prompt(sanitized["prompt"])
if scan_error:
return web.json_response({"error": scan_error}, status=400)
job = _cron_update(job_id, sanitized)
if not job:
return web.json_response({"error": "Job not found"}, status=404)
_notify_cron_provider_jobs_changed()
return web.json_response({"job": job})
except Exception as e:
return web.json_response({"error": _redact_api_error_text(e)}, status=500)
async def _handle_delete_job(self, request: "web.Request") -> "web.Response":
"""DELETE /api/jobs/{job_id} — delete a cron job."""
auth_err = self._check_auth(request)
if auth_err:
return auth_err
cron_err = self._check_jobs_available()
if cron_err:
return cron_err
job_id, id_err = self._check_job_id(request)
if id_err:
return id_err
try:
success = _cron_remove(job_id)
if not success:
return web.json_response({"error": "Job not found"}, status=404)
_notify_cron_provider_jobs_changed()
return web.json_response({"ok": True})
except Exception as e:
return web.json_response({"error": _redact_api_error_text(e)}, status=500)
async def _handle_pause_job(self, request: "web.Request") -> "web.Response":
"""POST /api/jobs/{job_id}/pause — pause a cron job."""
auth_err = self._check_auth(request)
if auth_err:
return auth_err
cron_err = self._check_jobs_available()
if cron_err:
return cron_err
job_id, id_err = self._check_job_id(request)
if id_err:
return id_err
try:
job = _cron_pause(job_id)
if not job:
return web.json_response({"error": "Job not found"}, status=404)
_notify_cron_provider_jobs_changed()
return web.json_response({"job": job})
except Exception as e:
return web.json_response({"error": _redact_api_error_text(e)}, status=500)
async def _handle_resume_job(self, request: "web.Request") -> "web.Response":
"""POST /api/jobs/{job_id}/resume — resume a paused cron job."""
auth_err = self._check_auth(request)
if auth_err:
return auth_err
cron_err = self._check_jobs_available()
if cron_err:
return cron_err
job_id, id_err = self._check_job_id(request)
if id_err:
return id_err
try:
job = _cron_resume(job_id)
if not job:
return web.json_response({"error": "Job not found"}, status=404)
_notify_cron_provider_jobs_changed()
return web.json_response({"job": job})
except Exception as e:
return web.json_response({"error": _redact_api_error_text(e)}, status=500)
async def _handle_run_job(self, request: "web.Request") -> "web.Response":
"""POST /api/jobs/{job_id}/run — trigger immediate execution."""
auth_err = self._check_auth(request)
if auth_err:
return auth_err
draining = self._draining_response()
if draining is not None:
return draining
cron_err = self._check_jobs_available()
if cron_err:
return cron_err
job_id, id_err = self._check_job_id(request)
if id_err:
return id_err
# Optional transient per-run context forwarded from a standalone
# `hermes cron run` / cronjob(action='run', prompt=...) — same length
# cap and strict injection scan as a stored job prompt.
extra_prompt = None
try:
body = await request.json()
except Exception:
body = None
if isinstance(body, dict):
raw_prompt = body.get("prompt")
if raw_prompt is not None:
extra_prompt = str(raw_prompt)
if len(extra_prompt) > self._MAX_PROMPT_LENGTH:
return web.json_response(
{"error": f"Prompt must be ≤ {self._MAX_PROMPT_LENGTH} characters"},
status=400,
)
if extra_prompt and _scan_cron_prompt is not None:
scan_error = _scan_cron_prompt(extra_prompt)
if scan_error:
return web.json_response({"error": scan_error}, status=400)
extra_prompt = extra_prompt or None
try:
job = _cron_trigger(job_id, extra_prompt=extra_prompt)
if not job:
return web.json_response({"error": "Job not found"}, status=404)
return web.json_response({"job": job})
except Exception as e:
return web.json_response({"error": _redact_api_error_text(e)}, status=500)
async def _handle_cron_fire(self, request: "web.Request") -> "web.Response":
"""POST /api/cron/fire — Chronos managed-cron fire webhook (NAS → agent).
Authenticated by a NAS-minted JWT (verified via the pluggable
fire-verifier), NOT API_SERVER_KEY — NAS holds no API server key, and
this is the only inbound that can trigger remote job execution, so it
gets its own purpose-scoped token check.
Returns 202 + runs the job in the background so a long agent turn never
trips NAS's HTTP timeout. The store CAS claim inside fire_due guards
against double-fire on a NAS/scheduler retry.
"""
from hermes_cli.config import cfg_get, load_config
from plugins.cron_providers.chronos.verify import get_fire_verifier
auth = request.headers.get("Authorization", "")
token = auth[7:].strip() if auth.startswith("Bearer ") else ""
cfg = load_config()
verifier = get_fire_verifier()
verify_kwargs = dict(
token=token,
expected_audience=cfg_get(cfg, "cron", "chronos", "expected_audience", default=""),
jwks_or_key=cfg_get(cfg, "cron", "chronos", "nas_jwks_url", default="") or None,
issuer=cfg_get(cfg, "cron", "chronos", "portal_url", default="") or None,
)
try:
if asyncio.iscoroutinefunction(verifier):
claims = await verifier(**verify_kwargs)
else:
# The verifier resolves the NAS signing key from a JWKS URL,
# which is a synchronous HTTP GET on a cache miss (cold client
# or a rotated kid) — keep that blocking I/O off the event loop
# so a slow or rate-limited portal can't stall every other
# adapter sharing this loop. Same hardening the platform HTTP
# event verifier already got.
claims = await asyncio.to_thread(verifier, **verify_kwargs)
except Exception:
# Fail closed: a crashing verifier must never admit a fire — this
# is the only inbound that can trigger remote job execution.
logger.exception("cron fire: verifier crashed; rejecting token")
claims = None
if claims is None:
logger.warning(
"cron fire: rejected invalid token: %s",
self._request_audit_log_suffix(request),
)
return web.json_response({"error": "invalid fire token"}, status=401)
draining = self._draining_response()
if draining is not None:
return draining
with _reserve_pending_api_work(self) as reservation:
try:
body = await request.json()
except Exception:
body = {}
job_id = (body or {}).get("job_id")
if not job_id:
return web.json_response({"error": "missing job_id"}, status=400)
from cron.scheduler_provider import (
provider_supports_split_fire,
resolve_cron_scheduler,
)
provider = resolve_cron_scheduler()
loop = asyncio.get_running_loop()
# Live adapters for delivery parity with the built-in ticker
# (gateway/run.py passes runner.adapters to the in-process
# scheduler). Without them, _deliver_result cannot resolve a live
# transport, so E2EE platforms and relay-fronted logical platforms
# (whose only send path IS the live relay adapter — no native
# credential exists) fail with "platform 'X' not
# configured/enabled" on every external-provider fire even though
# the same job delivers fine under the built-in ticker.
runner = self.gateway_runner or request.app.get("gateway_runner")
if runner is None:
try:
from gateway.run import _gateway_runner_ref
runner = _gateway_runner_ref()
except Exception:
runner = None
adapters = getattr(runner, "adapters", None) or None
if not provider_supports_split_fire(provider):
# Legacy single-phase provider: it overrides the documented
# ``fire_due`` hook (custom claim/re-arm/telemetry) but
# inherits the base ``claim_fire`` — driving it through the
# split claim path would silently bypass that override.
task = asyncio.create_task(
asyncio.to_thread(
provider.fire_due,
job_id,
adapters=adapters,
loop=loop,
)
)
reservation["detached"] = True
task.add_done_callback(
lambda _task: _release_pending_api_work(self, reservation)
)
try:
self._background_tasks.add(task)
task.add_done_callback(self._background_tasks.discard)
except (TypeError, AttributeError):
pass
return web.json_response(
{"status": "accepted", "job_id": job_id}, status=202
)
# Persist the attempt and exact store owner before acknowledging NAS.
# A failure here is retryable and the reservation remains attached.
try:
claimed_job = await asyncio.to_thread(provider.claim_fire, job_id)
except Exception as exc:
logger.error("cron fire admission failed for %s: %s", job_id, exc)
return web.json_response(
{"error": "cron fire admission failed", "job_id": job_id},
status=503,
)
if claimed_job is None:
return web.json_response(
{"status": "duplicate", "job_id": job_id},
status=200,
)
task = asyncio.create_task(
asyncio.to_thread(
provider.fire_claimed,
claimed_job,
adapters=adapters,
loop=loop,
)
)
reservation["detached"] = True
task.add_done_callback(
lambda _task: _release_pending_api_work(self, reservation)
)
try:
self._background_tasks.add(task)
task.add_done_callback(self._background_tasks.discard)
except (TypeError, AttributeError):
pass
return web.json_response({"status": "accepted", "job_id": job_id}, status=202)
# ------------------------------------------------------------------
# Output extraction helper
# ------------------------------------------------------------------
@staticmethod
def _build_response_conversation_history(
conversation_history: List[Dict[str, Any]],
user_message: Any,
result: Dict[str, Any],
final_response: Any,
) -> List[Dict[str, Any]]:
"""Build the stored Responses transcript without duplicating history.
When context compression occurs during a turn the agent returns a
compressed full transcript in ``result["messages"]`` (starting with a
summary) and sets ``result["_compressed"] = True``. Because the
compressed transcript does not share the input ``conversation_history``
prefix, the normal turn-start detection fails and old code would
concatenate the uncompressed history on front, bloating the stored
context and re-triggering compression on every subsequent request.
"""
prior = list(conversation_history)
current_user = {"role": "user", "content": user_message}
agent_messages = result.get("messages") if isinstance(result, dict) else None
if isinstance(agent_messages, list) and agent_messages:
turn_start = APIServerAdapter._response_messages_turn_start_index(
conversation_history,
user_message,
result,
)
if turn_start:
return list(agent_messages)
# turn_start == 0: agent_messages does not start with prior.
# This can happen because compression rewrote the transcript
# (summary prefix replaces original history), OR because
# agent_messages only carries the current turn without prior.
# The ``_compressed`` flag (set by _run_agent after compaction)
# distinguishes — skip the concatenation and use the compressed
# transcript directly.
if result.get("_compressed"):
return list(agent_messages)
full_history = prior
full_history.append(current_user)
full_history.extend(agent_messages)
return full_history
full_history = prior
full_history.append(current_user)
full_history.append({"role": "assistant", "content": final_response})
return full_history
@staticmethod
def _response_messages_turn_start_index(
conversation_history: List[Dict[str, Any]],
user_message: Any,
result: Dict[str, Any],
) -> int:
"""Detect transcript-shaped result["messages"] and return turn start."""
agent_messages = result.get("messages") if isinstance(result, dict) else None
if not isinstance(agent_messages, list) or not agent_messages:
return 0
prior = list(conversation_history)
current_user = {"role": "user", "content": user_message}
expected_prefix = prior + [current_user]
if agent_messages[:len(expected_prefix)] == expected_prefix:
return len(expected_prefix)
if prior and agent_messages[:len(prior)] == prior:
return len(prior)
return 0
@classmethod
def _turn_transcript_messages(
cls,
conversation_history: List[Dict[str, Any]],
user_message: Any,
result: Dict[str, Any],
) -> List[Dict[str, Any]]:
"""Return this turn's assistant/tool messages in client-safe shape.
The streaming SSE contract delivers all assistant text as
``assistant.delta`` events under one ``message_id`` interleaved with
``tool.*`` events, and a single ``assistant.completed`` carrying only
the final reply. A client that accumulates deltas into one buffer
cannot reconstruct *intermediate* assistant text segments that preceded
tool calls — so when the page is re-opened mid/post-stream those
segments appear lost, even though state.db persisted them correctly.
Emitting the authoritative per-turn transcript on ``run.completed`` lets
any SSE consumer reconcile its live view against ground truth without a
separate ``GET /messages`` round-trip. Purely additive: clients that
ignore the field are unaffected. Refs #34703.
"""
agent_messages = result.get("messages") if isinstance(result, dict) else None
if not isinstance(agent_messages, list) or not agent_messages:
return []
start = cls._response_messages_turn_start_index(
conversation_history, user_message, result
)
turn = agent_messages[start:]
out: List[Dict[str, Any]] = []
for msg in turn:
if not isinstance(msg, dict):
continue
if msg.get("role") not in {"assistant", "tool"}:
continue
# _message_response projects compaction scaffolding itself and
# marks pure handoffs display_kind == "hidden"; classifying here
# first would re-run the content classifier (a full content
# flatten + prefix scan) a second time per message.
projected = cls._message_response(msg)
if projected.get("display_kind") == "hidden":
continue
out.append(projected)
return out
@staticmethod
def _extract_output_items(result: Dict[str, Any], start_index: int = 0) -> List[Dict[str, Any]]:
"""
Build the output item array from the agent's messages.
Walks *result["messages"]* starting at *start_index* and emits:
- ``function_call`` items for each tool_call on assistant messages
- ``function_call_output`` items for each tool-role message
- a final ``message`` item with the assistant's text reply
"""
items: List[Dict[str, Any]] = []
messages = result.get("messages", [])
if start_index > 0:
messages = messages[start_index:]
for msg in messages:
role = msg.get("role")
if role == "assistant" and msg.get("tool_calls"):
for tc in msg["tool_calls"]:
func = tc.get("function", {})
items.append({
"id": f"fc_{uuid.uuid4().hex[:24]}",
"type": "function_call",
# These calls were already executed server-side by the
# Hermes agent; they are replayed for structured tool
# UI only. Mark them completed (matching the SSE
# streaming path) so OpenAI clients don't interpret
# them as pending calls the client must execute.
"status": "completed",
"name": func.get("name", ""),
"arguments": func.get("arguments", ""),
"call_id": tc.get("id", ""),
})
elif role == "tool":
items.append({
"id": f"fco_{uuid.uuid4().hex[:24]}",
"type": "function_call_output",
"status": "completed",
"call_id": msg.get("tool_call_id", ""),
"output": msg.get("content", ""),
})
# Final assistant message
final = result.get("final_response", "")
if not final:
final = _redact_api_error_text(result.get("error", "(No response generated)"))
items.append({
"type": "message",
"role": "assistant",
"content": [
{
"type": "output_text",
"text": final,
}
],
})
return items
# ------------------------------------------------------------------
# Agent execution
# ------------------------------------------------------------------
def _concurrency_limited_response(self) -> Optional["web.Response"]:
"""Return a 429 response if the concurrent-run cap is reached, else None.
The cap bounds total in-flight agent activity across every
agent-serving endpoint. Reuse the same adapter-owned work count that
shutdown draining uses, including an admitted request before it reaches
agent/task bookkeeping. Stream queues are transport state and may
disappear while their underlying run remains active, so they must not
define run concurrency. A configured value of 0 disables the cap.
"""
limit = self._max_concurrent_runs
if limit <= 0:
return None
inflight = self.active_agent_work_count()
# The current request owns one reservation until it hands off to
# _run_agent() or /v1/runs task registration. It must not consume its
# own last available slot; other admitted requests remain counted.
reservation = _api_agent_request_reservation.get()
if reservation and reservation["active"]:
inflight -= 1
if inflight >= limit:
return web.json_response(
_openai_error(
f"Too many concurrent runs (max {limit})",
err_type="rate_limit_error",
code="rate_limit_exceeded",
),
status=429,
headers={"Retry-After": "1"},
)
return None
@staticmethod
def _bind_api_server_session(
*,
chat_id: str = "",
session_key: str = "",
session_id: str = "",
browser_control_principal: str = "",
browser_control_transport_family: str = "",
) -> list:
"""Bind session contextvars for an API-server agent run.
This is the SINGLE structural chokepoint every API-server agent-entry
path must use to seed session context — it hardwires
``platform="api_server"`` and ``async_delivery=False`` so a new route
physically cannot reintroduce the silent-no-op bug (#10760) by
forgetting to mark the channel as non-delivering. There is no
``async_delivery`` parameter to get wrong; the stateless HTTP path can
never wake the agent after the turn ends, on ANY route.
Returns reset tokens; pass them to ``clear_session_vars`` in a
``finally`` block (the binding is request-scoped and must not outlive
the turn — a session resumed later on a delivering interface, e.g. the
CLI or a gateway platform, re-binds fresh and is NOT blocked).
"""
from gateway.session_context import set_session_vars
return set_session_vars(
platform="api_server",
chat_id=chat_id,
session_key=session_key,
session_id=session_id,
browser_control_principal=browser_control_principal,
browser_control_transport_family=browser_control_transport_family,
async_delivery=False,
cron_session="",
)
async def _run_agent(
self,
user_message: str,
conversation_history: List[Dict[str, str]],
ephemeral_system_prompt: Optional[str] = None,
session_id: Optional[str] = None,
stream_delta_callback=None,
tool_progress_callback=None,
tool_start_callback=None,
tool_complete_callback=None,
agent_ref: Optional[list] = None,
active_run_id: Optional[str] = None,
gateway_session_key: Optional[str] = None,
requested_model: Optional[str] = None,
requested_provider: Optional[str] = None,
model_options: Optional[Dict[str, Any]] = None,
route: Optional[Dict[str, Any]] = None,
session_model: Optional[str] = None,
requested_runtime: Optional[Dict[str, Any]] = None,
route_source: str = "global",
confirmed_runtime_lock: bool = False,
bind_declared_conversation: bool = False,
) -> tuple:
"""
Create an agent and run a conversation in a thread executor.
Returns ``(result_dict, usage_dict)`` where *usage_dict* contains
``input_tokens``, ``output_tokens`` and ``total_tokens``.
*route* is an optional ``model_routes`` entry (resolved from the
request's ``model`` field) that overrides the global model/provider
for this specific request.
*session_model* is a raw model persisted on a native API session
row. It is used only when the persisted value did not resolve to a
``model_routes`` alias — see ``_create_agent`` for precedence.
*requested_runtime* / *route_source* / *confirmed_runtime_lock*
carry the Browser model-lock contract: when a confirmed lock is
active the completed agent's actual provider/model must match the
locked selection or the turn fails, and the response carries
sanitized ``runtime`` metadata reporting actual vs requested.
If *agent_ref* is a one-element list, the AIAgent instance is stored
at ``agent_ref[0]`` before ``run_conversation`` begins. This allows
callers (e.g. the SSE writer) to call ``agent.interrupt()`` from
another thread to stop in-progress LLM calls.
If *active_run_id* is supplied, the same live agent is registered in
``_active_run_agents`` while the turn is running so API clients can
call run-scoped control endpoints such as ``/v1/runs/{run_id}/steer``.
"""
loop = asyncio.get_running_loop()
# Capture before hopping to the executor — ContextVars do not follow
# run_in_executor threads, so the profile scope must be re-entered
# inside _run() from this explicit value.
request_profile = _api_request_profile.get()
request_browser_control_principal = (
_api_request_browser_control_principal.get()
)
request_browser_control_transport_family = (
_api_request_browser_control_transport_family.get()
)
def _run():
from gateway.session_context import clear_session_vars
with self._profile_scope(request_profile):
tokens = self._bind_api_server_session(
chat_id=session_id or "",
session_key=gateway_session_key or session_id or "",
session_id=session_id or "",
browser_control_principal=request_browser_control_principal,
browser_control_transport_family=(
request_browser_control_transport_family
),
)
agent = None
try:
agent = self._create_agent(
ephemeral_system_prompt=ephemeral_system_prompt,
session_id=session_id,
stream_delta_callback=stream_delta_callback,
tool_progress_callback=tool_progress_callback,
tool_start_callback=tool_start_callback,
tool_complete_callback=tool_complete_callback,
gateway_session_key=gateway_session_key,
requested_model=requested_model,
requested_provider=requested_provider,
model_options=model_options,
route=route,
session_model=session_model,
confirmed_runtime_lock=confirmed_runtime_lock,
)
if agent_ref is not None:
agent_ref[0] = agent
if active_run_id:
self._active_run_agents[active_run_id] = agent
effective_task_id = session_id or str(uuid.uuid4())
# Baseline for selective background-process reaping on
# SSE client disconnect — mirrors gateway/run.py's
# gateway-turn cleanup (#76115); this API-server surface
# runs its own agent lifecycle and doesn't go through
# TurnRunner, so it needs its own baseline.
_publish_turn_process_ownership(agent, effective_task_id)
# Shutdown interrupt coverage (#63529). Registering here,
# once, covers every _run_agent() caller — the same reason
# the _ProviderAuthResolutionError handler below lives here
# rather than in each route. Only two callers pass
# ``agent_ref``, and only /v1/runs has a run_id, so neither
# is a usable hook for the rest.
self._shutdown_interruptible_agents[id(agent)] = agent
result = agent.run_conversation(
user_message=user_message,
conversation_history=conversation_history,
task_id=effective_task_id,
)
usage = {
"input_tokens": getattr(agent, "session_prompt_tokens", 0) or 0,
"output_tokens": getattr(agent, "session_completion_tokens", 0) or 0,
"total_tokens": getattr(agent, "session_total_tokens", 0) or 0,
}
# Include the effective session ID in the result so callers
# (e.g. X-Hermes-Session-Id header) can track compression-
# triggered session rotations. (#16938)
_eff_sid = getattr(agent, "session_id", session_id)
if isinstance(_eff_sid, str) and _eff_sid:
result["session_id"] = _eff_sid
# Signal whether context compression occurred during this turn
# so _build_response_conversation_history can skip the
# prior-concatenation path and store the compressed transcript
# directly. Rotation mode changes agent.session_id; in-place
# mode sets _last_compaction_in_place (see #38763).
_compacted_in_place = bool(getattr(agent, "_last_compaction_in_place", False))
_session_rotated = (
isinstance(_eff_sid, str) and isinstance(session_id, str)
and _eff_sid != session_id
)
if _compacted_in_place or _session_rotated:
result["_compressed"] = True
include_runtime = bool(
requested_runtime
or route
or confirmed_runtime_lock
or (route_source and route_source != "global")
)
if include_runtime:
runtime = dict(getattr(agent, "_hermes_api_runtime", {}) or {})
raw_provider = getattr(agent, "provider", "")
raw_model = getattr(agent, "model", "")
actual_provider = (
self._clean_runtime_id(raw_provider, max_len=80)
if isinstance(raw_provider, str)
else ""
)
actual_model = (
self._clean_runtime_id(raw_model)
if isinstance(raw_model, str)
else ""
)
if actual_provider:
runtime["provider"] = actual_provider
else:
runtime.setdefault("provider", "")
if actual_model:
runtime["model"] = actual_model
else:
runtime.setdefault("model", "")
if confirmed_runtime_lock:
expected_provider = self._clean_runtime_id(
(route or {}).get("provider")
or (requested_runtime or {}).get("provider"),
max_len=80,
)
expected_model = self._clean_runtime_id(
(route or {}).get("model")
or (requested_runtime or {}).get("model")
)
mismatched = (
(expected_provider and actual_provider != expected_provider)
or (expected_model and actual_model != expected_model)
)
if mismatched:
raise RuntimeError(
"confirmed model lock runtime mismatch: "
f"expected provider={expected_provider or '<unspecified>'} "
f"model={expected_model or '<unspecified>'}; "
f"actual provider={actual_provider or '<unknown>'} "
f"model={actual_model or '<unknown>'}"
)
if requested_runtime:
runtime["requested"] = {
"provider": self._clean_runtime_id((requested_runtime or {}).get("provider"), max_len=80),
"model": self._clean_runtime_id((requested_runtime or {}).get("model")),
}
runtime["route_source"] = route_source or runtime.get("route_source") or "global"
runtime = self._sanitize_runtime_metadata(
runtime=runtime,
requested_runtime=requested_runtime,
route_source=route_source or "global",
model_lock=("confirmed" if confirmed_runtime_lock else ""),
)
if isinstance(result, dict):
result["runtime"] = runtime
usage["runtime"] = runtime
return result, usage
except _ProviderAuthResolutionError as exc:
# Only _ProviderAuthResolutionError — raised exclusively
# where _resolve_runtime_agent_kwargs() is called inside
# _create_agent() — means a provider auth/credential
# failure. Catching bare RuntimeError here would
# mislabel unrelated RuntimeErrors from
# run_conversation() (e.g. "Failed to recreate closed
# OpenAI client") as auth failures. Matches run.py's
# response shape (final_response text, no HTTP error).
# Previously this propagated unhandled:
# /v1/chat/completions caught it as an undifferentiated
# "Internal server error" 500, and
# /api/sessions/{id}/chat[/stream] didn't catch it at
# all (raw aiohttp 500, no JSON body). Handling it
# here, once, covers every _run_agent() caller;
# /v1/runs has its own branch in its executor.
logger.warning("Provider authentication failed for session=%s: %s",
session_id or "", exc)
return (
{
"final_response": f"⚠️ Provider authentication failed: {exc}",
"messages": [],
"api_calls": 0,
"tools": [],
},
{"input_tokens": 0, "output_tokens": 0, "total_tokens": 0},
)
finally:
# Turn finished (success, auth failure, or crash) — clear
# ownership markers so a disconnect landing after this
# point can't reap background work this turn left
# running on purpose. Mirrors the same race-window guard
# in gateway/run.py's _run_sync_with_timeout_lifecycle.
if active_run_id:
self._active_run_agents.pop(active_run_id, None)
if agent is not None:
_clear_turn_process_ownership(agent)
# Symmetric with the registration above: the turn is
# over, so it must not be interrupted by a later
# shutdown. pop() is a no-op when _create_agent
# succeeded but the turn never reached registration.
self._shutdown_interruptible_agents.pop(id(agent), None)
# Record the declared conversation on the row the turn
# actually ended on — ``agent.session_id`` already
# carries a mid-turn compression rotation (#16938), so
# the next reply resolves the live transcript rather
# than its retired parent. Opt-in: only the routes that
# resolve their session id from the declared key
# (/v1/responses, /v1/runs) record one, so no other
# caller's rows change shape.
if bind_declared_conversation:
self._bind_declared_conversation(
getattr(agent, "session_id", None) or session_id,
gateway_session_key,
)
clear_session_vars(tokens)
self._activate_admitted_request()
self._inflight_agent_runs += 1
try:
return await loop.run_in_executor(None, _run)
finally:
self._inflight_agent_runs -= 1
# ------------------------------------------------------------------
# /v1/runs — structured event streaming
# ------------------------------------------------------------------
_RUN_STREAM_TTL = 300 # seconds before orphaned runs are swept
_RUN_STATUS_TTL = 3600 # seconds to retain terminal run status for polling
def _set_run_status(self, run_id: str, status: str, **fields: Any) -> Dict[str, Any]:
return _api_runs._set_run_status(self, run_id, status, **fields)
def _make_run_event_callback(self, run_id: str, loop: "asyncio.AbstractEventLoop"):
return _api_runs._make_run_event_callback(
self,
run_id,
loop,
_api_server=sys.modules[__name__],
)
def _run_idempotency_scope(self, request: "web.Request") -> str:
return _api_runs._run_idempotency_scope(
self,
request,
_api_server=sys.modules[__name__],
)
@staticmethod
def _room_grant_token(request: "web.Request") -> str:
return _room_grants._room_grant_token(request)
def _room_grant_secret(self) -> bytes:
return _room_grants._room_grant_secret(self)
def _room_grant_claims(
self,
request: "web.Request",
*,
permission: str,
) -> dict[str, Any]:
return _room_grants._room_grant_claims(
self,
request,
permission=permission,
)
def _check_run_auth(
self,
request: "web.Request",
*,
permission: str,
) -> "web.Response | None":
return _api_runs._check_run_auth(
self,
request,
permission=permission,
_api_server=sys.modules[__name__],
)
async def _ensure_hosted_member_session(self, dispatch: Any) -> str:
return await _room_dispatch._ensure_hosted_member_session(self, dispatch)
async def _normalize_room_dispatch(
self,
request: "web.Request",
body: Any,
) -> tuple[Any, "web.Response | None"]:
return await _room_dispatch._normalize_room_dispatch(
self,
request,
body,
_api_server=sys.modules[__name__],
)
async def _handle_room_member_invitation(
self, request: "web.Request"
) -> "web.Response":
return await _room_grants._handle_room_member_invitation(
self,
request,
_openai_error=_openai_error,
_api_request_profile=_api_request_profile,
)
async def _handle_room_member_capabilities(
self, request: "web.Request"
) -> "web.Response":
return await _room_grants._handle_room_member_capabilities(
self,
request,
_openai_error=_openai_error,
_api_request_profile=_api_request_profile,
)
async def _handle_room_member_grant_refresh(
self, request: "web.Request"
) -> "web.Response":
return await _room_grants._handle_room_member_grant_refresh(
self,
request,
_openai_error=_openai_error,
_api_request_profile=_api_request_profile,
)
async def _handle_room_member_grant_revoke(
self, request: "web.Request"
) -> "web.Response":
return await _room_grants._handle_room_member_grant_revoke(
self,
request,
_openai_error=_openai_error,
_api_request_profile=_api_request_profile,
)
def _durable_run_status(
self, request: "web.Request", run_id: str
) -> Dict[str, Any] | None:
return _api_runs._durable_run_status(self, request, run_id)
@_admit_api_agent_request
async def _handle_runs(self, request: "web.Request") -> "web.Response":
"""POST /v1/runs — start an agent run, return run_id immediately."""
return await _api_runs._handle_runs(
self,
request,
_api_server=sys.modules[__name__],
)
def _request_owns_run(self, request: "web.Request", run_id: str) -> bool:
return _api_runs._request_owns_run(self, request, run_id)
def _release_run_owner_if_forgotten(self, run_id: str) -> None:
_api_runs._release_run_owner_if_forgotten(self, run_id)
async def _handle_get_run(self, request: "web.Request") -> "web.Response":
"""GET /v1/runs/{run_id} — return pollable run status for external UIs."""
return await _api_runs._handle_get_run(
self,
request,
_api_server=sys.modules[__name__],
)
async def _handle_run_events(self, request: "web.Request") -> "web.StreamResponse":
"""GET /v1/runs/{run_id}/events — stream structured lifecycle events."""
return await _api_runs._handle_run_events(
self,
request,
_api_server=sys.modules[__name__],
)
async def _handle_run_approval(self, request: "web.Request") -> "web.Response":
"""POST /v1/runs/{run_id}/approval — resolve a pending approval."""
return await _api_runs._handle_run_approval(
self,
request,
_api_server=sys.modules[__name__],
)
async def _handle_steer_run(self, request: "web.Request") -> "web.Response":
"""POST /v1/runs/{run_id}/steer — inject guidance into a running agent."""
return await _api_runs._handle_steer_run(
self,
request,
_api_server=sys.modules[__name__],
)
async def _handle_stop_run(self, request: "web.Request") -> "web.Response":
"""POST /v1/runs/{run_id}/stop — interrupt a running agent."""
return await _api_runs._handle_stop_run(
self,
request,
_api_server=sys.modules[__name__],
)
async def _sweep_orphaned_runs(self) -> None:
return await _api_runs._sweep_orphaned_runs(self)
def _sweep_orphaned_runs_once(self, now: Optional[float] = None) -> None:
return _api_runs._sweep_orphaned_runs_once(self, now)
# ------------------------------------------------------------------
# BasePlatformAdapter interface
# ------------------------------------------------------------------
def _api_key_passes_startup_guard(self) -> bool:
"""Return True when API_SERVER_KEY is present and strong enough to start."""
if not self._api_key:
logger.error(
"[%s] Refusing to start: API_SERVER_KEY is required for the API server, "
"including loopback-only binds on %s.",
self.name, self._host,
)
return False
try:
from hermes_cli.auth import has_usable_secret
except Exception as exc:
# Fail CLOSED. This guard is the only thing between a guessable
# key and a terminal-capable endpoint, so "the check could not be
# run" must not resolve to "start anyway" — the same posture
# tools/credential_files.py takes when its deny-list cannot be
# consulted.
logger.error(
"[%s] Refusing to start: API_SERVER_KEY strength could not be "
"verified (%s: %s), and this endpoint dispatches "
"terminal-capable agent work. Repair the installation before "
"starting the API server on %s.",
self.name, type(exc).__name__, exc, self._host,
)
return False
if not has_usable_secret(self._api_key, min_length=16):
logger.error(
"[%s] Refusing to start: API_SERVER_KEY is a "
"placeholder or too short (<16 chars). This endpoint "
"dispatches terminal-capable agent work — a guessable "
"key is remote code execution. Generate a strong secret "
"(e.g. `openssl rand -hex 32`) and set API_SERVER_KEY "
"before starting the API server on %s.",
self.name, self._host,
)
return False
return True
async def connect(self, *, is_reconnect: bool = False) -> bool:
"""Start the aiohttp web server."""
if not AIOHTTP_AVAILABLE:
logger.warning("[%s] aiohttp not installed", self.name)
return False
with self._session_db_cache_lock:
self._session_db_cache_closed = False
if not self._api_key_passes_startup_guard():
# A rejected API_SERVER_KEY is a configuration error, not a
# transient blip — the key will not become valid on its own. A
# bare ``return False`` makes the reconnect watcher in
# gateway.run treat it as retryable and loop forever at the
# backoff cap, re-instantiating the adapter (and its
# ResponseStore sqlite connection) every retry (#38803: ~501
# leaked connections / 1002 fds over 2.5 days until EMFILE took
# the whole gateway down). Non-retryable drops it from the
# reconnect queue — same treatment as the port-conflict guard
# (api_server_port_in_use). The guard already logged the
# specific rejection reason just above.
self._set_fatal_error(
"api_server_key_invalid",
"API_SERVER_KEY was rejected by the startup guard (missing, "
"placeholder/too short, or strength unverifiable — see the "
"error logged above). Generate a strong secret (e.g. "
"`openssl rand -hex 32`), set API_SERVER_KEY, then "
"`/platform resume api_server`.",
retryable=False,
)
return False
try:
mws = [
mw
for mw in (
self._make_profile_prefix_middleware(),
cors_middleware,
body_limit_middleware,
security_headers_middleware,
)
if mw is not None
]
self._app = web.Application(middlewares=mws, client_max_size=MAX_REQUEST_BYTES)
assert self._app is not None
# Native routes + multiplex /p/<profile>/… mirrors. Same handlers;
# the profile-prefix middleware validates the prefix and scopes
# config/credentials to that profile when multiplexing is on.
for method, path, handler in self._http_route_table():
self._app.router.add_route(method, path, handler)
self._app.router.add_route(method, f"/p/{{profile}}{path}", handler)
# Store the adapter after native routes are registered. Local Hermes-Relay
# bootstrap shims use this key as a feature-detection hook; registering
# native routes first lets those shims no-op instead of shadowing the
# upstream session-control handlers.
self._app["api_server_adapter"] = self
if self.gateway_runner is not None:
self._app["gateway_runner"] = self.gateway_runner
# Start background sweep to clean up orphaned (unconsumed) run streams
sweep_task = asyncio.create_task(self._sweep_orphaned_runs())
try:
self._background_tasks.add(sweep_task)
except TypeError:
pass
if hasattr(sweep_task, "add_done_callback"):
sweep_task.add_done_callback(self._background_tasks.discard)
# Loud warning when a network-accessible API server runs against an
# unsandboxed local terminal backend. The API server can drive the
# agent's terminal/file tools as the host user; on a public bind
# that is the exact surface the hermes-0day campaign abused to write
# ~/.hermes/config.yaml and plant persistence. Sandboxing (Docker /
# remote backend) contains the blast radius. Warn, don't refuse —
# the operator may have an external firewall / strong key.
if is_network_accessible(self._host):
try:
from hermes_cli.config import load_config as _load_cfg
_backend = (
((_load_cfg() or {}).get("terminal") or {}).get(
"backend", "local"
)
)
except Exception:
_backend = "local"
if str(_backend).lower() == "local":
logger.warning(
"[%s] API server is network-accessible (%s) AND the "
"terminal backend is 'local' (unsandboxed). Agent work "
"dispatched through this endpoint runs as the host user "
"with full terminal/file access. Strongly consider a "
"sandboxed backend (terminal.backend: docker) and "
"firewalling this port to trusted networks only.",
self.name, self._host,
)
# Plugin-registered native handlers (aiohttp web.Application —
# router routes). Wired before AppRunner.setup() freezes the router.
self._wire_plugin_handlers(self._app)
self._runner = web.AppRunner(self._app)
await self._runner.setup()
# Bind directly instead of probing 127.0.0.1 first — the old
# single-family pre-probe raced the real bind and reported a
# TIME_WAIT socket as "in use" (#10297), failing gateway
# restarts for up to ~60s.
#
# SO_REUSEADDR is platform-dependent (same rationale as the
# webhook adapter, #65482):
# - macOS (BSD semantics): two sockets with SO_REUSEADDR can
# silently split traffic while both report success — disable.
# - Linux: SO_REUSEADDR only permits rebinding past TIME_WAIT
# (a second live listener needs SO_REUSEPORT, never set), so
# keep the default (enabled) for instant restart rebinds.
self._site = web.TCPSite(
self._runner,
self._host,
self._port,
reuse_address=False if sys.platform == "darwin" else None,
)
try:
await self._site.start()
except OSError as exc:
await self._runner.cleanup()
self._runner = None
self._site = None
if getattr(exc, "errno", None) == errno.EADDRINUSE:
# A port conflict is a configuration error, not a
# transient blip — another process holds the port for
# its lifetime. A bare ``return False`` makes the
# reconnect watcher in gateway.run treat it as retryable
# and loop forever at the backoff cap (observed: 1568+
# retries over 5 days across multi-profile setups all
# defaulting to the same port, #52132), filling
# errors.log and leaking the adapter's ResponseStore
# fds each retry. Non-retryable drops it from the
# reconnect queue; the operator recovers with
# ``/platform resume api_server`` after changing the port.
self._set_fatal_error(
"api_server_port_in_use",
f"Port {self._port} already in use. Set "
f"platforms.api_server.port in config.yaml to a "
f"different value, then `/platform resume api_server`.",
retryable=False,
)
logger.error(
"[%s] Could not bind %s:%d: %s. Set a different port in "
"config.yaml: platforms.api_server.port",
self.name, self._host, self._port, exc,
)
return False
self._mark_connected()
logger.info(
"[%s] API server listening on http://%s:%d (model: %s)",
self.name, self._host, self._port, self._model_name,
)
return True
except Exception as e:
logger.error("[%s] Failed to start API server: %s", self.name, e)
return False
async def disconnect(self) -> None:
"""Stop the aiohttp web server and release all owned resources.
Closes the ResponseStore SQLite connection in addition to stopping
the aiohttp web server. Without this, every adapter instance leaks
2 file descriptors (the database file and its WAL sidecar) — the
reconnect loop in ``gateway.run`` constructs a fresh adapter on
every retry, so 2 fds/retry × 300s backoff cap ≈ 12 fds/hour, which
exhausts the default 2560 fd limit after ~12h of failed reconnects
and turns the whole gateway into a zombie
(OSError: [Errno 24] Too many open files, #37011).
"""
self._mark_disconnected()
if self._response_store is not None:
try:
self._response_store.close()
except Exception:
logger.debug(
"Failed to close response store for %s", self.name, exc_info=True,
)
_api_runs._close_run_state(self)
try:
if self._site:
await self._site.stop()
self._site = None
if self._runner:
await self._runner.cleanup()
self._runner = None
finally:
self._close_cached_session_dbs()
self._app = None
logger.info("[%s] API server stopped", self.name)
async def send(
self,
chat_id: str,
content: str,
reply_to: Optional[str] = None,
metadata: Optional[Dict[str, Any]] = None,
) -> SendResult:
"""
Not used — HTTP request/response cycle handles delivery directly.
"""
return SendResult(success=False, error="API server uses HTTP request/response, not send()")
async def get_chat_info(self, chat_id: str) -> Dict[str, Any]:
"""Return basic info about the API server."""
return {
"name": "API Server",
"type": "api",
"host": self._host,
"port": self._port,
}
Reference in New Issue View Git Blame Copy Permalink