refactor(agent/transports): collapse chat_completions sanitize/base-kwargs helpers and codex build_kwargs plumbing; wire output fuzz-identical

This commit is contained in:
Teknium
2026-09-02 18:21:51 -07:00
parent c94ced6225
commit 0e66a40d09
2 changed files with 190 additions and 266 deletions
+115 -179
View File
@@ -5,7 +5,8 @@ provider-specific work lives in build_kwargs (max_tokens, reasoning, extra_body)
"""
import json
from typing import Any, Dict
from typing import Any
from urllib.parse import urlparse
from agent.lmstudio_reasoning import resolve_lmstudio_effort
from agent.reasoning_effort import (
@@ -17,9 +18,9 @@ from agent.prompt_builder import DEVELOPER_ROLE_MODELS
from agent.transports.base import ProviderTransport
from agent.transports.types import NormalizedResponse, ToolCall, Usage
# xAI reserves the function name ``tool_search`` for its server-side tool and
# rejects client declarations of it (HTTP 400, #95003); alias it on the wire and
# map back in normalize_response. Value matches the Codex-side alias (#83122).
# xAI reserves ``tool_search`` for its server-side tool and rejects client
# declarations of it (HTTP 400); alias it on the wire and map back in
# normalize_response. Value matches the Codex-side alias.
_XAI_TOOL_SEARCH_ALIAS = "hermes_tool_search"
# Persistence-only / cross-transport message keys that strict OpenAI-compatible
@@ -32,14 +33,10 @@ _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.
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"})``.
Returns ``(rewritten_tools, alias_map)`` where alias_map maps each alias
emitted by THIS request back to ``tool_search``. If a real tool already
holds ``hermes_tool_search``, the bridge takes a ``_2``/``_3`` suffix.
If a real tool already holds ``hermes_tool_search``, the bridge takes a ``_2``/``_3`` suffix.
"""
from agent.transports.codex import _alias_reserved_tools
@@ -51,9 +48,7 @@ def _rename_tool_search_bridge_for_xai(
def _static_prompt_instructions(messages: list[dict[str, Any]]) -> str:
"""Stable leading system/developer prefix used for cache routing (later messages are conversation state)."""
if not messages or not isinstance(messages[0], dict):
return ""
first = messages[0]
first = messages[0] if messages and isinstance(messages[0], dict) else {}
if first.get("role") not in {"system", "developer"}:
return ""
content = first.get("content")
@@ -73,17 +68,16 @@ def _add_prompt_cache_key(
"""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 context-compression session rotation (#79017). A caller-supplied key
is authoritative but is bounded to OpenAI's 64-char wire cap in place.
survives context-compression session rotation. A caller-supplied key is
authoritative but is bounded to OpenAI's 64-char wire cap in place. Shares the
Responses transport's hash + scope normalization so equivalent prefixes hit
one bucket across modes.
"""
# Share the Responses transport's hash + scope normalization so equivalent
# prefixes hit one bucket across modes without merging unrelated sessions (#78941).
from agent.transports.codex import (
_bound_prompt_cache_key_field, _cache_scope_from_session_id, _content_cache_key
)
extra_body = api_kwargs.get("extra_body")
containers = [c for c in (api_kwargs, extra_body) if isinstance(c, dict) and "prompt_cache_key" in c]
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)
@@ -91,43 +85,34 @@ def _add_prompt_cache_key(
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),
_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
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 (#89503)."""
"""Clamp Hermes' extended effort set (``ultra``) to the OpenAI-compat wire vocabulary."""
if not isinstance(reasoning_config, dict):
return reasoning_config
effort = str(reasoning_config.get("effort") or "").strip().lower()
if not effort:
return reasoning_config
clamped = clamp_effort(effort, OPENAI_COMPAT_WIRE_EFFORTS)
if clamped != effort:
return {**reasoning_config, "effort": clamped}
return reasoning_config
clamped = clamp_effort(effort, OPENAI_COMPAT_WIRE_EFFORTS) if effort else effort
return {**reasoning_config, "effort": clamped} if clamped != effort else reasoning_config
def _build_gemini_thinking_config(model: str, reasoning_config: dict | None) -> dict | None:
"""Translate Hermes/OpenRouter-style reasoning config to Gemini thinkingConfig."""
if reasoning_config is None or not isinstance(reasoning_config, dict):
if not isinstance(reasoning_config, dict):
return None
normalized_model = (model or "").strip().lower()
if normalized_model.startswith("google/"):
normalized_model = normalized_model.split("/", 1)[1]
normalized_model = (model or "").strip().lower().removeprefix("google/")
# ``thinking_config`` is Gemini-only; Gemma/PaLM on the same provider reject
# the field with HTTP 400 even as ``{"includeThoughts": False}`` (#17426).
# the field with HTTP 400 even as ``{"includeThoughts": False}``.
if not normalized_model.startswith("gemini"):
return None
if reasoning_config.get("enabled") is False:
return {"includeThoughts": False}
effort = str(reasoning_config.get("effort", "medium") or "medium").strip().lower()
if effort == "none":
if reasoning_config.get("enabled") is False or effort == "none":
return {"includeThoughts": False}
thinking_config: Dict[str, Any] = {"includeThoughts": True}
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
@@ -148,7 +133,7 @@ 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] = {}
translated: dict[str, Any] = {}
include, level, budget = config.get("includeThoughts"), config.get("thinkingLevel"), config.get("thinkingBudget")
if isinstance(include, bool):
translated["include_thoughts"] = include
@@ -181,8 +166,6 @@ def _is_openai_api_base_url(base_url: Any) -> bool:
field and must stay opt-in via ``supports_prompt_cache_key``.
"""
try:
from urllib.parse import urlparse
return (urlparse(str(base_url or "").strip()).hostname or "").lower() == "api.openai.com"
except Exception:
return False
@@ -198,20 +181,13 @@ def _model_consumes_thought_signature(model: Any) -> bool:
return "gemini" in m or "gemma" in m
def _thinking_disabled(reasoning_config: Any) -> bool:
return bool(reasoning_config and isinstance(reasoning_config, dict) and reasoning_config.get("enabled") is False)
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"
sanitized and isinstance(sanitized[0], dict) and sanitized[0].get("role") == "system"
and any(p in model_lower for p in DEVELOPER_ROLE_MODELS)
):
sanitized = list(sanitized)
sanitized[0] = {**sanitized[0], "role": "developer"}
return [{**sanitized[0], "role": "developer"}, *sanitized[1:]]
return sanitized
@@ -228,49 +204,69 @@ def _apply_max_tokens(api_kwargs: dict, model: str, reasoning_config: Any, param
api_kwargs["max_tokens"] = params["anthropic_max_output"]
def _base_kwargs(model: str, sanitized: list, tools: Any, params: dict) -> dict[str, Any]:
"""Shared ``{model, messages[, timeout][, tools]}`` scaffold for both build paths."""
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: a profile's ``fixed_temperature`` wins
over the caller's, and ``OMIT_TEMPERATURE`` sends none at all.
"""
api_kwargs: dict[str, Any] = {"model": model, "messages": sanitized}
timeout = params.get("timeout")
if timeout is not None:
api_kwargs["timeout"] = timeout
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]:
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"),
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 _msg_strip_keys(msg: dict) -> list:
"""Keys to drop from a message: persistence sidecars plus any ``_``-prefixed Hermes scaffolding marker."""
return [k for k in msg if k in _STRIP_MSG_KEYS or (isinstance(k, str) and k.startswith("_"))]
def _sanitize_message(msg: Any, strip_extra_content: bool) -> dict | None:
"""Sanitized copy of ``msg``, or None when nothing needs stripping.
def _tc_strip_keys(tc: dict, strip_extra_content: bool) -> list:
keys = [k for k in _STRIP_TC_KEYS if k in tc]
if strip_extra_content and "extra_content" in tc:
keys.append("extra_content")
return keys
def _invalid_assistant_tool_calls(msg: dict, tool_calls: Any) -> bool:
"""``tool_calls: []`` / ``tool_calls: null`` on an assistant message — strict providers reject both (#58755)."""
return (
msg.get("role") == "assistant"
and "tool_calls" in msg
and (tool_calls is None or (isinstance(tool_calls, list) and not tool_calls))
)
Drops persistence sidecars, ``_``-prefixed Hermes scaffolding markers, tool-call
``call_id``/``response_item_id`` (and ``extra_content`` unless a Gemini target),
and an assistant ``tool_calls: []`` / ``null`` (strict providers reject both).
"""
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("_"))]
out_msg = {k: v for k, v in msg.items() if k not in strip_keys}
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 strip_extra_content and "extra_content" in tc:
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):
@@ -278,7 +274,7 @@ class ChatCompletionsTransport(ProviderTransport):
# Wire-alias provenance of the most recent request: ``{alias: original}``.
# ``None`` = no request recorded (normalize-only call sites) -> fall back to
# the static alias; ``{}`` = last request emitted no aliases (#95003).
# the static alias; ``{}`` = last request emitted no aliases.
_last_wire_aliases: dict[str, str] | None = None
@property
@@ -288,41 +284,10 @@ class ChatCompletionsTransport(ProviderTransport):
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).
Codex sidecars, ``tool_name``, ``_``-prefixed markers, native-transport
block sidecars, and tool-call ``call_id``/``response_item_id`` are always
dropped; ``extra_content`` is dropped unless ``model`` is Gemini-family.
Returns the input list unchanged when nothing needs sanitizing.
"""
strip_extra_content = not _model_consumes_thought_signature(kwargs.get("model"))
def sanitize(msg: Any) -> "dict | None":
"""Sanitized copy of ``msg``, or None when nothing needs stripping."""
if not isinstance(msg, dict):
return None
strip_keys = _msg_strip_keys(msg)
out_msg = dict(msg)
for key in strip_keys:
out_msg.pop(key, None)
tool_calls = msg.get("tool_calls")
copied_tool_calls = None
if _invalid_assistant_tool_calls(msg, 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):
keys = _tc_strip_keys(tc, strip_extra_content) if isinstance(tc, dict) else []
if keys:
if copied_tool_calls is None:
copied_tool_calls = list(tool_calls)
copied_tc = dict(tc)
for key in keys:
copied_tc.pop(key, None)
copied_tool_calls[tc_idx] = copied_tc
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
sanitized_pairs = [(m, sanitize(m)) for m in messages]
sanitized_pairs = [(m, _sanitize_message(m, strip_extra_content)) 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]
@@ -332,8 +297,7 @@ class ChatCompletionsTransport(ProviderTransport):
return tools
def build_kwargs(
self, model: str, messages: list[dict[str, Any]], tools: list[dict[str, Any]] | None = None,
**params,
self, model: str, messages: list[dict[str, Any]], tools: list[dict[str, Any]] | None = None, **params,
) -> dict[str, Any]:
"""Build chat.completions.create() kwargs.
@@ -350,11 +314,13 @@ class ChatCompletionsTransport(ProviderTransport):
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 = _thinking_disabled(reasoning_config)
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).
@@ -366,18 +332,16 @@ class ChatCompletionsTransport(ProviderTransport):
)
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 params.get("is_lmstudio", False) and params.get("supports_reasoning", False):
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)
provider_name = str(params.get("provider_name") or "").strip().lower()
base_url = params.get("base_url")
provider_prefs = params.get("provider_preferences")
if provider_prefs and is_openrouter:
extra_body["provider"] = provider_prefs
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 = params.get("openrouter_min_coding_score")
@@ -391,22 +355,19 @@ class ChatCompletionsTransport(ProviderTransport):
extra_body["thinking"] = {"type": "disabled" if thinking_off else "enabled"}
# LM Studio is handled above via top-level reasoning_effort.
if params.get("supports_reasoning", False) and not params.get("is_lmstudio", False):
if supports_reasoning and not is_lmstudio:
if params.get("is_github_models", False):
gh_reasoning = params.get("github_reasoning_extra")
if gh_reasoning is not None:
extra_body["reasoning"] = gh_reasoning
if params.get("github_reasoning_extra") is not None:
extra_body["reasoning"] = params["github_reasoning_extra"]
else:
_effort = "medium"
if reasoning_config and isinstance(reasoning_config, dict):
_effort = reasoning_config.get("effort", "medium") or "medium"
_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.
if thinking_off or _effort == "none":
extra_body["reasoning"] = {"enabled": False, "effort": "none"}
else:
extra_body["reasoning"] = {"enabled": True, "effort": _effort}
if provider_name == "gemini":
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)
@@ -419,43 +380,28 @@ class ChatCompletionsTransport(ProviderTransport):
elif raw_thinking_config:
extra_body["thinking_config"] = raw_thinking_config
additions = params.get("extra_body_additions")
if additions:
extra_body.update(additions)
if params.get("extra_body_additions"):
extra_body.update(params["extra_body_additions"])
if extra_body:
api_kwargs["extra_body"] = extra_body
overrides = params.get("request_overrides")
if overrides:
api_kwargs.update(overrides)
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(params.get("base_url")),
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."""
from providers.base import OMIT_TEMPERATURE
sanitized = _swap_developer_role(profile.prepare_messages(sanitized), (model or "").lower())
api_kwargs: dict[str, Any] = {"model": model, "messages": sanitized}
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"]
api_kwargs.update(_base_kwargs(model, sanitized, tools, params))
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),
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"),
@@ -464,21 +410,18 @@ class ChatCompletionsTransport(ProviderTransport):
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,
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)
overrides = params.get("request_overrides")
if overrides:
for k, v in overrides.items():
if k == "extra_body" and isinstance(v, dict):
extra_body.update(v)
else:
api_kwargs[k] = v
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
@@ -494,8 +437,7 @@ class ChatCompletionsTransport(ProviderTransport):
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)),
api_kwargs, sanitized, params, supports_prompt_cache_key=bool(getattr(profile, "supports_prompt_cache_key", False)),
)
def normalize_response(self, response: Any, **kwargs) -> NormalizedResponse:
@@ -544,32 +486,26 @@ class ChatCompletionsTransport(ProviderTransport):
except Exception:
break
tc_provider_data["extra_content"] = extra
tool_calls.append(
ToolCall(
id=getattr(tc, "id", None), name=function_name,
arguments=function_arguments if function_arguments is not None else "{}",
provider_data=tc_provider_data or None,
)
)
tool_calls.append(ToolCall(
id=getattr(tc, "id", None), name=function_name,
arguments=function_arguments if function_arguments is not None else "{}",
provider_data=tc_provider_data or None,
))
usage = None
if hasattr(response, "usage") and response.usage:
usage = Usage.from_openai(response.usage)
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.
model_extra = getattr(msg, "model_extra", None) or {}
model_extra = model_extra if isinstance(model_extra, dict) else {}
reasoning = getattr(msg, "reasoning", None)
reasoning_content = getattr(msg, "reasoning_content", None)
if reasoning_content is None:
reasoning_content = model_extra.get("reasoning_content")
provider_data: Dict[str, Any] = {}
provider_data: dict[str, Any] = {}
if reasoning_content is not None:
provider_data["reasoning_content"] = reasoning_content
rd = getattr(msg, "reasoning_details", None)
if rd:
provider_data["reasoning_details"] = rd
if getattr(msg, "reasoning_details", None):
provider_data["reasoning_details"] = msg.reasoning_details
# OpenAI structured refusal: ``message.refusal`` set, ``content`` empty.
# Proxies fronting Anthropic/Bedrock surface Claude refusals this way; without
@@ -589,7 +525,7 @@ class ChatCompletionsTransport(ProviderTransport):
return NormalizedResponse(
content=content, tool_calls=tool_calls, finish_reason=finish_reason,
reasoning=reasoning, usage=usage, provider_data=provider_data or None,
reasoning=getattr(msg, "reasoning", None), usage=usage, provider_data=provider_data or None,
)
def validate_response(self, response: Any) -> bool:
@@ -604,7 +540,7 @@ class ChatCompletionsTransport(ProviderTransport):
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 (#61871)
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
+75 -87
View File
@@ -8,7 +8,7 @@ import hashlib
import json
import logging
import re
from typing import Any, Callable, Dict, List, Optional, Tuple
from typing import Any, Callable, Optional
from agent.reasoning_effort import (
ACTUAL_RELAY_EFFORTS, XAI_GROK46_EFFORTS, XAI_LEGACY_EFFORTS, clamp_effort,
@@ -36,9 +36,7 @@ def _bounded_prompt_cache_key(value: Any) -> Optional[str]:
key = "" if value is None else str(value).strip()
if not key:
return None
if len(key) <= 64:
return key
return "pck_" + hashlib.sha256(key.encode("utf-8", errors="replace")).hexdigest()[:24]
return key if len(key) <= 64 else "pck_" + hashlib.sha256(key.encode("utf-8", errors="replace")).hexdigest()[:24]
def _bound_prompt_cache_key_field(container: Any) -> None:
@@ -51,9 +49,12 @@ def _bound_prompt_cache_key_field(container: Any) -> None:
container.pop("prompt_cache_key", None)
def _str_headers(existing: Any) -> Dict[str, str]:
"""Copy an ``extra_headers`` dict with str-coerced keys/values."""
return {str(k): str(v) for k, v in existing.items() if k and v is not None} if isinstance(existing, dict) else {}
def _merge_extra_headers(kwargs: dict[str, Any], **headers: str) -> None:
"""Merge ``headers`` into a str-coerced copy of ``kwargs['extra_headers']`` (SDK kwarg -> HTTP headers)."""
existing = kwargs.get("extra_headers")
merged = {str(k): str(v) for k, v in existing.items() if k and v is not None} if isinstance(existing, dict) else {}
merged.update(headers)
kwargs["extra_headers"] = merged
# Client-side ``web_search`` on xAI Responses collides with Grok's native
@@ -62,8 +63,8 @@ def _str_headers(existing: Any) -> Dict[str, str]:
_XAI_CLIENT_WEB_SEARCH_ALIAS = "hermes_web_search"
# OpenCode /v1/responses rejects client tools using these names (HTTP 400
# "custom function name 'X' is reserved", #85589); xAI rejects ``tool_search``
# (reserved for Grok's native Tool Search, #95003). Aliased as hermes_<name>.
# "custom function name 'X' is reserved"); xAI rejects ``tool_search``
# (reserved for Grok's native Tool Search). Aliased as hermes_<name>.
_OPENCODE_RESERVED_TOOL_NAMES = ("web_search", "search_files")
_XAI_RESERVED_TOOL_NAMES = ("tool_search",)
_RESERVED_TOOL_ALIAS_PREFIX = "hermes_"
@@ -78,7 +79,7 @@ _LEGACY_ALIAS_FALLBACK = {
_LEGACY_ALIAS_FALLBACK[_XAI_CLIENT_WEB_SEARCH_ALIAS] = "web_search"
def _is_opencode_responses_backend(params: Dict[str, Any]) -> bool:
def _is_opencode_responses_backend(params: dict[str, Any]) -> bool:
"""True for opencode-zen/go providers, ``opencode-*`` families, or opencode.ai hosts."""
try:
from hermes_cli.models import opencode_provider_family
@@ -96,10 +97,10 @@ def _is_opencode_responses_backend(params: Dict[str, Any]) -> bool:
def _alias_reserved_tools(
response_tools: List[Dict[str, Any]], reserved_names: Tuple[str, ...],
response_tools: list[dict[str, Any]], reserved_names: tuple[str, ...],
name_of: Callable[[dict], Any] = lambda t: t.get("name"),
rename: Callable[[dict, str], dict] = lambda t, alias: {**t, "name": alias},
) -> Tuple[List[Dict[str, Any]], Dict[str, str]]:
) -> tuple[list[dict[str, Any]], dict[str, str]]:
"""Alias provider-reserved function names on the wire.
Returns ``(rewritten_tools, {alias: original_name})``. An alias that is
@@ -107,8 +108,8 @@ def _alias_reserved_tools(
duplicating a wire name. ``name_of``/``rename`` adapt the tool shape
(Responses ``{name}`` by default; chat_completions passes ``function.name``).
"""
rewritten: List[Dict[str, Any]] = []
alias_map: Dict[str, str] = {}
rewritten: list[dict[str, Any]] = []
alias_map: dict[str, str] = {}
taken = {name_of(tool) for tool in response_tools if isinstance(tool, dict) and name_of(tool)}
for tool in response_tools:
name = name_of(tool) if isinstance(tool, dict) else None
@@ -131,7 +132,7 @@ def _xai_prefers_native_web_search() -> bool:
Resolves via the web-search registry (config ``web.search_backend`` /
``web.backend``), falling back to the legacy ``_get_search_backend`` probe.
Fails closed to native (True) on any error — preserves the #48108 fix.
Fails closed to native (True) on any error.
"""
try:
from agent.web_search_registry import get_active_search_provider
@@ -147,17 +148,16 @@ def _xai_prefers_native_web_search() -> bool:
return True
def _alias_wire_tools(
response_tools: Any, params: Dict[str, Any], is_xai_responses: bool
) -> Tuple[Any, Dict[str, str]]:
def _alias_wire_tools(response_tools: Any, params: dict[str, Any], is_xai_responses: bool) -> tuple[Any, dict[str, str]]:
"""Apply provider-reserved tool-name aliasing; returns ``(tools, {alias: original})``.
The alias map is wire provenance for THIS request — normalize_response
reverses only these. xAI: a client function named ``web_search`` collides
with Grok's native search (#48108); native mode swaps it 1:1 for the
built-in, client mode keeps Hermes dispatch under an alias.
with Grok's native search; native mode swaps it 1:1 for the built-in,
client mode keeps Hermes dispatch under an alias.
"""
wire_aliases: Dict[str, str] = {}
wire_aliases: dict[str, str] = {}
def is_client_web_search(t: Any) -> bool:
return isinstance(t, dict) and t.get("name") == "web_search"
@@ -178,7 +178,7 @@ def _alias_wire_tools(
return response_tools, wire_aliases
def _resolve_reasoning(model: str, params: Dict[str, Any]) -> Tuple[Any, bool]:
def _resolve_reasoning(model: str, params: dict[str, Any]) -> tuple[Any, bool]:
"""``(effort, enabled)`` for the request, effort clamped to the endpoint's vocabulary.
Wire vocabularies live in agent.reasoning_effort; clamp_effort picks the
@@ -209,13 +209,6 @@ def _resolve_reasoning(model: str, params: Dict[str, Any]) -> Tuple[Any, bool]:
return clamp_effort(reasoning_effort, supported), reasoning_enabled
def _merge_extra_headers(kwargs: Dict[str, Any], **headers: str) -> None:
"""Merge str-coerced ``headers`` into ``kwargs['extra_headers']`` (SDK kwarg -> HTTP headers)."""
merged = _str_headers(kwargs.get("extra_headers"))
merged.update(headers)
kwargs["extra_headers"] = merged
_EXTENDED_PROMPT_CACHE_MODELS = (
"gpt-5.5-pro", "gpt-5.5", "gpt-5.4", "gpt-5.2",
"gpt-5.1-codex-max", "gpt-5.1-codex-mini", "gpt-5.1-chat-latest", "gpt-5.1-codex", "gpt-5.1",
@@ -243,9 +236,7 @@ def _default_prompt_cache_retention_for_request(model: str, base_url: Any) -> Op
return "24h" if _EXTENDED_PROMPT_CACHE_MODEL_RE.search(normalized) else None
def _content_cache_key(
instructions: str, tools: Optional[List[Dict[str, Any]]], scope_id: str = ""
) -> Optional[str]:
def _content_cache_key(instructions: str, tools: Optional[list[dict[str, Any]]], scope_id: str = "") -> Optional[str]:
"""``pck_<sha256[:24]>`` of (scope_id, instructions, name-sorted tools), or None if nothing static.
The key is a routing hint only. ``scope_id`` keeps unrelated sessions off one
@@ -256,14 +247,12 @@ def _content_cache_key(
tools_part = ""
if tools:
sorted_tools = sorted(
(t for t in tools if isinstance(t, dict)),
key=lambda t: str(t.get("name") or t.get("type") or ""),
(t for t in tools if isinstance(t, dict)), key=lambda t: str(t.get("name") or t.get("type") or ""),
)
tools_part = json.dumps(sorted_tools, sort_keys=True, ensure_ascii=False, separators=(",", ":"))
# \x00 separators so a boundary can't be forged by content containing the same bytes.
content = f"{scope_id}\x00{instructions or ''}\x00{tools_part}"
digest = hashlib.sha256(content.encode("utf-8", errors="replace")).hexdigest()[:24]
return f"pck_{digest}"
return "pck_" + hashlib.sha256(content.encode("utf-8", errors="replace")).hexdigest()[:24]
def _profile_declared_efforts(provider: Any, model: Optional[str], base_url: Any = None) -> Optional[tuple]:
@@ -293,7 +282,7 @@ def _profile_declared_efforts(provider: Any, model: Optional[str], base_url: Any
return None if declared is None else tuple(declared)
def _is_azure_foundry_responses(params: Dict[str, Any]) -> bool:
def _is_azure_foundry_responses(params: dict[str, Any]) -> bool:
"""True for Microsoft Foundry's Responses API (provider id, else host match — not substring)."""
from utils import base_url_host_matches
@@ -302,7 +291,7 @@ def _is_azure_foundry_responses(params: Dict[str, Any]) -> bool:
return base_url_host_matches(str(params.get("base_url") or ""), "services.ai.azure.com")
def _is_post_tool_replay(messages: Optional[List[Dict[str, Any]]]) -> bool:
def _is_post_tool_replay(messages: Optional[list[dict[str, Any]]]) -> bool:
"""True when ``messages`` end on a tool-result run issued by the preceding assistant turn.
Azure Foundry rejects only this post-tool follow-up shape when encrypted
@@ -356,6 +345,13 @@ def _native_compaction_active(context_management: Any) -> bool:
return isinstance(context_management, list) and bool(context_management)
def _coerce_timeout(timeout: Any) -> Optional[float]:
"""Finite positive number -> float; anything else (None, bool, str, inf) -> None."""
if isinstance(timeout, (int, float)) and not isinstance(timeout, bool) and 0 < float(timeout) < float("inf"):
return float(timeout)
return None
class ResponsesApiTransport(ProviderTransport):
"""Transport for api_mode='codex_responses'."""
@@ -367,44 +363,45 @@ class ResponsesApiTransport(ProviderTransport):
_last_issuer_kind: Optional[str] = None
# ``{wire_alias: original_name}`` of the most recent build_kwargs call. None
# = no request built (use legacy map); {} = no aliases sent, no reverse rewrite.
_last_wire_aliases: Optional[Dict[str, str]] = None
_last_wire_aliases: Optional[dict[str, str]] = None
@property
def api_mode(self) -> str:
return "codex_responses"
def _resolve_issuer_kind(self, params: Dict[str, Any]) -> str:
"""Classify the current Responses endpoint from transport params."""
def _resolve_issuer_kind(self, params: dict[str, Any]) -> str:
"""Classify the current Responses endpoint from transport params (stashed for normalize_response)."""
from agent.codex_responses_adapter import _classify_responses_issuer
return _classify_responses_issuer(
self._last_issuer_kind = _classify_responses_issuer(
is_xai_responses=params.get("is_xai_responses") is True,
is_github_responses=params.get("is_github_responses") is True,
is_codex_backend=params.get("is_codex_backend") is True,
base_url=params.get("base_url"),
)
return self._last_issuer_kind
def convert_messages(self, messages: List[Dict[str, Any]], **kwargs) -> Any:
def convert_messages(self, messages: list[dict[str, Any]], **kwargs) -> Any:
"""Convert OpenAI chat messages to Responses API input items."""
from agent.codex_responses_adapter import _chat_messages_to_responses_input
issuer = self._resolve_issuer_kind(kwargs)
self._last_issuer_kind = issuer
return _chat_messages_to_responses_input(
messages, is_xai_responses=kwargs.get("is_xai_responses") is True,
is_github_responses=kwargs.get("is_github_responses") is True,
replay_encrypted_reasoning=bool(kwargs.get("replay_encrypted_reasoning", True)),
current_issuer_kind=issuer,
current_issuer_kind=self._resolve_issuer_kind(kwargs),
native_compaction_eligible=_native_compaction_active(kwargs.get("context_management")),
)
def convert_tools(self, tools: List[Dict[str, Any]]) -> Any:
def convert_tools(self, tools: Optional[list[dict[str, Any]]]) -> Any:
"""Convert OpenAI tool schemas to Responses API function definitions."""
from agent.codex_responses_adapter import _responses_tools
return _responses_tools(tools)
def build_kwargs(
self, model: str, messages: List[Dict[str, Any]],
tools: Optional[List[Dict[str, Any]]] = None, **params,
) -> Dict[str, Any]:
self, model: str, messages: list[dict[str, Any]], tools: Optional[list[dict[str, Any]]] = None, **params,
) -> dict[str, Any]:
"""Build Responses API kwargs (calls convert_messages/convert_tools internally).
params: instructions, reasoning_config ({effort, enabled}), session_id
@@ -414,7 +411,6 @@ class ResponsesApiTransport(ProviderTransport):
base_url, is_github_responses, is_codex_backend, is_xai_responses,
github_reasoning_extra, context_management, replay_encrypted_reasoning.
"""
from agent.codex_responses_adapter import _chat_messages_to_responses_input, _responses_tools
from run_agent import DEFAULT_AGENT_IDENTITY
instructions = params.get("instructions", "")
@@ -422,26 +418,23 @@ class ResponsesApiTransport(ProviderTransport):
if not instructions and messages and messages[0].get("role") == "system":
instructions = str(messages[0].get("content") or "").strip()
payload_messages = messages[1:]
if not instructions:
instructions = DEFAULT_AGENT_IDENTITY
instructions = instructions or DEFAULT_AGENT_IDENTITY
is_github_responses = params.get("is_github_responses") is True
is_codex_backend = params.get("is_codex_backend") is True
is_xai_responses = params.get("is_xai_responses") is True
replay_encrypted_reasoning = bool(params.get("replay_encrypted_reasoning", True))
# Foundry 400s on encrypted-reasoning replay only in the post-tool
# follow-up turn; suppress replay for exactly that shape.
if replay_encrypted_reasoning and _is_azure_foundry_responses(params) and _is_post_tool_replay(payload_messages):
replay_encrypted_reasoning = False
replay_encrypted_reasoning = bool(params.get("replay_encrypted_reasoning", True)) and not (
_is_azure_foundry_responses(params) and _is_post_tool_replay(payload_messages)
)
# Single source of truth: the same predicate decides whether
# context_management goes out AND whether the converter may replay a checkpoint.
context_management = params.get("context_management")
native_compaction_active = _native_compaction_active(context_management)
issuer_kind = self._resolve_issuer_kind(params)
self._last_issuer_kind = issuer_kind
reasoning_effort, reasoning_enabled = _resolve_reasoning(model, params)
response_tools, self._last_wire_aliases = _alias_wire_tools(_responses_tools(tools), params, is_xai_responses)
response_tools, self._last_wire_aliases = _alias_wire_tools(self.convert_tools(tools), params, is_xai_responses)
# Lazy: provider plugins import this transport during model_metadata init.
from agent.model_metadata import strip_codex_context_variant_suffix as _strip_ctx_variant
@@ -449,17 +442,15 @@ class ResponsesApiTransport(ProviderTransport):
# ``-900k`` picker variants are Hermes-side aliases; the backend knows only the base slug.
"model": _strip_ctx_variant(model),
"instructions": instructions,
"input": _chat_messages_to_responses_input(
payload_messages, is_xai_responses=is_xai_responses,
is_github_responses=is_github_responses,
replay_encrypted_reasoning=replay_encrypted_reasoning,
current_issuer_kind=issuer_kind,
native_compaction_eligible=native_compaction_active,
"input": self.convert_messages(
payload_messages, is_xai_responses=is_xai_responses, is_github_responses=is_github_responses,
replay_encrypted_reasoning=replay_encrypted_reasoning, base_url=params.get("base_url"),
is_codex_backend=is_codex_backend, context_management=context_management,
),
"store": False,
}
# ``tools`` MUST be omitted when empty: the openai SDK iterates it without
# a None guard before any HTTP request (#32892).
# a None guard before any HTTP request.
if response_tools:
kwargs["tools"] = response_tools
kwargs["tool_choice"] = "auto"
@@ -469,9 +460,9 @@ class ResponsesApiTransport(ProviderTransport):
session_id = params.get("session_id")
# Cache key is content-addressed (instructions + tools) within a logical
# scope — cache_scope_id survives compression rotation (#79017), and the
# cron per-fire timestamp is stripped. session_id itself stays untouched
# for transcript isolation (Codex ``session_id`` header below).
# scope — cache_scope_id survives compression rotation, and the cron
# per-fire timestamp is stripped. session_id itself stays untouched for
# transcript isolation (Codex ``session_id`` header below).
_cache_scope = _cache_scope_from_session_id(params.get("cache_scope_id") or session_id)
cache_key = _content_cache_key(instructions, response_tools, _cache_scope) or _cache_scope
# xAI takes prompt_cache_key in extra_body (below); GitHub Models opts out entirely.
@@ -482,32 +473,31 @@ class ResponsesApiTransport(ProviderTransport):
if cache_retention:
kwargs.setdefault("prompt_cache_retention", cache_retention)
include = ["reasoning.encrypted_content"] if replay_encrypted_reasoning else []
if reasoning_enabled and is_xai_responses:
from agent.model_metadata import grok_supports_reasoning_effort
kwargs["include"] = ["reasoning.encrypted_content"] if replay_encrypted_reasoning else []
kwargs["include"] = include
# xAI 400s on ``reasoning.effort`` for models outside the allowlist.
if grok_supports_reasoning_effort(model):
kwargs["reasoning"] = {"effort": reasoning_effort}
elif reasoning_enabled:
if is_github_responses:
github_reasoning = params.get("github_reasoning_extra")
if github_reasoning is not None:
kwargs["reasoning"] = github_reasoning
if params.get("github_reasoning_extra") is not None:
kwargs["reasoning"] = params["github_reasoning_extra"]
else:
kwargs["reasoning"] = {"effort": reasoning_effort, "summary": "auto"}
kwargs["include"] = ["reasoning.encrypted_content"] if replay_encrypted_reasoning else []
kwargs["include"] = include
elif not is_github_responses and not is_xai_responses:
kwargs["include"] = []
request_overrides = params.get("request_overrides")
if request_overrides:
kwargs.update(request_overrides)
if params.get("request_overrides"):
kwargs.update(params["request_overrides"])
_bound_prompt_cache_key_field(kwargs)
# Older xAI models reject ``service_tier`` (HTTP 400); only Grok 4.6
# accepts Priority Processing (#28490, #84799).
# accepts Priority Processing.
if is_xai_responses:
from agent.model_metadata import is_grok_46_family
@@ -515,16 +505,16 @@ class ResponsesApiTransport(ProviderTransport):
kwargs.pop("service_tier", None)
# Forward per-request timeout to the SDK (providers.<id>.request_timeout_seconds).
timeout = kwargs.get("timeout", params.get("timeout"))
if isinstance(timeout, (int, float)) and not isinstance(timeout, bool) and 0 < float(timeout) < float("inf"):
kwargs["timeout"] = float(timeout)
timeout = _coerce_timeout(kwargs.get("timeout", params.get("timeout")))
if timeout is not None:
kwargs["timeout"] = timeout
else:
kwargs.pop("timeout", None)
if is_codex_backend:
# Codex rejects body-level extra_headers, but the SDK kwarg maps to
# HTTP headers. ``session_id`` = raw physical id (transcript identity);
# ``x-client-request-id`` mirrors the body cache key so both agree (#78941).
# ``x-client-request-id`` mirrors the body cache key so both agree.
final_cache_key = kwargs.get("prompt_cache_key") or _bounded_prompt_cache_key(_cache_scope)
headers = {}
if session_id:
@@ -533,14 +523,12 @@ class ResponsesApiTransport(ProviderTransport):
headers["x-client-request-id"] = final_cache_key
if headers:
_merge_extra_headers(kwargs, **headers)
max_tokens = params.get("max_tokens")
if max_tokens is not None and not is_codex_backend:
kwargs["max_output_tokens"] = max_tokens
elif params.get("max_tokens") is not None:
kwargs["max_output_tokens"] = params["max_tokens"]
if is_xai_responses and session_id:
# Scoped like the body key so cron's per-fire timestamp doesn't pin
# each fire to a different xAI backend server (#78941).
# each fire to a different xAI backend server.
_merge_extra_headers(kwargs, **{"x-grok-conv-id": _cache_scope})
# xAI reads prompt_cache_key from the body; sent via extra_body so it
# survives SDK builds whose Responses.stream() dropped the typed kwarg.