agent/auxiliary_client.py
Rama main · réplica de lectura del repositorio original.
"""Shared auxiliary client router for side tasks (compression, search, vision, ...).
Text auto chain: main provider+model → OpenRouter → Nous Portal → custom endpoint →
native Anthropic → direct API-key providers → None. Vision auto chain: main
provider (if a supported vision backend) → OpenRouter → Nous → Anthropic → custom.
``auxiliary.free_only`` restricts the OpenRouter lane to ``:free`` SKUs. Codex OAuth is
in neither chain (undocumented, shifting allow-list): main provider or explicit
``auxiliary.<task>.provider`` only. HTTP 402 in call_llm() falls through the chain.
"""
import contextlib
import contextvars
import functools
import hashlib
import inspect
import json
import logging
import os
import re
import threading
import time
import uuid
from types import SimpleNamespace
from typing import Any, Callable, Dict, List, NamedTuple, Optional, Tuple, TYPE_CHECKING, Union
from urllib.parse import urlparse, parse_qs, urlunparse
from agent.local_prompt_progress import (
PromptReadingClock, current_prompt_reading_hook, prompt_progress, prompt_reading_hook, with_prompt_progress)
from agent import auxiliary_call_outcome, auxiliary_error_policy as aux_errors
from agent.error_classifier import (
FailoverReason,
UNSUPPORTED_PARAM_MARKERS,
is_reasoning_field_rejection,
is_reasoning_required_rejection,
)
from agent.errors import InvalidProviderResponseError
from agent.auxiliary_fallback_recovery import reapply_repairs
from agent.auxiliary_managed_local import _is_managed_local_endpoint
from agent.auxiliary_reasoning_floor import remember_reasoning_floor, with_reasoning_floor
from agent.auxiliary_profile_gates import (
PROFILE_REASONING_KEYS, contains_profile_reasoning_fields, profile_forwards_max_tokens,
)
from agent.auxiliary_structured_output import _route_key, _without_structured_output_format, remember_structured_output_rejection
from agent.codex_headers import (
CODEX_AUX_BASE_URL as _CODEX_AUX_BASE_URL,
apply_required_codex_headers as _apply_required_codex_headers,
codex_cloudflare_headers as _codex_cloudflare_headers,
is_official_codex_base_url as _is_official_codex_base_url,
)
from agent.codex_runtime import _codex_event_has_content
from agent.sdk_transform_bypass import bypass_chat_sdk_request_transform
# `openai.OpenAI` is imported lazily (~240 ms cold); `OpenAI` below is a proxy
# so in-module calls, `auxiliary_client.OpenAI` reads and
# `patch("agent.auxiliary_client.OpenAI")` all keep working.
if TYPE_CHECKING:
from openai import OpenAI
_OPENAI_CLS_CACHE: Optional[type] = None
def _load_openai_cls() -> type:
"""Import and cache ``openai.OpenAI``."""
global _OPENAI_CLS_CACHE
if _OPENAI_CLS_CACHE is None:
from openai import OpenAI as _cls
_OPENAI_CLS_CACHE = _cls
return _OPENAI_CLS_CACHE
class _OpenAIProxy:
"""Lazy stand-in for ``openai.OpenAI``: forwards calls and isinstance checks, importing on first use."""
__slots__ = ()
def __call__(self, *args, **kwargs):
return _load_openai_cls()(*args, **kwargs)
def __instancecheck__(self, obj):
return isinstance(obj, _load_openai_cls())
def __repr__(self):
return "<lazy openai.OpenAI proxy>"
OpenAI = _OpenAIProxy()
# Availability probe mode: check_fns only need to know whether a client is RESOLVABLE, so
# inside `aux_probe_mode()` constructors return a stub instead of importing openai + building
# httpx/SSL (~0.3s on CLI startup). Stubs are never cached (see _store_cached_client).
_aux_probe_state = threading.local()
class _AuxProbeClientStub:
"""Non-functional placeholder returned while `aux_probe_mode` is active."""
__slots__ = ("_hermes_aux_backend_provider", "_hermes_aux_effective_provider", "api_key", "base_url")
def __init__(self, api_key: str = "", base_url: str = "") -> None:
self.api_key = api_key
self.base_url = base_url
self._hermes_aux_backend_provider = self._hermes_aux_effective_provider = ""
def __getattr__(self, name: str) -> Any:
# Loud failure if a probe stub ever leaks into a runtime call path.
raise RuntimeError(
f"_AuxProbeClientStub used as a real client (attribute {name!r}); "
"aux_probe_mode is for availability checks only")
def __repr__(self) -> str:
return "<aux availability-probe client stub>"
def _aux_probe_active() -> bool:
return bool(getattr(_aux_probe_state, "active", False))
@contextlib.contextmanager
def aux_probe_mode():
"""Resolve provider availability without constructing real SDK clients."""
prev = getattr(_aux_probe_state, "active", False)
_aux_probe_state.active = True
try:
yield
finally:
_aux_probe_state.active = prev
from agent.credential_pool import load_pool
from agent.credential_pool_model_cooldowns import pool_failure_reason
from agent.custom_entry_pool import entry_scoped_pool, pool_for_aux_provider
from agent.message_metadata import data_use_requirement
from agent.model_metadata import MINIMUM_CONTEXT_LENGTH, get_model_context_length
from hermes_cli.config import get_hermes_home
from hermes_cli.config_providers import _canonical_api_mode
from agent.auxiliary_health import (
_custom_health_base_url, _unhealthy_cache_key, fallback_candidate_quarantine_ttl,
)
from agent.auxiliary_unavailable import (
AuxiliaryClientUnavailable, clear_nous_credential_failure, missing_provider_credentials_message,
nous_credential_failure_detail, record_nous_credential_failure)
from hermes_constants import OPENROUTER_BASE_URL, hermes_home_key
from utils import base_url_host_matches, base_url_hostname, base_url_origin, base_url_path, env_float, is_truthy_value, model_forces_max_completion_tokens, normalize_proxy_env_vars
logger = logging.getLogger(__name__)
# resolve_provider_client fall-through dedup: misconfigured-provider warnings fire on every
# retry, so only the first per process surfaces. Separate sets let tests clear each branch.
_LOGGED_UNKNOWN_PROVIDER_KEYS: set = set()
_LOGGED_UNHANDLED_AUTHTYPE_KEYS: set = set()
_LOGGED_UNSUPPORTED_EXTPROC_KEYS: set = set()
_LOGGED_UNSUPPORTED_OAUTH_KEYS: set = set()
def _resolve_aux_verify(base_url: Optional[str]) -> Any:
"""httpx ``verify`` for an aux base_url, mirroring the main client (per-provider ``ssl_ca_cert`` /
``ssl_verify``; otherwise the OS trust store); any failure → httpx default (``True``)."""
try:
from agent.ssl_verify import resolve_httpx_verify
from hermes_cli.config import get_custom_provider_tls_settings, load_config_readonly
tls = get_custom_provider_tls_settings(str(base_url or ""), config=load_config_readonly())
return resolve_httpx_verify(
ca_bundle=tls.get("ssl_ca_cert"), ssl_verify=tls.get("ssl_verify"), base_url=str(base_url or ""))
except Exception:
return True
_WARNED_KEEPALIVE_IMPORT_SKEW = False
def _openai_http_client_kwargs(base_url: Optional[str], *, async_mode: bool = False) -> dict[str, Any]:
"""Inject keepalive httpx client with env-only proxy (not macOS system proxy)."""
try:
from agent.process_bootstrap import build_keepalive_http_client
client = build_keepalive_http_client(
str(base_url or ""), async_mode=async_mode, verify=_resolve_aux_verify(base_url))
except (ImportError, AttributeError):
# Version-skewed install (Desktop runtime lagging a git tree) lacks this helper:
# degrade to the SDK default httpx client rather than kill the job; warn once.
global _WARNED_KEEPALIVE_IMPORT_SKEW
if not _WARNED_KEEPALIVE_IMPORT_SKEW:
_WARNED_KEEPALIVE_IMPORT_SKEW = True
logger.warning(
"agent.process_bootstrap.build_keepalive_http_client is "
"unavailable — mixed/stale install detected (#64333). Falling "
"back to the SDK default HTTP client. Run `hermes update` (or "
"reinstall the Desktop app) to resync the runtime.")
client = None
return {"http_client": client} if client is not None else {}
def _create_openai_client(*, api_key: str, base_url: str, **kwargs: Any) -> Any:
if _aux_probe_active():
# Availability probe: resolved credentials/base_url are the answer.
return _AuxProbeClientStub(api_key=api_key, base_url=base_url)
kwargs = {**_openai_http_client_kwargs(base_url), **kwargs}
_apply_required_codex_headers(kwargs, access_token=api_key, base_url=base_url)
# Hermes owns aux retry/fallback policy; the SDK default (max_retries=2) would triple
# wall time on a hung endpoint before Hermes sees one failure.
# Hermes owns auxiliary retry + provider/model fallback policy (the same-provider transient retry in
# call_llm plus the except-chain fallback). The OpenAI SDK's own default (max_retries=2 → up to 3
# attempts) silently multiplies the effective wall time of every aux call by 3× on a slow/hung endpoint,
# so a 120s timeout can stall ~360s before Hermes sees a single failure (issue #54465). Disable
# SDK-internal retries by default and let Hermes control the budget; explicit callers can still override
# via kwargs.
kwargs.setdefault("max_retries", 0)
return OpenAI(api_key=api_key, base_url=base_url, **kwargs)
# Interrupt protection for atomic aux tasks: a compression summary killed by an ordinary
# gateway interrupt degrades to a static marker, so a thread-local flag marks such calls
# protected. Explicit host cancel (Ctrl+C, /stop) still overrides it, timeouts still fire.
# ── Interrupt protection for atomic auxiliary tasks ────────────────────── Some auxiliary tasks must NOT be
# aborted mid-flight by a gateway interrupt (e.g. an incoming user message while the agent is busy). Context
# compression is the prime case: if the summary LLM call is interrupted part-way, compression falls back to
# a static "summary unavailable" marker and the real handoff is lost (#23975). A thread-local flag lets such
# a task mark its in-flight LLM call as interrupt-protected; the Codex Responses stream's cancellation check
# honors it. TIMEOUTS still fire (a hung call must die), and all OTHER aux tasks (vision, web_extract,
# title_generation, …) remain freely interruptible.
_aux_interrupt_protection = threading.local()
class AuxiliaryExplicitCancellation(BaseException):
"""Frozen signal that an auxiliary attempt was explicitly hard-cancelled. ``BaseException`` so broad
``except Exception`` retry/fallback code never treats a host stop as a transport failure; ``cause``
is immutable class data so nothing re-queries a mutable host Event after the transport unwound."""
cause = "explicit_host_cancel"
def __init__(self) -> None:
super().__init__("auxiliary request explicitly cancelled by host")
def _aux_interrupt_protected() -> bool:
return bool(getattr(_aux_interrupt_protection, "active", False))
def _aux_interrupt_cancel_requested() -> bool:
"""Return whether an explicit host cancel overrides aux protection."""
check = _capture_aux_cancel_check()
return _captured_aux_cancel_requested(check) if check is not None else False
@contextlib.contextmanager
def aux_interrupt_protection(active: bool = True, cancel_check=None, cancel_event=None):
"""Mark this thread's aux LLM call interrupt-protected (re-entrant-safe). ``cancel_check`` /
``cancel_event`` keep an explicit host hard-cancel path (Event preferred); nested scopes inherit both."""
prev = getattr(_aux_interrupt_protection, "active", False)
prev_cancel_check = getattr(_aux_interrupt_protection, "cancel_check", None)
prev_cancel_event = getattr(_aux_interrupt_protection, "cancel_event", None)
_aux_interrupt_protection.active = active
if callable(cancel_check):
_aux_interrupt_protection.cancel_check = cancel_check
if cancel_event is not None and callable(getattr(cancel_event, "is_set", None)):
_aux_interrupt_protection.cancel_event = cancel_event
try:
yield
finally:
_aux_interrupt_protection.active = prev
_aux_interrupt_protection.cancel_check = prev_cancel_check
_aux_interrupt_protection.cancel_event = prev_cancel_event
def _capture_aux_cancel_check() -> Optional[Callable[[], Any]]:
"""Capture the current explicit-cancel source on the owning request thread."""
is_set = getattr(getattr(_aux_interrupt_protection, "cancel_event", None), "is_set", None)
if callable(is_set):
return is_set
# Return the callable itself so attempt-local decision objects keep begin_timeout_cleanup().
check = getattr(_aux_interrupt_protection, "cancel_check", None)
return check if callable(check) else None
def _captured_aux_cancel_requested(cancel_check: Callable[[], Any]) -> bool:
"""Read a request-thread cancellation source without leaking its failures."""
try:
return bool(cancel_check())
except Exception:
logger.debug("captured aux cancel check failed", exc_info=True)
return False
class _AuxiliaryCancellationDecision:
"""Atomically choose explicit cancellation or provider timeout per attempt."""
def __init__(self, source_cancel_check: Callable[[], Any]) -> None:
self._source_cancel_check = source_cancel_check
self._lock = threading.Lock()
self._outcome = "active"
def __call__(self) -> bool:
with self._lock:
if self._outcome == "active" and _captured_aux_cancel_requested(self._source_cancel_check):
self._outcome = "cancelled"
return self._outcome == "cancelled"
def begin_timeout_cleanup(self) -> bool:
"""Return whether timeout won and destructive cleanup is permitted."""
with self._lock:
if self._outcome == "active":
cancelled = _captured_aux_cancel_requested(self._source_cancel_check)
self._outcome = "cancelled" if cancelled else "timed_out"
return self._outcome == "timed_out"
# Forward-progress hooks for streamed aux calls: a fixed host deadline kills a SLOW model
# streaming a big summary as hard as a HUNG one, so wire consumers tick the progress hook only
# for non-empty payloads and the host extends its deadline while tokens move. Thread-local:
# the call and its stream consumption run on the installing thread.
_aux_progress = threading.local()
_aux_dispatch = threading.local()
_aux_provider_response = threading.local()
# Absolute monotonic deadline of the waiting HOST. The stream's own ceiling
# (_aux_stream_total_ceiling, >= the host's and started later) would otherwise leave an
# orphaned stream still billing after every host-ceiling timeout.
# Absolute wall-clock deadline (time.monotonic) of the HOST waiting for this auxiliary call, when it has one
# (#99692). Liveness alone is not enough: a host also stops waiting at its own total ceiling, and the
# streamed consumer below bounds itself only by _aux_stream_total_ceiling() — a budget derived from the aux
# request timeout, which is >= the host ceiling for every configured value AND starts counting later. So the
# stream that outlives its abandoned host is not an edge case; it is the guaranteed outcome of every
# total-ceiling timeout.
_aux_stream_deadline = threading.local()
def _tick_hook(local: threading.local, label: str) -> None:
"""Call the thread-local hook installed on ``local``, if any. Never raises."""
hook = getattr(local, "hook", None)
if hook is None:
return
try:
hook()
except Exception:
logger.debug("aux %s hook failed", label, exc_info=True)
def _notify_aux_progress() -> None:
"""Tick the installed forward-progress hook, if any."""
_tick_hook(_aux_progress, "progress")
def _notify_aux_dispatch() -> None:
"""Record an actual provider dispatch without claiming response progress."""
_tick_hook(_aux_dispatch, "dispatch")
def _notify_aux_timing_response() -> None:
"""Record a content-free frame (keepalive/empty delta): counts toward
``time_to_first_progress_ms`` but must not reset a compression inactivity fence."""
_tick_hook(_aux_provider_response, "provider response")
def _notify_aux_provider_response() -> None:
"""Record a provider response/chunk, then preserve the liveness signal."""
_notify_aux_timing_response()
_notify_aux_progress()
def _aux_progress_active() -> bool:
return getattr(_aux_progress, "hook", None) is not None
def _field(obj: Any, key: str, default: Any = None) -> Any:
"""Field access for wire objects that may be dicts or SDK/SimpleNamespace objects."""
val = obj.get(key) if isinstance(obj, dict) else getattr(obj, key, None)
return default if val is None else val
def _anthropic_event_has_content(event: Any) -> bool:
"""Whether an Anthropic stream event carries a non-empty payload."""
event_type = _field(event, "type")
if event_type == "content_block_delta":
delta = _field(event, "delta")
return any(bool(_field(delta, f)) for f in ("text", "thinking", "partial_json", "signature", "citation"))
if event_type == "content_block_start":
block = _field(event, "content_block")
return _field(block, "type") == "tool_use" and any(bool(_field(block, f)) for f in ("id", "name"))
return False
def _anthropic_aux_stream_event_hook() -> Callable[[Any], None]:
"""Per-event callback for the Anthropic aux wire: progress only for substantive payloads
(keepalives must not keep a stalled summary alive), stop at the host deadline or explicit
cancel. The ``TimeoutError`` text must say "timed out" so the classifier reads it as a timeout."""
host_deadline = _current_aux_stream_deadline()
started = time.monotonic()
def _on_event(event: Any) -> None:
if _anthropic_event_has_content(event):
_notify_aux_provider_response()
else:
_notify_aux_timing_response()
if _aux_interrupt_cancel_requested():
raise AuxiliaryExplicitCancellation()
if host_deadline is not None and time.monotonic() >= host_deadline:
raise TimeoutError(
"Anthropic auxiliary stream timed out at the host compression "
f"deadline after {time.monotonic() - started:.0f}s (the caller already stopped waiting)")
return _on_event
# A dead stream fails at the no-progress window (first token AND between tokens); a live
# stream re-arms per event, bounded by _aux_stream_total_ceiling().
_AUX_STREAM_NO_PROGRESS_TIMEOUT_SECONDS = 60.0
@contextlib.contextmanager
def _aux_thread_local_hook(local: threading.local, hook):
"""Install one thread-local hook, restoring the prior on exit (non-callable = passthrough)."""
previous = getattr(local, "hook", None)
local.hook = hook if callable(hook) else previous
try:
yield
finally:
local.hook = previous
@contextlib.contextmanager
def aux_progress_hook(hook):
"""Install *hook* as the current thread's aux forward-progress callback (None = passthrough)."""
with _aux_thread_local_hook(_aux_progress, hook):
yield
def _current_aux_stream_deadline() -> Optional[float]:
"""The waiting host's absolute monotonic deadline, if one is installed (read live from a callable)."""
value = getattr(_aux_stream_deadline, "value", None)
return value() if callable(value) else value
@contextlib.contextmanager
def aux_stream_deadline(deadline: float | Callable[[], float | None] | None):
"""Publish the host's absolute ``time.monotonic()`` deadline to the stream consumer.
``None`` is a passthrough; re-entrant-safe. Host->worker return leg of the progress hook:
without it the isolated provider daemon streams to its own ceiling after the host stopped
waiting, billing a summary the commit fence refuses.
``8207862212`` releases the compression OWNER when the fence is cancelled, but the isolated provider
daemon (:func:`_run_protected_sync_provider_call`) that holds the socket keeps streaming to its own
``_aux_stream_total_ceiling`` budget — >= the host's ceiling by construction — billing an abandoned
summary the commit fence is already guaranteed to refuse, and stacking one fresh orphan per turn on a
session that compression never managed to shrink. See #99692.
"""
previous = getattr(_aux_stream_deadline, "value", None)
_aux_stream_deadline.value = deadline if isinstance(deadline, (int, float)) or callable(deadline) else previous
try:
yield
finally:
_aux_stream_deadline.value = previous
def _run_protected_sync_provider_call(callback: Callable[[dict[str, Any]], Any], kwargs: dict[str, Any]) -> Any:
"""Run one protected provider callback in an attempt-isolated daemon thread.
Aux clients are process-shared and cannot be closed to wake one request, so the callback (incl.
stream aggregation) runs in a daemon while the owner polls cancellation; on cancel the owner
unwinds at once; the daemon stops at its next streamed chunk, else under the provider timeout in ``kwargs``
(it owns no transcript/commit state, never holds the session lock). Unprotected / no cancel source: direct.
"""
source_cancel_check = _capture_aux_cancel_check()
if not _aux_interrupt_protected() or not callable(source_cancel_check):
return callback(kwargs)
# One linearized outcome per attempt: the host Event is reused/cleared on later turns and
# the Codex timeout Timer may race owner polling — same lock for both.
cancel_check = _AuxiliaryCancellationDecision(source_cancel_check)
if cancel_check():
raise AuxiliaryExplicitCancellation()
# Thread-locals do not cross into the daemon: timing hooks fire from the thread running
# the callback, and the host deadline is inert unless carried along.
progress_hook = getattr(_aux_progress, "hook", None)
dispatch_hook = getattr(_aux_dispatch, "hook", None)
provider_response_hook = getattr(_aux_provider_response, "hook", None)
host_deadline = getattr(_aux_stream_deadline, "value", None)
reading_hook = current_prompt_reading_hook()
# #99692: the stream is consumed on the daemon below, and thread-locals do not cross that boundary — an
# owner-thread-only deadline would leave the fix inert on exactly the path large-session compression
# takes (protected call + hard-cancel source installed).
provider_context = contextvars.copy_context()
done = threading.Event()
outcome: dict[str, Any] = {}
def _provider_worker() -> None:
try:
with (
aux_progress_hook(progress_hook),
_aux_thread_local_hook(_aux_dispatch, dispatch_hook),
_aux_thread_local_hook(_aux_provider_response, provider_response_hook),
aux_stream_deadline(host_deadline), prompt_reading_hook(reading_hook),
aux_interrupt_protection(cancel_check=cancel_check),
):
outcome["result"] = callback(kwargs)
except BaseException as exc:
outcome["exception"] = exc
finally:
done.set()
threading.Thread(
target=provider_context.run, args=(_provider_worker,), name="hermes-protected-aux-provider",
daemon=True).start()
while True:
# Check cancel before AND after each wait so it wins when result publication and the
# host Event land in the same polling interval.
if _captured_aux_cancel_requested(cancel_check):
raise AuxiliaryExplicitCancellation()
if not done.wait(0.02):
continue
if _captured_aux_cancel_requested(cancel_check):
raise AuxiliaryExplicitCancellation()
exception = outcome.get("exception")
if exception is not None:
raise exception
return outcome.get("result")
def _client_declares(client_obj: Any, flag: str) -> bool:
"""Whether ``client_obj`` (or its class) sets ``flag`` truthy; absent → False. Capability declaration,
not isinstance, so out-of-tree clients can opt out of wrappers unimported (cf. SUPPORTS_HERMES_TOOL_CALLS)."""
try:
return bool(getattr(client_obj, flag, False))
except Exception:
return False
def _safe_isinstance(obj: Any, maybe_type: Any) -> bool:
"""Return False instead of raising when a patched symbol is not a type."""
try:
return isinstance(obj, maybe_type)
except TypeError:
return False
def _extract_url_query_params(url: str):
"""Extract query params from URL, return (clean_url, default_query dict or None)."""
parsed = urlparse(url)
if parsed.query:
return urlunparse(parsed._replace(query="")), {k: v[0] for k, v in parse_qs(parsed.query).items()}
return url, None
# Warn only once per process about stale OPENAI_BASE_URL.
_stale_base_url_warned = False
# Local OpenAI-compatible servers (Ollama, vLLM, llama.cpp) route through the generic custom
# provider — mirrors hermes_cli.auth._PROVIDER_ALIASES. Without this group an explicit
# ``provider: ollama`` aux lane matches no registry entry and raises a misleading
# ``OLLAMA_API_KEY`` error instead of using the lane's base_url (#106010).
_LOCAL_SERVER_ALIASES = {
"ollama": "custom", "vllm": "custom", "llamacpp": "custom",
"llama.cpp": "custom", "llama-cpp": "custom",
}
_ALIAS_TABLES: dict[str, tuple[int, dict[str, str]]] = {} # home key -> (registry generation, table)
def _provider_alias_table() -> dict[str, str]:
"""The alias table the main provider path resolves against (hermes_cli.auth), per home and registry
generation: plugin aliases are per home. A hand-copied mirror rots silently (#106010); local-server
names stay in: they guard the /v1 tail and the custom-branch no-key-borrow rule."""
from providers import bound_registry_key
home, generation = bound_registry_key()
cached = _ALIAS_TABLES.get(home)
if cached is None or cached[0] != generation:
cached = (generation, dict(_LOCAL_SERVER_ALIASES))
with contextlib.suppress(Exception):
from hermes_cli.auth import _plugin_aliases
cached[1].update(_plugin_aliases())
_ALIAS_TABLES[home] = cached
return cached[1]
def _normalize_aux_provider(provider: Optional[str]) -> str:
normalized = (provider or "auto").strip().lower()
if normalized.startswith("custom:"):
suffix = normalized.split(":", 1)[1].strip()
if not suffix:
return "custom"
normalized = suffix
if normalized == "codex":
return "openai-codex"
if normalized == "main":
# Resolve to the actual main provider so named custom providers work.
main_prov = (_read_main_provider() or "").strip().lower()
if not main_prov or main_prov in {"auto", "main"}:
return "custom"
normalized = main_prov
return _provider_alias_table().get(normalized, normalized)
# Sentinel from _fixed_temperature_for_model(): callers strip ``temperature`` entirely.
# Kimi/Moonshot manage it server-side — any value can conflict with gateway mode selection.
OMIT_TEMPERATURE: object = object()
def _bare_model(model: Optional[str]) -> str:
"""Lowercased model slug with any ``vendor/`` prefix stripped."""
return (model or "").strip().lower().rsplit("/", 1)[-1]
def _is_kimi_model(model: Optional[str]) -> bool:
"""True for any Kimi / Moonshot model that manages temperature server-side."""
bare = _bare_model(model)
return bare.startswith("kimi-") or bare == "kimi"
def _is_arcee_trinity_thinking(model: Optional[str]) -> bool:
"""True for Arcee Trinity Large Thinking (direct or via OpenRouter)."""
return _bare_model(model) == "trinity-large-thinking"
# Codex OAuth hard-caps gpt-5.4/5.5/5.6 and gpt-6 Astra at 272K (raw API/OpenRouter expose 1.05M);
# the default 50% trigger would compact at ~136K, so raise to 85% (~231K).
_CODEX_GPT54_GPT55_COMPACTION_THRESHOLD = 0.85
# gpt-5.3-codex-spark: Codex-OAuth-only, native 128K; 70% (~90K) leaves summary headroom.
_CODEX_SPARK_COMPACTION_THRESHOLD = 0.70
def _is_codex_gpt54_or_gpt55(model: Optional[str], provider: Optional[str] = None) -> bool:
"""True for gpt-5.4/5.5/5.6, gpt-6 Sol/Terra/Luna, gpt-6 Astra (and the Daybreak Sol alias) on the Codex OAuth route only.
Other routes expose a larger window for the same slug and keep the user's threshold.
Prefix-matched so ``-pro`` and dated snapshots track every 272K-capped family; ``-900k``
picker variants are excluded. Astra is substring-matched (any slug containing ``astra``
without ``900k``). Name kept for the ``compression.codex_gpt55_autoraise`` key.
"""
bare = _codex_route_bare_model(model, provider)
if bare is None:
return False
from agent.model_metadata import is_codex_context_variant
if is_codex_context_variant(bare):
return False
if "astra" in bare:
return "900k" not in bare
return bare == "gpt-daybreak-blue-latest" or any(
bare == fam or bare.startswith(fam + "-") or bare.startswith(fam + ".")
for fam in ("gpt-5.4", "gpt-5.5", "gpt-5.6", "gpt-6-sol", "gpt-6.1-sol", "gpt-6-luna"))
def _codex_route_bare_model(model: Optional[str], provider: Optional[str]) -> Optional[str]:
"""Lowercased bare model slug when ``provider`` is the Codex OAuth route, else None."""
return _bare_model(model) if (provider or "").strip().lower() == "openai-codex" else None
def _is_codex_spark(model: Optional[str], provider: Optional[str] = None) -> bool:
"""True for ``gpt-5.3-codex-spark`` on the Codex OAuth route (the slug exists nowhere else)."""
return _codex_route_bare_model(model, provider) == "gpt-5.3-codex-spark"
def _is_openai_default_temperature_only(model: Optional[str]) -> bool:
"""True for OpenAI reasoning families that 400 (``unsupported_value``) on any non-default
``temperature``: gpt-5.x (incl. dated snapshots, ``-pro``, ``-codex``), o1/o3/o4. The
``gpt-5-chat`` non-reasoning line still accepts it (#51083)."""
bare = _bare_model(model)
return bare.startswith(("gpt-5", "o1", "o3", "o4")) and not bare.startswith("gpt-5-chat")
# Routes (host + model) that rejected ``temperature`` at runtime; the next call omits it up front
# instead of paying the 400 round-trip again (the retry alone left #51083's first call to time out).
_TEMPERATURE_REJECTED_ROUTES: set = set()
def remember_temperature_rejection(
provider: Optional[str], base_url: Optional[str], rejected_kwargs: dict[str, Any], error: BaseException,
) -> None:
_TEMPERATURE_REJECTED_ROUTES.add((_route_key(provider, base_url), _bare_model(rejected_kwargs.get("model"))))
def _fixed_temperature_for_model(
model: Optional[str], base_url: Optional[str] = None, provider: Optional[str] = None,
) -> Optional[float] | object:
"""``OMIT_TEMPERATURE`` (drop the key; Kimi/Moonshot, OpenAI reasoning families, Gemini 3.6+, routes
that already rejected it), a fixed ``float``, or ``None``."""
if _is_kimi_model(model):
logger.debug("Omitting temperature for Kimi model %r (server-managed)", model)
return OMIT_TEMPERATURE
from agent.gemini_native_adapter import gemini_sampling_deprecated
if _is_openai_default_temperature_only(model) or gemini_sampling_deprecated(_bare_model(model)):
logger.debug("Omitting temperature for %r (default-only, or Gemini 3.6+ ignores sampling)", model)
return OMIT_TEMPERATURE
if (_route_key(provider, base_url), _bare_model(model)) in _TEMPERATURE_REJECTED_ROUTES:
logger.debug("Omitting temperature for %r (route rejected it earlier)", model)
return OMIT_TEMPERATURE
return 0.5 if _is_arcee_trinity_thinking(model) else None
def _compression_threshold_for_model(
model: Optional[str], provider: Optional[str] = None, *,
allow_codex_gpt55_autoraise: bool = True,
) -> Optional[float]:
"""Per-model/route compression threshold override (fraction of context used), or None.
Arcee Trinity Large Thinking → 0.75 (preserve reasoning context); Codex-route gpt-5.4/5.5/5.6/Astra
→ 0.85, gated by ``allow_codex_gpt55_autoraise``; Codex-route gpt-5.3-codex-spark → 0.70, ungated.
"""
if _is_arcee_trinity_thinking(model):
return 0.75
if allow_codex_gpt55_autoraise and _is_codex_gpt54_or_gpt55(model, provider):
return _CODEX_GPT54_GPT55_COMPACTION_THRESHOLD
if _is_codex_spark(model, provider):
return _CODEX_SPARK_COMPACTION_THRESHOLD
return None
# Aux "fast tier" families, fastest first (measured p50 titling latency). Matched as substrings
# against the LIVE /v1/models catalog because pinned ids rot; rolling "-latest" aliases lead.
_FAST_MODEL_FAMILIES: tuple = (
"gpt-mini-latest", "gpt-nano-latest", "claude-haiku-latest", "gemini-flash-latest",
"gpt-5.4-nano", "gpt-5.4-mini", "gpt-5-mini", "haiku-4.5", "gemini-3.6-flash", "flash-lite",
"-nano", "-mini", "-flash", "haiku",
)
# Disqualifiers: reasoning variants think before answering; ":batch" is a queue; ":free" tiers
# are rate-limited and slowest; embedders/modality endpoints match a rung but cannot answer.
_FAST_MODEL_EXCLUDE: tuple = (
"thinking", "reason", "-r1", "minilm", ":batch", ":free",
"o1-", "o3-", "o4-", "codex", "audio", "-vl", "embed",
"-tts", "-transcribe", "-realtime", "-image", "-search-preview",
)
def _model_recency_key(model_id: str) -> tuple:
"""Sort key putting a family's newest release first: digit runs compare numerically (plain
string order picks ``gpt-3.5-mini`` over ``gpt-5.4-mini`` and breaks at 9 vs 10)."""
# re.split with one capturing group alternates text, number, text, …
return tuple(
(1, float(part), "") if index % 2 else (0, 0.0, part)
for index, part in enumerate(re.split(r"(\d+(?:\.\d+)?)", model_id.lower())) if part)
def _fast_model_from_catalog(provider_id: str) -> str:
"""Newest ``_FAST_MODEL_FAMILIES`` match from the provider's live (cached) catalog.
"" when the catalog is unavailable or holds no small model (caller falls through to the
curated default). Never raises; the fetch is memory+disk cached.
"""
is_nous = provider_id.strip().lower() == "nous"
try:
from hermes_cli.auth import resolve_api_key_provider_credentials
from hermes_cli.models_pricing import fetch_models_with_pricing
from providers import get_provider_profile
# Most /v1/models endpoints are authenticated; an anonymous 401 would read as "no small
# model" and pin the curated default forever.
api_key, base_url = "", ""
try:
creds = resolve_api_key_provider_credentials(provider_id) or {}
api_key = str(creds.get("api_key", "")).strip()
base_url = str(creds.get("base_url", "")).strip()
except Exception:
# Not an API-key provider, or nothing configured; anonymous fetch may still work.
logger.debug("No credentials for %s catalog", provider_id, exc_info=True)
if not api_key and is_nous:
# Nous is OAuth (resolver raises); anonymous reads return the full catalog.
try:
from hermes_cli.models_pricing import _resolve_nous_pricing_credentials
api_key, base_url = _resolve_nous_pricing_credentials()
except Exception:
logger.debug("No Nous credentials for catalog", exc_info=True)
if not base_url:
base_url = str(getattr(get_provider_profile(provider_id), "base_url", "") or "")
base_url = base_url.rstrip("/")
if not base_url:
return ""
base_url = base_url.removesuffix("/v1")
# Nous-only args must match the pickers' or the seeded cache loses sale chrome and
# policy-catalog expiry.
_nous_kwargs = {}
if is_nous:
from hermes_cli.models_pricing import _NOUS_CATALOG_TTL_SECONDS
_nous_kwargs = {"include_sale_original": True, "cache_ttl_seconds": _NOUS_CATALOG_TTL_SECONDS}
catalog = fetch_models_with_pricing(
api_key=api_key or None, base_url=base_url, timeout=3.0, **_nous_kwargs) or {}
except Exception:
logger.debug("Fast-model catalog lookup failed for %s", provider_id, exc_info=True)
return ""
ids = sorted((str(m) for m in catalog), key=_model_recency_key, reverse=True)
if is_nous:
# Narrow catalog ids by org policy, as the pickers do.
try:
from hermes_cli.models_pricing import nous_policy_allowed_ids, restrict_to_nous_policy
ids = restrict_to_nous_policy(ids, nous_policy_allowed_ids())
except Exception:
logger.debug("Nous policy filter unavailable", exc_info=True)
for family in _FAST_MODEL_FAMILIES:
for model_id in ids:
lowered = model_id.lower()
if family in lowered and not any(x in lowered for x in _FAST_MODEL_EXCLUDE):
return model_id
return ""
# Default auxiliary models for direct API-key providers (cheap/fast for side tasks)
def _get_aux_model_for_provider(provider_id: str, *, prefer_fast: bool = False) -> str:
"""Cheap auxiliary model for a provider.
Ladder: (``prefer_fast`` only) live-catalog family match, then ``ProviderProfile.resolve_aux_model``;
then ``default_aux_model`` (curated); then the legacy dict. ``prefer_fast`` is opt-in (titling)
so other callers keep their static behaviour and cache keys.
"""
profile = None
with contextlib.suppress(Exception):
from providers import get_provider_profile
profile = get_provider_profile(provider_id)
picked = ""
if prefer_fast:
picked = _fast_model_from_catalog(provider_id)
if not picked and profile is not None:
try:
picked = profile.resolve_aux_model() or ""
except Exception:
logger.debug("resolve_aux_model failed for %s", provider_id, exc_info=True)
if not picked and profile is not None and profile.default_aux_model:
picked = profile.default_aux_model
if not picked:
picked = _API_KEY_PROVIDER_AUX_MODELS_FALLBACK.get(provider_id, "")
# Rungs 2-4 are policy-blind; a blocked pick is refused at request time, so drop it and
# let the caller keep the main model.
if picked and provider_id.strip().lower() == "nous":
try:
from hermes_cli.models_pricing import nous_policy_allowed_ids, restrict_to_nous_policy
allowed = nous_policy_allowed_ids()
if allowed and not restrict_to_nous_policy([picked], allowed):
return ""
except Exception:
logger.debug("Nous policy check unavailable", exc_info=True)
return picked
# Fallback for providers without ProviderProfile.default_aux_model (plus some pinned here).
# New providers should set default_aux_model instead.
_API_KEY_PROVIDER_AUX_MODELS_FALLBACK: dict[str, str] = {
"gemini": "gemini-3.6-flash", "zai": "glm-4.5-flash", "kimi-coding": "kimi-k2-turbo-preview",
"stepfun": "step-3.5-flash", "kimi-coding-cn": "kimi-k2-turbo-preview",
"gmi": "google/gemini-3.1-flash-lite-preview", "anthropic": "claude-haiku-4-5-20251001",
"ai-gateway": "google/gemini-3-flash", "opencode-zen": "gemini-3-flash", "opencode-go": "glm-5",
"kilocode": "google/gemini-3.6-flash", "ollama-cloud": "nemotron-3-nano:30b",
"tencent-tokenhub": "hy4-preview", "tencent-tokenplan": "hy4-preview",
# No "deepinfra": its aux model lives on the ProviderProfile (read first).
}
# Legacy alias for callers not yet using _get_aux_model_for_provider().
_API_KEY_PROVIDER_AUX_MODELS: dict[str, str] = _API_KEY_PROVIDER_AUX_MODELS_FALLBACK
# Tasks that may opt into ``auxiliary.<task>.prefer_fast_model``.
_FAST_MODEL_TASKS: frozenset = frozenset({"title_generation"})
def _task_prefers_fast_model(task: Optional[str]) -> bool:
"""Return whether an eligible task explicitly opts into fast-model routing."""
return task in _FAST_MODEL_TASKS and is_truthy_value(
_get_auxiliary_task_config(task).get("prefer_fast_model"), default=False)
# Dedicated vision models for direct providers whose main chat model differs. zai: glm-5.3-flash
# is the only image-capable GLM id served on every Z.AI surface (pay-as-you-go and Coding Plan,
# global and CN); the former glm-5v-turbo pin 404s / 1211 "Unknown Model" on the coding endpoints
# (#111429). ZaiProfile has no default_vision_model(), so dropping the pin would route vision to
# the user's text-only chat model and skip Z.AI entirely.
_PROVIDER_VISION_MODELS: dict[str, str] = {"xiaomi": "mimo-v2.5", "zai": "glm-5.3-flash"}
def _resolve_provider_vision_default(provider: str) -> Optional[str]:
"""Provider default vision model id, or None: static ``_PROVIDER_VISION_MODELS`` (vision-only
names absent from any catalog) win, else ``ProviderProfile.default_vision_model()``."""
static = _PROVIDER_VISION_MODELS.get(provider)
if static:
return static
try:
from providers import get_provider_profile
profile = get_provider_profile(provider)
return profile.default_vision_model() if profile is not None else None
except Exception:
return None
# Endpoints that reject image input: vision auto-detect skips these to the aggregator chain
# instead of returning a client that 404s (Kimi Coding Plan Anthropic wire has no image_in).
_PROVIDERS_WITHOUT_VISION: frozenset = frozenset({"kimi-coding", "kimi-coding-cn"})
# OpenRouter app attribution (always sent). `X-Title` is what the dashboard reads.
_OR_HEADERS_BASE = {
"HTTP-Referer": "https://hermes-agent.nousresearch.com",
"X-Title": "Hermes Agent",
"X-OpenRouter-Categories": "productivity,cli-agent",
}
def _apply_user_default_headers(headers: dict | None) -> dict | None:
"""Merge user ``model.default_headers`` onto resolved headers (user wins; ``model.extra_headers``
alias wins over both). Mirrors ``AIAgent._apply_user_default_headers`` so a custom endpoint behind a
WAF rejecting ``User-Agent`` / ``X-Stainless-*`` works for aux calls. SECURITY: never log values."""
try:
from hermes_cli.config import cfg_get, load_config
_cfg = load_config()
user_headers = cfg_get(_cfg, "model", "default_headers")
alias_headers = cfg_get(_cfg, "model", "extra_headers")
if isinstance(alias_headers, dict) and alias_headers:
user_headers = {**(user_headers if isinstance(user_headers, dict) else {}), **alias_headers}
except Exception:
return headers
if not isinstance(user_headers, dict) or not user_headers:
return headers
merged = dict(headers or {})
merged.update({str(k): str(v) for k, v in user_headers.items() if v is not None})
return merged or headers
def build_or_headers(or_config: dict | None = None) -> dict:
"""OpenRouter headers, plus response-cache headers when enabled.
Precedence env > config > default: ``HERMES_OPENROUTER_CACHE`` overrides
``openrouter.response_cache``; ``HERMES_OPENROUTER_CACHE_TTL`` (1-86400 s) overrides
``openrouter.response_cache_ttl``. ``or_config=None`` reads from disk.
"""
headers = dict(_OR_HEADERS_BASE)
if or_config is None:
try:
from hermes_cli.config import load_config_readonly
or_config = load_config_readonly().get("openrouter", {})
except Exception:
or_config = {}
env_cache = os.environ.get("HERMES_OPENROUTER_CACHE", "").strip().lower()
if not (env_cache in {"1", "true", "yes", "on"} if env_cache else or_config.get("response_cache", False)):
return headers
headers["X-OpenRouter-Cache"] = "true"
env_ttl = os.environ.get("HERMES_OPENROUTER_CACHE_TTL", "").strip()
if env_ttl:
if env_ttl.isdigit() and 1 <= int(env_ttl) <= 86400:
headers["X-OpenRouter-Cache-TTL"] = str(int(env_ttl))
else:
ttl = or_config.get("response_cache_ttl", 300)
if isinstance(ttl, (int, float)) and 1 <= ttl <= 86400:
headers["X-OpenRouter-Cache-TTL"] = str(int(ttl))
return headers
# NVIDIA NIM cloud billing attribution; host-gated because NVIDIA_BASE_URL may be a local NIM.
_NVIDIA_NIM_CLOUD_HEADERS = {"X-BILLING-INVOKE-ORIGIN": "HermesAgent"}
def build_nvidia_nim_headers(base_url: str | None) -> dict:
"""Return NVIDIA NIM cloud attribution headers for build.nvidia.com traffic."""
return dict(_NVIDIA_NIM_CLOUD_HEADERS) if base_url_host_matches(str(base_url or ""), "integrate.api.nvidia.com") else {}
# Vercel AI Gateway attribution (HTTP-Referer → referrerUrl, X-Title → appName).
from hermes_cli.version_info import hermes_user_agent
def _ai_gateway_headers() -> dict[str, str]:
return {"HTTP-Referer": "https://hermes-agent.nousresearch.com", "X-Title": "Hermes Agent",
"User-Agent": hermes_user_agent("HermesAgent")}
# Nous Portal attribution tags, built per call. They come from agent.portal_tags so the client=
# marker tracks the canonical base version — never inline a literal here.
from agent.portal_tags import nous_portal_tags as _nous_portal_tags
# _OPENROUTER_MODEL MUST stay a :free SKU (matching the free_only warning): this lane engages
# silently, and a paid default meant spend the user never opted into. User-configured values
# are honored untouched (_warn_paid_lane_once fires).
_OPENROUTER_MODEL = "nvidia/nemotron-3-ultra-550b-a55b:free"
_NOUS_MODEL = "google/gemini-3.6-flash"
_NOUS_DEFAULT_BASE_URL = "https://inference-api.nousresearch.com/v1"
_ANTHROPIC_DEFAULT_BASE_URL = "https://api.anthropic.com"
_AUTH_JSON_PATH = get_hermes_home() / "auth.json"
_AUTH_JSON_PATH_AT_IMPORT = _AUTH_JSON_PATH
def _auth_json_path():
"""Active profile's ``auth.json`` at call time (a patched ``_AUTH_JSON_PATH`` still wins). The
import-time constant is the LAUNCH profile's; under multiplexing a secondary's auxiliary calls
would otherwise authenticate to Nous with the default profile's token."""
from hermes_cli.auth import _auth_file_path
return _AUTH_JSON_PATH if _AUTH_JSON_PATH != _AUTH_JSON_PATH_AT_IMPORT else _auth_file_path()
# Hosts exposing BOTH ``…/anthropic`` and a sibling OpenAI ``…/v1``. Matched on the URL *host*
# only: unconditional rewrites break Anthropic-only gateways.
_DUAL_SURFACE_ANTHROPIC_HOST_SUFFIXES = ("minimax.io", "minimax.chat", "minimaxi.com")
_DUAL_SURFACE_ANTHROPIC_HOST_PREFIXES = ("api.minimax.",)
def _is_dual_surface_anthropic_host(url: str) -> bool:
"""True when the URL's host is a known dual-surface (MiniMax-family) host."""
try:
host = (urlparse(url).hostname or "").lower()
except ValueError:
return False
return any(
host == suffix or host.endswith("." + suffix) for suffix in _DUAL_SURFACE_ANTHROPIC_HOST_SUFFIXES
) or any(host.startswith(prefix) for prefix in _DUAL_SURFACE_ANTHROPIC_HOST_PREFIXES)
def _to_openai_base_url(base_url: str) -> str:
"""Normalize dual-surface Anthropic URLs to their OpenAI-compatible sibling.
MiniMax-family: ``/anthropic`` → ``/v1``; ZAI Coding Plan → ``/coding/paas/v4`` (the general
endpoint bills separately); Kimi Code ``/coding`` → ``/coding/v1`` (the OpenAI SDK path 404s
without it). Anthropic-only gateways keep their path.
"""
url = str(base_url or "").strip().rstrip("/")
if base_url_hostname(url) == "api.actual.inc":
from hermes_cli.auth import normalize_actual_base_url
return normalize_actual_base_url(url)
if url.endswith("/anthropic"):
if base_url_host_matches(url, "open.bigmodel.cn") or base_url_host_matches(url, "api.z.ai"):
rewritten = url[: -len("/anthropic")] + "/coding/paas/v4"
logger.debug("Auxiliary client: rewrote ZAI base URL %s → %s", url, rewritten)
return rewritten
if _is_dual_surface_anthropic_host(url):
rewritten = url[: -len("/anthropic")] + "/v1"
logger.debug("Auxiliary client: rewrote dual-surface base URL %s → %s", url, rewritten)
return rewritten
logger.debug(
"Auxiliary client: keeping Anthropic-only base URL %s (no dual-surface host match)", url)
return url
if base_url_host_matches(url, "api.kimi.com") and url.endswith("/coding"):
rewritten = url + "/v1"
logger.debug("Auxiliary client: rewrote Kimi base URL %s → %s", url, rewritten)
return rewritten
return url
def _load_pool_with_credentials(provider: str, note: str = "") -> Optional[Any]:
"""``load_pool(provider)`` when it has credentials, else None (never raises)."""
try:
pool = load_pool(provider)
except Exception as exc:
logger.debug("Auxiliary client: could not load pool for %s%s: %s", provider, note, exc)
return None
return pool if pool and pool.has_credentials() else None
def _select_pool_entry(provider: str, model: Optional[str] = None) -> tuple[bool, Optional[Any]]:
"""Return (pool_exists_for_provider, selected_entry)."""
pool = _load_pool_with_credentials(provider)
if pool is None:
return False, None
try:
return True, pool.select(model=_normalize_resolved_model(model, provider)) if model else pool.select()
except Exception as exc:
logger.debug("Auxiliary client: could not select pool entry for %s: %s", provider, exc)
return True, None
def _peek_pool_entry(provider: str, pool: Any = None, model: Optional[str] = None) -> Optional[Any]:
"""Best-effort current/next pool entry without mutating selection order.
``pool`` skips the disk re-read when the caller already loaded it.
"""
if pool is None:
pool = _load_pool_with_credentials(provider, " (peek)")
if pool is None:
return None
try:
peek_fn = getattr(pool, "peek", None)
if model and callable(peek_fn): # the pool checks its current entry against the model's cooldowns
with contextlib.suppress(TypeError): # a pool whose peek takes no model: current, then peek
return peek_fn(model=_normalize_resolved_model(model, provider))
current_fn = getattr(pool, "current", None)
current = current_fn() if callable(current_fn) else None
if current is not None:
return current
if callable(peek_fn):
return peek_fn()
except Exception as exc:
logger.debug("Auxiliary client: could not peek pool entry for %s: %s", provider, exc)
return None
def _pool_runtime_api_key(entry: Any) -> str:
# runtime_api_key handles provider-specific fallback (e.g. agent_key for nous); None entry → "".
key = getattr(entry, "runtime_api_key", None) or getattr(entry, "access_token", "")
return str(key or "").strip()
def _pool_runtime_base_url(entry: Any, fallback: str = "") -> str:
if entry is None:
return str(fallback or "").strip().rstrip("/")
if getattr(entry, "provider", None) == "nous":
# Canonical auth-layer reader so the env override shares one normalization path.
from hermes_cli.auth import _nous_inference_env_override
env_url = _nous_inference_env_override()
if env_url:
return env_url
# runtime_base_url is provider-aware; fall back for non-PooledCredential entries.
url = (getattr(entry, "runtime_base_url", None) or getattr(entry, "inference_base_url", None)
or getattr(entry, "base_url", None) or fallback)
return str(url or "").strip().rstrip("/")
# Hosts the aux Anthropic path may be pointed at via model.base_url; anything else falls back
# to the Anthropic default so a foreign host never leaks in.
_ANTHROPIC_COMPATIBLE_HOSTS = frozenset({"api.anthropic.com"})
def _is_anthropic_compatible_host(url: str) -> bool:
"""True for native Anthropic hosts and gateways serving Messages under a ``/anthropic`` path
(same convention as runtime_provider / ``_wrap_if_needed``), so a configured ``model.base_url``
whose gateway holds auth is not discarded. A bare non-Anthropic base_url is False."""
if not url:
return False
try:
parsed = urlparse(url)
if (parsed.hostname or "").strip().lower().rstrip(".") in _ANTHROPIC_COMPATIBLE_HOSTS:
return True
path = (parsed.path or "").rstrip("/").lower()
return path.endswith("/anthropic") or path.endswith("/anthropic/v1")
except Exception:
return False
def _nous_min_key_ttl_seconds() -> int:
try:
return max(60, int(os.getenv("HERMES_NOUS_MIN_KEY_TTL_SECONDS", "1800")))
except (TypeError, ValueError):
return 1800
def _scoped_key_env(name: str) -> str:
"""Read a provider API key (or its paired base-URL) env var through the profile secret scope.
In agent turns the scope's verdict is authoritative (a scoped miss must not borrow another
profile's key); only the unscoped default-profile path (``UnscopedSecretError``) reads
``os.environ`` -- any other scope failure propagates instead of borrowing the ambient env.
"""
if not name:
return ""
from agent.secret_scope import UnscopedSecretError, get_secret
try:
return (get_secret(name) or "").strip()
except UnscopedSecretError:
return (os.getenv(name) or "").strip()
def _attempt_stream_socket(stream: Any) -> Any:
"""The raw socket under an SDK event stream (``stream.response`` is the ``httpx.Response``;
httpcore publishes its connection as the ``network_stream`` extension), or None."""
from agent.agent_runtime_helpers import _socket_from_stream
extensions = getattr(getattr(stream, "response", None), "extensions", None)
network_stream = extensions.get("network_stream") if isinstance(extensions, dict) else None
return _socket_from_stream(network_stream) if network_stream is not None else None
def _close_quietly(target: Any, failure_note: Optional[str]) -> None:
"""Call ``target.close()`` if present; a failure is debug-logged under ``failure_note`` (silent when None)."""
close = getattr(target, "close", None)
if callable(close):
try:
close()
except Exception:
if failure_note:
logger.debug("Codex auxiliary: %s", failure_note, exc_info=True)
# The context compressor keys its retry-ladder classification on this text (#124077).
CODEX_STREAM_STALL_MARKER = "stream stalled"
class _CodexStreamGuard:
"""Progress-aware deadline + FD-safe timeout watchdog for one Codex aux stream attempt.
(1) The first substantive payload must arrive within ``no_progress_timeout`` or we fail fast into
the caller's retry/fallback chain (a dead or keepalive-only zombie must not hold the budget);
(2) each substantive event re-arms that window (keepalive/lifecycle frames do NOT, mirroring
commit-fence gating) so a live stream is never killed by an absolute total; (3) a hard ceiling
from ``_aux_stream_total_ceiling`` still terminates a pathological drip.
"""
def __init__(
self, client: Any, total_timeout: Optional[float],
no_progress_timeout: Optional[float] = None,
):
self._client = client
self.total_timeout = total_timeout
self._start = time.monotonic()
# Task-scoped override (auxiliary.<task>.no_progress_timeout, #108104); falls back to the
# built-in default when unset or not a positive number.
if isinstance(no_progress_timeout, (int, float)) and no_progress_timeout > 0:
self.no_progress_timeout = float(no_progress_timeout)
else:
self.no_progress_timeout = _AUX_STREAM_NO_PROGRESS_TIMEOUT_SECONDS
# Progress-aware stream deadlines (supersedes the old single absolute kill at ``total_timeout``).
# Three regimes: 1. First token: the stream must produce its first substantive payload within
# ``no_progress_timeout`` (60s default) or we fail fast and let the caller's normal retry/fallback
# chain run — a dead (or keepalive-only zombie) Codex stream no longer holds the full 300s
# compression budget before falling back (masoria report, Aug 2026: 3 stacked 300s waits -> 20+ min
# stuck on "Summarizing"). 2. Streaming: every substantive event re-arms the deadline by
# ``no_progress_timeout`` — a live stream is never killed by an absolute total, so a long reasoning
# summary that is actually producing tokens completes instead of timing out at 300s and falling back
# (#54915's original complaint, fixed properly). Keepalive/lifecycle frames do NOT re-arm, mirroring
# the commit-fence progress gating (#96707). 3. Hard ceiling: an absolute backstop from
# ``_aux_stream_total_ceiling`` (max(600s, 4x configured timeout) — the same bound the streamed
# chat.completions path uses) so a pathological one-token-per-59s drip still terminates.
if total_timeout is not None:
self.no_progress_timeout = min(self.no_progress_timeout, float(total_timeout))
self.hard_deadline = self._start + _aux_stream_total_ceiling(total_timeout)
# The waiting host's absolute deadline clamps the ceiling so the watchdog Timer severs
# the socket the instant the host stops waiting — a stream blocked between events
# can't be stopped by a per-event check.
host_deadline = _current_aux_stream_deadline()
if isinstance(host_deadline, (int, float)) and host_deadline < self.hard_deadline:
self.hard_deadline = float(host_deadline)
self._deadline_lock = threading.Lock()
self._progress_deadline = self._start + self.no_progress_timeout
self.saw_content = threading.Event()
self.timed_out = threading.Event()
# Set only when the timeout WON (not when the owner hard-cancelled first): tells the
# owner's ``finally`` the shared client's FDs still need a real close.
self.timeout_release_pending = threading.Event()
self.stream_finished = threading.Event()
self._timer = None
# The owner may return on hard cancel while this attempt is still blocked in the SDK
# stream. Timer threads don't inherit the worker's thread-local protection state, so
# freeze the hard-cancel source before creating the timer.
self._protected_cancel_check = _capture_aux_cancel_check() if _aux_interrupt_protected() else None
self._attempt_stream_lock = threading.Lock()
self._attempt_stream: Any = None
# The request-driving thread owns the transport FDs — see _close_client_on_timeout.
self._owner_tid = threading.get_ident()
def effective_deadline(self) -> float:
with self._deadline_lock:
return min(self.hard_deadline, self._progress_deadline)
def cancel_requested(self) -> bool:
"""True when the frozen hard-cancel source says the owner already cancelled."""
check = self._protected_cancel_check
return callable(check) and _captured_aux_cancel_requested(check)
def adopt_stream(self, stream: Any) -> None:
with self._attempt_stream_lock:
self._attempt_stream = stream
def release_stream(self, stream: Any) -> None:
"""Owner-side: close the attempt stream silently and forget it."""
_close_quietly(stream, None)
with self._attempt_stream_lock:
self._attempt_stream = None
def close_attempt_stream(self, failure_note: str) -> None:
"""Wake only this attempt's stream, never the shared client. The owner thread closes it;
any other thread only ``shutdown()``s its socket, since ``Stream.close()`` would release the FD
under the owner's ``SSL_read`` (#70773, #130115). Socketless streams are closed as before."""
with self._attempt_stream_lock:
stream = self._attempt_stream
if stream is None:
return
if threading.get_ident() != self._owner_tid:
sock = _attempt_stream_socket(stream)
if sock is not None:
from agent.agent_runtime_helpers import _shutdown_socket
_shutdown_socket(sock)
return
_close_quietly(stream, failure_note)
def record_progress(self) -> None:
"""Substantive payload re-arms the no-progress window; the hard ceiling never moves."""
with self._deadline_lock:
self._progress_deadline = time.monotonic() + self.no_progress_timeout
def timeout_message(self) -> str:
elapsed = time.monotonic() - self._start
if time.monotonic() >= self.hard_deadline:
return f"Codex auxiliary Responses stream exceeded {self.hard_deadline - self._start:.1f}s hard ceiling"
if not self.saw_content.is_set():
return (
"Codex auxiliary Responses stream produced no output "
f"within {float(self.no_progress_timeout):.1f}s (no-progress timeout, {elapsed:.1f}s elapsed)")
return (
f"Codex auxiliary Responses {CODEX_STREAM_STALL_MARKER}: no new output "
f"for {float(self.no_progress_timeout):.1f}s ({elapsed:.1f}s elapsed)")
def _close_client_on_timeout(self) -> None:
begin_timeout_cleanup = getattr(self._protected_cancel_check, "begin_timeout_cleanup", None)
if callable(begin_timeout_cleanup):
timeout_won = bool(begin_timeout_cleanup())
else:
timeout_won = not self.cancel_requested()
# Publish transport timeout only after the attempt-local decision is fixed, so owner
# polling cannot observe completion in between.
self.timed_out.set()
if not timeout_won:
# Owner already hard-cancelled. The OpenAI client is process-shared, so never
# close/evict it here; wake only this attempt's stream if responses.create()
# returned one, else rely on the bounded SDK timeout.
self.close_attempt_stream("cancelled attempt stream close during timeout failed")
return
# FD-ownership contract: only the thread driving the request may ``close()`` this
# client's FDs. From a stranger thread (the watchdog Timer) only ``shutdown()`` is
# FD-safe — ``close()`` releases the raw TLS fd while the owner's OpenSSL BIO still
# caches it, the kernel recycles it (e.g. into a SQLite handle), and the owner's TLS
# flush corrupts that file. The owner does the real close in its ``finally``.
# This callback has two callers — ``_check_cancelled`` on the owning thread, and the daemon watchdog
# ``threading.Timer``, which is a stranger thread. The owning thread performs the real close in the
# ``finally`` below, which is where the FD release belongs. See #70773.
self.timeout_release_pending.set()
if threading.get_ident() == self._owner_tid:
_close_quietly(self._client, "client close during timeout failed")
else:
try:
from agent.agent_runtime_helpers import force_close_tcp_sockets
shutdown_count = force_close_tcp_sockets(self._client)
logger.info(
"Codex auxiliary client aborted (timeout, tcp_force_closed=%d, "
"deferred_close=stranger_thread)", shutdown_count)
except Exception:
logger.debug("Codex auxiliary: client abort during timeout failed", exc_info=True)
# Socket shutdown only wakes a reader on a REAL transport; the owner may be blocked
# inside the SDK's event stream (or a socketless test double). Wake the
# attempt-owned stream too — from this thread that is a shutdown of its socket,
# never a close (see close_attempt_stream).
self.close_attempt_stream("attempt stream close during stranger-thread timeout failed")
# The aux client cache wraps this same client; drop the entry so the next aux call
# doesn't reuse the dead transport and fail fast.
try:
# After we close the httpx transport above, the cache must drop that entry — otherwise the next
# auxiliary call (compression retry, memory flush, etc.) reuses the dead client and fails fast
# with a connection error. See issue #23432.
_evict_cached_client_instance(self._client)
except Exception:
logger.debug("Codex auxiliary: cache eviction on timeout failed", exc_info=True)
def check_cancelled(self) -> None:
if self.total_timeout is not None and time.monotonic() >= self.effective_deadline():
if not self.timed_out.is_set():
self._close_client_on_timeout()
raise TimeoutError(self.timeout_message())
try:
from tools.interrupt import is_interrupted
# Protected atomic aux tasks (compression) must not abort on a mid-flight gateway
# interrupt (degraded fallback marker); explicit host cancel has its own exception.
if _aux_interrupt_cancel_requested():
raise AuxiliaryExplicitCancellation()
# Explicit host cancellation has its own frozen exception; timeouts above still fire and other
# aux tasks remain interruptible. See #23975.
if is_interrupted() and not _aux_interrupt_protected():
raise InterruptedError("Codex auxiliary Responses stream interrupted")
except InterruptedError:
raise
except Exception:
# Interrupt state is best-effort UX; never a new failure mode.
pass
def _watchdog_fire(self) -> None:
# Re-armable: if progress moved the deadline forward, reschedule instead of killing a
# live stream.
remaining = self.effective_deadline() - time.monotonic()
if remaining > 0:
if not (self.timed_out.is_set() or self.stream_finished.is_set()):
self._arm_timer(remaining)
return
self._close_client_on_timeout()
def _arm_timer(self, delay: float) -> None:
self._timer = t = threading.Timer(delay, self._watchdog_fire)
t.daemon = True
t.start()
def start(self) -> None:
"""Arm the watchdog (when a total timeout exists) and run the first cancel check."""
if self.total_timeout:
self._arm_timer(max(self.effective_deadline() - time.monotonic(), 0.0))
self.check_cancelled()
def on_event(self, _event: Any) -> None:
# TTFP telemetry records every frame, but forward progress (compression commit fence,
# no-progress window) counts only substantive payloads — keepalives must not re-arm,
# so a zombie stream dies at the same window as a dead connection.
# #93650: keep bulk wire-format payload out of the SDK's GIL-holding request transform on auxiliary
# calls too.
if _codex_event_has_content(_event):
self.record_progress()
self.saw_content.set()
_notify_aux_provider_response()
else:
_notify_aux_timing_response()
self.check_cancelled()
def finish(self) -> None:
"""Owner ``finally``: stop the watchdog and release FDs a stranger-thread timeout only shut down."""
self.stream_finished.set()
if self._timer is not None:
self._timer.cancel()
# Gated on timeout_release_pending, NOT timed_out: after a hard-cancel the shared
# client must stay usable for other sessions.
if self.timeout_release_pending.is_set():
_close_quietly(self._client, "owner-thread close after timeout failed")
class _CodexCompletionsAdapter:
"""Drop-in shim routing chat.completions.create() kwargs through Codex Responses streaming."""
def __init__(self, real_client: OpenAI, model: str):
self._client = real_client
self._model = model
def _build_responses_kwargs(self, kwargs: dict[str, Any]) -> tuple[dict[str, Any], str, Any]:
"""chat.completions kwargs → Responses API kwargs, ``(resp_kwargs, model, timeout)``; mirrors codex.py::build_kwargs."""
# Separate system/instructions from replayable conversation messages, then route the rest through
# the SINGLE shared chat->Responses converter used by the main agent transport
# (agent/transports/codex.py). Maintaining a private conversion loop here let chat-style messages
# with role="tool" leak straight into Responses input[] — which the Responses API rejects with
# "Invalid value: 'tool'. Supported values are: 'assistant', 'system', 'developer', and 'user'."
# (issue #5709, hit hard by flush_memories() / compression replaying real session history that
# includes assistant tool_calls + role="tool" results). The shared converter encodes assistant tool
# calls as `function_call` items and tool results as `function_call_output` items with a valid
# call_id, so every Responses path normalizes tool history identically and cannot drift.
from agent.codex_responses_adapter import (
_chat_messages_to_responses_input,
_classify_responses_issuer,
_responses_tools,
_role_message_item,
_wire_model_identity,
classify_responses_route,
)
from agent.transports.codex import _alias_wire_tools
model = kwargs.get("model", self._model)
wire_model = _wire_model_identity(model)
host = str(getattr(self._client, "base_url", "") or "")
# Same route classifier as the main transport, so the issuer stamp matches what it minted.
route = classify_responses_route(SimpleNamespace(provider=None, base_url=host))
is_xai = route.is_xai_responses
tools = kwargs.get("tools")
if tools:
# xAI Responses rejects ``pattern``/``format`` JSON Schema keywords (400); strip for
# chat_completion_helpers.py parity. Deep-copy first — sanitizers mutate inner dicts
# in place and would strip the caller's tool registry.
try:
import copy as _copy
from tools.schema_sanitizer import strip_pattern_and_format, strip_slash_enum
tools = _copy.deepcopy(list(tools))
tools, _ = strip_pattern_and_format(tools)
tools, _ = strip_slash_enum(tools)
except Exception as exc:
logger.warning(
"Auxiliary client: failed to sanitize tool schemas for "
"Codex/xAI Responses path: %s", exc,
)
# Tool schemas go through the SAME converter (``strict: False``) and the SAME reserved-name
# aliasing (OpenCode / Perplexity / xAI) as agent/transports/codex.py::build_kwargs. A private
# list here sent raw names without ``strict``, so title/compression/MoA calls 400ed on routes
# where the main loop worked (#114260). The alias map is request-local: it rides on the payload
# (``_wire_aliases``, popped in ``create()``), never on the instance — aux adapters are shared.
wire_tools, wire_aliases = _alias_wire_tools(
_responses_tools(tools),
{"provider": getattr(self._client, "_hermes_aux_effective_provider", "") or None, "base_url": host},
is_xai,
)
renamed = {original: alias for alias, original in wire_aliases.items()}
# System → ``instructions``; the rest goes through the SINGLE shared chat→Responses
# converter (a private loop here once let role="tool" leak into input[]; the shared one
# encodes tool history as function_call/function_call_output).
instructions = "You are a helpful assistant."
replay_messages: list[dict[str, Any]] = []
for msg in kwargs.get("messages", []):
content = msg.get("content") or ""
if msg.get("role", "user") == "system":
instructions = content if isinstance(content, str) else str(content)
continue
# Replayed history names each tool the way THIS request declares it.
if renamed and msg.get("tool_calls"):
msg = {**msg, "tool_calls": [
{**tc, "function": {**tc["function"], "name": renamed[tc["function"]["name"]]}}
if isinstance(tc, dict) and (tc.get("function") or {}).get("name") in renamed else tc
for tc in msg["tool_calls"]
]}
replay_messages.append(msg)
# Endpoints that bind replayed codex_message_items ids to a backend connection 401 on a stale id
# (``replay_item_ids=False``) — same guard as build_kwargs. Aux calls
# never send ``context_management`` (main-turn feature): no compaction checkpoint.
# Auxiliary calls (context compression, flush_memories, MoA aggregation) go through this adapter
# instead of agent/transports/codex.py's build_kwargs, so they need the same guard applied
# independently. See #32716.
# Aux requests run their own model; stamp/filter reasoning provenance against it, not the main agent's.
issuer_kind = _classify_responses_issuer(base_url=host, **route._asdict())
input_items = _chat_messages_to_responses_input(
replay_messages, drop_replay_item_ids=not route.endpoint.replay_item_ids,
current_issuer_kind=issuer_kind,
current_issuer_model=wire_model, native_compaction_eligible=False,
)
resp_kwargs: dict[str, Any] = {
# Codex only knows the base slug; strip the Hermes ``-900k`` picker suffix.
"model": wire_model, "instructions": instructions,
"input": input_items or [_role_message_item("user", "")], "store": False,
}
# Forward the chat.completions timeout; otherwise a Codex stream can sit behind a
# dead-looking CLI until the user force-interrupts.
timeout = kwargs.get("timeout")
if timeout is not None:
resp_kwargs["timeout"] = timeout
# Per-request HTTP headers (OpenCode session affinity, provider request headers) map to real
# headers via the SDK kwarg — forward them.
if isinstance(kwargs.get("extra_headers"), dict) and kwargs["extra_headers"]:
resp_kwargs["extra_headers"] = dict(kwargs["extra_headers"])
# The Codex endpoint rejects max_output_tokens/temperature (400) — omit.
extra_body = kwargs.get("extra_body") or {}
if isinstance(extra_body, dict):
# service_tier (fast mode) is a top-level Responses field; xAI's endpoint rejects it.
service_tier = extra_body.get("service_tier")
if isinstance(service_tier, str) and service_tier.strip() and not is_xai:
resp_kwargs["service_tier"] = service_tier.strip()
reasoning_cfg = extra_body.get("reasoning")
if isinstance(reasoning_cfg, dict):
# Shared per-model vocabulary with the main transport ("max" only where the model publishes it; "minimal"/"ultra"
# clamp to a listed level; ``()`` = the model takes no ``reasoning`` field at all — gpt-4o/4.1 on api.openai.com,
# #76255). ``enabled: False`` goes on the wire as ``effort: none`` where the vocabulary has it,
# since an omitted field leaves the model's default effort on (#75227).
from agent.reasoning_effort import clamp_effort
from agent.transports.codex import _codex_efforts_for_route
supported = _codex_efforts_for_route(model, host, is_codex_backend=route.is_codex_backend)
if supported and reasoning_cfg.get("enabled") is not False:
# Truthy-only: Codex 400s on e.g. {"effort": null}, so falsy → default.
effort = clamp_effort(reasoning_cfg.get("effort") or "medium", supported)
resp_kwargs["reasoning"] = {"effort": effort, "summary": "auto"}
resp_kwargs["include"] = ["reasoning.encrypted_content"]
elif "none" in supported and not is_xai:
resp_kwargs["reasoning"] = {"effort": "none"}
if wire_tools:
resp_kwargs["tools"] = wire_tools
if wire_aliases:
resp_kwargs["_wire_aliases"] = wire_aliases
# Response normalization is route-sensitive too (popped in ``create()``, never sent).
resp_kwargs["_issuer_kind"] = issuer_kind
# Stable prompt-cache routing: key is content-addressed from the static prefix
# (instructions + tool schemas) so it survives across turns, scoped by the owning
# conversation (rotation-stable logical scope, else the physical session id). Skip the
# key where the main transport does: xAI takes it in extra_body, ``prompt_cache_key=False`` endpoints opt out.
try:
# Reuse the Responses transport's single authoritative hash algorithm and session-scope
# normalization so equivalent static prefixes route to the same cache bucket across modes,
# without concentrating unrelated sessions into one shared bucket (see #78941).
from agent.transports.codex import _cache_scope_from_session_id, _content_cache_key
from agent.transports.codex import _default_prompt_cache_retention_for_request
if not is_xai and route.endpoint.prompt_cache_key and "prompt_cache_key" not in resp_kwargs:
scope = _cache_scope_from_session_id(
_runtime_main_value("cache_scope") or _runtime_main_value("session_id")
)
cache_key = _content_cache_key(resp_kwargs["instructions"], resp_kwargs.get("tools"), scope)
if cache_key:
resp_kwargs["prompt_cache_key"] = cache_key
if "prompt_cache_retention" not in resp_kwargs:
cache_retention = _default_prompt_cache_retention_for_request(model, host)
if cache_retention:
resp_kwargs["prompt_cache_retention"] = cache_retention
except Exception:
logger.debug("Codex auxiliary: prompt_cache_key derivation skipped", exc_info=True)
# Last, like the main transport: caller extra_body must not put a rejected Astra field back.
from agent.transports.codex import _sanitize_astra_request_kwargs
_sanitize_astra_request_kwargs(resp_kwargs, model, host)
return resp_kwargs, model, timeout
def create(self, **kwargs) -> Any:
from hermes_cli.providers import is_actual_route
if is_actual_route(
getattr(self._client, "_hermes_aux_effective_provider", ""),
str(getattr(self._client, "base_url", "") or ""),
):
raise ValueError(
"Actual requests require Chat Completions; refusing to call /responses."
)
# Low-level ``responses.create(stream=True)`` and assemble the final response ourselves
# from ``response.output_item.done``: the high-level ``responses.stream()`` rebuilds from
# ``response.completed.response.output``, which Codex returns as ``null`` (SDK crash).
resp_kwargs, model, timeout = self._build_responses_kwargs(kwargs)
wire_aliases = resp_kwargs.pop("_wire_aliases", None) or {}
issuer_kind = resp_kwargs.pop("_issuer_kind", None)
issuer_model = str(resp_kwargs.get("model") or model)
total_timeout = timeout if isinstance(timeout, (int, float)) and timeout > 0 else None
guard = _CodexStreamGuard(self._client, total_timeout, no_progress_timeout=kwargs.get("no_progress_timeout"))
deadline_hit = False
try:
guard.start()
from agent.codex_runtime import _consume_codex_event_stream
from agent.sdk_transform_bypass import bypass_sdk_request_transform
# Keep bulk wire payload out of the SDK's GIL-holding request transform.
stream_kwargs = bypass_sdk_request_transform({**resp_kwargs, "stream": True})
event_stream = self._client.responses.create(**stream_kwargs)
guard.adopt_stream(event_stream)
# The timer may fire while responses.create() is blocked; if the cancelled attempt
# had no stream to close then, close it now that it is attempt-owned — never the shared client.
if guard.timed_out.is_set() and guard.cancel_requested():
guard.close_attempt_stream("late cancelled attempt stream close failed")
try:
# Some Codex-compatible hosts accept ``stream=True`` but return a completed
# Responses object (not iterable) — don't hand it to the consumer.
if hasattr(event_stream, "output"):
final = event_stream
else:
final = _consume_codex_event_stream(
event_stream, model=issuer_model, on_event=guard.on_event
)
finally:
guard.release_stream(event_stream)
if final is None:
raise RuntimeError("Codex auxiliary Responses stream did not return a final response")
from agent.auxiliary_codex_response import _parse_codex_final_response
text_parts, tool_calls_raw, usage, finish_reason = _parse_codex_final_response(
final, issuer_kind=issuer_kind, issuer_model=issuer_model,
)
# Undo only the aliases THIS request emitted, before the call reaches Hermes dispatch.
for tc in tool_calls_raw or ():
if tc.function.name in wire_aliases:
tc.function.name = wire_aliases[tc.function.name]
except Exception as exc:
if not guard.timed_out.is_set():
logger.debug("Codex auxiliary Responses API call failed: %s", exc)
raise
deadline_hit = True
finally:
guard.finish()
if deadline_hit: # raised outside the handler, with no context: the deadline is the verdict
raise TimeoutError(guard.timeout_message())
# Shape the result like chat.completions.
message = SimpleNamespace(
role="assistant", content="".join(text_parts).strip() or None,
tool_calls=tool_calls_raw or None,
)
choice = SimpleNamespace(index=0, message=message, finish_reason=finish_reason)
return SimpleNamespace(choices=[choice], model=model, usage=usage)
class _ChatShim:
"""Exposes ``client.chat.completions.create()`` over a sync or async adapter."""
def __init__(self, adapter: Any):
self.completions = adapter
class _AsyncCompletionsAdapter:
"""Async adapter: runs the sync adapter's ``create`` via asyncio.to_thread()."""
def __init__(self, sync_adapter: Any):
self._sync = sync_adapter
async def create(self, **kwargs) -> Any:
import asyncio
return await asyncio.to_thread(self._sync.create, **kwargs)
class _AsyncAuxiliaryClientBase:
"""Async-compatible wrapper matching AsyncOpenAI.chat.completions.create().
Mirrors ``_real_client`` (when the sync wrapper has one) so cache eviction by
leaf OpenAI client drops this async entry too instead of reusing a closed transport.
"""
def __init__(self, sync_wrapper: Any):
self.chat = _ChatShim(_AsyncCompletionsAdapter(sync_wrapper.chat.completions))
self.api_key = sync_wrapper.api_key
self.base_url = sync_wrapper.base_url
if hasattr(sync_wrapper, "_real_client"):
# Mirror the sync wrapper's _real_client so cache eviction by leaf OpenAI client (e.g.
# _close_client_on_timeout in #23482) drops this async entry too. Without this, sync and async
# cache entries diverge on poisoning: the sync entry is evicted but the async entry keeps
# reusing the closed transport, failing every subsequent async aux call with 'Connection error'
# until the gateway restarts.
self._real_client = sync_wrapper._real_client
_AsyncAnthropicCompletionsAdapter = _AsyncCompletionsAdapter # imported by tests
class CodexAuxiliaryClient:
"""OpenAI-client-compatible wrapper routing through the Codex Responses API (.api_key/.base_url for introspection)."""
def __init__(self, real_client: OpenAI, model: str):
self._real_client = real_client
self.chat = _ChatShim(_CodexCompletionsAdapter(real_client, model))
self.api_key = real_client.api_key
self.base_url = real_client.base_url
def close(self):
self._real_client.close()
class AsyncCodexAuxiliaryClient(_AsyncAuxiliaryClientBase):
pass
def _translate_anthropic_response_format(anthropic_kwargs: dict[str, Any], response_format: Any) -> None:
"""Merge an OpenAI response format into Anthropic ``output_config``."""
if not isinstance(response_format, dict):
return
format_type = response_format.get("type")
if format_type == "json_schema":
json_schema = response_format.get("json_schema")
if not isinstance(json_schema, dict) or "schema" not in json_schema:
return
schema = json_schema["schema"]
elif format_type == "json_object":
# Anthropic SDK has no schema-less JSON mode; only ``json_schema``.
schema = {"type": "object"}
else:
return
output_config = anthropic_kwargs.get("output_config")
if not isinstance(output_config, dict):
output_config = {}
anthropic_kwargs["output_config"] = output_config
output_config["format"] = {"type": "json_schema", "schema": schema}
class _AnthropicCompletionsAdapter:
"""OpenAI-client-compatible adapter for Anthropic Messages API."""
def __init__(self, real_client: Any, model: str, is_oauth: bool = False, base_url: str | None = None):
self._client = real_client
self._model = model
self._is_oauth = is_oauth
# Caller URL first; fall back to the SDK client's host only for Nous Portal — a blanket
# fallback would flip MiniMax/Zhipu aux adapters to third-party handling (strips thinking sigs).
self._base_url = base_url or None
if not self._base_url:
candidate = str(getattr(real_client, "base_url", "") or "") or None
if candidate:
with contextlib.suppress(Exception):
from agent.anthropic_endpoints import _is_nous_portal_endpoint
if _is_nous_portal_endpoint(candidate):
self._base_url = candidate
def create(self, **kwargs) -> Any:
from agent.anthropic_adapter import build_anthropic_kwargs, create_anthropic_message
from agent.transports import get_transport
model = kwargs.get("model", self._model)
# ZAI's Anthropic endpoint rejects max_tokens on vision models (code 1210);
# callers signal this via _skip_zai_max_tokens.
if kwargs.pop("_skip_zai_max_tokens", False):
max_tokens = None
else:
max_tokens = kwargs.get("max_tokens") or kwargs.get("max_completion_tokens")
temperature = kwargs.get("temperature")
# Reasoning priority: explicit per-call _reasoning_config (MoA per-slot) wins over
# extra_body.reasoning; build_anthropic_kwargs translates to ``thinking``.
reasoning_cfg = kwargs.get("_reasoning_config")
if reasoning_cfg is None:
_eb = kwargs.get("extra_body")
_rc = _eb.get("reasoning") if isinstance(_eb, dict) else None
if isinstance(_rc, dict):
reasoning_cfg = _rc
# OpenAI tool_choice (str or dict) → Anthropic-style name/mode string.
tool_choice = kwargs.get("tool_choice")
if isinstance(tool_choice, dict):
choice_type = str(tool_choice.get("type", "")).lower()
if choice_type == "function":
tool_choice = tool_choice.get("function", {}).get("name")
else:
tool_choice = choice_type if choice_type in {"auto", "required", "none"} else None
elif not isinstance(tool_choice, str):
tool_choice = None
anthropic_kwargs = build_anthropic_kwargs(
model=model, messages=kwargs.get("messages", []), tools=kwargs.get("tools"),
max_tokens=max_tokens, reasoning_config=reasoning_cfg, tool_choice=tool_choice,
is_oauth=self._is_oauth,
# Portal routes on ``anthropic/<slug>`` ids and replays signed thinking
# keyed off base_url; omitting it breaks Portal model resolution.
base_url=self._base_url,
)
# Opus 4.7+ rejects non-default temperature/top_p/top_k; build_anthropic_kwargs
# also strips these as a safety net — keep both layers.
if temperature is not None:
from agent.anthropic_adapter import _forbids_sampling_params
if not _forbids_sampling_params(model):
anthropic_kwargs["temperature"] = temperature
# Per-request HTTP headers (OpenCode session affinity); Messages create/stream accept them too.
if isinstance(kwargs.get("extra_headers"), dict) and kwargs["extra_headers"]:
anthropic_kwargs["extra_headers"] = {
**(anthropic_kwargs.get("extra_headers") or {}),
**kwargs["extra_headers"],
}
# response_format: top-level gets the same translation as the extra_body form; when both
# are present the extra_body form wins. Passthrough excludes ``reasoning``/``response_format``
# (already TRANSLATED to native fields — raw would 400 on strict gateways) and ``_`` Hermes plumbing.
# The Messages body is built from a fixed allow-list of kwargs (``timeout`` included, below): an
# unlisted kwarg is dropped, so a schema contract became prompt compliance (#85626 review, point 2).
top_level_response_format = kwargs.get("response_format")
if top_level_response_format is not None:
_translate_anthropic_response_format(anthropic_kwargs, top_level_response_format)
caller_extra_body = kwargs.get("extra_body")
if caller_extra_body and isinstance(caller_extra_body, dict):
_translate_anthropic_response_format(anthropic_kwargs, caller_extra_body.get("response_format"))
passthrough = {
k: v for k, v in caller_extra_body.items()
if k not in {"reasoning", "response_format"} and not str(k).startswith("_")
}
if passthrough:
existing = anthropic_kwargs.get("extra_body") or {}
if not isinstance(existing, dict):
existing = {}
anthropic_kwargs["extra_body"] = {**existing, **passthrough}
if isinstance(kwargs.get("timeout"), (int, float)) and kwargs["timeout"] > 0: # else the client's 900 s read
anthropic_kwargs["timeout"] = kwargs["timeout"]
response = create_anthropic_message(
self._client,
anthropic_kwargs,
# Record provider-response timing every event, but tick forward progress only for
# substantive payloads so keepalives can't hold a stalled summary open. None keeps
# the fast get_final_message path.
on_stream_event=(_anthropic_aux_stream_event_hook() if _aux_progress_active() else None),
)
_nr = get_transport("anthropic_messages").normalize_response(response, strip_tool_prefix=self._is_oauth)
usage = None
if hasattr(response, "usage") and response.usage:
prompt_tokens = getattr(response.usage, "input_tokens", 0) or 0
completion_tokens = getattr(response.usage, "output_tokens", 0) or 0
usage = SimpleNamespace(
prompt_tokens=prompt_tokens, completion_tokens=completion_tokens,
total_tokens=getattr(response.usage, "total_tokens", 0) or (prompt_tokens + completion_tokens),
)
# ToolCall already duck-types as OpenAI shape via properties.
choice = SimpleNamespace(
index=0,
message=SimpleNamespace(content=_nr.content, tool_calls=_nr.tool_calls, reasoning=_nr.reasoning),
finish_reason=_nr.finish_reason,
)
return SimpleNamespace(choices=[choice], model=model, usage=usage)
class AnthropicAuxiliaryClient:
"""OpenAI-client-compatible wrapper over a native Anthropic client."""
def __init__(self, real_client: Any, model: str, api_key: str, base_url: str, is_oauth: bool = False):
self._real_client = real_client
self.chat = _ChatShim(_AnthropicCompletionsAdapter(real_client, model, is_oauth=is_oauth, base_url=base_url))
self.api_key = api_key
self.base_url = base_url
def close(self):
close_fn = getattr(self._real_client, "close", None)
if callable(close_fn):
close_fn()
class AsyncAnthropicAuxiliaryClient(_AsyncAuxiliaryClientBase):
pass
class _BedrockCompletionsAdapter:
"""Translates ``chat.completions.create(**kwargs)`` into Bedrock Converse."""
def __init__(self, region: str, model: str):
self._region = region
self._model = model
def create(self, **kwargs) -> Any:
from agent.bedrock_adapter import call_converse
model = kwargs.get("model", self._model)
max_tokens = kwargs.get("max_tokens") or kwargs.get("max_completion_tokens")
# OpenAI accepts ``stop`` as str or list; Converse requires a list.
stop = kwargs.get("stop")
if isinstance(stop, str):
stop = [stop]
if kwargs.get("tool_choice") is not None:
# Converse toolChoice isn't wired through call_converse(); surface the drop.
logger.debug(
"BedrockAuxiliaryClient: tool_choice=%r not supported by the "
"Converse shim — ignored.", kwargs.get("tool_choice"),
)
if kwargs.get("stream"):
# Converse streaming isn't wired here; call_llm's streaming consumer
# detects a final object and downgrades to non-live output.
logger.debug(
"BedrockAuxiliaryClient: stream=True requested for %s — returning a complete response "
"(Converse shim does not stream); caller downgrades to non-streaming.", model,
)
response = call_converse(
region=self._region, model=model, messages=kwargs.get("messages", []), tools=kwargs.get("tools"),
# Converse specifically defaults to the model maximum when omitted.
# Truthiness mirrors the Anthropic shim: explicit 0 means omit.
max_tokens=int(max_tokens) if max_tokens else None, temperature=kwargs.get("temperature"),
top_p=kwargs.get("top_p"), stop_sequences=stop,
)
# Converse is complete-response here: mark provider progress only after
# return so TTFP reflects real Bedrock latency, not dispatch/setup.
_notify_aux_provider_response()
return response
class BedrockAuxiliaryClient:
"""OpenAI-client-compatible wrapper over AWS Bedrock Converse API."""
def __init__(self, region: str, model: str):
self._region = region
self._model = model
self.chat = _ChatShim(_BedrockCompletionsAdapter(region, model))
self.api_key = "aws-sdk"
self.base_url = f"https://bedrock-runtime.{region}.amazonaws.com"
def close(self):
pass
class AsyncBedrockAuxiliaryClient(_AsyncAuxiliaryClientBase):
pass
def _endpoint_speaks_anthropic_messages(base_url: str) -> bool:
"""True if ``base_url`` speaks Anthropic Messages, not OpenAI chat.completions.
Mirrors ``hermes_cli.runtime_provider._detect_api_mode_for_url`` so aux and main agree: any
``/anthropic`` URL (MiniMax, Zhipu, LiteLLM), ``api.kimi.com/coding`` (chat 404s), ``api.anthropic.com``.
"""
normalized = (base_url or "").strip().lower().rstrip("/")
if not normalized:
return False
if urlparse(normalized).path.rstrip("/").endswith(("/anthropic", "/anthropic/v1")):
return True
hostname = base_url_hostname(normalized)
return hostname == "api.anthropic.com" or bool(hostname == "api.kimi.com" and "/coding" in normalized)
def _maybe_wrap_anthropic(
client_obj: Any, model: str, api_key: str, base_url: str, api_mode: Optional[str] = None
) -> Any:
"""Rewrap a plain OpenAI client in ``AnthropicAuxiliaryClient`` when the endpoint speaks Anthropic Messages.
Single transport-correction chokepoint at the end of every ``resolve_provider_client`` branch; returns
``client_obj`` unchanged for probe stubs/specialized adapters, OpenAI-wire, explicit non-Anthropic
``api_mode``, or missing ``anthropic`` SDK.
"""
# Anthropic/Bedrock/Codex wrappers, plus any client declaring HERMES_SKIP_TRANSPORT_WRAP
# (native/ACP shims, in-tree or plugin), must never be re-dispatched through a wire adapter —
# a class-attribute declaration rather than isinstance so this hot path never imports them.
if (
isinstance(client_obj, _AuxProbeClientStub)
or _safe_isinstance(client_obj, (AnthropicAuxiliaryClient, BedrockAuxiliaryClient, CodexAuxiliaryClient))
or _client_declares(client_obj, "HERMES_SKIP_TRANSPORT_WRAP")
):
return client_obj
# Explicit non-anthropic api_mode wins over URL heuristics.
if api_mode != "anthropic_messages" and (api_mode or not _endpoint_speaks_anthropic_messages(base_url)):
return client_obj
try:
from agent.anthropic_adapter import build_anthropic_client
except ImportError:
logger.warning(
"Endpoint %s speaks Anthropic Messages but the anthropic SDK is "
"not installed — falling back to OpenAI-wire (will likely 404).",
base_url,
)
return client_obj
try:
real_client = build_anthropic_client(api_key, base_url)
except Exception as exc:
logger.warning(
"Failed to build Anthropic client for %s (%s) — falling back to "
"OpenAI-wire client.", base_url, exc,
)
return client_obj
logger.debug(
"Auxiliary transport: wrapping client in AnthropicAuxiliaryClient "
"(model=%s, base_url=%s, api_mode=%s)",
model, base_url[:60] if base_url else "", api_mode or "auto-detected",
)
return AnthropicAuxiliaryClient(real_client, model, api_key, base_url, is_oauth=False)
def _read_nous_auth() -> Optional[dict]:
"""Nous provider state dict from the credential pool or ~/.hermes/auth.json; None when not active with tokens."""
pool_present, entry = _select_pool_entry("nous")
if pool_present:
if entry is None:
return None
return {
"access_token": getattr(entry, "access_token", ""),
"refresh_token": getattr(entry, "refresh_token", None),
"agent_key": getattr(entry, "agent_key", None),
"inference_base_url": _pool_runtime_base_url(entry, _NOUS_DEFAULT_BASE_URL),
"portal_base_url": getattr(entry, "portal_base_url", None),
"client_id": getattr(entry, "client_id", None),
"scope": getattr(entry, "scope", None),
"token_type": getattr(entry, "token_type", "Bearer"),
"source": "pool",
}
try:
auth_path = _auth_json_path()
if not auth_path.is_file():
return None
data = json.loads(auth_path.read_text(encoding="utf-8-sig"))
if data.get("active_provider") != "nous":
return None
provider = data.get("providers", {}).get("nous", {})
# Must have at least an access_token or agent_key.
if not provider.get("agent_key") and not provider.get("access_token"):
return None
return provider
except Exception as exc:
logger.debug("Could not read Nous auth: %s", exc)
return None
def _nous_api_key(provider: dict) -> str:
"""Extract a usable Nous inference JWT from stored auth state."""
from hermes_cli.auth import _nous_invoke_jwt_is_usable
for token_key, expiry_key in (("agent_key", "agent_key_expires_at"), ("access_token", "expires_at")):
token = provider.get(token_key)
if not isinstance(token, str) or not token.strip():
continue
if _nous_invoke_jwt_is_usable(token, scope=provider.get("scope"), expires_at=provider.get(expiry_key)):
return token
return ""
def _resolve_nous_pool_runtime_api(*, force_refresh: bool = False) -> Optional[tuple[str, str]]:
"""Resolve Nous auxiliary credentials from the selected pool entry."""
try:
from hermes_cli.auth import _agent_key_is_usable
pool = load_pool("nous")
except Exception as exc:
logger.debug("Auxiliary Nous pool credential resolution failed: %s", exc)
return None
if not pool or not pool.has_credentials():
return None
try:
entry = pool.select()
except Exception as exc:
logger.debug("Auxiliary Nous pool selection failed: %s", exc)
return None
if entry is None:
return None
def _entry_state(e: Any) -> dict[str, Any]:
return {k: getattr(e, k, None) for k in (
"agent_key", "agent_key_expires_at", "access_token", "expires_at", "scope")}
if force_refresh or not _agent_key_is_usable(_entry_state(entry), _nous_min_key_ttl_seconds()):
try:
refreshed = pool.try_refresh_current()
except Exception as exc:
logger.debug("Auxiliary Nous pool refresh failed: %s", exc)
refreshed = None
if refreshed is None:
return None
entry = refreshed
api_key = _nous_api_key(_entry_state(entry))
base_url = _pool_runtime_base_url(entry, _NOUS_DEFAULT_BASE_URL)
if not api_key or not base_url:
return None
return api_key, base_url
def _resolve_nous_runtime_api(
*, force_refresh: bool = False, stale_access_token: Optional[str] = None
) -> Optional[tuple[str, str]]:
"""Fresh Nous runtime credentials (pool first, then auth store + JWT refresh) — mirrors the main
agent's 401 recovery. ``stale_access_token`` is the bearer that just 401'd; with ``force_refresh``
it lets the auth store adopt a sibling process's rotation instead of re-POSTing the shared grant."""
pooled = _resolve_nous_pool_runtime_api(force_refresh=force_refresh)
if pooled is not None:
return pooled
try:
from hermes_cli.auth import resolve_nous_runtime_credentials
creds = resolve_nous_runtime_credentials(
timeout_seconds=env_float("HERMES_NOUS_TIMEOUT_SECONDS", 15),
force_refresh=force_refresh,
stale_access_token=stale_access_token or None,
)
except Exception as exc:
# Kept at WARNING (once per message) and remembered: the ladder falls back silently, and
# without this the goal judge only ever saw "judge error: RuntimeError" (#42177).
record_nous_credential_failure(exc)
return None
clear_nous_credential_failure()
return _creds_pair(creds)
def _creds_pair(creds: dict[str, Any]) -> Optional[tuple[str, str]]:
"""``(api_key, base_url)`` from a runtime-credentials dict, or None when either is missing."""
api_key = str(creds.get("api_key") or "").strip()
base_url = str(creds.get("base_url") or "").strip().rstrip("/")
if not api_key or not base_url:
return None
return api_key, base_url
def _resolve_xai_oauth_for_aux(model: Optional[str] = None) -> Optional[tuple[str, str]]:
"""Fresh xAI OAuth (api_key, base_url) for aux clients, or None.
Pool first (some xAI OAuth logins exist only as pool entries), then the singleton auth-store resolver.
"""
try:
from hermes_cli.auth import DEFAULT_XAI_OAUTH_BASE_URL, _xai_validate_inference_base_url
pool = load_pool("xai-oauth")
if pool and pool.has_credentials():
entry = pool.select(model=model)
if entry is not None:
api_key = str(
getattr(entry, "runtime_api_key", None) or getattr(entry, "access_token", "") or ""
).strip()
_url = lambda v: str(v or "").strip().rstrip("/")
base_url = _xai_validate_inference_base_url(
_url(_scoped_key_env("HERMES_XAI_BASE_URL"))
or _url(_scoped_key_env("XAI_BASE_URL"))
or _url(getattr(entry, "runtime_base_url", None))
or _url(getattr(entry, "base_url", None)),
fallback=DEFAULT_XAI_OAUTH_BASE_URL,
)
if api_key and base_url:
return api_key, base_url
except Exception as exc:
logger.debug("Auxiliary xAI OAuth pool credential resolution failed: %s", exc)
try:
from hermes_cli.auth import resolve_xai_oauth_runtime_credentials
creds = resolve_xai_oauth_runtime_credentials()
except Exception as exc:
logger.debug("Auxiliary xAI OAuth runtime credential resolution failed: %s", exc)
return None
return _creds_pair(creds)
def _resolve_codex_credential_and_base(model: Optional[str] = None) -> tuple[Optional[str], str]:
"""``(token, base_url)`` taken from ONE authority, so a Codex key is only ever sent to the host
it belongs to (#121486): the profile-scoped ``HERMES_CODEX_BASE_URL`` wins; otherwise a pooled
key goes where that pool entry routes (row URL / ``model.base_url``) and the auth.json OAuth
token goes to the ChatGPT default. ``(None, <base>)`` without a usable token."""
override = _codex_base_url_override()
pool_present, entry = _select_pool_entry("openai-codex", model) # a model-scoped cooldown spares siblings
if pool_present:
token = _pool_runtime_api_key(entry)
if token:
if override:
return token, override
# Same route rule as the chat path (row URL, else ``model.base_url``); never empty.
from hermes_cli.auth_codex import _codex_pool_route_base_url
return token, _codex_pool_route_base_url(_pool_runtime_base_url(entry))
# No usable pool token: auth.json only (re-selecting could pair another row's key with the default).
return _read_codex_singleton_token(), override or _CODEX_AUX_BASE_URL
def _read_codex_singleton_token() -> Optional[str]:
"""The profile's auth.json Codex access token (expired JWTs skipped), else None."""
try:
from hermes_cli.auth import _read_codex_tokens
access_token = _read_codex_tokens().get("tokens", {}).get("access_token")
if not isinstance(access_token, str) or not access_token.strip():
return None
# Expired JWTs would block the auto chain and prevent fallback to working providers.
try:
import base64
payload = access_token.split(".")[1]
payload += "=" * (-len(payload) % 4)
exp = json.loads(base64.urlsafe_b64decode(payload)).get("exp", 0)
if exp and time.time() > exp:
logger.debug("Codex access token expired (exp=%s), skipping", exp)
return None
except Exception:
pass # Non-JWT token or decode error — use as-is
return access_token.strip()
except Exception as exc:
logger.debug("Could not read Codex auth for auxiliary client: %s", exc)
return None
def _resolve_api_key_provider() -> tuple[Optional[OpenAI], Optional[str]]:
"""Try each API-key provider in PROVIDER_REGISTRY order; (client, model) or (None, None)."""
try:
from hermes_cli.auth import PROVIDER_REGISTRY, resolve_api_key_provider_credentials
except ImportError:
logger.debug("Could not import PROVIDER_REGISTRY for API-key fallback")
return None, None
from hermes_cli.auth_plugin_providers import canonical_registry_rows
for provider_id, pconfig in canonical_registry_rows():
if pconfig.auth_type != "api_key":
continue
if _is_provider_unhealthy(provider_id):
logger.debug("Auxiliary api-key chain: %s is unhealthy, skipping", provider_id)
continue
if provider_id == "anthropic":
# Explicit-config gate: Claude Code credentials must not silently become aux fallback.
with contextlib.suppress(ImportError):
from hermes_cli.auth import is_provider_explicitly_configured
if not is_provider_explicitly_configured("anthropic"):
continue
client, model = _try_anthropic()
return _tag_backend(client, "anthropic"), model
from providers import get_provider_profile
if getattr(get_provider_profile(provider_id), "ambient_credentials", False):
# Explicit-config gate: ambient credentials (another tool's login) must not silently become aux fallback (#114740).
from hermes_cli.auth import is_provider_explicitly_configured
if not is_provider_explicitly_configured(provider_id):
continue
model = _get_aux_model_for_provider(provider_id) or None
if model is None:
continue # skip provider if we don't know a valid aux model
pool_present, entry = _select_pool_entry(provider_id, model)
if pool_present:
api_key = _pool_runtime_api_key(entry)
if not api_key:
continue
raw_base_url = _pool_runtime_base_url(entry, pconfig.inference_base_url) or pconfig.inference_base_url
via = " via pool"
else:
creds = resolve_api_key_provider_credentials(provider_id)
api_key = str(creds.get("api_key", "")).strip()
if not api_key:
continue
raw_base_url = str(creds.get("base_url", "")).strip().rstrip("/") or pconfig.inference_base_url
via = ""
# The session's own endpoint wins for its provider: the key was issued for that gateway, and
# sending it to the registry default 401s, then quarantines the provider the main model is on.
runtime = _normalize_main_runtime(None)
if runtime.get("provider") == provider_id and runtime.get("base_url"):
raw_base_url = runtime["base_url"].rstrip("/")
if isinstance(runtime.get("api_key"), str) and runtime["api_key"]:
api_key = runtime["api_key"]
via = " (session endpoint)"
logger.debug("Auxiliary text client: %s (%s)%s", pconfig.name, model, via)
# Native Gemini, else OpenAI-wire + Anthropic rewrap.
base_url = _to_openai_base_url(raw_base_url)
if provider_id == "gemini":
from agent.gemini_native_adapter import GeminiNativeClient, is_native_gemini_base_url
if is_native_gemini_base_url(base_url):
return _tag_backend(GeminiNativeClient(api_key=api_key, base_url=base_url), provider_id), model
if base_url_host_matches(base_url, "api.kimi.com"):
headers = {"User-Agent": "claude-code/0.1.0"}
elif base_url_host_matches(base_url, "integrate.api.nvidia.com"):
headers = build_nvidia_nim_headers(base_url)
else:
headers = _route_client_headers(base_url, provider_id) or _profile_default_headers(provider_id)
extra = {"default_headers": headers} if headers else {}
merged = _apply_user_default_headers(extra.get("default_headers"))
if merged:
extra["default_headers"] = merged
client = _create_openai_client(api_key=api_key, base_url=base_url, **extra)
return _tag_backend(_maybe_wrap_anthropic(client, model, api_key, raw_base_url), provider_id), model
return None, None
def _tag_backend(client: Any, provider: str) -> Any:
"""Name the provider a discovery client was built for. Failures classify there; the request itself
keeps the route label's wire shape (``_hermes_aux_effective_provider`` stays the label)."""
if client is not None and isinstance(provider, str) and provider:
client._hermes_aux_backend_provider = provider
return client
def _route_client_headers(base_url: str, provider: str, *, is_vision: bool = False) -> Optional[dict]:
from agent.provider_route_hooks import route_answer
headers = route_answer("client_headers", provider=provider or "", base_url=base_url, has_images=is_vision)
return dict(headers) if headers else None
def _endpoint_default_headers(
base_url: str, provider: str, *, is_vision: bool = False, xai: bool = False,
) -> Optional[dict]:
"""Provider-specific client headers by endpoint host, merged with user ``model.default_headers``.
Kimi Code needs the claude-code User-Agent; a profile's ``client_headers`` (``is_vision`` = image
client), NVIDIA NIM and (optionally) xAI have their own; else the provider profile.
"""
if base_url_host_matches(base_url, "api.kimi.com"):
headers: dict = {"User-Agent": "claude-code/0.1.0"}
elif route := _route_client_headers(base_url, provider, is_vision=is_vision):
headers = route
elif base_url_host_matches(base_url, "integrate.api.nvidia.com"):
headers = dict(build_nvidia_nim_headers(base_url))
elif xai and base_url_host_matches(base_url, "x.ai"):
from tools.xai_http import hermes_xai_default_headers
headers = dict(hermes_xai_default_headers())
else:
headers = _profile_default_headers(provider) or {}
return _apply_user_default_headers(headers or None) or None
def _profile_default_headers(provider: str) -> Optional[dict]:
"""Client-level attribution headers from the provider profile (e.g. GMI User-Agent), or None."""
if not provider:
return None
with contextlib.suppress(Exception):
from providers import get_provider_profile
profile = get_provider_profile(provider)
if profile and profile.default_headers:
return dict(profile.default_headers)
return None
# Provider resolution helpers
_paid_lane_warned: set = set()
def _is_free_model(model: Optional[str]) -> bool:
"""True when ``model`` is a free SKU (``:free`` suffix or ``stealth/`` prefix) — naming-convention trust."""
if not model:
return False
normalized = str(model).strip()
return normalized.endswith(":free") or normalized.startswith("stealth/")
def _aux_openrouter_settings() -> tuple[bool, str]:
"""Read (free_only, openrouter_model) from config; (False, _OPENROUTER_MODEL) on failure."""
try:
from hermes_cli.config import cfg_get, load_config_readonly
cfg = load_config_readonly()
free_only = bool(cfg_get(cfg, "auxiliary", "free_only", default=False))
val = cfg_get(cfg, "auxiliary", "openrouter_model")
model = val.strip() if isinstance(val, str) and val.strip() else _OPENROUTER_MODEL
return free_only, model
except Exception:
return False, _OPENROUTER_MODEL
def _warn_paid_lane_once(model: str) -> None:
"""Log a WARNING the first time a non-free OpenRouter model is engaged."""
if model in _paid_lane_warned:
return
_paid_lane_warned.add(model)
logger.warning(
"Auxiliary client: PAID lane engaged for auxiliary task — OpenRouter fallback model %r is not "
"a :free SKU and may incur real spend. Set auxiliary.free_only: true to restrict auxiliary "
"fallbacks to free models, or auxiliary.openrouter_model to a :free model.", model,
)
def _try_openrouter(explicit_api_key: Optional[str | Callable[[], str]] = None, model: str | None = None,
explicit_base_url: Optional[str] = None) -> tuple[Optional[OpenAI], Optional[str]]:
free_only, cfg_model = _aux_openrouter_settings()
or_model = model or cfg_model
if free_only and not _is_free_model(or_model):
logger.warning(
"Auxiliary client: auxiliary.free_only is enabled but the OpenRouter fallback model %r is "
"not a :free SKU — skipping the OpenRouter fallback. Set auxiliary.openrouter_model to a "
":free model (e.g. nvidia/nemotron-3-ultra-550b-a55b:free) or disable auxiliary.free_only.",
or_model,
)
return None, None
if not _is_free_model(or_model):
_warn_paid_lane_once(or_model)
# A caller-supplied endpoint (fallback_providers entry, custom_providers entry) is
# authoritative over both the pool row and the canonical host (#121359).
override_url = (explicit_base_url or "").strip().rstrip("/")
pool_present, entry = _select_pool_entry("openrouter", or_model)
if pool_present:
or_key = explicit_api_key or _pool_runtime_api_key(entry)
if or_key:
base_url = override_url or _pool_runtime_base_url(entry, OPENROUTER_BASE_URL) or OPENROUTER_BASE_URL
logger.debug("Auxiliary client: OpenRouter via pool")
return _create_openai_client(
api_key=or_key, base_url=base_url, default_headers=build_or_headers()
), or_model
# Exhausted pool: fall through to OPENROUTER_API_KEY rather than fail.
logger.debug("Auxiliary client: OpenRouter pool exhausted, trying OPENROUTER_API_KEY")
or_key = explicit_api_key or _scoped_key_env("OPENROUTER_API_KEY")
if not or_key:
_mark_provider_unhealthy(
"openrouter", ttl=60, reason=_describe_openrouter_unavailable(or_model), level=logging.DEBUG)
return None, None
logger.debug("Auxiliary client: OpenRouter")
return _create_openai_client(
api_key=or_key, base_url=override_url or OPENROUTER_BASE_URL, default_headers=build_or_headers()
), or_model
def _describe_openrouter_unavailable(model: str | None = None) -> str:
"""Return the policy or credential reason OpenRouter was unavailable."""
free_only, cfg_model = _aux_openrouter_settings()
or_model = model or cfg_model
if free_only and not _is_free_model(or_model):
return (
f"auxiliary.free_only rejected non-free model {or_model!r}; "
"the request was skipped before provider availability checks"
)
pool_present, entry = _select_pool_entry("openrouter")
if pool_present:
if entry is None:
return "OpenRouter credential pool has no usable entries (credentials may be exhausted)"
if not _pool_runtime_api_key(entry):
return "OpenRouter credential pool entry is missing a runtime API key"
if not _scoped_key_env("OPENROUTER_API_KEY"):
return "OPENROUTER_API_KEY not set"
return "no usable OpenRouter credentials found"
def _try_nous(vision: bool = False) -> tuple[Optional[OpenAI], Optional[str]]:
nous = _read_nous_auth()
runtime = _resolve_nous_runtime_api(force_refresh=False)
if runtime is None and not nous:
logger.warning("Auxiliary Nous client unavailable: no Nous authentication found (run: hermes auth).")
_mark_provider_unhealthy("nous", ttl=60, reason="no Nous authentication found", level=logging.DEBUG)
return None, None
if runtime is None and nous:
logger.debug("Auxiliary Nous: runtime JWT refresh failed; checking stored auth.json token.")
if runtime is not None:
api_key, base_url = runtime
else:
api_key = _nous_api_key(nous or {})
if not api_key:
logger.warning(
"Auxiliary Nous client unavailable: no usable inference JWT found "
"(run: hermes auth add nous)."
)
_mark_provider_unhealthy("nous", ttl=60, reason="no usable Nous inference JWT", level=logging.DEBUG)
return None, None
base_url = str(
(nous or {}).get("inference_base_url") or _scoped_key_env("NOUS_INFERENCE_BASE_URL") or _NOUS_DEFAULT_BASE_URL
).rstrip("/")
with contextlib.suppress(Exception):
from agent.nous_rate_guard import nous_rate_limit_remaining
from hermes_cli.anon_auth import is_anonymous_request
anonymous = is_anonymous_request("nous", api_key)
remaining = nous_rate_limit_remaining(anonymous=anonymous)
if remaining is not None and remaining > 0:
logger.debug("Auxiliary: skipping Nous Portal (rate-limited, resets in %.0fs)", remaining)
# The health marker is provider-wide, so a full-length anonymous cooldown would
# outlive signing in mid-cooldown; bound it instead of re-resolving credentials
# (auth store lock, pool read) on every auxiliary call for the cooldown's duration.
_mark_provider_unhealthy(
"nous", ttl=min(remaining, 60.0) if anonymous else remaining,
reason="Nous Portal rate-limited", level=logging.INFO)
return None, None
lane = "vision" if vision else "text"
# The free tier's host serves exactly one model, for every lane: asking it for the Portal's
# recommended aux model is a guaranteed 429 ``model_not_free``. Pin the route's model instead.
# Vision rides the same id (the backing model is multimodal; a backing that is not answers
# the request with the upstream's own error, which the ladder handles like any other).
from hermes_cli.anon_auth import GUEST_MODEL, route_is_welcome_host
if route_is_welcome_host(base_url):
logger.debug("Auxiliary/%s: Nous free tier; using %s", lane, GUEST_MODEL)
return _create_openai_client(api_key=api_key, base_url=base_url), GUEST_MODEL
logger.debug("Auxiliary client: Nous Portal")
# Portal recommended-models is authoritative (tier-aware); _NOUS_MODEL when unreachable/null.
# Probes skip the lookup: exact model is irrelevant and it hits the network.
model = _NOUS_MODEL
if not _aux_probe_active():
try:
from hermes_cli.models import get_nous_recommended_aux_model
recommended = get_nous_recommended_aux_model(vision=vision)
if recommended:
model = recommended
logger.debug("Auxiliary/%s: using Portal-recommended model %s", lane, model)
else:
logger.debug("Auxiliary/%s: no Portal recommendation, falling back to %s", lane, model)
except Exception as exc:
logger.debug(
"Auxiliary/%s: recommended-models lookup failed (%s); "
"falling back to %s",
lane, exc, model,
)
return _create_openai_client(api_key=api_key, base_url=base_url), model
def _refresh_nous_recommended_model(*, vision: bool, stale_model: Optional[str]) -> Optional[str]:
"""Fresh Portal recommended model after a stale-model 404 (long-lived processes pin dropped models).
Returns the fresh recommendation, else ``_NOUS_MODEL``, whichever differs from ``stale_model``; None if neither.
"""
stale = (stale_model or "").strip().lower()
fresh: Optional[str] = None
try:
from hermes_cli.models import get_nous_recommended_aux_model
fresh = get_nous_recommended_aux_model(vision=vision, force_refresh=True)
except Exception as exc:
logger.debug("Nous recommended-model refresh failed (%s); using default %s", exc, _NOUS_MODEL)
if fresh and fresh.strip().lower() != stale:
return fresh
return _NOUS_MODEL if _NOUS_MODEL.strip().lower() != stale else None
def _read_main_field(field: str, *, readonly: bool, lower: bool = False) -> str:
"""Main ``model.<field>``: runtime override (``set_runtime_main``) first, then config.yaml.
The override wins so "active main model" gates see the live CLI/gateway runtime, not the persisted
default. ``readonly`` picks ``load_config_readonly`` (model/provider) vs ``load_config`` (api_key/base_url).
"""
override = _runtime_main_value(field)
if isinstance(override, str) and override.strip():
value = override.strip()
return value.lower() if lower else value
with contextlib.suppress(Exception):
from hermes_cli import config as _cfg_mod
cfg = (_cfg_mod.load_config_readonly if readonly else _cfg_mod.load_config)()
model_cfg = cfg.get("model", {})
if field == "model" and isinstance(model_cfg, str) and model_cfg.strip():
return model_cfg.strip()
if isinstance(model_cfg, dict):
value = model_cfg.get("default" if field == "model" else field, "")
if isinstance(value, str) and value.strip():
value = value.strip()
return value.lower() if lower else value
return ""
# Module-level callables (tests patch them): model/provider (lowercased) read the readonly config;
# api_key/base_url read the full config so ``custom`` aux tasks can inherit main creds.
_read_main_model = functools.partial(_read_main_field, "model", readonly=True)
_read_main_provider = functools.partial(_read_main_field, "provider", readonly=True, lower=True)
_read_main_api_key = functools.partial(_read_main_field, "api_key", readonly=False)
_read_main_base_url = functools.partial(_read_main_field, "base_url", readonly=False)
def _resolve_moa_aggregator(preset_name: Optional[str]) -> tuple[Optional[str], Optional[str]]:
"""MoA preset → aggregator (provider, model); (None, None) if unresolvable. None/"" = default preset.
"moa" is virtual — aux tasks skip the fan-out and use the aggregator slot; shared so lookup can't drift.
"""
try:
from hermes_cli.config import load_config
from hermes_cli.moa_config import resolve_moa_preset
preset = resolve_moa_preset(load_config().get("moa") or {}, preset_name or None)
agg = preset.get("aggregator") or {}
agg_provider = str(agg.get("provider") or "").strip()
agg_model = str(agg.get("model") or "").strip()
if agg_provider and agg_model and agg_provider.lower() != "moa":
return agg_provider, agg_model
except Exception:
logger.debug("MoA aggregator resolution failed for preset %r", preset_name, exc_info=True)
return None, None
def _read_main_model_for_aux() -> str:
"""Main model with MoA presets unwrapped to the aggregator's model; "" when unresolvable (a preset name would 400)."""
model = _read_main_model()
if (_read_main_provider() or "").strip().lower() == "moa":
_, agg_model = _resolve_moa_aggregator(model)
return agg_model or ""
return model
def _read_main_api_key_if_same_origin(aux_base_url: str) -> str | Callable[[], str]:
"""Main api_key only when *aux_base_url* has the main base_url's exact origin.
Unconditional inheritance would leak the credential to any misconfigured host; mismatch keeps ``no-key-required`` → 401.
Origin, not hostname: another scheme (``http://``) or port on the same host is a different endpoint.
Anchor and key come from ONE source: the live runtime a turn bound, else config.yaml. The
per-field readers fall back to config field by field, so a keyless or key_cmd live main would
pair its own base_url with config's key and send that key to the live endpoint.
Origins are compared before any key is read: the client cache calls this on every keyless
``custom`` lookup, and a mismatch must not pay for a config.yaml key read.
"""
aux_origin = base_url_origin(aux_base_url)
if not aux_origin[1]:
return ""
live = _normalize_main_runtime(None)
if live.get("base_url") or live.get("api_key"):
if aux_origin != base_url_origin(live.get("base_url", "")):
return ""
return live.get("api_key", "")
if aux_origin != base_url_origin(_read_main_base_url()):
return ""
return _read_main_api_key()
# The live main runtime, per context: overlapping gateway sessions make a process-global unsafe.
_RUNTIME_MAIN_CONTEXT: contextvars.ContextVar[Optional[dict[str, Any]]] = (
contextvars.ContextVar("auxiliary_runtime_main", default=None)
)
_RELAY_AUX_CALL_CONTEXT: contextvars.ContextVar[Optional[dict[str, Any]]] = (
contextvars.ContextVar("auxiliary_relay_call", default=None)
)
@contextlib.contextmanager
def _relay_aux_call_scope(args: tuple, kwargs: dict):
"""Bind a fresh relay call context for one auxiliary call; mark it failed on any exception."""
task = args[0] if args else kwargs.get("task")
token = _RELAY_AUX_CALL_CONTEXT.set({
"task": str(task or "unknown"),
"request_id": f"aux-{uuid.uuid4().hex}",
"attempt_count": 0,
"provider": "",
"model": "",
"response_model": None,
"api_mode": "chat_completions",
})
try:
yield
except BaseException as exc:
_fail_relay_auxiliary_call(exc)
raise
finally:
_RELAY_AUX_CALL_CONTEXT.reset(token)
def _relay_auxiliary_call(callback):
"""Give every physical retry in one auxiliary call a shared Relay identity."""
@functools.wraps(callback)
def wrapped(*args, **kwargs):
with _relay_aux_call_scope(args, kwargs):
return callback(*args, **kwargs)
return wrapped
def _relay_auxiliary_call_async(callback):
"""Async counterpart to :func:`_relay_auxiliary_call`."""
@functools.wraps(callback)
async def wrapped(*args, **kwargs):
with _relay_aux_call_scope(args, kwargs):
return await callback(*args, **kwargs)
return wrapped
def _set_relay_auxiliary_route(provider: str | None, model: str | None, api_mode: str | None) -> None:
context = _RELAY_AUX_CALL_CONTEXT.get()
if context is None:
return
context["provider"] = str(provider or "auxiliary")
context["model"] = str(model or "unknown")
context["response_model"] = None
context["api_mode"] = str(api_mode or "chat_completions")
def _record_route_info(
route_info: Optional[dict[str, str]], provider: Optional[str], model: Optional[str]
) -> None:
"""Expose the concrete route selected for one auxiliary call."""
if route_info is not None:
route_info["provider"] = provider or "auto"
route_info["model"] = model or "default"
def _relay_auxiliary_metadata(
*, provider: str | None = None, api_mode: str | None = None
) -> tuple[str, str, dict[str, Any]] | None:
context = _RELAY_AUX_CALL_CONTEXT.get()
if context is None:
return None
attempt_count = int(context.get("attempt_count") or 0)
context["attempt_count"] = attempt_count + 1
provider_name = str(provider or context.get("provider") or "auxiliary")
model_name = str(context.get("model") or "unknown")
return provider_name, model_name, {
"api_mode": str(api_mode or context.get("api_mode") or "chat_completions"),
"api_request_id": str(context["request_id"]),
"call_role": f"auxiliary:{context['task']}",
"retry_count": attempt_count,
"auxiliary_task": str(context["task"]),
}
def _provider_outcome(response: Any, error: Optional[Exception], judged: list) -> Any:
"""``response``, else raise the provider's own ``error``: a failure its attempt already classified (a judged
stream negotiation, a credit-limited 402) is unwrapped for Relay and the aux hooks and kept in ``judged``,
which the attempt boundary raises to recovery."""
if error is None:
return response
if isinstance(error, _ClassifiedAttempt):
judged.append(error)
error = error.failure.error
raise error
def _provider_callback(create: Callable[[dict[str, Any]], Any], judged: list) -> Callable[[dict[str, Any]], Any]:
def call(request: dict[str, Any]) -> Any:
judged.clear()
return _provider_outcome(*_outcome(create, request), judged)
return call
def _provider_callback_async(create: Callable[[dict[str, Any]], Any], judged: list) -> Callable[[dict[str, Any]], Any]:
async def call(request: dict[str, Any]) -> Any:
judged.clear()
return _provider_outcome(*await _aoutcome(create(request)), judged)
return call
def _relay_sync_completion(
client: Any, kwargs: dict[str, Any], *, provider: str | None = None,
api_mode: str | None = None, create: Callable[[dict[str, Any]], Any] | None = None,
) -> Any:
from agent.auxiliary_wire import prepare_chat_messages, relay_chat_metadata
kwargs = prepare_chat_messages(client, kwargs)
# The progress hook is installed per TASK, so every attempt (retries, recovery rungs, fallbacks)
# must stream through _create_with_progress or the compression watchdog sees silence (#98466).
# Recovery rungs / credential retries keep the task's ``no_progress_timeout`` window; the
# first-token window uses this attempt's provider (fallbacks name theirs) for its stale timeout.
relay_context = _RELAY_AUX_CALL_CONTEXT.get() or {}
task = relay_context.get("task")
relay_context["stream_provider"] = provider or relay_context.get("provider")
judged: list[_ClassifiedAttempt] = []
callback = _provider_callback(
create or (lambda request: _create_with_progress(client, request, task, provider=provider or "")), judged)
route = _relay_auxiliary_metadata(provider=provider, api_mode=api_mode)
# Isolate only the provider callback so the owning thread can unwind its lease/DB
# transaction on hard cancel without touching the shared client.
if route is None:
return _run_protected_sync_provider_call(callback, kwargs)
provider_name, fallback_model, metadata = route
from agent import relay_llm
from agent.auxiliary_hooks import run_with_aux_hooks
model_name = str(kwargs.get("model") or fallback_model)
try:
return run_with_aux_hooks(
lambda: relay_llm.execute_current(
kwargs, lambda request: _run_protected_sync_provider_call(callback, request),
name=provider_name, model_name=model_name, metadata=relay_chat_metadata(metadata),
defer_logical_completion=True,
),
aux_task=str(metadata.get("auxiliary_task") or ""), metadata=metadata, client=client, kwargs=kwargs,
provider=provider_name, model=model_name, api_mode=str(metadata.get("api_mode") or ""),
)
except Exception as exc:
if judged: # classified where it failed: recorded there, and Relay and the hooks saw the provider's error
raise judged[-1]
_note_relay_auxiliary_error(exc)
raise
async def _relay_async_completion(
client: Any, kwargs: dict[str, Any], *, provider: str | None = None,
api_mode: str | None = None, create: Callable[[dict[str, Any]], Any] | None = None,
) -> Any:
from agent.auxiliary_wire import prepare_chat_messages, relay_chat_metadata
kwargs = prepare_chat_messages(client, kwargs)
# Async twin of the seam default above (#98466).
judged: list[_ClassifiedAttempt] = []
callback = _provider_callback_async(
create or (lambda request: _acreate_with_progress(client, request, provider=provider or "")), judged)
route = _relay_auxiliary_metadata(provider=provider, api_mode=api_mode)
if route is None:
return await callback(kwargs)
provider_name, fallback_model, metadata = route
from agent import relay_llm
from agent.auxiliary_hooks import arun_with_aux_hooks
model_name = str(kwargs.get("model") or fallback_model)
try:
return await arun_with_aux_hooks(
lambda: relay_llm.execute_current_async(
kwargs, callback, name=provider_name, model_name=model_name,
metadata=relay_chat_metadata(metadata), defer_logical_completion=True,
),
aux_task=str(metadata.get("auxiliary_task") or ""), metadata=metadata, client=client, kwargs=kwargs,
provider=provider_name, model=model_name, api_mode=str(metadata.get("api_mode") or ""),
)
except Exception as exc:
if judged:
raise judged[-1]
_note_relay_auxiliary_error(exc)
raise
def _relay_sync_stream(
client: Any, kwargs: dict[str, Any], *, provider: str | None = None, api_mode: str | None = None
) -> Any:
from agent.auxiliary_wire import prepare_chat_messages, relay_chat_metadata
kwargs = prepare_chat_messages(client, kwargs)
# The bypass runs inside the provider callback, AFTER Relay has seen (and possibly
# rewritten) the real conversation; applying it to `kwargs` would hand Relay an empty one.
create = lambda request: client.chat.completions.create(**bypass_chat_sdk_request_transform(request, client))
route = _relay_auxiliary_metadata(provider=provider, api_mode=api_mode)
if route is None:
return create(kwargs)
provider_name, fallback_model, metadata = route
from agent import relay_llm
from agent.auxiliary_hooks import run_with_aux_hooks
model_name = str(kwargs.get("model") or fallback_model)
return run_with_aux_hooks(
lambda: relay_llm.stream_current(
kwargs, create, name=provider_name, model_name=model_name, finalizer=dict,
metadata=relay_chat_metadata(metadata), completed_response_predicate=lambda value: hasattr(value, "choices"),
),
aux_task=str(metadata.get("auxiliary_task") or ""), metadata=metadata, client=client, kwargs=kwargs,
provider=provider_name, model=model_name, api_mode=str(metadata.get("api_mode") or ""), streaming=True,
)
def _runtime_main_value(field: str) -> Any:
"""Read one field of the current context's main runtime ("" when unbound)."""
runtime = _RUNTIME_MAIN_CONTEXT.get()
return (runtime.get(field) or "") if isinstance(runtime, dict) else ""
def set_runtime_main(
provider: str, model: str, *, requested_provider: str = "", base_url: str = "",
api_key: Any = "", api_mode: str = "", auth_mode: str = "", session_id: str = "",
cache_scope: str = "",
) -> contextvars.Token:
"""Record the current context's live main runtime for auxiliary routing.
Context-local so concurrent gateway sessions don't clobber each other.
``cache_scope`` is the rotation-stable logical cache scope (compression- lineage root —
agent/prompt_cache_scope.py) resolved once per turn by turn_context; auxiliary Responses calls prefer it
over ``session_id`` for prompt_cache_key derivation (#79017).
"""
runtime = {
"provider": (provider or "").strip().lower(),
"requested_provider": (requested_provider or "").strip().lower(),
"model": (model or "").strip(),
"base_url": (base_url or "").strip(),
"api_key": _normalize_api_key(api_key),
"api_mode": (api_mode or "").strip(),
"auth_mode": (auth_mode or "").strip().lower(),
"session_id": (session_id or "").strip(),
"cache_scope": (cache_scope or "").strip(),
}
return _RUNTIME_MAIN_CONTEXT.set(runtime)
def reset_runtime_main(token: contextvars.Token) -> None:
"""Restore the runtime binding that preceded one scoped turn."""
if token is None:
return
try:
_RUNTIME_MAIN_CONTEXT.reset(token)
except (RuntimeError, ValueError):
pass # Tokens can't be reset from a copied Context (workers inherit values, not token ownership).
@contextlib.contextmanager
def scoped_runtime_main(main_runtime: Optional[dict[str, Any]]):
"""Temporarily bind an explicit runtime for the current context."""
runtime = _normalize_main_runtime(main_runtime)
token = _RUNTIME_MAIN_CONTEXT.set(runtime or None)
try:
yield runtime
finally:
_RUNTIME_MAIN_CONTEXT.reset(token)
def clear_runtime_main() -> None:
"""Clear the runtime override in the current context."""
_RUNTIME_MAIN_CONTEXT.set(None)
def _resolve_custom_runtime() -> tuple[Optional[str], Optional[str], Optional[str]]:
"""Resolve the active custom/main endpoint like the main CLI (env OPENAI_BASE_URL or config-saved)."""
try:
from hermes_cli.auth import AuthError
from hermes_cli.runtime_provider import resolve_runtime_provider
runtime = resolve_runtime_provider(requested="custom")
except AuthError as exc:
# Bare 'custom' with nothing configured fails fast in the main resolver: there is no
# custom endpoint, so do NOT fall through to a stale env OPENAI_BASE_URL that the main
# resolver deliberately never consults.
logger.debug("Auxiliary client: no custom endpoint configured: %s", exc)
return None, None, None
except Exception as exc:
logger.debug("Auxiliary client: custom runtime resolution failed: %s", exc)
runtime = None
if not isinstance(runtime, dict):
# Base URL is per-profile like the key one line below (a scoped key must not hit the default's proxy).
openai_base = _scoped_key_env("OPENAI_BASE_URL").rstrip("/")
if not openai_base:
return None, None, None
runtime = {"base_url": openai_base, "api_key": _scoped_key_env("OPENAI_API_KEY")}
custom_base = runtime.get("base_url")
custom_key = runtime.get("api_key")
custom_mode = runtime.get("api_mode")
if not isinstance(custom_base, str) or not custom_base.strip():
return None, None, None
custom_base = custom_base.strip().rstrip("/")
if base_url_host_matches(custom_base, "openrouter.ai"):
return None, None, None # requested='custom' falls back to OpenRouter when unconfigured.
# Local servers (Ollama, vLLM, ...) ignore auth but the SDK needs a non-empty key.
# Use a placeholder key — the OpenAI SDK requires a non-empty string but local servers ignore the
# Authorization header. Same fix as cli.py _ensure_runtime_credentials() (PR #2556).
if not isinstance(custom_key, str) or not custom_key.strip():
custom_key = "no-key-required"
if not isinstance(custom_mode, str) or not custom_mode.strip():
custom_mode = None
return custom_base, custom_key.strip(), custom_mode
def _current_custom_base_url() -> str:
custom_base, _, _ = _resolve_custom_runtime()
return custom_base or ""
def _validate_proxy_env_urls() -> None:
"""Fail fast on malformed proxy env URLs (a shell typo like ``:6153export`` otherwise surfaces as a cryptic httpx ``Invalid port``)."""
from urllib.parse import urlparse
normalize_proxy_env_vars()
for key in ("HTTPS_PROXY", "HTTP_PROXY", "ALL_PROXY", "https_proxy", "http_proxy", "all_proxy"):
value = str(os.environ.get(key) or "").strip()
if not value:
continue
try:
parsed = urlparse(value)
if parsed.scheme:
_ = parsed.port # raises ValueError for e.g. '6153export'
except ValueError as exc:
raise RuntimeError(
f"Malformed proxy environment variable {key}={value!r}. "
"Fix or unset your proxy settings and try again."
) from exc
def _validate_base_url(base_url: str) -> None:
"""Reject obviously broken custom endpoint URLs before they reach httpx."""
from urllib.parse import urlparse
candidate = str(base_url or "").strip()
if not candidate or candidate.startswith("acp://"):
return
try:
parsed = urlparse(candidate)
if parsed.scheme in {"http", "https"}:
_ = parsed.port # raises ValueError for malformed ports
except ValueError as exc:
raise RuntimeError(
f"Malformed custom endpoint URL: {candidate!r}. "
"Run `hermes setup` or `hermes model` and enter a valid http(s) base URL."
) from exc
def _try_custom_endpoint() -> tuple[Optional[Any], Optional[str]]:
runtime = _resolve_custom_runtime()
custom_base, custom_key, custom_mode = (*runtime, None) if len(runtime) == 2 else runtime
if not custom_base or not custom_key:
return None, None
if custom_base.lower().startswith(_CODEX_AUX_BASE_URL.lower()):
return None, None
model = _read_main_model_for_aux() or "gpt-4o-mini"
logger.debug("Auxiliary client: custom endpoint (%s, api_mode=%s)", model, custom_mode or "chat_completions")
_clean_base, _dq = _extract_url_query_params(custom_base)
_extra = {"default_query": _dq} if _dq else {}
# User model.default_headers override SDK fingerprint headers (as on the main client) for strict gateways/WAFs.
_custom_headers = _apply_user_default_headers(None)
if _custom_headers:
_extra["default_headers"] = _custom_headers
if custom_mode == "codex_responses":
real_client = _create_openai_client(api_key=custom_key, base_url=_clean_base, **_extra)
return CodexAuxiliaryClient(real_client, model), model
if custom_mode == "anthropic_messages":
# OAuth identity only when the host is exactly api.anthropic.com (key_cmd callable included,
# #114967); third-party Anthropic-compatible gateways never get it.
try:
from agent.anthropic_adapter import build_anthropic_client
from agent.anthropic_credentials import anthropic_route_is_oauth
real_client = build_anthropic_client(custom_key, custom_base)
except ImportError:
logger.warning(
"Custom endpoint declares api_mode=anthropic_messages but the "
"anthropic SDK is not installed — falling back to OpenAI-wire."
)
return _create_openai_client(api_key=custom_key, base_url=_clean_base, **_extra), model
return AnthropicAuxiliaryClient(real_client, model, custom_key, custom_base,
is_oauth=anthropic_route_is_oauth(custom_base, custom_key)), model
# URL-based anthropic detection for custom endpoints without explicit api_mode.
_fallback_client = _create_openai_client(api_key=custom_key, base_url=_clean_base, **_extra)
return _maybe_wrap_anthropic(_fallback_client, model, custom_key, custom_base, custom_mode), model
def _build_xai_oauth_aux_client(model: str) -> tuple[Optional[Any], Optional[str]]:
"""CodexAuxiliaryClient for xAI Grok OAuth (Responses API); (None, None) if not authed.
Caller must pass an explicit model — a pinned Grok default would rot as xAI's allowlist drifts.
"""
if not model:
logger.warning(
"Auxiliary client: xai-oauth requested without a model; "
"pass model explicitly (auxiliary.<task>.model in config.yaml)."
)
return None, None
resolved = _resolve_xai_oauth_for_aux(model)
if resolved is None:
return None, None
api_key, base_url = resolved
logger.debug("Auxiliary client: xAI OAuth (%s via Responses API)", model)
from tools.xai_http import hermes_xai_default_headers
real_client = _create_openai_client(
api_key=api_key, base_url=base_url, default_headers=hermes_xai_default_headers()
)
return CodexAuxiliaryClient(real_client, model), model
def _codex_base_url_override() -> str:
"""Profile-scoped ``HERMES_CODEX_BASE_URL`` (same read as the API-key env vars: under a
multiplexer the routed profile's .env decides the endpoint, never a sibling's process env)."""
return _scoped_key_env("HERMES_CODEX_BASE_URL").rstrip("/")
def _build_codex_client(model: str) -> tuple[Optional[Any], Optional[str]]:
"""CodexAuxiliaryClient for an explicit model; (None, None) without a Codex OAuth token.
No auto-selected default: the Codex model allow-list is undocumented and drifts.
"""
if not model:
logger.warning(
"Auxiliary client: openai-codex requested without a model; "
"pass model explicitly (auxiliary.<task>.model in config.yaml)."
)
return None, None
codex_token, base_url = _resolve_codex_credential_and_base(model)
if not codex_token:
return None, None
logger.debug("Auxiliary client: Codex OAuth (%s via Responses API)", model)
real_client = _create_openai_client(
api_key=codex_token, base_url=base_url,
default_headers=_codex_cloudflare_headers(codex_token, base_url=base_url),
)
return CodexAuxiliaryClient(real_client, model), model
def _try_azure_foundry(
*, model: Optional[str] = None, explicit_api_key: Optional[str] = None,
explicit_base_url: Optional[str] = None, api_mode: Optional[str] = None,
) -> tuple[Optional[Any], Optional[str]]:
"""Azure Foundry aux client via the main agent's ``_resolve_azure_foundry_runtime`` (api_key vs Entra
callable bearer, per-model api_mode, base_url overrides). Returns ``(client, model)`` or ``(None, None)``."""
try:
from hermes_cli.runtime_provider import _resolve_azure_foundry_runtime
from hermes_cli.auth import AuthError
from hermes_cli.config import load_config_readonly
except ImportError:
return None, None
try:
cfg = load_config_readonly()
model_cfg = cfg.get("model") if isinstance(cfg, dict) else {}
if not isinstance(model_cfg, dict):
model_cfg = {}
except Exception:
model_cfg = {}
try:
runtime = _resolve_azure_foundry_runtime(
requested_provider="azure-foundry", model_cfg=model_cfg,
explicit_api_key=explicit_api_key, explicit_base_url=explicit_base_url,
target_model=model,
)
except AuthError as exc:
logger.debug("Auxiliary azure-foundry: %s", exc)
return None, None
except Exception as exc:
logger.debug("Auxiliary azure-foundry runtime error: %s", exc)
return None, None
api_key = runtime.get("api_key")
base_url = str(runtime.get("base_url", "") or "")
runtime_api_mode = api_mode or runtime.get("api_mode") or "chat_completions"
# api_key may be a callable token provider; bail only on None/"".
if not (callable(api_key) or api_key) or not base_url:
return None, None
final_model = _normalize_resolved_model(model or str(model_cfg.get("default") or ""), "azure-foundry")
if not final_model:
# No fallback aux model for Azure (needs a deployment name): let the auto chain fall through instead of 404ing.
logger.debug(
"Auxiliary azure-foundry: no model resolved (model=%r, default=%r)",
model, model_cfg.get("default"),
)
return None, None
# The SDK drops api-version query params from the base URL; pass via default_query.
_clean_base, _dq = _extract_url_query_params(base_url)
extra: dict[str, Any] = {"default_query": _dq} if _dq else {}
client = _create_openai_client(api_key=api_key, base_url=_clean_base, **extra)
if runtime_api_mode == "codex_responses":
return CodexAuxiliaryClient(client, final_model), final_model
if runtime_api_mode == "anthropic_messages":
# api_key forwarded verbatim (string or Entra callable; build_anthropic_client installs the bearer hook).
return _maybe_wrap_anthropic(client, final_model, api_key, base_url, runtime_api_mode), final_model
return client, final_model
def _try_anthropic(explicit_api_key: Optional[str | Callable[[], str]] = None,
explicit_base_url: Optional[str] = None,
requested_model: Optional[str] = None) -> tuple[Optional[Any], Optional[str]]:
try:
from agent.anthropic_adapter import build_anthropic_client
from agent.anthropic_credentials import resolve_anthropic_token
except ImportError:
return None, None
model = _get_aux_model_for_provider("anthropic") or "claude-haiku-4-5-20251001"
# A model-scoped cooldown spares the key's other models.
pool_present, entry = _select_pool_entry("anthropic", requested_model or model)
if pool_present and entry is not None:
token = explicit_api_key or _pool_runtime_api_key(entry)
else:
# Pool absent/empty: legacy resolver so a dead pool entry can't wedge aux tasks when a standalone credential exists.
entry = None
token = explicit_api_key or resolve_anthropic_token()
if not token:
return None, None
# Honor config.yaml model.base_url only when provider is anthropic AND the URL is
# Anthropic-compatible; a foreign host (Codex, OpenRouter) would 401 every aux call.
base_url = _pool_runtime_base_url(entry, _ANTHROPIC_DEFAULT_BASE_URL) if pool_present else _ANTHROPIC_DEFAULT_BASE_URL
with contextlib.suppress(Exception):
from hermes_cli.config import load_config_readonly
cfg = load_config_readonly()
model_cfg = cfg.get("model")
if isinstance(model_cfg, dict):
cfg_provider = str(model_cfg.get("provider") or "").strip().lower()
if cfg_provider == "anthropic":
cfg_base_url = (model_cfg.get("base_url") or "").strip().rstrip("/")
if cfg_base_url and _is_anthropic_compatible_host(cfg_base_url):
base_url = cfg_base_url
# A caller-supplied endpoint (fallback_providers entry, custom_providers entry) wins over
# both the pool row and config.yaml, under the same Anthropic-compatible host rule the
# primary path applies. A foreign host is REFUSED outright rather than silently demoted to
# the canonical host: continuing would send the explicit credential to a target the caller
# did not ask for (#121359).
override_url = (explicit_base_url or "").strip().rstrip("/")
if override_url:
if not _is_anthropic_compatible_host(override_url):
logger.warning(
"Auxiliary client: refusing anthropic explicit base_url %r — not an "
"Anthropic-compatible host; no client built and no request sent.",
override_url,
)
return None, None
base_url = override_url
from agent.anthropic_credentials import _is_oauth_token
is_oauth = _is_oauth_token(token)
if _aux_probe_active():
# Probe: token + adapter import resolved; skip real client construction.
return _AuxProbeClientStub(api_key="", base_url=base_url), model
logger.debug("Auxiliary client: Anthropic native (%s) at %s (oauth=%s)", model, base_url, is_oauth)
try:
real_client = build_anthropic_client(token, base_url)
except ImportError:
return None, None # Adapter imports fine but the anthropic SDK itself is missing.
return AnthropicAuxiliaryClient(real_client, model, token, base_url, is_oauth=is_oauth), model
_MAIN_RUNTIME_FIELDS = ("provider", "model", "base_url", "api_key", "api_mode", "auth_mode")
_MAIN_RUNTIME_CONTEXT_FIELDS = _MAIN_RUNTIME_FIELDS + (
"requested_provider", "session_id", "cache_scope",
)
def _normalize_main_runtime(main_runtime: Optional[dict[str, Any]]) -> dict[str, Any]:
"""Return a sanitized copy of a live main-runtime override.
``api_key`` may be a zero-arg callable (Entra ID token provider, accepted by the OpenAI SDK)
— preserved as-is so aux clients share main-agent auth.
"""
if main_runtime is None:
main_runtime = _RUNTIME_MAIN_CONTEXT.get()
if not isinstance(main_runtime, dict):
return {}
normalized: dict[str, Any] = {}
for field in _MAIN_RUNTIME_CONTEXT_FIELDS:
value = main_runtime.get(field)
if field == "api_key" and callable(value) and not isinstance(value, str):
normalized[field] = value
elif isinstance(value, str) and value.strip():
normalized[field] = value.strip()
for identity_field in ("provider", "requested_provider"):
identity = normalized.get(identity_field)
if isinstance(identity, str):
normalized[identity_field] = identity.lower()
return normalized
def _get_provider_chain() -> list[tuple]:
"""Ordered provider detection chain, built at call time so ``_try_*`` patches are picked up.
``openai-codex`` is deliberately absent (shifting allow-list breaks guessed-model fallback).
"""
return [
("openrouter", _try_openrouter), ("nous", _try_nous),
("local/custom", _try_custom_endpoint), ("api-key", _resolve_api_key_provider),
]
# "Recently 402'd" unhealthy-provider cache: a depleted provider stays so for hours, so hiding it
# for a TTL saves an RTT per aux call. In-process only (profiles may use different keys).
_AUX_UNHEALTHY_TTL_SECONDS = 600 # 10 minutes
_aux_unhealthy_until: dict[Any, float] = {}
_aux_unhealthy_logged_at: dict[Any, float] = {}
_aux_unhealthy_reason: dict[Any, str] = {}
# resolved_provider / explicit-config names → chain labels.
_AUX_UNHEALTHY_LABEL_ALIASES = {
"openrouter": "openrouter", "nous": "nous", "custom": "local/custom",
"local/custom": "local/custom", "openai-codex": "openai-codex", "codex": "openai-codex",
}
def _normalize_chain_label(provider: str) -> str:
"""resolved_provider → chain label; unknown API-key providers fall back to the lowercased input."""
if not provider:
return ""
p = str(provider).strip().lower()
return _AUX_UNHEALTHY_LABEL_ALIASES.get(p, p)
_AUX_UNHEALTHY_PAYMENT_REASON = "payment / credit error"
def _mark_provider_unhealthy(
provider: str, ttl: Optional[float] = None, *, base_url: Optional[str] = None,
reason: str = _AUX_UNHEALTHY_PAYMENT_REASON, level: int = logging.WARNING, model: Optional[str] = None,
) -> None:
"""Hide one provider endpoint until the TTL expires. ``reason`` is what the log says and what the
skip line echoes: absent credentials are an expected state (DEBUG), a confirmed 402 is a fault
(WARNING) — the old fixed payment wording sent local-only users chasing billing (#64144)."""
label = _normalize_chain_label(provider)
if not label:
return
key = _unhealthy_cache_key(label, base_url, model)
label = f"{label} model {model}" if model else label
ttl = _AUX_UNHEALTHY_TTL_SECONDS if ttl is None else ttl
expires_at = time.time() + ttl
_aux_unhealthy_until[key] = expires_at
_aux_unhealthy_reason[key] = reason
logger.log(
level,
"Auxiliary: marking %s unhealthy for %ds (%s). "
"Subsequent auxiliary calls will skip it until %s.",
label, int(ttl), reason, time.strftime("%H:%M:%S", time.localtime(expires_at)),
)
def _is_provider_unhealthy(label: str, base_url: Optional[str] = None, model: Optional[str] = None) -> bool:
"""True iff this provider endpoint (or that ``model`` on it) is unhealthy and unexpired; lazily evicts expired entries."""
if not label:
return False
if model and _is_provider_unhealthy(label, base_url):
return True
key = _unhealthy_cache_key(label, base_url, model)
expires_at = _aux_unhealthy_until.get(key)
if expires_at is None:
return False
if time.time() >= expires_at:
_aux_unhealthy_until.pop(key, None)
_aux_unhealthy_logged_at.pop(key, None)
_aux_unhealthy_reason.pop(key, None)
return False
return True
def _log_skip_unhealthy(
label: str, task: Optional[str] = None, *, base_url: Optional[str] = None, model: Optional[str] = None,
) -> None:
"""Log a skipped unhealthy provider (or provider model) at most once per minute per key."""
now = time.time()
key = _unhealthy_cache_key(label, base_url)
key = key if key in _aux_unhealthy_until else _unhealthy_cache_key(label, base_url, model)
if now - _aux_unhealthy_logged_at.get(key, 0.0) >= 60:
_aux_unhealthy_logged_at[key] = now
expires_at = _aux_unhealthy_until.get(key, now)
logger.info(
"Auxiliary %s: skipping %s (%s, retry in %ds)",
task or "call", label, _aux_unhealthy_reason.get(key, _AUX_UNHEALTHY_PAYMENT_REASON),
max(0, int(expires_at - now)),
)
def _reset_aux_unhealthy_cache() -> None:
"""Clear the unhealthy cache (tests / explicit user reset)."""
_aux_unhealthy_until.clear()
_aux_unhealthy_logged_at.clear()
_aux_unhealthy_reason.clear()
def _contains_any(text: str, needles: tuple[str, ...]) -> bool:
"""True when any needle is a substring of ``text``."""
return any(kw in text for kw in needles)
def _nous_portal_account_has_fresh_paid_access() -> bool:
"""Return True only when the fresh Nous account API says paid access is allowed."""
try:
from hermes_cli.nous_account import get_nous_portal_account_info
return get_nous_portal_account_info(force_fresh=True).paid_service_access is True
except Exception as exc:
logger.debug("Auxiliary Nous paid-entitlement refresh check failed: %s", exc)
return False
_DEFAULT_TRANSIENT_RETRIES = 2
_TRANSIENT_RETRY_BACKOFF_BASE = 1.0 # Backoff base (seconds); overridable so tests can zero it out.
def _transient_retry_count() -> int:
"""Same-provider retries for a transient blip: ``auxiliary.transient_retries``
(default 2), clamped to [0, 6]; config-read failures fall back to default."""
try:
from hermes_cli.config import cfg_get, load_config
val = cfg_get(load_config(), "auxiliary", "transient_retries")
return _DEFAULT_TRANSIENT_RETRIES if val is None else max(0, min(int(val), 6))
except Exception:
return _DEFAULT_TRANSIENT_RETRIES
def _is_unsupported_parameter_error(exc: Exception, param: str) -> bool:
"""Provider 400 for an unsupported request parameter: the parameter name plus a generic
unsupported/unknown/unrecognized marker, so call sites can retry without the key."""
param_lower = (param or "").lower()
if not param_lower:
return False
err_lower = str(exc).lower()
return param_lower in err_lower and _contains_any(err_lower, UNSUPPORTED_PARAM_MARKERS)
def _is_structured_output_rejection(exc: Exception) -> bool:
"""Provider 400/422 rejecting the structured-output field, on either wire: OpenAI ``response_format``
(incl. vLLM's ``guided_grammar``/xgrammar failures) or Anthropic ``output_config.format`` ("Extra inputs
are not permitted"). Callers tolerate an unconstrained reply, so the reaction is one retry without it."""
status = getattr(exc, "status_code", None)
if status is not None and status not in {400, 422}:
return False
err_lower = str(exc).lower()
# vLLM grammar-backend failures name the translated parameter, not ours.
if _contains_any(err_lower, ("guided_grammar", "xgrammar", "compile_grammar_error")):
return True
if "extra inputs are not permitted" in err_lower and (
"response_format" in err_lower or "output_config" in err_lower
):
return True
if "response_format" in err_lower and "unavailable" in err_lower:
return True
# Gateways that validate the request body with a strict pydantic model reject the
# OBJECT-form json_schema by shape ("str type expected" on response_format.json_schema,
# 422) rather than by naming the feature. The field is what they refuse; the retry
# without it is the same remedy, so treat the shape error as a rejection too.
if "response_format" in err_lower and "json_schema" in err_lower:
return True
# Gemini native names its own generationConfig keys, never ours: "Function calling with a response
# mime type: 'application/json' is unsupported" (pre-Gemini-3 + tools via a proxy), or an
# "Unknown name"/"Invalid value" 400 on response_schema / response_json_schema for a schema the
# surface cannot express. Same remedy: one retry without the format.
if _contains_any(err_lower, ("response mime type", "response_schema", "response_json_schema")):
return True
return _is_unsupported_parameter_error(exc, "response_format") or _is_unsupported_parameter_error(exc, "output_config")
def _is_reasoning_field_rejection(exc: Exception) -> bool:
"""Provider 400 rejecting a reasoning wire control by name (``reasoning_effort``, ``reasoning``,
``thinking``/``think``). Chat-only models behind OpenAI-compatible relays reject the top-level
``reasoning_effort: none`` a disabled ``reasoning_config`` projects on the custom profile
("Unrecognized request argument supplied: reasoning_effort", #112781); the route default is
the right answer for such a model, so the reaction is one retry without any reasoning field."""
status = getattr(exc, "status_code", None)
if status is not None and status not in {400, 422}:
return False
return is_reasoning_field_rejection(str(exc))
def _is_reasoning_required_rejection(exc: Exception) -> bool:
"""Provider 400 refusing to switch reasoning OFF ("Reasoning is mandatory for this endpoint and cannot
be disabled"): the field is understood, only the disable is refused, so the rung steps the effort up to
the floor instead of dropping the field (agent/auxiliary_reasoning_floor.py)."""
status = getattr(exc, "status_code", None)
if status is not None and status not in {400, 422}:
return False
return is_reasoning_required_rejection(str(exc))
def _without_reasoning_fields(kwargs: dict) -> Optional[dict]:
"""Copy *kwargs* without reasoning wire controls (top-level ``reasoning_effort``, the adapter's
private ``_reasoning_config`` and every ``extra_body`` reasoning key); None when nothing was
removed, so call sites don't retry an unchanged request."""
retry_kwargs = dict(kwargs)
changed = retry_kwargs.pop("reasoning_effort", None) is not None
changed = retry_kwargs.pop("_reasoning_config", None) is not None or changed
extra_body = retry_kwargs.get("extra_body")
if isinstance(extra_body, dict):
remaining = {k: v for k, v in extra_body.items() if str(k).strip().lower() not in PROFILE_REASONING_KEYS}
if len(remaining) != len(extra_body):
if remaining:
retry_kwargs["extra_body"] = remaining
else:
retry_kwargs.pop("extra_body", None)
changed = True
return retry_kwargs if changed else None
# Tasks on a user-visible critical path (compression blocks resuming an oversized session; vision
# stalls the serialised turn queue). A same-provider retry after a full-budget timeout costs another
# whole ``timeout`` window, so they skip straight to fallback; fast blips still retry.
# Fast blips (a streaming-close or a 5xx) still retry, since those are cheap. See issue #54465 for the
# compression case. Title generation joins them because its retries multiplied the user's
# ``auxiliary.title_generation.timeout`` (~4x: three full windows plus backoff) on a slow local
# model, and the auto-title thread outlived the deadline the user thought they had set (#89445, #66251).
_TIMEOUT_NO_RETRY_TASKS = frozenset({"compression", "vision", "title_generation"})
def _should_skip_same_provider_retry(task: Optional[str], failure: aux_errors.Failure) -> bool:
"""True when a transient error on a critical-path task should go straight to fallback.
Carve-out: a fast first-token fail (dead stream within the no-progress window, zero output —
see ``_timeout_message``) is cheap and keeps the same-provider retry; mid-stream stalls and
hard-ceiling timeouts skip to fallback.
"""
return (task in _TIMEOUT_NO_RETRY_TASKS and failure.verdict.error_context.get("transport") == "timeout"
and "no-progress timeout" not in str(failure.error))
def _evict_cached_clients(provider: str) -> None:
"""Drop this profile's cached auxiliary clients for a provider so fresh creds are used.
Scoped to the calling profile (``hermes_home_key()`` is the first key slot): a rotation in
one profile must not drop another profile's client for the same provider in a multiplexing
gateway, since that profile's credentials did not change. Entries are popped, not closed:
a concurrent caller may be mid-request on the shared client (closing it raises ReadError /
"client has been closed" for them); the dropped client is retired by GC like the FIFO
overflow path in ``_get_cached_client``.
"""
normalized = _normalize_aux_provider(provider)
home = hermes_home_key()
with _client_cache_lock:
for key in [key for key in _client_cache
if key[0] == home and _normalize_aux_provider(str(key[1])) == normalized]:
_client_cache.pop(key, None)
def _evict_cached_client_instance(target: Any) -> bool:
"""Drop cache entries whose stored client (or its ``_real_client``) is *target*; True if any evicted.
Used when a cached client is poisoned (closed transport after a timeout). Async wrappers must
expose the same ``_real_client`` as their sync sibling or the async entry survives.
"""
if target is None:
return False
evicted = False
with _client_cache_lock:
for key, entry in list(_client_cache.items()):
cached = entry[0] if entry is not None else None
if cached is not None and (cached is target or getattr(cached, "_real_client", None) is target):
del _client_cache[key]
evicted = True
return evicted
def _pool_credential_digest(pool: Any, entry: Any = None) -> str:
"""Secret-safe digest over the runtime key(s) backing the cache hint.
When ``entry`` (the peeked selection) carries a key, only THAT key is digested: rotating a
sibling account must not churn this entry's cached client. Otherwise every pool entry
contributes, taken WITHOUT availability filters (``pool.entries()``, not ``peek()``): an entry
benched by a per-model cooldown still counts, so a token rotated while the cooldown is active
changes the digest too (#113022 point 5). Only the blake2b of the keys enters the cache key.
"""
keys = [k for k in (_pool_runtime_api_key(entry),) if k] if entry is not None else []
if not keys:
try:
entries_fn = getattr(pool, "entries", None)
entries = entries_fn() if callable(entries_fn) else []
except Exception as exc:
logger.debug("Auxiliary client: could not list pool entries for digest: %s", exc)
return ""
keys = sorted(k for k in (_pool_runtime_api_key(e) for e in entries or ()) if k)
if not keys:
return ""
digest = hashlib.blake2b(digest_size=16)
for key in keys:
digest.update(key.encode("utf-8"))
digest.update(b"\0")
return digest.hexdigest()
def _pool_cache_hint(provider: str, *, main_runtime: Optional[dict[str, Any]] = None) -> str:
"""Return a cache discriminator that follows the pooled credential, not just its entry id.
``<provider>:<entry id>:<digest of that entry's key>`` when ``peek()`` names an entry,
``<provider>::<digest of every entry's key>`` when every entry is filtered out (cooldown)
but keyed entries exist. A token rotated or
refreshed by another process yields a new client instead of a 401 round trip on the cached
one; the stale entry ages out through the non-closing FIFO cap (#113022).
"""
normalized = _normalize_aux_provider(provider)
if normalized == "auto":
runtime = _normalize_main_runtime(main_runtime)
normalized = _normalize_aux_provider(runtime.get("provider") or _read_main_provider())
if normalized in {"", "auto", "custom"}:
return ""
pool = _load_pool_with_credentials(normalized, " (cache hint)")
if pool is None:
return ""
entry = _peek_pool_entry(normalized, pool)
digest = _pool_credential_digest(pool, entry)
entry_id = str(getattr(entry, "id", "") or "").strip() if entry is not None else ""
if not entry_id and not digest:
return ""
return f"{normalized}:{entry_id}:{digest}"
# Ordered (host, provider) tables for inferring a backend from a client base URL (then profile ``hostnames``).
_POOL_PROVIDER_BY_HOST = (
("chatgpt.com", "openai-codex"), ("openrouter.ai", "openrouter"),
("inference-api.nousresearch.com", "nous"), ("api.anthropic.com", "anthropic"),
("api.kimi.com", "kimi-coding"), ("api.x.ai", "xai-oauth"),
)
_AUTH_REFRESH_PROVIDER_BY_HOST = (
("chatgpt.com", "openai-codex"),
("api.anthropic.com", "anthropic"), ("inference-api.nousresearch.com", "nous"),
# An aux call that inherits the main xai-oauth route arrives as "auto"; without this row the
# 403 bad-credentials rung skipped the refresh and benched the only grant (#84845).
("api.x.ai", "xai-oauth"),
)
def _provider_for_host(base_url: str, table: tuple[tuple[str, str], ...]) -> Optional[str]:
"""First provider in ``table`` whose host matches ``base_url``, else a profile claiming the host."""
for host, provider in table:
if base_url_host_matches(base_url, host):
return provider
from providers import profile_for_host
return getattr(profile_for_host(base_url), "name", None)
def _relay_path(base_url: str) -> str:
"""The path an aux client built for ``base_url`` reaches (dual-surface ``/anthropic`` mapped), ``/v1`` dropped."""
return base_url_path(_to_openai_base_url(base_url)).removesuffix("/v1")
def _recoverable_pool_provider(
resolved_provider: str, client: Any, main_runtime: Optional[dict[str, Any]] = None, *, same_path: bool = False,
) -> Optional[str]:
"""Infer which provider pool can recover the current auxiliary client.
None when the client targets a different host than the session's configured endpoint for that
provider: a rejection there says nothing about the key, so rotating/quarantining it would kill a
working credential (Miho report — proxy users)."""
normalized = _normalize_aux_provider(resolved_provider)
base = str(getattr(client, "base_url", "") or "")
runtime = _normalize_main_runtime(main_runtime)
rt_base = str(runtime.get("base_url") or "")
rt_key = runtime.get("api_key")
client_key = getattr(client, "api_key", None)
# Only the SESSION's own key is shielded, and only when it was sent somewhere other than the
# session's origin (scheme+host+port — a port or HTTPS→HTTP change is a different trust boundary).
# An independently owned auxiliary pool keeps rotating at its own origin.
if (base and rt_base and normalized == runtime.get("provider")
and isinstance(rt_key, str) and rt_key and client_key == rt_key
and base_url_origin(base) != base_url_origin(rt_base)):
logger.info("Auxiliary: %s rejected the session key at %s, but the session's endpoint is %s — "
"endpoint mismatch, not a dead key; skipping credential rotation",
normalized, base_url_hostname(base), base_url_hostname(rt_base))
return None
if normalized not in {"", "auto", "custom"}: # a pinned ``custom:<name>`` keeps its entry's identity
return pinned if (pinned := str(resolved_provider or "").strip().lower()).startswith("custom:") else normalized
known = _provider_for_host(base, _POOL_PROVIDER_BY_HOST)
if known is not None:
return known
# Providers outside the table (e.g. opencode-go): match base URL against registered
# api_key providers so pool rotation works for them too.
if main_runtime:
runtime = _normalize_main_runtime(main_runtime)
rt_provider = runtime.get("provider", "")
if rt_provider and rt_provider not in {"", "auto", "custom"}:
with contextlib.suppress(Exception):
from hermes_cli.auth import PROVIDER_REGISTRY
pconfig = PROVIDER_REGISTRY.get(rt_provider)
if pconfig and getattr(pconfig, "auth_type", None) == "api_key":
# The pool's key was issued for the endpoint the main runtime actually uses; a
# rejection at any other host (registry default vs configured proxy) says nothing
# about that key, so it must not be marked exhausted.
rt_base = str(runtime.get("base_url") or getattr(pconfig, "inference_base_url", "") or "").rstrip("/")
# ``same_path``: naming the backend that sent a request (not the pool to rotate) also needs the
# origin and path, as one host can serve two products (OpenCode Zen at /zen, Go at /zen/go).
if rt_base and base_url_host_matches(base, base_url_hostname(rt_base)) and (not same_path or (
base_url_origin(base), _relay_path(base)) == (base_url_origin(rt_base), _relay_path(rt_base))):
return rt_provider
return None
def _recover_provider_pool(provider: str, exc: Exception, *, failed_api_key: str = "",
verdict: Optional[aux_errors.ClassifiedError] = None) -> bool:
"""Try same-provider credential-pool recovery for auxiliary calls.
``failed_api_key`` lets mark_exhausted_and_rotate identify the right pool entry even if
another process already rotated (current() would be None).
"""
normalized = _normalize_aux_provider(provider)
try:
pool = pool_for_aux_provider(provider, normalized, load_pool)
except Exception as load_exc:
logger.debug("Auxiliary client: could not load pool for %s recovery: %s", normalized, load_exc)
return False
if not pool or not pool.has_credentials():
return False
if verdict is None: # the ladder hands in the failed request's own verdict: hooks fire once per request
verdict = aux_errors.classify_attempt(exc, normalized)
status_code = verdict.status_code
gated = verdict.reason is FailoverReason.model_entitlement
def _rotate(fallback_status: int) -> bool:
# Sized as the main loop's recover_with_credential_pool sizes it: real status, pool failure reason, model.
error_context: dict[str, Any] = {"message": str(exc)}
if status_code is not None:
error_context["status_code"] = status_code
next_entry = pool.mark_exhausted_and_rotate(
status_code=fallback_status if status_code is None else status_code,
error_context=error_context, api_key_hint=failed_api_key or None,
failure_reason=pool_failure_reason(verdict.reason, verdict.error_context, verdict.billing_unverified),
model=verdict.model or None,
)
if next_entry is None:
return False
_evict_cached_clients(normalized)
return True
if aux_errors.auth_refresh(verdict):
if pool.try_refresh_current() is not None:
_evict_cached_clients(normalized)
return True
return _rotate(401)
if not verdict.should_rotate_credential: # e.g. an overloaded server or an upstream model throttle
return False
return _rotate(402 if aux_errors.benches_provider(verdict) else 400 if gated else 429)
def _prepare_same_provider_retry(
*, task: Optional[str], resolved_provider: str, resolved_model: Optional[str],
resolved_base_url: Optional[str], resolved_api_key: Optional[str],
resolved_api_mode: Optional[str], main_runtime: Optional[dict[str, Any]],
final_model: Optional[str], messages: list, temperature: Optional[float],
max_tokens: Optional[int], tools: Optional[list], effective_timeout: float,
effective_extra_body: dict, reasoning_config: Optional[dict], async_mode: bool,
extra_headers: Optional[dict[str, str]] = None, repairs: tuple = (),
) -> tuple[Any, dict[str, Any]]:
"""Rebuild (client, request kwargs) for a same-provider retry after credential recovery. ``repairs``: the
parameter rungs' strips the failed request already carried, re-applied so a stripped field stays stripped."""
if task == "vision":
effective_provider, retry_client, retry_model = resolve_vision_provider_client(
provider=resolved_provider, model=final_model, base_url=resolved_base_url,
api_key=resolved_api_key, async_mode=async_mode,
)
else:
retry_client, retry_model = _get_cached_client(
resolved_provider, resolved_model, async_mode=async_mode, base_url=resolved_base_url,
api_key=resolved_api_key, api_mode=resolved_api_mode, main_runtime=main_runtime, task=task,
)
effective_provider = _effective_provider_for_client(retry_client, resolved_provider)
if retry_client is None:
raise RuntimeError(
f"Auxiliary {task or 'call'}: provider {resolved_provider} could not be rebuilt after recovery"
)
retry_base = str(getattr(retry_client, "base_url", "") or "")
retry_kwargs = _build_call_kwargs(
effective_provider or resolved_provider, retry_model or final_model, messages,
temperature=temperature, max_tokens=max_tokens, tools=tools, timeout=effective_timeout,
extra_body=effective_extra_body, reasoning_config=reasoning_config,
base_url=retry_base or resolved_base_url, task=task,
)
# Preserve per-request attribution headers (a profile's ``request_headers``) across the rebuilt-client
# retry: dropping them would let a recovery retry silently lose capability gating (#60293).
if extra_headers:
retry_kwargs["extra_headers"] = dict(extra_headers)
retry_kwargs = reapply_repairs(retry_kwargs, repairs) # a stripped field stays stripped
if _is_anthropic_compat_endpoint(resolved_provider, retry_base):
retry_kwargs["messages"] = _convert_openai_images_to_anthropic(retry_kwargs["messages"])
return retry_client, retry_kwargs
class _ClassifiedAttempt(Exception):
"""A failed request, raised where it was classified on the client/model it was sent to (a rebuilt same-provider
retry, a judged stream negotiation, a credit-limited 402): recovery reads ``failure`` instead of classifying
again, and ``_rung`` re-raises its error. Relay and the aux hooks only see that error (``_provider_outcome``)."""
def __init__(self, failure: aux_errors.Failure) -> None:
super().__init__(str(failure.error))
self.failure = failure
def _retry_same_provider_sync(*, resolved_provider: str, resolved_api_mode: Optional[str], task: Optional[str], **prep) -> Any:
retry_client, retry_kwargs = _prepare_same_provider_retry(
task=task, resolved_provider=resolved_provider, resolved_api_mode=resolved_api_mode, async_mode=False, **prep,
)
resp, exc = _outcome(lambda: _validate_llm_response(_relay_sync_completion(
retry_client, retry_kwargs, provider=resolved_provider, api_mode=resolved_api_mode), task))
if exc is not None:
raise _ClassifiedAttempt(_classify_failure(exc, resolved_provider, retry_client, retry_kwargs.get("model")))
return resp
async def _retry_same_provider_async(*, resolved_provider: str, resolved_api_mode: Optional[str], task: Optional[str], **prep) -> Any:
import asyncio
retry_client, retry_kwargs = _prepare_same_provider_retry(
task=task, resolved_provider=resolved_provider, resolved_api_mode=resolved_api_mode, async_mode=True, **prep,
)
async def _send() -> Any:
resp = await _relay_async_completion(retry_client, retry_kwargs, provider=resolved_provider, api_mode=resolved_api_mode)
return await asyncio.to_thread(_validate_llm_response, resp, task)
resp, exc = await _aoutcome(_send())
if exc is not None:
raise _ClassifiedAttempt(_classify_failure(exc, resolved_provider, retry_client, retry_kwargs.get("model")))
return resp
def _creds_have_api_key(creds: dict[str, Any]) -> bool:
return bool(str(creds.get("api_key", "") or "").strip())
def _refresh_codex_credentials() -> bool:
from hermes_cli.auth import resolve_codex_runtime_credentials
return _creds_have_api_key(resolve_codex_runtime_credentials(force_refresh=True))
def _refresh_nous_credentials() -> bool:
from hermes_cli.auth import resolve_nous_runtime_credentials
return _creds_have_api_key(resolve_nous_runtime_credentials(
timeout_seconds=env_float("HERMES_NOUS_TIMEOUT_SECONDS", 15), force_refresh=True
))
def _refresh_anthropic_credentials(failed_api_key: str = "") -> bool:
from agent.anthropic_credentials import read_claude_code_credentials, _refresh_oauth_token
token = failed_api_key
if not token:
return False
pool = load_pool("anthropic")
if pool.entry_id_for_api_key(token):
return pool.try_refresh_matching(api_key_hint=token) is not None
creds = read_claude_code_credentials()
# Never spend an ambient login's refresh rotation for another request's key.
if isinstance(creds, dict) and creds.get("accessToken") == token and creds.get("refreshToken"):
return bool(_refresh_oauth_token(creds))
return False
def _refresh_xai_oauth_credentials() -> bool:
"""Pool-level refresh first, then the singleton auth-store resolver."""
pool = load_pool("xai-oauth")
if pool and pool.has_credentials():
pool.select()
refreshed = pool.try_refresh_current()
if refreshed is not None and str(getattr(refreshed, "runtime_api_key", "") or "").strip():
return True
from hermes_cli.auth import resolve_xai_oauth_runtime_credentials
return _creds_have_api_key(resolve_xai_oauth_runtime_credentials(force_refresh=True))
def _refresh_vertex_credentials() -> bool:
"""Mirrors run_agent's Vertex refresh; the cache key ignores the rotating bearer, so
without the eviction that follows, a ~1h-expired aux Vertex client 401s forever."""
from agent.vertex_adapter import get_vertex_config
token, base_url = get_vertex_config()
return bool(isinstance(token, str) and token.strip() and isinstance(base_url, str) and base_url.strip())
# Each refresher returns True when a usable credential exists; the caller then evicts cached clients.
_CREDENTIAL_REFRESHERS: dict[str, Callable[..., bool]] = {
"openai-codex": _refresh_codex_credentials,
"nous": _refresh_nous_credentials, "anthropic": _refresh_anthropic_credentials,
"xai-oauth": _refresh_xai_oauth_credentials, "vertex": _refresh_vertex_credentials,
}
def _refresh_provider_credentials(provider: str, *, failed_api_key: str = "") -> bool:
"""Refresh short-lived credentials for OAuth-backed auxiliary providers."""
normalized = _normalize_aux_provider(provider)
refresher = _CREDENTIAL_REFRESHERS.get(normalized)
if refresher is None:
from providers import get_provider_profile
profile = get_provider_profile(normalized)
if profile is None:
return False
refresher = lambda: profile.refresh_runtime_credentials(reason="auxiliary_unauthorized")
try:
if not (refresher(failed_api_key) if normalized == "anthropic" else refresher()):
return False
_evict_cached_clients(normalized)
return True
except Exception as exc:
logger.debug("Auxiliary provider credential refresh failed for %s: %s", normalized, exc)
return False
def _auth_refresh_provider_for_route(
resolved_provider: Optional[str], client_base_url: str, effective_provider: str = "",
) -> str:
"""Provider whose short-lived credentials should be refreshed; auto-routed calls keep
``resolved_provider == "auto"``, so infer the backend from the client's base URL."""
normalized = _normalize_aux_provider(resolved_provider)
if normalized and normalized != "auto":
return normalized
host_provider = _provider_for_host(client_base_url, _AUTH_REFRESH_PROVIDER_BY_HOST)
# The auto client already knows it runs on the host's API-key sibling (``xai`` on api.x.ai):
# an XAI_API_KEY 401 must not spend a stale ``xai-oauth`` grant's refresh and switch routes.
if host_provider and host_provider == f"{_normalize_aux_provider(effective_provider)}-oauth":
return normalized
return host_provider or normalized
def _fallback_chain_entry(task: Optional[str], fb_label: str) -> Optional[dict[str, Any]]:
"""Resolve the ``fallback_chain`` entry a ``fallback_chain[<i>](<provider>)`` label points at,
or None when the label is not a configured-chain candidate or the index no longer resolves."""
if not task or not fb_label:
return None
m = re.match(r"fallback_chain\[(\d+)\]", fb_label)
if not m:
return None
try:
chain = _get_auxiliary_task_config(task).get("fallback_chain")
entry = chain[int(m.group(1))] if isinstance(chain, list) else None
except Exception:
return None
return entry if isinstance(entry, dict) else None
def _coerce_positive_timeout(raw: Any) -> Optional[float]:
"""Coerce a config ``timeout`` to a positive float, or None (rejects bools, which are ints)."""
if isinstance(raw, (int, float)) and not isinstance(raw, bool) and raw > 0:
return float(raw)
return None
def _fallback_entry_timeout(task: Optional[str], fb_label: str) -> Optional[float]:
"""Per-entry ``timeout`` for a configured fallback candidate, or None (keep the task-level
timeout). Inheriting the primary's deadline used to kill healthy-but-slower fallbacks.
A fallback candidate previously inherited the exact timeout the primary provider was called with. When
that deadline was tuned for the primary (or the primary simply consumed its whole budget before failing
over), the fallback aborted on the same clock even when independently healthy — a 163k-token compression
that needs ~90s on the fallback died at the primary's 30s deadline every turn (#62452).
"""
entry = _fallback_chain_entry(task, fb_label)
return _coerce_positive_timeout(entry.get("timeout") if entry else None)
def _fallback_provider_from_label(label: str) -> str:
"""Recover the provider identifier from a fallback display label."""
match = re.match(r"(?:fallback_chain\[\d+\]|fallback_providers\[\d+\]|main-agent)\(([^)]+)\)$", label or "")
return match.group(1).strip() if match else str(label or "").strip()
class _FallbackDestination(NamedTuple):
provider: str
base_url: str
api_mode: Optional[str]
model: Optional[str]
def _complete_fallback_destination(
provider: str, base_url: str, api_mode: Optional[str], model: Optional[str]
) -> _FallbackDestination:
if not api_mode:
if _endpoint_speaks_anthropic_messages(base_url):
api_mode = "anthropic_messages"
else:
with contextlib.suppress(Exception):
from hermes_cli.runtime_provider import resolve_runtime_provider
runtime = resolve_runtime_provider(
requested=provider, explicit_base_url=base_url or None, target_model=model or ""
)
api_mode = str(runtime.get("api_mode") or "").strip() or None
return _FallbackDestination(provider, base_url, api_mode, model)
def _fallback_destination_from_entry(
entry: dict[str, Any], fb_client: Any, fb_model: Optional[str]
) -> _FallbackDestination:
provider = str(entry.get("provider") or "").strip()
base_url = str(entry.get("base_url") or getattr(fb_client, "base_url", "") or "").strip()
api_mode = str(entry.get("api_mode") or entry.get("transport") or "").strip() or None
model = fb_model or str(entry.get("model") or "").strip() or None
return _complete_fallback_destination(provider, base_url, api_mode, model)
def _fallback_destination(
task: Optional[str], fb_client: Any, fb_model: Optional[str], fb_label: str
) -> _FallbackDestination:
"""Route identity of a fallback request: attached destination, else configured entry, else label."""
attached = getattr(fb_client, "_hermes_fallback_destination", None)
if isinstance(attached, _FallbackDestination):
return attached
entry = _fallback_chain_entry(task, fb_label)
if entry is not None:
return _fallback_destination_from_entry(entry, fb_client, fb_model)
return _complete_fallback_destination(
_fallback_provider_from_label(fb_label), str(getattr(fb_client, "base_url", "") or ""), None, fb_model,
)
def _replan_synchronous_cache_sections(
messages: list, tools: Optional[list], *, destination: _FallbackDestination
) -> tuple[list, list]:
"""Plan one destination at the configured TTL (1h must not regress to 5m); one-shots never cache."""
from agent.agent_runtime_helpers import configured_cache_ttl, plan_cache_sections_for_destination
from agent.prompt_caching import is_one_shot_request
return plan_cache_sections_for_destination(
messages, tools, provider=destination.provider, base_url=destination.base_url,
api_mode=destination.api_mode or "", model=destination.model or "", cache_ttl=configured_cache_ttl(),
cache_disabled=True if is_one_shot_request(messages, tools) else None,
)
def _fallback_request_kwargs(
destination: _FallbackDestination, *, task: Optional[str], messages: list,
tools: Optional[list], temperature: Optional[float], max_tokens: Optional[int],
effective_timeout: float, effective_extra_body: dict, reasoning_config: Optional[dict],
fallback_entry: dict, task_config: dict, apply_fast_lane: bool,
) -> dict[str, Any]:
"""Build request kwargs for one fallback destination (cache-section replan + fast-lane cap)."""
fallback_max_tokens, fallback_extra_body = max_tokens, effective_extra_body
if apply_fast_lane:
fallback_max_tokens, fallback_extra_body = _compression_fast_lane_controls(
task, actual_provider=destination.provider, actual_model=destination.model,
requested_provider=fallback_entry.get("provider"),
requested_model=fallback_entry.get("model"), route_config=fallback_entry,
leak_guard_config=task_config, max_tokens=max_tokens, extra_body=effective_extra_body,
)
fallback_messages, fallback_tools = _replan_synchronous_cache_sections(messages, tools, destination=destination)
fb_kwargs = _build_call_kwargs(
destination.provider, destination.model, fallback_messages,
temperature=temperature, max_tokens=fallback_max_tokens, tools=fallback_tools, timeout=effective_timeout,
extra_body=fallback_extra_body, reasoning_config=reasoning_config, base_url=destination.base_url, task=task)
return fb_kwargs
def _plan_fallback_candidate(
fb_client: Any, fb_model: Optional[str], fb_label: str, *, task: Optional[str],
effective_timeout: float, apply_fast_lane: bool, **request,
) -> tuple[_FallbackDestination, dict[str, Any], Callable[[str, Any, Optional[str]], dict[str, Any]]]:
"""Resolve the destination + first-attempt kwargs for a fallback candidate.
Returns ``(destination, kwargs, rebuild)`` where ``rebuild(provider, client, model)`` produces
kwargs for the credential-refreshed retry destination. A configured-chain entry's own
``timeout`` overrides ``effective_timeout``.
"""
fb_timeout = _fallback_entry_timeout(task, fb_label)
if fb_timeout is not None and fb_timeout != effective_timeout:
logger.info(
"Auxiliary %s: %s using its configured timeout %.0fs "
"(task-level was %.0fs)",
task or "call", fb_label, fb_timeout, effective_timeout,
)
effective_timeout = fb_timeout
destination = _fallback_destination(task, fb_client, fb_model, fb_label)
task_config = _get_auxiliary_task_config(task) if task == "compression" else {}
fallback_entry = _fallback_chain_entry(task, fb_label) or {}
common = dict(
task=task, effective_timeout=effective_timeout, fallback_entry=fallback_entry,
task_config=task_config, apply_fast_lane=apply_fast_lane, **request,
)
def _rebuild(provider: str, client: Any, model: Optional[str]) -> tuple[_FallbackDestination, dict[str, Any]]:
retry_destination = _FallbackDestination(
provider, destination.base_url or str(getattr(client, "base_url", "") or ""),
destination.api_mode, model or destination.model,
)
return retry_destination, _fallback_request_kwargs(retry_destination, **common)
return destination, _fallback_request_kwargs(destination, **common), _rebuild
def _quarantine_fallback_candidate(
task: Optional[str], fb_label: str, fb_provider: str, fb_err: Exception, *,
base_url: str = "", tag: str = "", reason: Optional[str] = None, model: Optional[str] = None,
) -> None:
"""The candidate cannot serve this walk (``reason`` = its ``aux_errors.fallback_label``,
None = dead token): mark it unhealthy so the ordered re-walk skips it and the caller moves on to
the next entry. Transient classes get a short hold, payment/quota and dead tokens the long one;
a model the route lacks or refuses marks only that ``model``, its siblings stay eligible."""
_mark_provider_unhealthy(
fb_provider or fb_label, ttl=fallback_candidate_quarantine_ttl(reason),
base_url=base_url, reason=reason or "stale fallback credential",
model=model if reason in (aux_errors.MODEL_NOT_FOUND, aux_errors.MODEL_INCOMPATIBLE) else None)
why = f"is out of capacity ({reason})" if reason else "has a stale/unrefreshable credential"
logger.warning("Auxiliary %s%s: fallback candidate %s %s (%s) — skipping to next fallback",
task or "call", tag, fb_label, why, fb_err)
def _plan_fallback_auth_retry(
destination: _FallbackDestination,
rebuild: Callable[[str, Any, Optional[str]], tuple[_FallbackDestination, dict[str, Any]]], *,
async_mode: bool,
failed_api_key: str = "",
) -> tuple[str, Optional[tuple[Any, dict[str, Any], _FallbackDestination]]]:
"""After an auth error on a fallback candidate: refresh credentials and rebuild the request.
Returns ``(refresh_provider, retry)``; ``retry`` = ``(client, kwargs, destination)`` or None."""
fb_provider = _auth_refresh_provider_for_route(destination.provider, destination.base_url)
refresh_kwargs = {"failed_api_key": failed_api_key} if fb_provider == "anthropic" else {}
if fb_provider not in {"auto", "", None} and _refresh_provider_credentials(fb_provider, **refresh_kwargs):
retry_client, retry_model = _get_cached_client(
fb_provider, destination.model, **({"async_mode": True} if async_mode else {}),
base_url=destination.base_url or None, api_mode=destination.api_mode,
)
if retry_client is not None:
retry_destination, retry_kwargs = rebuild(fb_provider, retry_client, retry_model)
return fb_provider, (retry_client, retry_kwargs, retry_destination)
return fb_provider, None
def _fallback_candidate_ladder(fb_client: Any, fb_kwargs: dict[str, Any], destination: _FallbackDestination,
rebuild: Callable[..., Any], *, task: Optional[str], fb_label: str, tag: str, async_mode: bool):
"""One fallback candidate, shared by the sync and async helpers: on an auth error refresh its
credentials and retry once with a rebuilt client; if that also auth-fails, quarantine the
provider and return None so the caller moves on. Any other failure another candidate could
avoid also quarantines and returns None so the ordered chain advances (#106367); other errors raise."""
resp, failure, repairs = yield _LadderStep("attempt", (fb_client, fb_kwargs, destination))
fb_provider = destination.provider
if failure is not None and aux_errors.auth_refresh(failure.verdict):
fb_provider, retry = _plan_fallback_auth_retry(
destination, rebuild, async_mode=async_mode, failed_api_key=getattr(fb_client, "api_key", ""))
if retry is not None: # re-apply the first attempt's strips; a failure after any strip is after a spent rung
resp, failure, more = yield _LadderStep("attempt", (retry[0], reapply_repairs(retry[1], repairs), retry[2]))
destination, repairs = retry[2], (*repairs, *more) # a failure from here is the refreshed destination's
if failure is not None and (retry is None or aux_errors.auth_refresh(failure.verdict)):
_quarantine_fallback_candidate(task, fb_label, fb_provider, failure.error, base_url=destination.base_url, tag=tag)
return None
if failure is None:
return resp
reason = aux_errors.fallback_label(failure, rung_spent=bool(repairs))
if reason is None:
raise failure.error
_quarantine_fallback_candidate(
task, fb_label, fb_provider, failure.error, base_url=destination.base_url, tag=tag, reason=reason,
model=failure.verdict.model or destination.model)
return None
def _candidate_classification_provider(fb_client: Any, fb_model: Optional[str], fb_label: str,
task: Optional[str]) -> str:
destination = _fallback_destination(task, fb_client, fb_model, fb_label)
return _classification_provider(fb_client, destination.provider)
def _call_fallback_candidate_sync(
fb_client: Any, fb_model: Optional[str], fb_label: str, *, task: Optional[str], messages: list,
temperature: Optional[float], max_tokens: Optional[int], tools: Optional[list],
effective_timeout: float, effective_extra_body: dict, reasoning_config: Optional[dict],
) -> Optional[Any]:
"""Call one fallback candidate with stale-credential recovery (``_fallback_candidate_ladder``).
``effective_timeout`` is the task-level deadline; a configured-chain candidate with its own ``timeout``
entry gets that instead, so a fallback tuned differently from the primary is allowed its own budget
(#62452).
"""
destination, fb_kwargs, rebuild = _plan_fallback_candidate(
fb_client, fb_model, fb_label, task=task, effective_timeout=effective_timeout,
apply_fast_lane=True, messages=messages, tools=tools, temperature=temperature,
max_tokens=max_tokens, effective_extra_body=effective_extra_body,
reasoning_config=reasoning_config,
)
provider = _candidate_classification_provider(fb_client, fb_model, fb_label, task)
def _send(client: Any, request_kwargs: dict[str, Any], dest: _FallbackDestination) -> Any:
return _validate_llm_response(
_relay_sync_completion(
client, request_kwargs, provider=dest.provider, api_mode=dest.api_mode,
create=lambda request: _create_with_progress(
client, request, task, provider=provider,
force_stream=_provider_requires_stream(dest.provider, dest.base_url),
),
),
task,
)
from agent.auxiliary_fallback_recovery import send_with_parameter_rungs
def _attempt(step: _LadderStep) -> Any:
client, request_kwargs, dest = step.args
return send_with_parameter_rungs(
lambda c, kw: _send(c, kw, dest), client, request_kwargs, task=task, provider=provider,
base_url=dest.base_url)
return _drive_ladder(_fallback_candidate_ladder(
fb_client, fb_kwargs, destination, rebuild, task=task, fb_label=fb_label, tag="", async_mode=False),
_attempt)
async def _call_fallback_candidate_async(
fb_client: Any, fb_model: Optional[str], fb_label: str, *, task: Optional[str], messages: list,
temperature: Optional[float], max_tokens: Optional[int], tools: Optional[list],
effective_timeout: float, effective_extra_body: dict, reasoning_config: Optional[dict], classify_as: str,
) -> Optional[Any]:
"""Async mirror of :func:`_call_fallback_candidate_sync` (no fast-lane cap on this wire)."""
destination, fb_kwargs, rebuild = _plan_fallback_candidate(
fb_client, fb_model, fb_label, task=task, effective_timeout=effective_timeout,
apply_fast_lane=False, messages=messages, tools=tools, temperature=temperature,
max_tokens=max_tokens, effective_extra_body=effective_extra_body,
reasoning_config=reasoning_config,
)
provider = classify_as
async def _send(client: Any, request_kwargs: dict[str, Any], dest: _FallbackDestination) -> Any:
import asyncio
return await asyncio.to_thread(
_validate_llm_response,
await _relay_async_completion(client, request_kwargs, provider=dest.provider, api_mode=dest.api_mode,
create=lambda request: _acreate_with_progress(client, request, provider=provider)),
task,
)
from agent.auxiliary_fallback_recovery import send_with_parameter_rungs_async
async def _attempt(step: _LadderStep) -> Any:
client, request_kwargs, dest = step.args
return await send_with_parameter_rungs_async(
lambda c, kw: _send(c, kw, dest), client, request_kwargs, task=task, provider=provider,
base_url=dest.base_url)
return await _drive_ladder_async(_fallback_candidate_ladder(
fb_client, fb_kwargs, destination, rebuild, task=task, fb_label=fb_label, tag=" (async)", async_mode=True), _attempt)
def _try_payment_fallback(
failed_provider: str, task: str | None = None, reason: str = "payment error", *,
failed_base_url: str = "", failure_scope: Any = None, main_runtime: Optional[dict[str, Any]] = None, skip_lanes=(),
) -> tuple[Optional[Any], Optional[str], str]:
"""Try the auto-detection chain after a payment/credit or connection error, skipping the failed
provider (and the main-provider path when it maps to the same backend) and the walk's ``skip_lanes``."""
skip = failed_provider.lower().strip()
# The SESSION's provider decides whether discovery is allowed: a live `/model xai-oauth` session
# over a persisted ``provider: auto`` is a selection, so the disk value alone is not the answer.
main_provider = _normalize_main_runtime(main_runtime).get("provider") or _read_main_provider()
if not _discovery_chain_allowed(main_provider, task):
return None, None, ""
skip_labels = {skip}
if main_provider and main_provider.lower() in skip:
skip_labels.add(main_provider.lower())
skip_chain_labels = {_normalize_chain_label(s) for s in skip_labels}
skip_backend = _failed_backend_skip(
failed_provider, None, failed_base_url=failed_base_url, failure_scope=failure_scope)
tried = []
for label, try_fn in _get_provider_chain():
candidate_base_url = _custom_health_base_url(label)
if (not failed_base_url and label in skip_chain_labels) or skip_backend(
label, None, candidate_base_url):
continue
if _is_provider_unhealthy(label, candidate_base_url):
_log_skip_unhealthy(label, task, base_url=candidate_base_url)
tried.append(f"{label} (unhealthy)")
continue
client, model = try_fn()
if (client is not None and not _is_provider_unhealthy(label, candidate_base_url, model)
and _fallback_lane(client, model, label) not in skip_lanes):
logger.info("Auxiliary %s: %s on %s — falling back to %s (%s)",
task or "call", reason, failed_provider, label, model or "default")
return client, model, label
tried.append(label)
logger.warning("Auxiliary %s: %s on %s and no fallback available (tried: %s)",
task or "call", reason, failed_provider, ", ".join(tried))
return None, None, ""
def _fallback_lane(client: Any, model: Optional[str], label: str) -> tuple[str, Optional[str], str]:
return label, model, str(getattr(client, "base_url", "") or "") # one candidate's identity within a walk
def _failed_backend_skip(
failed_provider: str, failed_model: Optional[str], *, failed_base_url: str = "",
failure_scope: Any = None,
) -> Callable[..., bool]:
"""Predicate ``skip(provider, model, base_url="")`` → True when a candidate must be skipped for the failed
route. Scope: ``failed_model`` → model-scoped (only that deployment; timeout/connection/rate-limit);
None → credential-wide (whole provider; auth/payment)."""
from agent.backend_identity import BackendIdentity, FailureScope, should_skip_candidate
skip_model = (failed_model or "").strip().lower() or None
failed_ident = BackendIdentity.build(
provider=failed_provider, model=skip_model, base_url=failed_base_url)
failure_scope = failure_scope or (FailureScope.MODEL if skip_model else FailureScope.CREDENTIAL)
def _skip(provider: str, model: Optional[str], base_url: str = "") -> bool:
return should_skip_candidate(
BackendIdentity.build(provider=provider, model=model, base_url=base_url), failed_ident, failure_scope,
)
return _skip
def _try_main_agent_model_fallback(
failed_provider: str, task: str | None = None, reason: str = "error",
failed_model: Optional[str] = None, failed_base_url: str = "", failure_scope: Any = None,
) -> tuple[Optional[Any], Optional[str], str]:
"""Last-resort fallback to the main agent provider + model after the configured chain is exhausted.
``failed_model`` scoping per ``_failed_backend_skip``; same-URL custom endpoints serve many models,
so a hung aux model says nothing about the main model's health. Returns (client, model, label) or (None, None, "")."""
main_provider = (_read_main_provider() or "").strip()
main_model = (_read_main_model() or "").strip()
if main_provider.lower() == "moa":
# MoA virtual provider: fall back to the preset's aggregator (the acting model).
_agg_provider, _agg_model = _resolve_moa_aggregator(main_model)
if not _agg_provider or not _agg_model:
return None, None, ""
main_provider, main_model = _agg_provider, _agg_model
if not main_provider or not main_model or main_provider.lower() in {"auto", ""}:
return None, None, ""
if task == "vision" and (
main_provider in _PROVIDERS_WITHOUT_VISION or not _main_model_supports_vision(main_provider, main_model)):
# Same capability gate as the auto-route (_vision_main_provider_client): handing an image to a
# text-only main model turns a transient 429 into a guaranteed 400 (#108349).
logger.info("Auxiliary vision: %s on %s — main agent provider %s accepts no image input, not falling back",
reason, failed_provider, main_provider)
return None, None, ""
main_base_url = _custom_health_base_url(main_provider)
if _failed_backend_skip(
failed_provider, failed_model, failed_base_url=failed_base_url,
failure_scope=failure_scope)(main_provider, main_model, main_base_url):
return None, None, ""
if _is_provider_unhealthy(main_provider, main_base_url, main_model):
_log_skip_unhealthy(main_provider, task, base_url=main_base_url, model=main_model)
return None, None, ""
try:
client, resolved_model = resolve_provider_client(provider=main_provider, model=main_model)
except Exception:
client, resolved_model = None, None
if client is None:
return None, None, ""
label = f"main-agent({main_provider})"
logger.info("Auxiliary %s: %s on %s — falling back to main agent model %s (%s)",
task or "call", reason, failed_provider, label, resolved_model or main_model)
return client, resolved_model or main_model, label
# Context-window screening for runtime fallback chains: the startup feasibility check filters
# too-small aux models; runtime chains must too, or compression stops at a reachable-but-too-small
# candidate. ``None`` (unknown) passes through.
# ── Context-window screening for runtime fallback chains (issue #52392) ── When the runtime auxiliary
# fallback chain selects a candidate that is reachable but has a context window smaller than the compression
# task requires, the call errors out instead of continuing to the next, viable candidate. The startup
# feasibility check in ``agent.conversation_compression.check_compression_model_feasibility`` already
# filters too-small auxiliary models at startup, but the runtime fallback chain
# (``_try_configured_fallback_chain`` and ``_try_main_fallback_chain``) does not apply the same filter, so
# compression can stop at the first alive door even if the room behind it is too small. The helpers below
# screen each candidate by its effective context window before it is returned. ``None`` results from
# ``get_model_context_length`` are passed through (we cannot prove a model is too small, so we do not block
# it). This preserves the existing fallback surface for unrecognised/custom models while closing the gap on
# the well-known ones.
def _task_minimum_context_length(task: Optional[str]) -> Optional[int]:
"""Minimum context length for an auxiliary task; None = no floor (only ``compression`` has one)."""
return MINIMUM_CONTEXT_LENGTH if task == "compression" else None
def _candidate_context_window(provider: str, model: str, base_url: str = "", api_key: str = "") -> Optional[int]:
"""Best-effort context window for a fallback candidate; ``None`` = unknown (never raises; callers pass it through)."""
if not model:
return None
try:
ctx = get_model_context_length(model, base_url=base_url, api_key=api_key, provider=provider)
except Exception as exc:
logger.debug("Auxiliary fallback: could not resolve context window for %s/%s: %s", provider, model, exc)
return None
return ctx if isinstance(ctx, int) and ctx > 0 else None
def _context_too_small(
entry: dict[str, Any], provider: str, model: str, min_ctx: Optional[int], *,
task: Optional[str], label: str, name_model: bool = False,
) -> Optional[str]:
"""Screen one fallback candidate by context window; returns the ``tried`` note when it is too small."""
if min_ctx is None:
return None
fb_ctx = _candidate_context_window(
provider, model, base_url=str(entry.get("base_url") or ""), api_key=_fallback_entry_api_key(entry) or "")
if fb_ctx is None or fb_ctx >= min_ctx:
return None
if name_model:
logger.info("Auxiliary %s: skipping %s (%s context=%d < min=%d), continuing chain",
task, label, model, fb_ctx, min_ctx)
else:
logger.info("Auxiliary %s: skipping %s (context=%d < min=%d), continuing chain",
task or "call", label, fb_ctx, min_ctx)
return f"{label} (context too small: {fb_ctx}<{min_ctx})"
def _try_configured_fallback_chain(
task: str, failed_provider: str, reason: str = "error", failed_model: Optional[str] = None, *,
failed_base_url: str = "", failure_scope: Any = None, skip_lanes=(),
) -> tuple[Optional[Any], Optional[str], str]:
"""Try auxiliary.<task>.fallback_chain entries in order (each needs ``provider``; model/base_url/api_key optional).
``failed_model`` scoping per ``_failed_backend_skip`` (sibling models on the same provider still run after a
model-scoped failure); this walk's ``skip_lanes`` are passed over. Returns (client, model, label)."""
if not task:
return None, None, ""
chain = _get_auxiliary_task_config(task).get("fallback_chain")
if not chain or not isinstance(chain, list):
return None, None, ""
skip = _failed_backend_skip(
failed_provider, failed_model, failed_base_url=failed_base_url, failure_scope=failure_scope)
tried = []
min_ctx = _task_minimum_context_length(task)
for i, entry in enumerate(chain):
if not isinstance(entry, dict):
continue
fb_provider = str(entry.get("provider", "")).strip()
if not fb_provider:
continue
fb_model_raw = str(entry.get("model", "")).strip()
fb_base_url = _custom_health_base_url(fb_provider, entry.get("base_url"))
if skip(fb_provider, fb_model_raw, fb_base_url):
continue
if _is_provider_unhealthy(fb_provider, fb_base_url, fb_model_raw):
_log_skip_unhealthy(fb_provider, task, base_url=fb_base_url, model=fb_model_raw)
tried.append(f"fallback_chain[{i}]({fb_provider}) (unhealthy)")
continue
fb_model = fb_model_raw or None
label = f"fallback_chain[{i}]({fb_provider})"
try:
fb_client, resolved_model = _resolve_fallback_entry(entry)
except Exception:
fb_client, resolved_model = None, None
if fb_client is not None and _fallback_lane(fb_client, resolved_model or fb_model, label) not in skip_lanes:
too_small = _context_too_small(
entry, fb_provider, resolved_model, min_ctx, task=task, label=label, name_model=True,
) if resolved_model else None
if too_small:
tried.append(too_small)
continue
logger.info("Auxiliary %s: %s on %s — configured fallback to %s (%s)",
task, reason, failed_provider, label, resolved_model or fb_model or "default")
return fb_client, resolved_model or fb_model, label
tried.append(label)
if tried:
logger.debug("Auxiliary %s: configured fallback_chain exhausted (tried: %s)", task, ", ".join(tried))
return None, None, ""
def _try_configured_fallback_for_unavailable_client(
task: Optional[str], failed_provider: str
) -> tuple[Optional[Any], Optional[str], str]:
"""Task fallback_chain when an explicit aux provider cannot build a client (no key/OAuth/pool creds);
stops at the per-task chain — the main-agent model stays the runtime last resort."""
explicit = (failed_provider or "").strip().lower()
if not task or not explicit or explicit in {"auto"}:
return None, None, ""
return _try_configured_fallback_chain(task, explicit, reason="provider unavailable")
def _fallback_entry_api_key(entry: dict[str, Any]) -> Optional[str]:
"""Resolve inline or env-backed API key via the secret-scope-aware resolver (no raw os.getenv under multiplexing)."""
from hermes_cli.fallback_config import resolve_entry_api_key
return resolve_entry_api_key(entry)
def _resolve_fallback_entry(entry: dict[str, Any]) -> tuple[Optional[Any], Optional[str]]:
"""Resolve one fallback entry through the central provider router."""
provider = str(entry.get("provider") or "").strip()
model = str(entry.get("model") or "").strip() or None
if not provider or not model:
return None, None
client, resolved_model = resolve_provider_client(
provider, model=model, explicit_base_url=str(entry.get("base_url") or "").strip() or None,
explicit_api_key=_fallback_entry_api_key(entry),
api_mode=str(entry.get("api_mode") or entry.get("transport") or "").strip() or None,
)
if client is not None:
with contextlib.suppress(Exception):
client._hermes_fallback_destination = _fallback_destination_from_entry(entry, client, resolved_model)
return client, resolved_model
def _try_main_fallback_chain(
task: Optional[str], failed_provider: str = "", reason: str = "error", *,
failed_model: Optional[str] = None, failed_base_url: str = "", failure_scope: Any = None, skip_lanes=(),
) -> tuple[Optional[Any], Optional[str], str]:
"""Top-level main-agent fallback chain for a ``provider: auto`` auxiliary call: auto tasks honour the
user's main fallback policy before the built-in discovery chain; read via ``get_fallback_chain`` so
``fallback_providers`` and legacy ``fallback_model`` keep the main agent's order."""
try:
from hermes_cli.config import load_config_readonly
from hermes_cli.fallback_config import get_fallback_chain
chain = get_fallback_chain(load_config_readonly())
except Exception as exc:
logger.debug("Auxiliary %s: could not load main fallback chain: %s", task or "call", exc)
return None, None, ""
if not chain:
return None, None, ""
skip = _failed_backend_skip(
failed_provider, failed_model, failed_base_url=failed_base_url, failure_scope=failure_scope)
tried: list[str] = []
min_ctx = _task_minimum_context_length(task)
for i, entry in enumerate(chain):
if not isinstance(entry, dict):
continue
fb_provider = str(entry.get("provider") or "").strip()
fb_model = str(entry.get("model") or "").strip()
if not fb_provider or not fb_model:
continue
fb_norm = fb_provider.lower()
label = f"fallback_providers[{i}]({fb_provider})"
fb_base_url = _custom_health_base_url(fb_provider, entry.get("base_url"))
if fb_norm == "auto" or skip(fb_provider, fb_model, fb_base_url):
tried.append(f"{label} (skipped)")
continue
if _is_provider_unhealthy(fb_norm, fb_base_url, fb_model):
_log_skip_unhealthy(fb_norm, task, base_url=fb_base_url, model=fb_model)
tried.append(f"{label} (unhealthy)")
continue
try:
fb_client, resolved_model = _resolve_fallback_entry(entry)
except Exception as exc:
logger.debug("Auxiliary %s: main fallback %s failed to resolve: %s", task or "call", label, exc)
fb_client, resolved_model = None, None
if fb_client is not None and _fallback_lane(fb_client, resolved_model or fb_model, fb_provider) not in skip_lanes:
too_small = _context_too_small(
entry, fb_provider, resolved_model or fb_model, min_ctx, task=task, label=label,
)
if too_small:
tried.append(too_small)
continue
logger.info("Auxiliary %s: %s on %s — main fallback chain to %s (%s)",
task or "call", reason, failed_provider or "auto", label, resolved_model or fb_model)
return fb_client, resolved_model or fb_model, fb_provider
tried.append(label)
if tried:
logger.debug("Auxiliary %s: main fallback chain exhausted (tried: %s)", task or "call", ", ".join(tried))
return None, None, ""
def _warn_stale_openai_base_url(runtime_provider: str) -> None:
"""Warn once when OPENAI_BASE_URL is set but config.yaml names a non-custom provider (a stale
~/.hermes/.env value after `hermes model` poisons routing)."""
global _stale_base_url_warned
if _stale_base_url_warned:
return
_env_base = os.getenv("OPENAI_BASE_URL", "").strip()
_cfg_provider = runtime_provider or _read_main_provider()
if (_env_base and _cfg_provider and _cfg_provider != "custom" and not _cfg_provider.startswith("custom:")):
logger.warning(
"OPENAI_BASE_URL is set (%s) but model.provider is '%s'. "
"Auxiliary clients may route to the wrong endpoint. "
"Run: hermes model to reconfigure, or remove "
"OPENAI_BASE_URL from ~/.hermes/.env",
_env_base, _cfg_provider,
)
_stale_base_url_warned = True
def _main_route_target(runtime: dict[str, Any], task: Optional[str]) -> tuple[str, str, str, Any, str]:
"""Step-1 target: (provider, model, base_url, api_key, api_mode) of the main runtime, after the
fast-model opt-in and the MoA aggregator substitution."""
main_provider = str(runtime.get("provider", "") or _read_main_provider() or "")
main_model = str(runtime.get("model") or _read_main_model() or "")
runtime_base_url = str(runtime.get("base_url") or "")
runtime_api_key = runtime.get("api_key", "")
runtime_api_mode = str(runtime.get("api_mode") or "")
# Latency-critical tasks (titling only) opt in to the provider's fast model. Opt-in only:
# every settings surface defines "auto" as the main model.
if _task_prefers_fast_model(task) and main_provider and main_provider not in {"auto", ""}:
fast_model = _get_aux_model_for_provider(main_provider, prefer_fast=True)
if fast_model and fast_model != main_model:
logger.debug("Auxiliary task %s: preferring fast model %s over main model %s",
task, fast_model, main_model)
main_model = fast_model
# MoA virtual provider: the preset name is not a wire model; run aux on the aggregator and drop
# the facade's "moa://local" base_url / placeholder key so it uses its own credentials.
if main_provider == "moa":
_agg_provider, _agg_model = _resolve_moa_aggregator(main_model)
if _agg_provider and _agg_model:
main_provider, main_model = _agg_provider, _agg_model
runtime_base_url = runtime_api_key = runtime_api_mode = ""
return main_provider, main_model, runtime_base_url, runtime_api_key, runtime_api_mode
def _try_main_provider_route(
main_provider: str, main_model: str, runtime_base_url: str, runtime_api_key: Any, runtime_api_mode: str,
) -> Optional[tuple[Any, str, str]]:
"""Step 1: route aux onto the main provider + main model; None if unusable."""
if not (main_provider and main_model and main_provider not in {"auto", ""}):
return None
resolved_provider = main_provider
explicit_base_url = runtime_base_url or None
health_base_url = _custom_health_base_url(main_provider, explicit_base_url)
explicit_api_key = None
if runtime_base_url and main_provider == "custom":
# Anonymous custom endpoint — pass through explicit base_url + api_key.
explicit_api_key = runtime_api_key or None
elif main_provider.startswith("custom:"):
# Named custom provider (custom_providers / providers dict entry).
_has_named_entry = False
with contextlib.suppress(ImportError):
from hermes_cli.runtime_provider import _get_named_custom_provider
_has_named_entry = _get_named_custom_provider(main_provider) is not None
if _has_named_entry:
# KEEP the full ``custom:<name>`` so the named arm honours the entry's api_mode
# (collapsing to "custom" strips /anthropic → 404s). base_url/api_key come from the entry.
explicit_base_url = None
elif runtime_base_url:
# Config-less named custom provider (live runtime only): anonymous custom arm + runtime key.
# See #34777.
resolved_provider = "custom"
explicit_api_key = runtime_api_key or None
elif runtime_api_key:
explicit_api_key = runtime_api_key
elif runtime_api_key:
# Pin aux to the main session's working key, not a re-selected (maybe exhausted) pool key.
explicit_api_key = runtime_api_key
# Skip if the main provider was recently 402'd (unhealthy TTL bounds the bypass).
main_chain_label = _normalize_chain_label(resolved_provider)
if main_chain_label and _is_provider_unhealthy(main_chain_label, health_base_url):
_log_skip_unhealthy(main_chain_label, base_url=health_base_url)
return None
client, resolved = resolve_provider_client(
resolved_provider, main_model, explicit_base_url=explicit_base_url,
explicit_api_key=explicit_api_key, api_mode=runtime_api_mode or None,
)
if client is None:
return None
logger.info("Auxiliary auto-detect: using main provider %s (%s)", main_provider, resolved or main_model)
return client, resolved or main_model, resolved_provider
def _discovery_chain_allowed(main_provider: str, task: Optional[str] = None) -> bool:
"""The built-in discovery chain is a convenience for installs with NO selected main provider.
Once the user picked one, every auxiliary route must be a provider they configured (main,
``auxiliary.<task>``, ``fallback_providers``); guessing "whatever else is logged in" bills an
account they never pointed this session at (xAI OAuth session with a dead token → every
compression silently charged to a Nous Portal balance)."""
if (main_provider or "").strip().lower() in {"", "auto"}:
return True
logger.warning(
"Auxiliary %s: main provider %s is unavailable and no fallback_chain / fallback_providers is "
"configured — refusing to guess another logged-in provider. Re-authenticate (`hermes model`) "
"or declare a fallback.", task or "call", main_provider)
return False
def _try_discovery_chain() -> tuple[Optional[OpenAI], Optional[str], str]:
"""Step 3: hardcoded aggregator/fallback chain, skipping unhealthy providers."""
tried = []
for label, try_fn in _get_provider_chain():
candidate_base_url = _custom_health_base_url(label)
if _is_provider_unhealthy(label, candidate_base_url):
_log_skip_unhealthy(label, base_url=candidate_base_url)
tried.append(f"{label} (unhealthy)")
continue
client, model = try_fn()
if client is not None:
if tried:
logger.info("Auxiliary auto-detect: using %s (%s) — skipped: %s",
label, model or "default", ", ".join(tried))
else:
logger.info("Auxiliary auto-detect: using %s (%s)", label, model or "default")
return client, model, label
tried.append(label)
logger.warning("Auxiliary auto-detect: no provider available (tried: %s). "
"Compression, summarization, and memory flush will not work. "
"Set OPENROUTER_API_KEY or configure a local model in config.yaml.", ", ".join(tried))
return None, None, ""
def _resolve_auto_route(
main_runtime: Optional[dict[str, Any]] = None, task: Optional[str] = None
) -> tuple[Optional[OpenAI], Optional[str], str]:
"""Full auto-detection chain, including the selected provider identity. Priority: (1) main provider +
main model, regardless of provider type ("auto" means "my main model for side tasks too"; explicit
per-task overrides still win); (2) configured fallback policy — task chain, then the main agent's
top-level chain; (3) OpenRouter → Nous → custom → Codex → API-key providers, only with no policy
and no working main client."""
runtime = _normalize_main_runtime(main_runtime)
_warn_stale_openai_base_url(runtime.get("provider", ""))
main_provider, main_model, base_url, api_key, api_mode = _main_route_target(runtime, task)
routed = _try_main_provider_route(main_provider, main_model, base_url, api_key, api_mode)
if routed is not None:
return routed
if task:
fb_client, fb_model, fb_label = _try_configured_fallback_chain(
task, main_provider or "auto", reason="main provider unavailable")
if fb_client is not None:
return fb_client, fb_model, _fallback_provider_from_label(fb_label)
fb_client, fb_model, fb_label = _try_main_fallback_chain(
task, main_provider or "auto", reason="main provider unavailable")
if fb_client is not None:
return fb_client, fb_model, fb_label
if not _discovery_chain_allowed(main_provider, task):
return None, None, ""
return _try_discovery_chain()
def _effective_provider_for_client(client: Any, fallback: str) -> str:
"""Return the concrete provider selected for an auto-routed client."""
effective_provider = getattr(client, "_hermes_aux_effective_provider", "")
if isinstance(effective_provider, str) and effective_provider:
return effective_provider
return str(fallback or "")
def _classification_provider(client: Any, fallback: str, main_runtime: Optional[dict[str, Any]] = None) -> str:
"""The provider a failure on ``client`` is classified on: the one discovery built it for, else its route. A route
label (``custom``, ``auto``, ``api-key``) names no profile: the pool recovery would act on stands in, so the one
verdict per request is already that backend's (an OpenCode Go upstream throttle keeps the key)."""
backend = getattr(client, "_hermes_aux_backend_provider", "")
if isinstance(backend, str) and backend:
return backend
label = _fallback_provider_from_label(_effective_provider_for_client(client, fallback))
if (route := _normalize_aux_provider(label)) not in {"auto", "custom", "api-key"}:
return label
return _recoverable_pool_provider("custom" if route == "custom" else "auto", client,
main_runtime=_normalize_main_runtime(main_runtime), same_path=True) or label
# Centralized Provider Router: resolve_provider_client() is the single entry point for building a configured
# client (auth, base URL, headers, API format) from (provider, model). Never read auth env vars ad-hoc.
def _to_async_client(sync_client, model: str, is_vision: bool = False):
"""Sync client → async counterpart, preserving Codex routing (``is_vision`` keeps the endpoint's vision headers)."""
from openai import AsyncOpenAI
if isinstance(sync_client, _AuxProbeClientStub):
return sync_client, model
if isinstance(sync_client, CodexAuxiliaryClient):
return AsyncCodexAuxiliaryClient(sync_client), model
if isinstance(sync_client, AnthropicAuxiliaryClient):
return AsyncAnthropicAuxiliaryClient(sync_client), model
if isinstance(sync_client, BedrockAuxiliaryClient):
return AsyncBedrockAuxiliaryClient(sync_client), model
with contextlib.suppress(ImportError):
from agent.gemini_native_adapter import GeminiNativeClient, AsyncGeminiNativeClient
if isinstance(sync_client, GeminiNativeClient):
return AsyncGeminiNativeClient(sync_client), model
# ACP shims (subprocess, not an HTTP pool) are already async-safe and opt out of the wrapper.
if _client_declares(sync_client, "HERMES_SKIP_ASYNC_WRAP"):
return sync_client, model
sync_base_url = str(sync_client.base_url)
# A key_cmd/Entra client keeps its credential in the SDK's per-request provider slot, not in
# ``.api_key`` (which stays ""); rebuilding from the snapshot alone ships NO Authorization
# header. Configured default_headers (a named entry's extra_headers) are likewise carried over,
# merged last exactly as the SDK merges them on the sync client. See #109595.
from agent.auxiliary_async_rebuild import async_api_key, configured_default_headers
async_kwargs = {"api_key": async_api_key(sync_client), "base_url": sync_base_url}
if base_url_host_matches(sync_base_url, "openrouter.ai"):
headers = _apply_user_default_headers(build_or_headers())
elif _is_official_codex_base_url(sync_base_url):
headers = _apply_user_default_headers(_codex_cloudflare_headers(sync_client.api_key, base_url=sync_base_url))
else:
# Provider for the profile-header fallback is inferred from the hostname.
try:
from agent.model_metadata import _infer_provider_from_url
inferred = _infer_provider_from_url(sync_base_url) or ""
except Exception:
inferred = ""
headers = _endpoint_default_headers(sync_base_url, inferred, is_vision=is_vision, xai=True)
headers = {**(headers or {}), **configured_default_headers(sync_client)}
if headers:
async_kwargs["default_headers"] = headers
_apply_required_codex_headers(async_kwargs, access_token=sync_client.api_key, base_url=sync_base_url)
async_kwargs = {**_openai_http_client_kwargs(sync_base_url, async_mode=True), **async_kwargs}
# Hermes owns the auxiliary retry/timeout budget; disable SDK-internal retries.
# See #54465.
async_kwargs.setdefault("max_retries", 0)
return AsyncOpenAI(**async_kwargs), model
def _normalize_resolved_model(model_name: Optional[str], provider: str) -> Optional[str]:
"""Normalize a resolved model for the provider that will receive it."""
if not model_name:
return model_name
try:
from hermes_cli.model_normalize import normalize_model_for_provider
return normalize_model_for_provider(model_name, provider)
except Exception:
return model_name
def _named_custom_api_key(custom_entry: dict[str, Any], provider: str, custom_base: str, model: Optional[str] = None) -> Any:
"""Credential for a named custom provider: inline api_key → key_env → key_cmd → credential pool → placeholder.
Aux resolves named custom providers here, not via _resolve_named_custom_runtime, so key_cmd must be
honoured at the same precedence or every aux call 401s."""
custom_key: Any = (custom_entry.get("api_key") or "").strip()
custom_key_env = (custom_entry.get("key_env") or custom_entry.get("api_key_env") or "").strip()
if not custom_key and custom_key_env:
custom_key = _scoped_key_env(custom_key_env)
custom_key_cmd = str(custom_entry.get("key_cmd", "") or "").strip()
if custom_key_cmd:
from agent.command_token_source import build_command_token_provider
custom_key = build_command_token_provider(custom_key_cmd, custom_entry.get("name") or provider) or custom_key
if not custom_key:
with contextlib.suppress(Exception):
from agent.credential_pool import custom_provider_pool_key_candidates
pool_name = custom_entry.get("provider_key") or custom_entry.get("name") or provider
for pool_key in custom_provider_pool_key_candidates(custom_base, pool_name):
try:
pool = load_pool(pool_key)
except Exception:
continue
if (pool := entry_scoped_pool(pool, pool_key, custom_entry)) is None or not pool.has_credentials():
continue
if (pool_entry := pool.select(model=model)) is None:
continue
pool_api_key = getattr(pool_entry, "runtime_api_key", None) or getattr(pool_entry, "access_token", "") or ""
if str(pool_api_key).strip():
custom_key = str(pool_api_key).strip()
break
return custom_key or "no-key-required"
def _build_bedrock_client(provider: str, model: Optional[str], *, raw_codex: bool) -> tuple[Optional[Any], Optional[str]]:
"""AWS Bedrock: Claude → Anthropic Bedrock SDK (prompt caching, thinking); bare in-Region OpenAI IDs
→ Mantle Responses; everything else, incl. OpenAI ``us.``/``global.`` profiles, → Converse API."""
try:
from agent.bedrock_adapter import (
has_aws_credentials, is_anthropic_bedrock_model, resolve_bedrock_runtime_region,
bedrock_openai_uses_mantle, bedrock_openai_base_url, resolve_bedrock_bearer_token,
configure_bedrock_openai_client_kwargs,
)
from agent.anthropic_adapter import build_anthropic_bedrock_client
except ImportError:
logger.warning("resolve_provider_client: bedrock requested but boto3, httpx/openai, or anthropic SDK not installed")
return None, None
if not has_aws_credentials():
logger.debug("resolve_provider_client: bedrock requested but no AWS credentials found")
return None, None
# Region must match the main runtime's resolution (bedrock.region in config first, then
# env/profile) so aux calls never leave the primary runtime's configured region.
# See #53880, #65076.
region = resolve_bedrock_runtime_region()
default_model = "anthropic.claude-haiku-4-5-20251001-v1:0"
final_model = _normalize_resolved_model(model or default_model, provider) or default_model
if bedrock_openai_uses_mantle(final_model):
# Module-level lazy ``OpenAI`` proxy on purpose so tests can patch("agent.auxiliary_client.OpenAI").
client_kwargs: dict[str, Any] = {
"api_key": resolve_bedrock_bearer_token() or "aws-sdk",
"base_url": bedrock_openai_base_url(region),
}
configure_bedrock_openai_client_kwargs(client_kwargs)
client = OpenAI(**client_kwargs)
logger.debug("resolve_provider_client: bedrock-openai (%s, %s)", final_model, region)
return (client if raw_codex else CodexAuxiliaryClient(client, final_model)), final_model
base_url = f"https://bedrock-runtime.{region}.amazonaws.com"
if is_anthropic_bedrock_model(final_model):
try:
real_client = build_anthropic_bedrock_client(region)
except ImportError as exc:
logger.warning("resolve_provider_client: cannot create Bedrock client: %s", exc)
return None, None
client = AnthropicAuxiliaryClient(real_client, final_model, api_key="aws-sdk", base_url=base_url)
logger.debug("resolve_provider_client: bedrock anthropic (%s, %s)", final_model, region)
else:
client = BedrockAuxiliaryClient(region, final_model)
logger.debug("resolve_provider_client: bedrock converse (%s, %s)", final_model, region)
return client, final_model
def _build_vertex_client(provider: str, model: Optional[str]) -> tuple[Optional[Any], Optional[str]]:
"""Google Vertex AI: Gemini via the OpenAI-compatible endpoint with an OAuth2 bearer (standard OpenAI client)."""
try:
from agent.vertex_adapter import get_vertex_config, has_vertex_credentials
except ImportError:
logger.warning("resolve_provider_client: vertex requested but google-auth not installed")
return None, None
if not has_vertex_credentials():
logger.debug("resolve_provider_client: vertex requested but no GCP credentials found")
return None, None
token, base_url = get_vertex_config()
if not token or not base_url:
logger.warning("resolve_provider_client: vertex requested but could not mint token / resolve project")
return None, None
final_model = _normalize_resolved_model(model or "google/gemini-3-flash-preview", provider)
try:
# Aliased import: a bare `from openai import OpenAI` would shadow the module-level lazy proxy.
from openai import OpenAI as _VertexOpenAI
client = _VertexOpenAI(api_key=token, base_url=base_url)
except Exception as exc:
logger.warning("resolve_provider_client: cannot create Vertex client: %s", exc)
return None, None
logger.debug("resolve_provider_client: vertex (%s)", final_model)
return client, final_model
class _ResolveRequest(NamedTuple):
"""Normalized resolve_provider_client() arguments shared by the per-provider branch helpers."""
provider: str
original_provider: str
model: Optional[str]
async_mode: bool
raw_codex: bool
explicit_base_url: Optional[str]
explicit_api_key: Optional[str | Callable[[], str]]
api_mode: Optional[str]
main_runtime: Optional[dict[str, Any]]
is_vision: bool
task: Optional[str]
_ResolveResult = tuple[Optional[Any], Optional[str]]
def _normalize_api_key(raw: Any) -> str | Callable[[], str]:
"""A key_cmd/Entra callable passes through uncalled; strings are stripped; anything else is ''."""
if callable(raw) and not isinstance(raw, str):
return raw
return raw.strip() if isinstance(raw, str) else ""
def _log_once_debug(seen: set, key: Any, msg: str, *args: Any) -> None:
"""Debug-log ``msg`` the first time ``key`` is seen so per-call retries stay silent."""
if key not in seen:
seen.add(key)
logger.debug(msg, *args)
def _is_actual_auxiliary_route(req: _ResolveRequest, base_url: str) -> bool:
from hermes_cli.auth import normalize_actual_base_url
from hermes_cli.providers import is_actual_route
from hermes_cli.route_identity import normalize_route_base_url
if is_actual_route(req.provider, base_url):
return True
runtime = _normalize_main_runtime(req.main_runtime)
return bool(
base_url
and is_actual_route(runtime.get("provider", ""), runtime.get("base_url", ""))
and normalize_route_base_url(normalize_actual_base_url(base_url))
== normalize_route_base_url(
normalize_actual_base_url(runtime.get("base_url", ""))
)
)
def _wrap_transport(req: _ResolveRequest, client_obj: Any, final_model_str: str,
base_url_str: str = "", api_key_str: str = ""):
"""Wrap a plain OpenAI client in the right transport adapter; specialized wrappers pass through.
Codex (Responses API): explicit ``api_mode=codex_responses``, else — with no
explicit api_mode — api.openai.com + codex model. Anthropic (Messages): ``api_mode=anthropic_messages``,
any ``/anthropic`` suffix, ``api.kimi.com/coding``, or ``api.anthropic.com``."""
if _is_actual_auxiliary_route(req, base_url_str):
client = (
client_obj._real_client
if isinstance(client_obj, CodexAuxiliaryClient)
else client_obj
)
client._hermes_aux_effective_provider = "actual"
return client
# OpenCode relay targets pick the wire per model; a task/provider-level api_mode is stale for
# every other model (#98799), so it is re-derived here like the main runtime does.
from agent.opencode_affinity import opencode_transport
_oc_mode, _oc_base = opencode_transport(req.provider, final_model_str, base_url_str)
if _oc_mode:
req, base_url_str = req._replace(api_mode=_oc_mode), _oc_base
needs_codex = not (
isinstance(client_obj, CodexAuxiliaryClient) or req.raw_codex
) and (
req.api_mode == "codex_responses"
or (
not req.api_mode
and base_url_hostname(base_url_str) == "api.openai.com"
and "codex" in (final_model_str or "").lower()
)
)
if needs_codex:
logger.debug("resolve_provider_client: wrapping client in CodexAuxiliaryClient "
"(api_mode=%s, model=%s, base_url=%s)",
req.api_mode or "auto-detected", final_model_str, base_url_str[:60] if base_url_str else "")
return CodexAuxiliaryClient(client_obj, final_model_str)
# A profile that declares the Messages wire (commandcode-anthropic) is on it whatever the URL
# looks like; the same declaration gates ``_reasoning_config`` in _build_call_kwargs.
api_mode = req.api_mode or _profile_declared_messages_wire(req.provider)
return _maybe_wrap_anthropic(client_obj, final_model_str, api_key_str, base_url_str, api_mode)
def _profile_declared_messages_wire(provider: str) -> Optional[str]:
"""``"anthropic_messages"`` when the registered profile declares that api_mode, else None."""
from providers import get_provider_profile
profile = get_provider_profile(str(provider or "").strip().lower())
return "anthropic_messages" if profile is not None and profile.api_mode == "anthropic_messages" else None
def _route_client(req: _ResolveRequest, client_obj: Any, final_model_str: Optional[str]) -> _ResolveResult:
"""Return (client, model), converting to the async wrapper when ``req.async_mode``."""
if req.async_mode:
return _to_async_client(client_obj, final_model_str, is_vision=req.is_vision)
return client_obj, final_model_str
def _route_or_warn(req: _ResolveRequest, client: Any, default: Optional[str], unavailable_msg: str, *args: Any) -> _ResolveResult:
"""Route ``client`` on ``req.model or default``; warn and return (None, None) when the provider produced no client."""
if client is None:
logger.warning(unavailable_msg, *args)
return None, None
return _route_client(req, client, _normalize_resolved_model(req.model or default, req.provider))
def _resolve_auto_branch(req: _ResolveRequest) -> _ResolveResult:
"""Auto: try all providers in priority order; tag the client with the effective provider (survives cache reuse)."""
client, resolved, effective_provider = _resolve_auto_route(main_runtime=req.main_runtime, task=req.task)
if client is None:
return None, None
model = req.model
# An OpenRouter-format model override won't work on a non-OpenRouter provider (e.g. local
# server); drop it for the provider's default.
if model and "/" in model and resolved and "/" not in resolved:
logger.debug("Dropping OpenRouter-format model %r for non-OpenRouter "
"auxiliary provider (using %r instead)", model, resolved)
model = None
routed_client, routed_model = _route_client(req, client, model or resolved)
if routed_client is not None and effective_provider:
try:
routed_client._hermes_aux_effective_provider = effective_provider
if routed_client is not client: # the async rebuild keeps the provider discovery built the client for
_tag_backend(routed_client, getattr(client, "_hermes_aux_backend_provider", None))
except (AttributeError, TypeError):
logger.debug("Auxiliary client %s cannot retain effective provider %s",
type(routed_client).__name__, effective_provider)
return routed_client, routed_model
def _resolve_openrouter_branch(req: _ResolveRequest) -> _ResolveResult:
"""OpenRouter."""
client, default = _try_openrouter(explicit_api_key=req.explicit_api_key, model=req.model,
explicit_base_url=req.explicit_base_url)
if client is None:
logger.warning("resolve_provider_client: openrouter requested but %s",
_describe_openrouter_unavailable(model=req.model))
return None, None
return _route_client(req, client, _normalize_resolved_model(req.model or default, req.provider))
def _resolve_nous_branch(req: _ResolveRequest) -> _ResolveResult:
"""Nous Portal (OAuth)."""
model = req.model
# Vision: caller flag, _PROVIDER_VISION_MODELS override, or a known vision id.
client, default = _try_nous(vision=(req.is_vision or model in _PROVIDER_VISION_MODELS.values()
or (model or "").strip().lower() == "mimo-v2-omni"))
if client is None:
logger.warning("resolve_provider_client: nous requested but Nous Portal not configured (run: hermes auth)")
return None, None
final_model = _normalize_resolved_model(model or default, req.provider)
# Dual-wire: anthropic/* → /v1/messages, else /chat/completions. Derive from the catalog id
# (not a stale api_mode) so aux matches the main agent.
from hermes_cli.providers import nous_api_mode
client = _maybe_wrap_anthropic(
client, final_model, str(getattr(client, "api_key", "") or ""),
str(getattr(client, "base_url", "") or ""), nous_api_mode(final_model),
)
return _route_client(req, client, final_model)
def _resolve_openai_codex_branch(req: _ResolveRequest) -> _ResolveResult:
"""OpenAI Codex (OAuth → Responses API)."""
model = req.model
if not model:
logger.warning("resolve_provider_client: openai-codex requested without a "
"model; pass model explicitly (e.g. model.model in config.yaml "
"or auxiliary.<task>.model for per-task aux routing).")
return None, None
no_token_msg = "resolve_provider_client: openai-codex requested but no Codex OAuth token found (run: hermes model)"
if req.raw_codex:
# Raw OpenAI client for callers needing responses.stream() (main agent loop).
codex_token, base_url = _resolve_codex_credential_and_base(model)
if not codex_token:
logger.warning(no_token_msg)
return None, None
raw_client = _create_openai_client(api_key=codex_token, base_url=base_url,
default_headers=_codex_cloudflare_headers(codex_token, base_url=base_url))
return raw_client, _normalize_resolved_model(model, req.provider)
client, default = _build_codex_client(model)
return _route_or_warn(req, client, default, no_token_msg)
def _resolve_xai_oauth_branch(req: _ResolveRequest) -> _ResolveResult:
"""xAI Grok OAuth (device code → Responses API). Without this branch xai-oauth falls to the generic
oauth_external arm, returns (None, None), and silently re-routes every aux task to the Step-2 fallback."""
client, default = _build_xai_oauth_aux_client(req.model)
return _route_or_warn(req, client, default,
"resolve_provider_client: xai-oauth requested but no xAI "
"OAuth token found (run: hermes model -> xAI Grok OAuth — SuperGrok / Premium+)")
def _resolve_custom_branch(req: _ResolveRequest) -> _ResolveResult:
"""Custom endpoint (OPENAI_BASE_URL + OPENAI_API_KEY)."""
provider, model, main_runtime = req.provider, req.model, req.main_runtime
# wrap_base: base for the Anthropic-wrap decision. anthropic_messages must keep the raw
# /anthropic base while the plain OpenAI client uses the /v1-rewritten custom_base (never
# /anthropic/chat/completions). Empty means "use custom_base".
custom_base = custom_key = wrap_base = ""
if req.explicit_base_url:
custom_base = _to_openai_base_url(req.explicit_base_url).strip()
# Ollama/vLLM/llama.cpp serve the OpenAI surface under /v1; a bare host posts to the
# native API and 404s (#106010). Only the alias group gets the tail — ``custom`` is verbatim.
if req.original_provider in _LOCAL_SERVER_ALIASES and not urlparse(custom_base).path.strip("/"):
custom_base = custom_base.rstrip("/") + "/v1"
if req.api_mode == "anthropic_messages":
wrap_base = (req.explicit_base_url or "").strip().rstrip("/")
if req.original_provider in _LOCAL_SERVER_ALIASES:
# SECURITY: a local-server alias never borrows OPENAI_API_KEY or the main key —
# the alias means "this is my own server"; sending an OpenAI secret to whatever
# host base_url names is never intended. Explicit api_key or the placeholder only.
custom_key = _normalize_api_key(req.explicit_api_key) or "no-key-required"
else:
custom_key = (
_normalize_api_key(req.explicit_api_key)
or _scoped_key_env("OPENAI_API_KEY")
or _read_main_api_key_if_same_origin(custom_base)
or "no-key-required" # local servers don't need auth
)
if not custom_base:
logger.warning("resolve_provider_client: explicit custom endpoint requested but base_url is empty")
return None, None
elif main_runtime:
# Reuse main_runtime's concrete base_url + api_key for a named custom provider;
# re-resolving from bare "custom" loses the name and lands on the wrong provider.
# Re-resolution loses the provider name and falls back to OpenRouter or a wrong API-key provider —
# the main agent already solved this, we just need to reuse its answer. (#45472)
_main_base = str(main_runtime.get("base_url") or "").strip().rstrip("/")
_main_key = _normalize_api_key(main_runtime.get("api_key"))
if _main_base and _main_key:
custom_base, custom_key = _main_base, _main_key
if custom_base and custom_key:
if _is_actual_auxiliary_route(req, custom_base):
from hermes_cli.auth import normalize_actual_base_url
custom_base = normalize_actual_base_url(custom_base)
final_model = _normalize_resolved_model(
model or (main_runtime.get("model") if main_runtime else None) or "gpt-4o-mini", provider,
)
_clean_base, _dq = _extract_url_query_params(custom_base)
extra = {"default_query": _dq} if _dq else {}
from hermes_cli.config import get_custom_provider_extra_headers # this endpoint's entry, as agent_init
_custom_headers = {**(_endpoint_default_headers(custom_base, provider, is_vision=req.is_vision) or {}),
**get_custom_provider_extra_headers(custom_base)}
if _custom_headers:
extra["default_headers"] = _custom_headers
client = _create_openai_client(api_key=custom_key, base_url=_clean_base, **extra)
client = _wrap_transport(req, client, final_model, wrap_base or custom_base, custom_key)
return _route_client(req, client, final_model)
# Try custom first, then API-key providers (Codex excluded here:
# falling through to Codex with no model is a stale-constant trap).
for try_fn in (_try_custom_endpoint, _resolve_api_key_provider):
client, default = try_fn()
if client is not None:
final_model = _normalize_resolved_model(model or default, provider)
# ``client.api_key`` may be a callable (Azure Entra bearer provider);
# wrapping decisions only need base_url + api_mode.
_raw_ckey = getattr(client, "api_key", "")
_ckey = "" if (callable(_raw_ckey) and not isinstance(_raw_ckey, str)) else str(_raw_ckey or "")
client = _wrap_transport(req, client, final_model, str(getattr(client, "base_url", "") or ""), _ckey)
return _route_client(req, client, final_model)
logger.warning("resolve_provider_client: custom/main requested but no endpoint credentials found")
return None, None
def _named_custom_openai_wire_client(custom_base: str, custom_key: Any, extra_headers: Optional[dict[str, str]] = None):
"""Plain OpenAI client on the /v1 equivalent of a named custom entry's base URL.
``extra_headers`` is the entry's own header block (gateway routing tags, proxy auth): the main
runtime lifts it onto every request, so aux must too (#109595). SECURITY: never log the values."""
_clean_base, _dq = _extract_url_query_params(_to_openai_base_url(custom_base))
_extra = {"default_query": _dq} if _dq else {}
_headers = _apply_user_default_headers(None)
if extra_headers:
_headers = {**(_headers or {}), **extra_headers}
if _headers:
_extra["default_headers"] = _headers
return _create_openai_client(api_key=custom_key, base_url=_clean_base, **_extra)
def _resolve_named_custom_branch(req: _ResolveRequest) -> Optional[_ResolveResult]:
"""Named custom provider (config.yaml providers dict / custom_providers list); None if no entry matches."""
from hermes_cli.runtime_provider import _get_named_custom_provider
provider = req.provider
# If the raw name is an alias (``kimi`` → ``kimi-coding``) and a custom_providers entry exists
# under it, the custom entry wins over alias rewriting. Only for aliases, so entries matching a
# canonical name (e.g. ``nous``) still defer to the built-in.
custom_entry = None
if req.original_provider and req.original_provider != provider:
custom_entry = _get_named_custom_provider(req.original_provider)
if custom_entry is None:
custom_entry = _get_named_custom_provider(provider)
if not custom_entry:
return None
# A per-task/explicit base_url or api_key composes OVER the named entry's defaults: the entry supplies
# whatever the caller left blank, never replaces what the caller set (compression prompts carry
# conversation history, so a silently swapped destination is a data-routing bug, not a nuisance).
custom_base = (req.explicit_base_url or custom_entry.get("base_url") or "").strip()
custom_key = _normalize_api_key(req.explicit_api_key) or _named_custom_api_key(custom_entry, provider, custom_base, req.model)
if custom_key == "no-key-required":
logger.warning("resolve_provider_client: named custom provider %r has no resolvable "
"api_key — request will be sent with placeholder no-key-required "
"and will 401 on auth-required endpoints", custom_entry.get("name") or provider)
# Actual's wire protocol takes precedence over persisted task/provider modes.
entry_api_mode = (req.api_mode or custom_entry.get("api_mode") or "").strip()
if _is_actual_auxiliary_route(req, custom_base):
from hermes_cli.auth import normalize_actual_base_url
custom_base = normalize_actual_base_url(custom_base)
entry_api_mode = "chat_completions"
if not custom_base:
logger.warning("resolve_provider_client: named custom provider %r has no base_url", provider)
return None, None
from hermes_cli.config import normalize_extra_headers
entry_headers = normalize_extra_headers(custom_entry.get("extra_headers"))
final_model = _normalize_resolved_model(
req.model
or custom_entry.get("model")
or (req.main_runtime.get("model") if req.main_runtime else None)
or _read_main_model_for_aux()
or "gpt-4o-mini",
provider,
)
# An OpenCode-family entry (``opencode-go-bridge``, #85589) persisted the api_mode of whichever
# model was selected at save time; the relay picks the wire per model (#98799).
from agent.opencode_affinity import opencode_transport
_oc_mode, _oc_base = opencode_transport(provider, final_model, custom_base)
if _oc_mode:
entry_api_mode, custom_base = _oc_mode, _oc_base
logger.debug("resolve_provider_client: named custom provider %r (%s, api_mode=%s)",
provider, final_model, entry_api_mode or "chat_completions")
# anthropic_messages: route via AnthropicAuxiliaryClient (mirrors _try_custom_endpoint);
# the Anthropic SDK sees the original (un-rewritten) URL.
# Mirrors the anonymous-custom branch in _try_custom_endpoint(). See #15033.
if entry_api_mode == "anthropic_messages":
try:
from agent.anthropic_adapter import build_anthropic_client
from agent.anthropic_credentials import anthropic_route_is_oauth
real_client = build_anthropic_client(custom_key, custom_base)
if entry_headers:
# Same entry headers as the two OpenAI-wire arms; ``with_options`` merges onto the
# beta/credential-Omit headers the builder installed (#109595).
real_client = real_client.with_options(default_headers=entry_headers)
except ImportError:
logger.warning("Named custom provider %r declares api_mode=anthropic_messages but the anthropic SDK "
"is not installed — falling back to OpenAI-wire.", provider)
return _route_client(req, _named_custom_openai_wire_client(custom_base, custom_key, entry_headers), final_model)
return _route_client(
req, AnthropicAuxiliaryClient(real_client, final_model, custom_key, custom_base,
is_oauth=anthropic_route_is_oauth(custom_base, custom_key)), final_model)
client = _named_custom_openai_wire_client(custom_base, custom_key, entry_headers)
# codex_responses, or auto-detect via _wrap_transport (which reads the task-level api_mode).
if entry_api_mode == "codex_responses":
client = CodexAuxiliaryClient(client, final_model)
else:
client = _wrap_transport(req, client, final_model, custom_base, custom_key)
return _route_client(req, client, final_model)
def _resolve_azure_foundry_branch(req: _ResolveRequest) -> _ResolveResult:
"""Azure Foundry via the runtime resolver: the generic PROVIDER_REGISTRY path only knows the static
AZURE_FOUNDRY_API_KEY env var, missing ``auth_mode: entra_id`` (callable bearer) and config base_url overrides."""
client, default_model = _try_azure_foundry(model=req.model, explicit_api_key=req.explicit_api_key,
explicit_base_url=req.explicit_base_url, api_mode=req.api_mode)
return _route_or_warn(req, client, default_model,
"resolve_provider_client: azure-foundry requested but "
"runtime resolution failed (run: hermes doctor for diagnostics)")
def _api_key_profile_supplied_client(provider: str, **client_kwargs: Any) -> Any | None:
"""Registered profile's own client for an ``api_key`` aux route, or ``None``.
Same registration seam as ``agent_runtime_helpers._provider_supplied_client`` (main agent)
and the ``external_process`` branch below: a profile whose wire protocol is not
OpenAI-over-HTTP overrides ``ProviderProfile.create_client()`` to supply its transport.
A profile that raises is logged and skipped — a third-party plugin can only fail to
provide a client, never take the auxiliary resolution down."""
try:
from providers import get_provider_profile
profile = get_provider_profile(provider)
except Exception:
return None
if profile is None:
return None
try:
return profile.create_client(**client_kwargs)
except Exception:
logger.warning("resolve_provider_client: provider profile %r failed to create an "
"auxiliary client; falling back to the standard client path",
provider, exc_info=True)
return None
def _resolve_api_key_branch(req: _ResolveRequest, pconfig: Any, resolve_creds: Callable) -> _ResolveResult:
"""PROVIDER_REGISTRY ``api_key`` providers (Anthropic via its own resolver), honouring explicit overrides."""
provider = req.provider
if provider == "anthropic":
client, default_model = _try_anthropic(explicit_api_key=req.explicit_api_key,
explicit_base_url=req.explicit_base_url, requested_model=req.model)
return _route_or_warn(req, client, default_model,
"resolve_provider_client: anthropic requested but no Anthropic credentials found")
creds = resolve_creds(provider)
api_key = str(creds.get("api_key", "")).strip()
# Explicit api_key override (fallback_model / custom_providers entry) lets callers
# authenticate where no built-in credential is registered for this alias.
api_key = _normalize_api_key(req.explicit_api_key) or api_key
raw_base_url = str(creds.get("base_url", "")).strip().rstrip("/") or pconfig.inference_base_url
if req.explicit_base_url:
raw_base_url = req.explicit_base_url.strip().rstrip("/")
if provider == "actual":
with contextlib.suppress(Exception):
from hermes_cli.auth import (
ACTUAL_LOCAL_NOAUTH_PLACEHOLDER, is_actual_local_base_url, normalize_actual_base_url
)
raw_base_url = normalize_actual_base_url(raw_base_url)
if not api_key and is_actual_local_base_url(raw_base_url):
api_key = ACTUAL_LOCAL_NOAUTH_PLACEHOLDER
if not api_key:
logger.debug("resolve_provider_client: provider %s has no API key configured (tried: %s)",
provider, ", ".join(pconfig.api_key_env_vars))
return None, None
base_url = _to_openai_base_url(raw_base_url)
# Explicit base_url override: a fallback_model/custom_providers entry pointing a built-in name elsewhere.
if req.explicit_base_url and provider != "actual":
base_url = _to_openai_base_url(req.explicit_base_url.strip().rstrip("/"))
final_model = _normalize_resolved_model(req.model or _get_aux_model_for_provider(provider), provider)
# Consulted before the built-in gemini/OpenAI ladder so a registered native transport wins (#112384).
profile_client = _api_key_profile_supplied_client(provider, api_key=api_key, base_url=base_url)
if profile_client is not None:
logger.debug("resolve_provider_client: %s native client from provider profile (%s)", provider, final_model)
return _route_client(req, profile_client, final_model)
if provider == "gemini":
from agent.gemini_native_adapter import GeminiNativeClient, is_native_gemini_base_url
if is_native_gemini_base_url(base_url):
client = GeminiNativeClient(api_key=api_key, base_url=base_url)
logger.debug("resolve_provider_client: %s (%s)", provider, final_model)
return _route_client(req, client, final_model)
headers = _endpoint_default_headers(base_url, provider, is_vision=req.is_vision, xai=True)
client = _create_openai_client(api_key=api_key, base_url=base_url, **({"default_headers": headers} if headers else {}))
from agent.provider_route_hooks import model_api_mode # a Responses-only model: call_llm() streams responses
from providers import get_provider_profile
profile = get_provider_profile(provider) if final_model and not req.raw_codex else None
if profile is not None and model_api_mode(profile, final_model) == "codex_responses":
logger.debug("resolve_provider_client: %s model %s needs Responses API — wrapping", provider, final_model)
client = CodexAuxiliaryClient(client, final_model)
# api_mode handling for any API-key provider (direct OpenAI + codex model) and Anthropic-wire
# endpoints (api.kimi.com/coding, /anthropic gateways) without per-provider branches.
client = _wrap_transport(req, client, final_model, raw_base_url, api_key)
logger.debug("resolve_provider_client: %s (%s)", provider, final_model)
return _route_client(req, client, final_model)
def _resolve_external_process_branch(req: _ResolveRequest, creds: dict[str, Any]) -> _ResolveResult:
"""PROVIDER_REGISTRY ``external_process`` providers, served via their registered profile."""
provider = req.provider
final_model = _normalize_resolved_model(
req.model or (req.main_runtime.get("model") if req.main_runtime else None) or _read_main_model_for_aux(),
provider,
)
# Keyed on the registered profile, not a provider name, so an out-of-tree ACP provider reaches
# the auxiliary path (compression, vision, background review) exactly like the in-tree one.
try:
from providers import get_provider_profile as _get_provider_profile
_extproc_profile = _get_provider_profile(provider)
except Exception:
_extproc_profile = None
if _extproc_profile is not None:
api_key = str(creds.get("api_key", "")).strip()
base_url = str(creds.get("base_url", "")).strip()
if not final_model:
logger.warning("resolve_provider_client: %s requested but no model was provided or configured", provider)
return None, None
if not api_key or not base_url:
logger.warning("resolve_provider_client: %s requested but external process credentials are incomplete", provider)
return None, None
try:
client = _extproc_profile.create_client(
api_key=api_key, base_url=base_url,
command=str(creds.get("command", "")).strip() or None, args=list(creds.get("args") or []))
except Exception:
logger.warning("resolve_provider_client: profile %r failed to create an external-process client",
provider, exc_info=True)
client = None
if client is not None:
logger.debug("resolve_provider_client: %s (%s)", provider, final_model)
return _route_client(req, client, final_model)
_log_once_debug(_LOGGED_UNSUPPORTED_EXTPROC_KEYS, provider,
"resolve_provider_client: external-process provider %s not "
"directly supported", provider)
return None, None
def _resolve_registry_branch(req: _ResolveRequest) -> _ResolveResult:
"""PROVIDER_REGISTRY providers, dispatched on ``auth_type``; unknown providers log once."""
provider = req.provider
try:
from hermes_cli.auth import (
PROVIDER_REGISTRY, resolve_api_key_provider_credentials,
resolve_external_process_provider_credentials,
)
except ImportError:
logger.debug("hermes_cli.auth not available for provider %s", provider)
return None, None
pconfig = PROVIDER_REGISTRY.get(provider)
if pconfig is None:
_log_once_debug(_LOGGED_UNKNOWN_PROVIDER_KEYS, provider,
"resolve_provider_client: unknown provider %r", provider)
return None, None
auth_type = pconfig.auth_type
if auth_type == "api_key":
return _resolve_api_key_branch(req, pconfig, resolve_api_key_provider_credentials)
if auth_type == "external_process":
return _resolve_external_process_branch(req, resolve_external_process_provider_credentials(provider))
from agent.auxiliary_client_registry import REGISTRY_AUTHTYPE_ARMS
arm = REGISTRY_AUTHTYPE_ARMS.get(auth_type)
if arm is not None:
return arm(req)
# The first occurrence surfaces a real schema-drift bug; per-call retries stay silent.
_log_once_debug(_LOGGED_UNHANDLED_AUTHTYPE_KEYS, (auth_type, provider),
"resolve_provider_client: unhandled auth_type %s for %s",
auth_type, provider)
return None, None
# Explicit providers with a dedicated branch; anything else falls through to named custom
# providers → azure-foundry → PROVIDER_REGISTRY (order preserved from the original if-chain).
_EXPLICIT_PROVIDER_BRANCHES: dict[str, Callable[[_ResolveRequest], _ResolveResult]] = {
"auto": _resolve_auto_branch,
"openrouter": _resolve_openrouter_branch,
"nous": _resolve_nous_branch,
"openai-codex": _resolve_openai_codex_branch,
"xai-oauth": _resolve_xai_oauth_branch,
"custom": _resolve_custom_branch,
}
def resolve_provider_client(
provider: str, model: str | None = None, async_mode: bool = False, raw_codex: bool = False,
explicit_base_url: str | None = None, explicit_api_key: Optional[str | Callable[[], str]] = None,
api_mode: str | None = None, main_runtime: Optional[dict[str, Any]] = None, is_vision: bool = False,
task: Optional[str] = None,
) -> tuple[Optional[Any], Optional[str]]:
"""Central router: return a configured client (auth, base URL, API format) for a provider + optional model.
The client always exposes ``.chat.completions.create()``; Codex/Responses providers get an adapter.
``provider``: built-in name, ``custom:<name>``, "custom" (OPENAI_BASE_URL + OPENAI_API_KEY) or "auto"
(full auto-detection chain). ``model=None`` → provider's default aux model. ``raw_codex`` → bare OpenAI
client for ``responses.stream()`` callers. ``api_mode`` forces "codex_responses"/"chat_completions"/
"anthropic_messages" instead of auto-detect. Returns (client, resolved_model) or (None, None)."""
_validate_proxy_env_urls()
# Keep the pre-alias name so a custom_providers entry named like a built-in alias
# (e.g. "kimi" → "kimi-coding") is still reachable via the named-custom branch.
original_provider = (provider or "").strip().lower()
provider = _normalize_aux_provider(provider)
api_mode = _canonical_api_mode(str(api_mode or "")).lower() or None
# MoA chokepoint: "moa" is not an HTTP provider; resolve to the aggregator so direct callers don't
# dead-end in unknown-provider. Unresolvable preset → leave untouched for the normal diagnostic.
if provider == "moa":
_agg_provider, _agg_model = _resolve_moa_aggregator(model)
if _agg_provider and _agg_model:
original_provider = _agg_provider.strip().lower()
provider = _normalize_aux_provider(_agg_provider)
model = _agg_model
# The moa:// facade endpoint/key belong to the virtual runtime, not the aggregator.
if explicit_base_url and str(explicit_base_url).lower().startswith("moa://"):
explicit_base_url = None
explicit_api_key = None
from agent.auxiliary_local_runtime import bare_llamacpp_endpoint
local = bare_llamacpp_endpoint(original_provider, explicit_base_url, explicit_api_key)
if local is not None:
if not local[0]:
logger.warning("resolve_provider_client: %s requested but no local llama.cpp server is running", original_provider)
return None, None
explicit_base_url, explicit_api_key = local
# Model for concrete providers: caller ``model`` → catalog default (empty for OAuth-gated providers whose
# lists drift) → configured main model (MoA → aggregator), keeping OAuth aux tasks off the Step-2 fallback.
# Excluded: ``auto`` (a stale main slug could pair with any picked provider) and Nous + vision (the
# Portal's tier-aware vision recommendation must win over a text-only model).
if not model and provider != "auto" and not (provider == "nous" and is_vision):
# ``auto`` is intentionally excluded: `_resolve_auto_route(main_runtime=...)` returns the model paired
# with the provider it actually selected. Pre-filling an auto call from `_read_main_model()` can
# leak a stale process-global runtime into a different provider (for example Claude model slug on
# Codex OAuth) and override that correctly resolved model. 1. ``model`` argument (caller knew what
# they wanted) 2. Provider's catalog default — cheap/fast model the provider registered via
# ``ProviderProfile.default_aux_model`` or the legacy ``_API_KEY_PROVIDER_AUX_MODELS_FALLBACK``
# dict. 3. User's main model from ``model.model`` in config.yaml. This is the load-bearing step for
# OAuth providers: an xai-oauth user with grok-4.3 configured gets grok-4.3 for title generation
# instead of silently dropping to whatever Step-2 fallback (#31845). When the main provider is MoA,
# ``_read_main_model_for_aux()`` substitutes the preset's aggregator model — the preset NAME is
# never a valid wire model id, so unset aux models default to the preset's acting model instead.
# Each provider branch below sees a non-empty ``model`` whenever the user has *anything* configured
# — no provider-specific empty-model guards needed. When the user has NOTHING configured (fresh
# install, main_model also empty), the branches still hit their own missing-credentials returns and
# ``_resolve_auto_route`` falls through to the Step-2 chain as before. Do NOT pre-fill a blank ``auto``
# request from the config/main default here. Claude model sent to Codex after the main lane fell
# back to gpt-5.5). Let _resolve_auto_route() return the actual current runtime model when the caller did
# not explicitly request one. (# compression-current-model) Nous + vision is the one carve-out: the
# branch below resolves its model from the Portal's tier-aware vision recommendation
# (``_try_nous(vision= True)``), and ``final_model = model or default`` means anything pre-filled
# here wins over that. The main chat model is routinely text-only (e.g. a ``:free`` chat SKU), so
# pre-filling it sends the image to a model that cannot accept one and the Portal 404s. Leave
# ``model`` unset and let the Portal slot through; only an explicit caller model may override it.
model = _get_aux_model_for_provider(provider) or _read_main_model_for_aux() or model
req = _ResolveRequest(
provider, original_provider, model, async_mode, raw_codex,
explicit_base_url, explicit_api_key, api_mode, main_runtime, is_vision, task,
)
branch = _EXPLICIT_PROVIDER_BRANCHES.get(provider)
alias_identity = original_provider.removeprefix("custom:")
# A configured provider whose name collides with a local-server alias is still a named
# provider. Resolve it before the alias's generic ``custom`` branch, which otherwise loses
# the configured base_url and can fall through to an unrelated credentialed provider.
if branch is None or alias_identity in _LOCAL_SERVER_ALIASES:
try:
result = _resolve_named_custom_branch(req)
except ImportError:
result = None
if result is not None:
return result
if branch is not None:
return branch(req)
if provider == "azure-foundry":
return _resolve_azure_foundry_branch(req)
return _resolve_registry_branch(req)
# ── Public API ──────────────────────────────────────────────────────────────
def get_text_auxiliary_client(task: str = "", *, main_runtime: Optional[dict[str, Any]] = None) -> tuple[Optional[OpenAI], Optional[str]]:
"""Return (client, default_model_slug) for text-only aux tasks; ``task`` selects auxiliary.<task> overrides."""
provider, model, base_url, api_key, api_mode = _resolve_task_provider_model(task or None)
return resolve_provider_client(
provider, model=model, explicit_base_url=base_url, explicit_api_key=api_key,
api_mode=api_mode, main_runtime=main_runtime,
)
_VISION_AUTO_PROVIDER_ORDER = ("openrouter", "nous", "deepinfra")
def _main_model_supports_vision(provider: str, model: Optional[str]) -> bool:
"""True when ``provider``/``model`` is known to accept image input; unknown capability → True (attempt the call)."""
try:
from agent.image_routing import _lookup_supports_vision
from hermes_cli.config import load_config_readonly
except ImportError:
return True
try:
supports = _lookup_supports_vision(provider, model, load_config_readonly())
except Exception: # pragma: no cover - defensive
return True
return True if supports is None else bool(supports)
def _normalize_vision_provider(provider: Optional[str]) -> str:
return _normalize_aux_provider(provider)
def _deepinfra_strict_vision_backend(model: Optional[str]) -> tuple[Optional[Any], Optional[str]]:
"""DeepInfra vision: default model is discovered live via default_vision_model() so no hardcoded id can rot."""
vision_model = model or _resolve_provider_vision_default("deepinfra")
if not vision_model:
logger.debug("Vision auto-detect: deepinfra catalog unreachable or returned no vision-tagged models — skipping")
return None, None
return resolve_provider_client("deepinfra", vision_model, is_vision=True)
# Strict (explicitly requested) vision backends by normalized provider name. nous MUST go
# through resolve_provider_client so anthropic/* picks wrap onto /v1/messages (a bare _try_nous
# client 404s). openai-codex has no safe default model; callers set auxiliary.<task>.model.
_STRICT_VISION_BACKENDS: dict[str, Callable[[Optional[str]], tuple[Optional[Any], Optional[str]]]] = {
"openrouter": lambda model: _try_openrouter(model=model),
"nous": lambda model: resolve_provider_client("nous", model, is_vision=True),
"openai-codex": lambda model: resolve_provider_client("openai-codex", model, is_vision=True),
"anthropic": lambda model: _try_anthropic(requested_model=model),
"deepinfra": _deepinfra_strict_vision_backend,
"custom": lambda model: _try_custom_endpoint(),
}
def _resolve_strict_vision_backend(provider: str, model: Optional[str] = None) -> tuple[Optional[Any], Optional[str]]:
backend = _STRICT_VISION_BACKENDS.get(_normalize_vision_provider(provider))
return backend(model) if backend is not None else (None, None)
def get_available_vision_backends() -> list[str]:
"""Available vision backends in auto-selection order (active provider → OpenRouter → Nous → DeepInfra).
Single source of truth for setup, tool gating, and runtime auto-routing.
"""
available: list[str] = []
main_provider = _read_main_provider()
if main_provider and main_provider not in {"auto", ""}:
if main_provider in _VISION_AUTO_PROVIDER_ORDER:
main_ok = _resolve_strict_vision_backend(main_provider)[0] is not None
else:
main_ok = resolve_provider_client(main_provider, _read_main_model())[0] is not None
if main_ok:
available.append(main_provider)
for p in _VISION_AUTO_PROVIDER_ORDER: # skip if already covered by main provider
if p not in available and _resolve_strict_vision_backend(p)[0] is not None:
available.append(p)
return available
def _finalize_vision_client(
resolved_provider: str, sync_client: Any, default_model: Optional[str],
resolved_model: Optional[str], async_mode: bool,
) -> tuple[Optional[str], Optional[Any], Optional[str]]:
"""Apply the explicit model override (and async wrapping) to a resolved vision client."""
if sync_client is None:
return resolved_provider, None, None
final_model = resolved_model or default_model
if async_mode:
async_client, async_model = _to_async_client(sync_client, final_model, is_vision=True)
return resolved_provider, async_client, async_model
return resolved_provider, sync_client, final_model
def _vision_main_provider_client(
main_provider: str, main_model: str, runtime: dict[str, Any], resolved_model: Optional[str],
resolved_api_mode: Optional[str],
) -> tuple[Optional[Any], Optional[str]]:
"""Auto-detect step 1: try the main provider; (None, None) falls through to the aggregator chain."""
# A provider vision default (static override or catalog discovery) is a *known* multimodal
# model; the pinned chat model usually isn't, so only fall back to it when no default exists.
provider_vision_default = _resolve_provider_vision_default(main_provider)
vision_model = provider_vision_default or main_model
if main_provider == "nous":
# Nous picks its vision model from Portal tier-aware slots inside _try_nous(vision=True);
# passing the chat model would override that and 404. Only auxiliary.vision.model may.
sync_client, default_model = _resolve_strict_vision_backend(main_provider, resolved_model or provider_vision_default)
if sync_client is None:
return None, None
logger.info("Vision auto-detect: using main provider %s (%s)", main_provider, default_model or resolved_model or main_model)
return sync_client, default_model
if main_provider in _PROVIDERS_WITHOUT_VISION: # endpoint rejects image input entirely
logger.debug("Vision auto-detect: skipping main provider %s (no vision support) — falling through to aggregator chain", main_provider)
return None, None
if not _main_model_supports_vision(main_provider, vision_model):
# Known text-only model. Log only the provider name (CodeQL clear-text-logging FPs).
logger.debug(
"Vision auto-detect: skipping main provider %s (reports no vision capability) — falling through to aggregator chain",
main_provider,
)
return None, None
# Custom endpoints carry no built-in base_url/api_key: recover the live main endpoint from
# set_runtime_main() or, with no live runtime recorded, the configured custom endpoint.
rpc_base_url = rpc_api_key = None
rpc_api_mode = resolved_api_mode
if main_provider == "custom" or main_provider.startswith("custom:"):
if runtime.get("base_url"):
custom_base, custom_key, custom_mode = runtime.get("base_url"), runtime.get("api_key") or None, runtime.get("api_mode")
else:
custom_base, custom_key, custom_mode = _resolve_custom_runtime()
if custom_base:
rpc_base_url, rpc_api_key = custom_base, custom_key
rpc_api_mode = resolved_api_mode or custom_mode or None
rpc_client, rpc_model = resolve_provider_client(
main_provider, vision_model, api_mode=rpc_api_mode, explicit_base_url=rpc_base_url,
explicit_api_key=rpc_api_key, main_runtime=runtime, is_vision=True)
if rpc_client is None:
return None, None
logger.info("Vision auto-detect: using main provider %s (%s)", main_provider, rpc_model or vision_model)
return rpc_client, rpc_model or vision_model
def _vision_auto_route(
runtime: dict[str, Any], resolved_model: Optional[str], resolved_api_mode: Optional[str],
async_mode: bool,
) -> tuple[Optional[str], Optional[Any], Optional[str]]:
"""Auto-detect order: 1. main provider + model, 2. OpenRouter, 3. Nous Portal, 4. DeepInfra, 5. stop."""
main_provider = str(runtime.get("provider") or _read_main_provider())
main_model = str(runtime.get("model") or _read_main_model())
if main_provider.strip().lower() == "moa":
# MoA main_model is a preset NAME, not a wire model — unwrap to the preset's aggregator
# slot. The moa:// facade endpoint belongs to the virtual provider, not the real one.
_agg_provider, _agg_model = _resolve_moa_aggregator(main_model)
if _agg_provider and _agg_model:
main_provider, main_model = _agg_provider, _agg_model
runtime = dict(runtime, base_url="", api_key="", api_mode="")
if main_provider and main_provider not in {"auto", "", "moa"}:
client, default_model = _vision_main_provider_client(main_provider, main_model, runtime, resolved_model, resolved_api_mode)
if client is not None:
return _finalize_vision_client(main_provider, client, default_model, resolved_model, async_mode)
# Aggregators use their dedicated vision model, not the user's main model.
for candidate in _VISION_AUTO_PROVIDER_ORDER:
if candidate == main_provider:
continue # already tried above
sync_client, default_model = _resolve_strict_vision_backend(candidate)
if sync_client is not None:
return _finalize_vision_client(candidate, sync_client, default_model, resolved_model, async_mode)
logger.debug("Auxiliary vision client: none available")
return None, None, None
# ZAI vision must use the OpenAI-compatible endpoint: the Anthropic wire rejects max_tokens on
# multimodal calls (error 1210).
_ZAI_OPENAI_VISION_URLS = ("https://open.bigmodel.cn/api/paas/v4", "https://api.z.ai/api/paas/v4")
def resolve_vision_provider_client(
provider: Optional[str] = None, model: Optional[str] = None, *, base_url: Optional[str] = None,
api_key: Optional[str] = None, async_mode: bool = False,
main_runtime: Optional[dict[str, Any]] = None,
) -> tuple[Optional[str], Optional[Any], Optional[str]]:
"""Resolve the client actually used for vision tasks.
Direct endpoint overrides beat provider selection; explicit providers may force
experimental backends; auto mode only tries backends known to work.
"""
runtime = _normalize_main_runtime(main_runtime)
requested, resolved_model, resolved_base_url, resolved_api_key, resolved_api_mode = _resolve_task_provider_model(
"vision", provider, model, base_url, api_key
)
# The raw name keeps a llama.cpp alias distinguishable from bare ``custom`` for the last resolve.
raw_requested, requested = requested, _normalize_vision_provider(requested)
if resolved_base_url:
provider_for_base_override = requested if requested and requested not in {"", "auto"} else "custom"
client, final_model = resolve_provider_client(
provider_for_base_override, model=resolved_model, async_mode=async_mode,
explicit_base_url=resolved_base_url, explicit_api_key=resolved_api_key,
api_mode=resolved_api_mode, main_runtime=runtime,
)
return provider_for_base_override, client, (final_model if client is not None else None)
if requested == "auto":
return _vision_auto_route(runtime, resolved_model, resolved_api_mode, async_mode)
if requested in _VISION_AUTO_PROVIDER_ORDER:
sync_client, default_model = _resolve_strict_vision_backend(requested, resolved_model)
return _finalize_vision_client(requested, sync_client, default_model, resolved_model, async_mode)
if requested == "zai":
for _zai_url in _ZAI_OPENAI_VISION_URLS:
client, final_model = _get_cached_client(
requested, resolved_model, async_mode, base_url=_zai_url,
api_key=resolved_api_key or None, api_mode="chat_completions", main_runtime=runtime,
is_vision=True,
)
if client is not None:
return _finalize_vision_client(requested, client, final_model, resolved_model, async_mode)
# Fallback: try without explicit base_url (old behavior)
client, final_model = _get_cached_client(
raw_requested, resolved_model, async_mode, api_mode=resolved_api_mode, main_runtime=runtime, is_vision=True,
)
return requested, client, (final_model if client is not None else None)
def auxiliary_max_tokens_param(value: int, *, model: Optional[str] = None) -> dict:
"""Max-tokens kwarg for the auxiliary provider: direct OpenAI, a custom endpoint whose profile declares
it, and newer OpenAI-family models (by ``model`` name: custom endpoints fronting gpt-5.x) need max_completion_tokens."""
from agent.provider_route_hooks import route_answer
_custom_base = _current_custom_base_url()
direct_openai_family = not _scoped_key_env("OPENROUTER_API_KEY") and _read_nous_auth() is None and (
base_url_hostname(_custom_base) == "api.openai.com"
or route_answer("max_tokens_param", base_url=_custom_base, model=model or "") == "max_completion_tokens")
if direct_openai_family or model_forces_max_completion_tokens(model):
return {"max_completion_tokens": value}
return {"max_tokens": value}
# ── Centralized LLM Call API: call_llm()/async_call_llm() own resolve → cached client → shape
# request → call → return. Every auxiliary LLM consumer should use these.
# Client cache: (provider, async_mode, base_url, api_key, api_mode, runtime_key) -> (client, default_model, loop)
# Loop identity is NOT part of the key: stale-loop entries are replaced in place on async hits,
# bounding growth to one entry per provider config (avoids fd accumulation in gateways).
# This bounds cache growth to one entry per unique provider config rather than one per (config ×
# event-loop), which previously caused unbounded fd accumulation in long-running gateway processes (#10200).
_client_cache: dict[tuple, tuple] = {}
_client_cache_lock = threading.Lock()
_CLIENT_CACHE_MAX_SIZE = 64 # safety belt — evict oldest when exceeded
class _CallableCacheDiscriminator:
"""Hash a credential callback by identity without exposing its state."""
__slots__ = ("_callback",)
def __init__(self, callback: Any) -> None:
self._callback = callback # retained so its id cannot be reused while cached
def __hash__(self) -> int:
return id(self._callback)
def __eq__(self, other: object) -> bool:
return isinstance(other, _CallableCacheDiscriminator) and self._callback is other._callback
def __repr__(self) -> str:
return "<callable-api-key>"
def _runtime_cache_discriminator(field: str, value: Any) -> Any:
"""Return a hashable, secret-safe runtime cache-key component."""
if field == "api_key" and callable(value):
return _CallableCacheDiscriminator(value)
if field == "api_key" and isinstance(value, str) and value:
return ("api-key-digest", hashlib.blake2b(value.encode("utf-8"), digest_size=16).digest())
return value
def _client_cache_key(
provider: str, *, async_mode: bool, base_url: Optional[str] = None,
api_key: Optional[str] = None, api_mode: Optional[str] = None,
main_runtime: Optional[dict[str, Any]] = None, is_vision: bool = False,
task: Optional[str] = None, model: Optional[str] = None,
) -> tuple:
runtime = _normalize_main_runtime(main_runtime)
# `auto` resolves through the main runtime and task-specific policy, so both join the key.
runtime_key = tuple(_runtime_cache_discriminator(f, runtime.get(f, "")) for f in _MAIN_RUNTIME_FIELDS) if provider == "auto" else ()
task_key = (task or "", _task_prefers_fast_model(task)) if provider == "auto" else ""
pool_hint = _pool_cache_hint(provider, main_runtime=main_runtime)
# Model MUST be in the key: concurrent calls to the same endpoint with different models would
# share an entry, and the second builder's _store_cached_client would close the first's client.
model_key = model or runtime.get("model", "")
api_key_key = _runtime_cache_discriminator("api_key", api_key or "")
# Profile home leads the key: callers that omit api_key (pool / Nous auth.json paths) would
# otherwise share one client across multiplex profiles holding different credentials.
return (hermes_home_key(), provider, async_mode, base_url or "", api_key_key, api_mode or "", runtime_key, is_vision, task_key, pool_hint, model_key,
_borrowed_main_credential_key(provider, base_url, api_key, runtime))
def _borrowed_main_credential_key(provider: str, base_url: Optional[str], api_key: Any, runtime: dict[str, Any]) -> tuple:
"""What a keyless ``custom`` route borrows from the main runtime when its client is built.
The client keeps that credential for its lifetime, so it joins the cache key: otherwise a
later runtime (another session, a ``/model`` switch) is served the earlier one's key.
"""
if _normalize_aux_provider(provider) != "custom":
return ()
if not base_url:
# This shape takes the runtime's endpoint and key even when an explicit key was passed.
return (runtime.get("base_url", ""), _runtime_cache_discriminator("api_key", runtime.get("api_key", "")))
# Same normalization as the client build, which treats a blank explicit key as keyless.
if _normalize_api_key(api_key):
return ()
borrowed = _read_main_api_key_if_same_origin(_to_openai_base_url(base_url).strip())
return (_runtime_cache_discriminator("api_key", borrowed),)
def _current_event_loop() -> Any:
"""``asyncio.get_event_loop()`` or None when no loop can be obtained (async cache-key binding)."""
try:
import asyncio as _aio
return _aio.get_event_loop()
except RuntimeError:
return None
def _store_cached_client(cache_key: tuple, client: Any, default_model: Optional[str], *, bound_loop: Any = None) -> None:
if isinstance(client, _AuxProbeClientStub):
return # probe stubs must never be cached — the next hit would get a dud client
with _client_cache_lock:
old_entry = _client_cache.get(cache_key)
if old_entry is not None and old_entry[0] is not client:
_close_cached_client(old_entry[0])
_client_cache[cache_key] = (client, default_model, bound_loop)
def _refresh_nous_auxiliary_client(
*, cache_provider: str, model: Optional[str], async_mode: bool, base_url: Optional[str] = None,
api_key: Optional[str] = None, api_mode: Optional[str] = None,
main_runtime: Optional[dict[str, Any]] = None, is_vision: bool = False,
lookup_model: Optional[str] = None, lookup_task: Optional[str] = None,
) -> tuple[Optional[Any], Optional[str]]:
"""Refresh Nous runtime creds, rebuild the client, and replace the cache entry.
``model`` is the resolved wire model stored as the entry's usable model and returned. The
cache KEY MUST be built from ``lookup_model``/``lookup_task`` — the model and task as passed
to ``_get_cached_client`` when the stale client was acquired — so the fresh client overwrites
the exact entry the stale one is served from. Keying on the resolved model or an empty task
would leave the expired client immortal and every auxiliary call 401ing forever.
See #56889.
For ``provider == "auto"`` the task participates in the cache key (task-specific fallback policy), so it
MUST be carried into the key here for the same reason as ``lookup_model``; otherwise an auto-provider
client refreshed on a 401 lands under the ``task=""`` key while the stale entry survives under the
task-scoped key (#58894).
"""
runtime = _resolve_nous_runtime_api(force_refresh=True, stale_access_token=api_key)
if runtime is None:
return None, model
fresh_key, fresh_bas