d159b6e0c3
#109964 made same-model cache-parity forks (background review, /btw) inherit the parent's resolved cache scope. That is the right call for content-addressed caches (Anthropic, DeepSeek, Gemini) and for OpenAI's routing-only prompt_cache_key. xAI is different: x-grok-conv-id / prompt_cache_key select ONE server-side conversation slot, so a fork of similar size that diverges early evicts the parent's slot and the parent's next call reads cold. Measured on grok-4.7 (xai-oauth Responses), ~160k context, fork of similar size diverging right after the system prompt, parent's next call after the fork: shared scope (current): 1,152 / 162,239 read (2/2 runs) derived scope (this PR): 162,176 / 162,239 read (2/2 runs) no fork (control): 177,536 / 177,639 read (2/2 runs) build_cache_parity_fork tags the fork (_prompt_cache_fork_tag). The resolver derives "<scope>::<tag>" for tagged agents on slot-keyed routes only (xai, xai-oauth, api.x.ai, OpenRouter x-ai/grok-*). The inherited scope is kept everywhere else. The tag is applied outside the memo, so a mid-run provider fallback re-evaluates. OpenRouter's Grok x-grok-conv-id now honours a fork scope over the ambient affinity/conversation scope (chat_completions threads cache_scope_id to the profile). session_id and transcript identity are unchanged. (cherry picked from commit 40fbb39d978648c76940cfb7835f77a42d3719b0)
693 lines
36 KiB
Python
693 lines
36 KiB
Python
"""OpenAI Chat Completions transport (default api_mode for OpenAI-compatible providers).
|
|
|
|
Messages/tools are already OpenAI-shaped, so convert_* are near-identity; the
|
|
provider-specific work lives in build_kwargs (max_tokens, reasoning, extra_body).
|
|
"""
|
|
|
|
import json
|
|
from typing import Any
|
|
from urllib.parse import urlparse
|
|
|
|
from agent.lmstudio_reasoning import resolve_lmstudio_effort
|
|
from agent.reasoning_effort import (
|
|
KIMI_K3_EFFORTS, KIMI_K3_OVERRIDES, OPENAI_COMPAT_WIRE_EFFORTS, TOKENHUB_EFFORTS, clamp_effort,
|
|
clamp_reasoning_config, kimi_supported_efforts, requested_effort,
|
|
)
|
|
from agent.message_sanitization import normalize_finish_reason as _normalize_finish_reason
|
|
from agent.moonshot_schema import is_moonshot_model, sanitize_moonshot_tools
|
|
from agent.prompt_builder import DEVELOPER_ROLE_MODELS
|
|
from agent.transports.base import ProviderTransport
|
|
from agent.transports.types import NormalizedResponse, ToolCall, Usage
|
|
|
|
# xAI reserves ``tool_search`` for its server-side tool (HTTP 400 on client
|
|
# declarations); aliased on the wire, mapped back in normalize_response.
|
|
# xAI's chat-completions API reserves the function name ``tool_search`` for its own server-side tool and
|
|
# rejects any request declaring a client function with that name (HTTP 400 "The function name tool_search is
|
|
# reserved for the tool_search tool", #95003). The Tool Search bridge (tools/tool_search.py) assembles its
|
|
# client-side discovery tool under the same literal name for every provider, so Grok providers are unusable
|
|
# whenever the bridge is active. Mirror the web_search treatment in transports/codex.py
|
|
# (_rename_client_web_search_for_xai): alias the wire declaration and map the alias back in
|
|
# normalize_response. The alias value matches _CODEX_TOOL_SEARCH_ALIAS from the Codex-side fix for the same
|
|
# reserved-name class (#83122) so the two transports stay consistent.
|
|
_XAI_TOOL_SEARCH_ALIAS = "hermes_tool_search"
|
|
|
|
# Persistence-only / cross-transport message keys that strict OpenAI-compatible
|
|
# providers reject with HTTP 400 ("Extra inputs are not permitted").
|
|
_STRIP_MSG_KEYS = (
|
|
"codex_reasoning_items", "codex_message_items", "tool_name", "effect_disposition", "timestamp",
|
|
"platform_message_id", "api_content", "anthropic_content_blocks", "bedrock_content_blocks",
|
|
)
|
|
_STRIP_TC_KEYS = ("call_id", "response_item_id")
|
|
_HIGH_EFFORTS = {"high", "xhigh", "max", "ultra"}
|
|
|
|
|
|
def _rename_tool_search_bridge_for_xai(tools: list[dict[str, Any]]) -> tuple[list[dict[str, Any]], dict[str, str]]:
|
|
"""Alias the client ``tool_search`` declaration for xAI; returns ``(tools, {alias: "tool_search"})``.
|
|
|
|
If a real tool already holds ``hermes_tool_search``, the bridge takes a ``_2``/``_3`` suffix.
|
|
"""
|
|
from agent.transports.codex import _alias_reserved_tools
|
|
|
|
return _alias_reserved_tools(
|
|
tools, ("tool_search",), name_of=lambda t: (t.get("function") or {}).get("name"),
|
|
rename=lambda t, alias: {**t, "function": {**t["function"], "name": alias}},
|
|
)
|
|
|
|
|
|
def _static_prompt_instructions(messages: list[dict[str, Any]]) -> str:
|
|
"""Stable leading system/developer prefix used for cache routing (later messages are conversation state)."""
|
|
first = messages[0] if messages and isinstance(messages[0], dict) else {}
|
|
if first.get("role") not in {"system", "developer"}:
|
|
return ""
|
|
content = first.get("content")
|
|
if isinstance(content, str):
|
|
return content
|
|
try:
|
|
return json.dumps(content, sort_keys=True, ensure_ascii=False, separators=(",", ":"))
|
|
except (TypeError, ValueError):
|
|
return str(content or "")
|
|
|
|
|
|
def _add_prompt_cache_key(
|
|
api_kwargs: dict[str, Any], *, messages: list[dict[str, Any]],
|
|
tools: list[dict[str, Any]] | None, supports_prompt_cache_key: bool,
|
|
session_id: str | None = None, cache_scope_id: str | None = None,
|
|
) -> None:
|
|
"""Add a content-addressed ``prompt_cache_key`` only for a capable endpoint.
|
|
|
|
``cache_scope_id`` (compression-lineage root) beats ``session_id`` so the key
|
|
survives compression rotation. A caller-supplied key is authoritative but is
|
|
bounded to OpenAI's 64-char cap in place. Shares the Responses transport's hash
|
|
so equivalent prefixes hit one bucket across modes.
|
|
|
|
``cache_scope_id``, when provided, is the rotation-stable logical scope (compression-lineage root —
|
|
agent/prompt_cache_scope.py) and takes precedence over the physical ``session_id`` so the key survives
|
|
context-compression session rotation (#79017).
|
|
"""
|
|
# Stable prompt-cache routing for the Codex/Responses aux path, mirroring the main transport
|
|
# (agent/transports/codex.py::build_kwargs, which sets prompt_cache_key =
|
|
# _content_cache_key(instructions, tools)). Without this, MoA acting-aggregator and other auxiliary
|
|
# Responses calls stay cache-cold while the main Responses transport is warm (issue #53735). The key is
|
|
# content-addressed from the static prefix (instructions + tool schemas) so it stays warm across
|
|
# turns/fires. Guard the top-level field the same way the main transport does: xAI Responses takes the
|
|
# key in extra_body (not top-level) and GitHub/Copilot Responses opts out of cache-key routing entirely
|
|
# — for those hosts, skip it here.
|
|
from agent.transports.codex import (
|
|
_bound_prompt_cache_key_field, _cache_scope_from_session_id, _content_cache_key
|
|
)
|
|
|
|
containers = [c for c in (api_kwargs, api_kwargs.get("extra_body")) if isinstance(c, dict) and "prompt_cache_key" in c]
|
|
if containers:
|
|
for c in containers:
|
|
_bound_prompt_cache_key_field(c)
|
|
return
|
|
if not supports_prompt_cache_key:
|
|
return
|
|
cache_key = _content_cache_key(
|
|
_static_prompt_instructions(messages), tools, _cache_scope_from_session_id(cache_scope_id or session_id),
|
|
)
|
|
if cache_key:
|
|
api_kwargs["prompt_cache_key"] = cache_key
|
|
|
|
|
|
_ROUTER_TIMEOUT_SHIM = "Connect timeout, please try again later."
|
|
|
|
|
|
def _has_positive_completion_tokens(usage: Any) -> bool:
|
|
"""Return whether a response usage object proves text was generated."""
|
|
for field in ("completion_tokens", "output_tokens"):
|
|
value = usage.get(field) if isinstance(usage, dict) else getattr(usage, field, None)
|
|
if isinstance(value, (int, float)) and not isinstance(value, bool) and value > 0:
|
|
return True
|
|
return False
|
|
|
|
|
|
def router_timeout_shim_may_follow(text: str) -> bool:
|
|
"""True while streamed text is still a prefix of the shim sentinel (hold it back until judged)."""
|
|
return bool(text) and _ROUTER_TIMEOUT_SHIM.startswith(text.lstrip())
|
|
|
|
|
|
def is_router_timeout_shim(response: Any) -> bool:
|
|
"""Recognize a router failure encoded as a successful ChatCompletion (#68396).
|
|
|
|
Some OpenAI-compatible routers answer an upstream connect timeout with HTTP 200 and the
|
|
sentinel as the sole assistant message. Only the exact sentinel, with no tool calls and no
|
|
positive ``completion_tokens``/``output_tokens`` proof of generation, is a shim — a model
|
|
that really produced those words keeps its usage evidence. Shared by every consumer of an
|
|
OpenAI-compatible response: ``validate_response``, the stream assembler, the
|
|
iteration-limit summary and the auxiliary ``_validate_llm_response``.
|
|
"""
|
|
choices = getattr(response, "choices", None)
|
|
if not isinstance(choices, list) or len(choices) != 1:
|
|
return False
|
|
message = getattr(choices[0], "message", None)
|
|
content = getattr(message, "content", None)
|
|
if not isinstance(content, str) or content.strip() != _ROUTER_TIMEOUT_SHIM:
|
|
return False
|
|
if getattr(message, "tool_calls", None):
|
|
return False
|
|
return not _has_positive_completion_tokens(getattr(response, "usage", None))
|
|
|
|
|
|
def _reasoning_config_for_model(model: str, reasoning_config: dict | None) -> dict | None:
|
|
"""Clamp Hermes' extended effort set (``ultra``) to the OpenAI-compat wire vocabulary.
|
|
|
|
Hermes' internal effort set extends the wire vocabulary with ``ultra`` (the /reasoning command documents
|
|
none..xhigh|max|ultra). OpenAI- compatible wires — OpenRouter chief among them — accept exactly
|
|
max|xhigh|high|medium|low|minimal|none and reject the extension with HTTP 400 (#89503). Clamp against
|
|
the declared wire vocabulary via the shared policy in ``agent.reasoning_effort``; provider profiles with
|
|
narrower sets clamp again downstream.
|
|
"""
|
|
return clamp_reasoning_config(reasoning_config, OPENAI_COMPAT_WIRE_EFFORTS)
|
|
|
|
|
|
def _build_gemini_thinking_config(model: str, reasoning_config: dict | None) -> dict | None:
|
|
"""Translate Hermes/OpenRouter-style reasoning config to Gemini thinkingConfig."""
|
|
if not isinstance(reasoning_config, dict):
|
|
return None
|
|
normalized_model = (model or "").strip().lower().removeprefix("google/")
|
|
# Gemini-only; Gemma/PaLM on the same provider 400 on the field even as ``{"includeThoughts": False}``.
|
|
# ``thinking_config`` is a Gemini-only request parameter. The same ``gemini`` provider also serves Gemma
|
|
# (and historically PaLM/Bard); those reject the field with HTTP 400 "Unknown name 'thinking_config':
|
|
# Cannot find field" — including the polite ``{"includeThoughts": False}`` form. Omit the field entirely
|
|
# on non-Gemini models. (#17426)
|
|
if not normalized_model.startswith("gemini"):
|
|
return None
|
|
effort = str(reasoning_config.get("effort", "medium") or "medium").strip().lower()
|
|
if reasoning_config.get("enabled") is False or effort == "none":
|
|
# ``includeThoughts: False`` only omits thought parts from the returned
|
|
# response; the model may still reason internally and bill thought
|
|
# tokens against maxOutputTokens, starving small budgets (title
|
|
# generation's 64 tokens). Set thinkingBudget to 0 to actually disable
|
|
# thinking on families that document it: Gemini 2.5 and 3+ (plus the
|
|
# ``gemini-flash-latest`` alias); future majors are added only when the
|
|
# API documents thinkingBudget for them. (#91927)
|
|
config: dict[str, Any] = {"includeThoughts": False}
|
|
if normalized_model == "gemini-flash-latest" or normalized_model.startswith(("gemini-2.5-", "gemini-3")):
|
|
config["thinkingBudget"] = 0
|
|
return config
|
|
thinking_config: dict[str, Any] = {"includeThoughts": True}
|
|
# Gemini 2.5 takes thinkingBudget; don't guess one from coarse effort levels.
|
|
if normalized_model.startswith("gemini-2.5-"):
|
|
return thinking_config
|
|
if effort not in {"minimal", "low", "medium", "high", "xhigh", "max", "ultra"}:
|
|
effort = "medium"
|
|
# Gemini 3 Flash documents low/medium/high thinking levels; Gemini 3 Pro
|
|
# is stricter (low/high). Clamp Hermes' wider effort set to what each
|
|
# family accepts so we never forward an undocumented level verbatim.
|
|
if normalized_model.startswith("gemini-3"):
|
|
if "flash" in normalized_model:
|
|
thinking_config["thinkingLevel"] = (
|
|
"low" if effort in {"minimal", "low"} else "high" if effort in _HIGH_EFFORTS else "medium"
|
|
)
|
|
elif "pro" in normalized_model:
|
|
thinking_config["thinkingLevel"] = "high" if effort in _HIGH_EFFORTS else "low"
|
|
return thinking_config
|
|
|
|
|
|
def _snake_case_gemini_thinking_config(config: dict | None) -> dict | None:
|
|
"""Convert Gemini thinking config keys to the OpenAI-compat field names."""
|
|
if not isinstance(config, dict) or not config:
|
|
return None
|
|
translated: dict[str, Any] = {}
|
|
include, level, budget = config.get("includeThoughts"), config.get("thinkingLevel"), config.get("thinkingBudget")
|
|
if isinstance(include, bool):
|
|
translated["include_thoughts"] = include
|
|
if isinstance(level, str) and level.strip():
|
|
translated["thinking_level"] = level.strip().lower()
|
|
if isinstance(budget, (int, float)):
|
|
translated["thinking_budget"] = int(budget)
|
|
return translated or None
|
|
|
|
|
|
def _raise_gemini_thinking_max_tokens(model: str, reasoning_config: dict | None, requested: Any) -> Any:
|
|
"""Raise Gemini output caps that thinking tokens (billed against max_tokens) would otherwise exhaust."""
|
|
thinking_config = _build_gemini_thinking_config(model, reasoning_config)
|
|
if not thinking_config:
|
|
return requested
|
|
from agent.gemini_native_adapter import _effective_gemini_max_output_tokens
|
|
|
|
return _effective_gemini_max_output_tokens(requested, thinking_config)
|
|
|
|
|
|
def _is_gemini_openai_compat_base_url(base_url: Any) -> bool:
|
|
normalized = str(base_url or "").strip().rstrip("/").lower()
|
|
return bool(normalized) and "generativelanguage.googleapis.com" in normalized and normalized.endswith("/openai")
|
|
|
|
|
|
def _is_openai_api_base_url(base_url: Any) -> bool:
|
|
"""True only for the exact api.openai.com host (implies ``prompt_cache_key`` support).
|
|
|
|
Not a substring match: Azure / strict compat endpoints stay opt-in via ``supports_prompt_cache_key``.
|
|
"""
|
|
try:
|
|
return (urlparse(str(base_url or "").strip()).hostname or "").lower() == "api.openai.com"
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
def _model_consumes_thought_signature(model: Any) -> bool:
|
|
"""True for Gemini-family targets, which require tool-call ``extra_content`` (thought_signature) replay.
|
|
|
|
Every other strict provider rejects it, so it is stripped for non-Gemini targets.
|
|
"""
|
|
m = str(model or "").lower()
|
|
return "gemini" in m or "gemma" in m
|
|
|
|
|
|
def _route_replays_reasoning_details(base_url: Any) -> bool:
|
|
"""True when the target route reads replayed ``reasoning_details`` (OpenRouter's unified
|
|
reasoning array, also consumed by the Nous Portal).
|
|
|
|
Every other chat-completions endpoint either ignores the field or, when its schema is
|
|
strict (Groq, Mistral, Cerebras, opencode relays: ``property 'reasoning_details' is
|
|
unsupported`` / ``Extra inputs are not permitted`` / ``no such field``), rejects the whole
|
|
request with HTTP 400/422 — so a reasoning turn produced earlier in the session wedges every
|
|
later turn once the model is switched (#70233). The stored history keeps the field; only the
|
|
wire copy drops it.
|
|
"""
|
|
from utils import base_url_host_matches
|
|
|
|
return base_url_host_matches(base_url, "openrouter.ai") or base_url_host_matches(base_url, "nousresearch.com")
|
|
|
|
|
|
def _has_replayable_thought_signature(extra_content: Any) -> bool:
|
|
"""Whether OpenRouter's Gemini sidecar contains a usable thought signature.
|
|
|
|
Gemini accepts the signature either directly or under its ``google``
|
|
namespace. Replaying an empty or non-string value makes a multimodal
|
|
request fail with ``Corrupted thought signature``; omit that sidecar while
|
|
leaving the stored history untouched.
|
|
"""
|
|
if not isinstance(extra_content, dict):
|
|
return False
|
|
candidate = extra_content.get("thought_signature")
|
|
google = extra_content.get("google")
|
|
if candidate is None and isinstance(google, dict):
|
|
candidate = google.get("thought_signature")
|
|
return isinstance(candidate, str) and bool(candidate.strip())
|
|
|
|
|
|
def _attr_or_model_extra(obj: Any, name: str) -> Any:
|
|
"""``obj.<name>``, else the same key from pydantic ``model_extra`` (some SDKs park fields there)."""
|
|
value = getattr(obj, name, None)
|
|
if value is None and hasattr(obj, "model_extra"):
|
|
value = (obj.model_extra if isinstance(obj.model_extra, dict) else {}).get(name)
|
|
return value
|
|
|
|
|
|
def _dump_extra_content(extra: Any) -> Any:
|
|
"""Plain-dict form of a pydantic ``extra_content``; older pydantic lacks ``warnings=``, so retry without it."""
|
|
if hasattr(extra, "model_dump"):
|
|
for dump_kwargs in ({"warnings": False}, {}):
|
|
try:
|
|
return extra.model_dump(**dump_kwargs)
|
|
except TypeError:
|
|
continue
|
|
except Exception:
|
|
break
|
|
return extra
|
|
|
|
|
|
def _pareto_score(raw: Any) -> float | None:
|
|
"""Coding-score floor for the Pareto router as a float in [0, 1], else None."""
|
|
try:
|
|
score = float(raw) if raw not in (None, "") else None
|
|
except (TypeError, ValueError):
|
|
return None
|
|
return score if score is not None and 0.0 <= score <= 1.0 else None
|
|
|
|
|
|
def _swap_developer_role(sanitized: list, model_lower: str) -> list:
|
|
"""GPT-5/Codex models take a ``developer`` role instead of ``system``."""
|
|
if (
|
|
sanitized and isinstance(sanitized[0], dict) and sanitized[0].get("role") == "system"
|
|
and any(p in model_lower for p in DEVELOPER_ROLE_MODELS)
|
|
):
|
|
return [{**sanitized[0], "role": "developer"}, *sanitized[1:]]
|
|
return sanitized
|
|
|
|
|
|
def _apply_max_tokens(api_kwargs: dict, model: str, reasoning_config: Any, params: dict, profile_max: Any = None) -> None:
|
|
"""Preserve internal task/recovery budgets and provider protocol exceptions."""
|
|
max_tokens_fn = params.get("max_tokens_param_fn")
|
|
for candidate in (params.get("ephemeral_max_output_tokens"), params.get("max_tokens")):
|
|
if candidate is not None and max_tokens_fn:
|
|
api_kwargs.update(max_tokens_fn(_raise_gemini_thinking_max_tokens(model, reasoning_config, candidate)))
|
|
return
|
|
if profile_max and max_tokens_fn:
|
|
api_kwargs.update(max_tokens_fn(_raise_gemini_thinking_max_tokens(model, reasoning_config, profile_max)))
|
|
|
|
|
|
|
|
def _base_kwargs(model: str, sanitized: list, tools: Any, params: dict, profile: Any = None) -> dict[str, Any]:
|
|
"""Shared ``{model, messages[, temperature][, timeout][, tools]}`` scaffold for both build paths.
|
|
|
|
``temperature`` is profile-path only: ``fixed_temperature`` beats the caller's; ``OMIT_TEMPERATURE`` sends none.
|
|
"""
|
|
api_kwargs: dict[str, Any] = {"model": model, "messages": sanitized}
|
|
if profile is not None:
|
|
from providers.base import OMIT_TEMPERATURE
|
|
|
|
if profile.fixed_temperature is OMIT_TEMPERATURE:
|
|
pass
|
|
elif profile.fixed_temperature is not None:
|
|
api_kwargs["temperature"] = profile.fixed_temperature
|
|
elif params.get("temperature") is not None:
|
|
api_kwargs["temperature"] = params["temperature"]
|
|
if params.get("timeout") is not None:
|
|
api_kwargs["timeout"] = params["timeout"]
|
|
if tools:
|
|
# Moonshot/Kimi uses a stricter JSON Schema flavor; rewriting here also covers aggregator routes.
|
|
api_kwargs["tools"] = sanitize_moonshot_tools(tools) if is_moonshot_model(model) else tools
|
|
return api_kwargs
|
|
|
|
|
|
def _finish_kwargs(api_kwargs: dict[str, Any], sanitized: list, params: dict, *, supports_prompt_cache_key: bool) -> dict[str, Any]:
|
|
"""Tail shared by both build paths: content-addressed prompt_cache_key, then return."""
|
|
_add_prompt_cache_key(
|
|
api_kwargs, messages=sanitized, tools=api_kwargs.get("tools"), supports_prompt_cache_key=supports_prompt_cache_key,
|
|
session_id=params.get("session_id"), cache_scope_id=params.get("cache_scope_id"),
|
|
)
|
|
return api_kwargs
|
|
|
|
|
|
def _sanitize_message(
|
|
msg: Any, strip_extra_content: bool, strip_reasoning_details: bool = False,
|
|
native_reasoning_details_type: str | None = None,
|
|
) -> dict | None:
|
|
"""Sanitized copy of ``msg``, or None when nothing needs stripping.
|
|
|
|
Drops persistence sidecars, ``_``-prefixed scaffolding markers, tool-call ``call_id`` /
|
|
``response_item_id`` (and ``extra_content`` unless Gemini), an assistant
|
|
``tool_calls: []`` / ``null`` (strict providers reject both), ``name``
|
|
on tool results (schema-valid only on user/assistant messages; strict
|
|
providers reject it with ``contains item with unknown key name``), and
|
|
``reasoning_details`` unless the route replays it (``_route_replays_reasoning_details``).
|
|
On a replaying route, private ``<provider>.native_assistant`` carriers still go only to the
|
|
profile that declared that exact type: another provider's signed replay is meaningless (or
|
|
rejected) elsewhere, and stored history keeps it for a return to the original provider.
|
|
"""
|
|
if not isinstance(msg, dict):
|
|
return None
|
|
strip_keys = [k for k in msg if k in _STRIP_MSG_KEYS or (isinstance(k, str) and k.startswith("_"))]
|
|
kept_details = None
|
|
if strip_reasoning_details and "reasoning_details" in msg:
|
|
strip_keys.append("reasoning_details")
|
|
elif isinstance(msg.get("reasoning_details"), list):
|
|
details = msg["reasoning_details"]
|
|
kept = [d for d in details if not (
|
|
isinstance(d, dict) and isinstance(d.get("type"), str)
|
|
and d["type"].endswith(".native_assistant") and d["type"] != native_reasoning_details_type)]
|
|
if len(kept) != len(details):
|
|
strip_keys.append("reasoning_details")
|
|
kept_details = kept
|
|
# ``name`` is schema-valid on user/assistant messages, so the removal is
|
|
# role-qualified: only tool results carry it illegally (strict providers
|
|
# reject with "contains item with unknown key name").
|
|
if msg.get("role") == "tool" and "name" in msg:
|
|
strip_keys.append("name")
|
|
out_msg = {k: v for k, v in msg.items() if k not in strip_keys}
|
|
if kept_details:
|
|
out_msg["reasoning_details"] = kept_details
|
|
tool_calls = msg.get("tool_calls")
|
|
copied_tool_calls = None
|
|
if msg.get("role") == "assistant" and "tool_calls" in msg and (tool_calls is None or (isinstance(tool_calls, list) and not tool_calls)):
|
|
out_msg.pop("tool_calls", None)
|
|
strip_keys.append("tool_calls")
|
|
elif isinstance(tool_calls, list):
|
|
for tc_idx, tc in enumerate(tool_calls):
|
|
if not isinstance(tc, dict):
|
|
continue
|
|
keys = [k for k in _STRIP_TC_KEYS if k in tc]
|
|
if "extra_content" in tc and (
|
|
strip_extra_content or not _has_replayable_thought_signature(tc["extra_content"])
|
|
):
|
|
keys.append("extra_content")
|
|
if keys:
|
|
if copied_tool_calls is None:
|
|
copied_tool_calls = list(tool_calls)
|
|
copied_tool_calls[tc_idx] = {k: v for k, v in tc.items() if k not in keys}
|
|
if copied_tool_calls is not None:
|
|
out_msg["tool_calls"] = copied_tool_calls
|
|
return out_msg if strip_keys or copied_tool_calls is not None else None
|
|
|
|
|
|
class ChatCompletionsTransport(ProviderTransport):
|
|
"""Transport for api_mode='chat_completions'."""
|
|
|
|
# ``{alias: original}`` of the most recent request. ``None`` = no request recorded
|
|
# (normalize-only call sites) -> static alias; ``{}`` = no aliases emitted.
|
|
_last_wire_aliases: dict[str, str] | None = None
|
|
|
|
@property
|
|
def api_mode(self) -> str:
|
|
return "chat_completions"
|
|
|
|
def convert_messages(self, messages: list[dict[str, Any]], **kwargs) -> list[dict[str, Any]]:
|
|
"""Strip internal fields that strict chat-completions providers reject (HTTP 400/422).
|
|
|
|
Returns the input list unchanged when nothing needs sanitizing.
|
|
"""
|
|
strip_extra_content = not _model_consumes_thought_signature(kwargs.get("model"))
|
|
# A profile declaring a native carrier type consumes replayed details by contract.
|
|
native_type = getattr(kwargs.get("provider_profile"), "native_reasoning_details_type", None) or None
|
|
strip_reasoning_details = not (native_type or _route_replays_reasoning_details(kwargs.get("base_url")))
|
|
sanitized_pairs = [(m, _sanitize_message(m, strip_extra_content, strip_reasoning_details, native_type))
|
|
for m in messages]
|
|
if all(s is None for _, s in sanitized_pairs):
|
|
return messages
|
|
return [m if s is None else s for m, s in sanitized_pairs]
|
|
|
|
def convert_tools(self, tools: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
|
"""Tools are already in OpenAI format — identity."""
|
|
return tools
|
|
|
|
def build_kwargs(
|
|
self, model: str, messages: list[dict[str, Any]], tools: list[dict[str, Any]] | None = None, **params,
|
|
) -> dict[str, Any]:
|
|
"""Build chat.completions.create() kwargs.
|
|
|
|
With ``provider_profile`` every quirk comes from the profile; the legacy flag
|
|
path below (is_kimi, is_openrouter, ...) is only reached for unregistered providers.
|
|
"""
|
|
_profile = params.get("provider_profile")
|
|
sanitized = self.convert_messages(messages, model=model, base_url=params.get("base_url"), provider_profile=_profile)
|
|
if _profile:
|
|
return self._build_kwargs_from_profile(_profile, model, sanitized, tools, params)
|
|
|
|
sanitized = _swap_developer_role(sanitized, params.get("model_lower", (model or "").lower()))
|
|
api_kwargs = _base_kwargs(model, sanitized, tools, params)
|
|
|
|
is_kimi = params.get("is_kimi", False)
|
|
is_lmstudio = params.get("is_lmstudio", False)
|
|
supports_reasoning = params.get("supports_reasoning", False)
|
|
reasoning_config = _reasoning_config_for_model(model, params.get("reasoning_config"))
|
|
_apply_max_tokens(api_kwargs, model, reasoning_config, params)
|
|
|
|
# Kimi / TokenHub / LM Studio: top-level reasoning_effort (unless thinking disabled).
|
|
thinking_off = isinstance(reasoning_config, dict) and reasoning_config.get("enabled") is False
|
|
_e = requested_effort(reasoning_config)
|
|
if is_kimi and not thinking_off:
|
|
# K3 = low/high/max (server default high), K2-era = low/medium/high (default medium).
|
|
_supported = kimi_supported_efforts(model)
|
|
is_k3 = _supported is KIMI_K3_EFFORTS
|
|
api_kwargs["reasoning_effort"] = (
|
|
("high" if is_k3 else "medium") if _e is None
|
|
else clamp_effort(_e, _supported, KIMI_K3_OVERRIDES if is_k3 else None)
|
|
)
|
|
if params.get("is_tokenhub", False) and not thinking_off:
|
|
api_kwargs["reasoning_effort"] = "high" if _e is None else clamp_effort(_e, TOKENHUB_EFFORTS)
|
|
if is_lmstudio and supports_reasoning:
|
|
_lm_effort = resolve_lmstudio_effort(reasoning_config, params.get("lmstudio_reasoning_options"))
|
|
if _lm_effort is not None:
|
|
api_kwargs["reasoning_effort"] = _lm_effort
|
|
|
|
extra_body: dict[str, Any] = {}
|
|
is_openrouter = params.get("is_openrouter", False)
|
|
base_url = params.get("base_url")
|
|
if is_openrouter and params.get("provider_preferences"):
|
|
extra_body["provider"] = params["provider_preferences"]
|
|
# Pareto Code router plugin (same shape as the OpenRouter profile path).
|
|
if is_openrouter and model == "openrouter/pareto-code":
|
|
_pareto_score_f = _pareto_score(params.get("openrouter_min_coding_score"))
|
|
if _pareto_score_f is not None:
|
|
extra_body["plugins"] = [{"id": "pareto-router", "min_coding_score": _pareto_score_f}]
|
|
if is_kimi:
|
|
extra_body["thinking"] = {"type": "disabled" if thinking_off else "enabled"}
|
|
|
|
# LM Studio is handled above via top-level reasoning_effort.
|
|
if supports_reasoning and not is_lmstudio:
|
|
if params.get("is_github_models", False):
|
|
if params.get("github_reasoning_extra") is not None:
|
|
extra_body["reasoning"] = params["github_reasoning_extra"]
|
|
else:
|
|
_effort = (reasoning_config.get("effort", "medium") or "medium") if reasoning_config and isinstance(reasoning_config, dict) else "medium"
|
|
# Honor explicit "thinking off" like the profile path — never re-enable it.
|
|
off = thinking_off or _effort == "none"
|
|
extra_body["reasoning"] = {"enabled": not off, "effort": "none" if off else _effort}
|
|
|
|
if str(params.get("provider_name") or "").strip().lower() == "gemini":
|
|
raw_thinking_config = _build_gemini_thinking_config(model, reasoning_config)
|
|
if _is_gemini_openai_compat_base_url(base_url):
|
|
thinking_config = _snake_case_gemini_thinking_config(raw_thinking_config)
|
|
if thinking_config:
|
|
openai_compat_extra = extra_body.get("extra_body", {})
|
|
openai_compat_extra["google"] = {**openai_compat_extra.get("google", {}), "thinking_config": thinking_config}
|
|
extra_body["extra_body"] = openai_compat_extra
|
|
elif raw_thinking_config:
|
|
extra_body["thinking_config"] = raw_thinking_config
|
|
|
|
if params.get("extra_body_additions"):
|
|
extra_body.update(params["extra_body_additions"])
|
|
if extra_body:
|
|
api_kwargs["extra_body"] = extra_body
|
|
if params.get("request_overrides"):
|
|
api_kwargs.update(params["request_overrides"])
|
|
return _finish_kwargs(
|
|
api_kwargs, sanitized, params,
|
|
supports_prompt_cache_key=bool(params.get("supports_prompt_cache_key")) or _is_openai_api_base_url(base_url),
|
|
)
|
|
|
|
def _build_kwargs_from_profile(self, profile, model, sanitized, tools, params):
|
|
"""Build API kwargs from a ProviderProfile — every quirk comes from the profile object."""
|
|
sanitized = _swap_developer_role(profile.prepare_messages(sanitized), (model or "").lower())
|
|
api_kwargs = _base_kwargs(model, sanitized, tools, params, profile=profile)
|
|
|
|
reasoning_config = _reasoning_config_for_model(model, params.get("reasoning_config"))
|
|
# Profiles fronting several backends override get_max_tokens() per model.
|
|
_apply_max_tokens(api_kwargs, model, reasoning_config, params, profile_max=profile.get_max_tokens(model))
|
|
|
|
extra_body_from_profile, top_level_from_profile = profile.build_api_kwargs_extras(
|
|
reasoning_config=reasoning_config, supports_reasoning=params.get("supports_reasoning", False),
|
|
qwen_session_metadata=params.get("qwen_session_metadata"), model=model,
|
|
base_url=params.get("base_url"), ollama_num_ctx=params.get("ollama_num_ctx"),
|
|
session_id=params.get("session_id"), cache_scope_id=params.get("cache_scope_id"),
|
|
)
|
|
api_kwargs.update(top_level_from_profile)
|
|
|
|
extra_body: dict[str, Any] = {}
|
|
profile_body = profile.build_extra_body(
|
|
session_id=params.get("session_id"), provider_preferences=params.get("provider_preferences"), model=model,
|
|
base_url=params.get("base_url"), reasoning_config=reasoning_config,
|
|
openrouter_min_coding_score=params.get("openrouter_min_coding_score"),
|
|
)
|
|
for part in (profile_body, extra_body_from_profile, params.get("extra_body_additions")):
|
|
if part:
|
|
extra_body.update(part)
|
|
for k, v in (params.get("request_overrides") or {}).items():
|
|
if k == "extra_body" and isinstance(v, dict):
|
|
extra_body.update(v)
|
|
else:
|
|
api_kwargs[k] = v
|
|
|
|
if extra_body:
|
|
# Native Gemini speaks Google's REST schema: OpenAI-style extra_body
|
|
# keys (tags, reasoning, provider, ...) are unknown fields -> HTTP 400.
|
|
# The native client only reads thinking_config, so drop everything else.
|
|
try:
|
|
from agent.gemini_native_adapter import is_native_gemini_base_url
|
|
_native_gemini = is_native_gemini_base_url(params.get("base_url"))
|
|
except Exception:
|
|
_native_gemini = False
|
|
if _native_gemini:
|
|
extra_body = {k: v for k, v in extra_body.items() if k in ("thinking_config", "thinkingConfig")}
|
|
if extra_body:
|
|
api_kwargs["extra_body"] = extra_body
|
|
return _finish_kwargs(
|
|
api_kwargs, sanitized, params, supports_prompt_cache_key=bool(getattr(profile, "supports_prompt_cache_key", False)),
|
|
)
|
|
|
|
def normalize_response(self, response: Any, **kwargs) -> NormalizedResponse:
|
|
"""Normalize an OpenAI ChatCompletion.
|
|
|
|
Gemini ``extra_content`` rides on ToolCall.provider_data; ``reasoning_content`` and
|
|
``reasoning_details`` stay distinct in provider_data because downstream reads them so.
|
|
"""
|
|
choice = response.choices[0]
|
|
msg = getattr(choice, "message", None)
|
|
_fr = getattr(choice, "finish_reason", None)
|
|
# Poolside returns int finish_reason; Gemini-fronting gateways return
|
|
# uppercase STOP / MAX_TOKENS — fold to the OpenAI contract here.
|
|
finish_reason = _normalize_finish_reason(str(_fr) if isinstance(_fr, int) else _fr) or "stop"
|
|
|
|
tool_calls = None
|
|
if getattr(msg, "tool_calls", None):
|
|
tool_calls = [tc for tc in (self._normalize_tool_call(tc) for tc in msg.tool_calls) if tc is not None]
|
|
|
|
usage = Usage.from_openai(response.usage) if hasattr(response, "usage") and response.usage else None
|
|
|
|
# Fields some SDKs park in pydantic ``model_extra`` rather than as attributes.
|
|
reasoning_content = _attr_or_model_extra(msg, "reasoning_content")
|
|
provider_data: dict[str, Any] = {}
|
|
if reasoning_content is not None:
|
|
provider_data["reasoning_content"] = reasoning_content
|
|
if getattr(msg, "reasoning_details", None):
|
|
provider_data["reasoning_details"] = msg.reasoning_details
|
|
|
|
# OpenAI structured refusal (``message.refusal`` set, ``content`` empty); without
|
|
# promotion the loop retries a deterministic refusal as an empty response.
|
|
content = getattr(msg, "content", None)
|
|
refusal = _attr_or_model_extra(msg, "refusal")
|
|
if isinstance(refusal, str) and refusal.strip():
|
|
provider_data["refusal"] = refusal
|
|
# Terminal ``content_filter`` only when the refusal is the sole payload.
|
|
if not (isinstance(content, str) and content.strip()) and not tool_calls:
|
|
content = refusal
|
|
if finish_reason in (None, "stop"):
|
|
finish_reason = "content_filter"
|
|
|
|
return NormalizedResponse(
|
|
content=content, tool_calls=tool_calls, finish_reason=finish_reason,
|
|
reasoning=getattr(msg, "reasoning", None), usage=usage, provider_data=provider_data or None,
|
|
)
|
|
|
|
def _normalize_tool_call(self, tc: Any) -> ToolCall | None:
|
|
"""One SDK tool call -> ToolCall; None when it lacks a function/name (matches Relay's codec)."""
|
|
tc_function = getattr(tc, "function", None)
|
|
name = getattr(tc_function, "name", None)
|
|
if tc_function is None or name is None:
|
|
return None
|
|
# Reverse only aliases THIS request emitted; a real ``hermes_tool_search`` tool stays itself.
|
|
alias_map = self._last_wire_aliases
|
|
if alias_map is None:
|
|
name = "tool_search" if name == _XAI_TOOL_SEARCH_ALIAS else name
|
|
else:
|
|
name = alias_map.get(name, name)
|
|
arguments = getattr(tc_function, "arguments", None)
|
|
extra = _attr_or_model_extra(tc, "extra_content")
|
|
return ToolCall(
|
|
id=getattr(tc, "id", None), name=name, arguments="{}" if arguments is None else arguments,
|
|
provider_data=None if extra is None else {"extra_content": _dump_extra_content(extra)},
|
|
)
|
|
|
|
def validate_response(self, response: Any) -> bool:
|
|
"""Check that response has valid choices and is not a router failure shim."""
|
|
if response is None or not getattr(response, "choices", None):
|
|
return False
|
|
return not is_router_timeout_shim(response)
|
|
|
|
def extract_cache_stats(self, response: Any) -> dict[str, int] | None:
|
|
"""Cache stats from prompt_tokens_details (OpenRouter/OpenAI) or DeepSeek's top-level prompt_cache_hit_tokens."""
|
|
usage = getattr(response, "usage", None)
|
|
if usage is None:
|
|
return None
|
|
details = getattr(usage, "prompt_tokens_details", None)
|
|
cached = getattr(details, "cached_tokens", 0) or 0 if details else 0
|
|
written = getattr(details, "cache_write_tokens", 0) or 0 if details else 0
|
|
cached = cached or getattr(usage, "prompt_cache_hit_tokens", 0) or 0 # DeepSeek native
|
|
return {"cached_tokens": cached, "creation_tokens": written} if cached or written else None
|
|
|
|
|
|
from agent.transports import register_transport # noqa: E402
|
|
|
|
register_transport("chat_completions", ChatCompletionsTransport)
|
|
|
|
|
|
# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
|
|
# Names external plugins imported from this module before the Sep 2026 decomposition.
|
|
# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
|
|
# The whole block is removed by reverting the commit that added it.
|
|
from typing import Dict # noqa: F401,E402
|
|
# ---- END PLUGIN-COMPAT ----
|