mirrors/hermes-agent
1
0
Fork 0
mirror of https://github.com/NousResearch/hermes-agent.git synced 2026-07-31 19:16:29 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

refactor(agent): single-owner call_id + reasoning_content sanitization policies (wire-parity verified)

Browse source
This commit is contained in:
teknium1 2026-07-29 12:07:40 -07:00 • committed by Teknium
parent 1ea2ea18fd
commit f8758dcaf8
5 changed files with 719 additions and 197 deletions
Download patch file Download diff file Expand all files Collapse all files

114
agent/agent_runtime_helpers.py
View file

@ -3331,89 +3331,17 @@ def intent_ack_continuation_enabled(agent) -> bool:
def copy_reasoning_content_for_api(agent, source_msg: dict, api_msg: dict) -> None:
"""Copy provider-facing reasoning fields onto an API replay message."""
if source_msg.get("role") != "assistant":
return
"""Copy provider-facing reasoning fields onto an API replay message.
needs_thinking_pad = agent._needs_thinking_reasoning_pad()
Forwarder — the strip-vs-repad POLICY is owned by
``agent.message_sanitization.apply_reasoning_content_policy`` (audit F4);
this only supplies the agent's cached provider-direction flag.
"""
from agent.message_sanitization import apply_reasoning_content_policy
# 1. Explicit reasoning_content already set.
#
# When the active provider enforces the thinking-mode echo-back
# (DeepSeek / Kimi / MiMo), preserve it verbatim — that includes their
# own space-placeholder written at creation time and any valid reasoning
# from the same provider. Sessions persisted BEFORE #17341 have
# empty-string placeholders pinned at creation time; DeepSeek V4 Pro
# rejects those with HTTP 400, so upgrade "" → " " on replay.
#
# When the active provider does NOT enforce echo-back, strip the field
# entirely. Strict OpenAI-compatible providers (Mistral, Cerebras, Groq,
# SambaNova, …) reject ANY reasoning_content key in input messages with
# HTTP 400/422 ("Extra inputs are not permitted"), even an empty string
# or a single-space pad. This is the cross-provider fallback case: a
# reasoning primary (DeepSeek/Kimi/MiMo) pads history with " ", then a
# fallback to a strict provider replays that pad and 422s. Stripping
# here covers the rebuild path; reapply_reasoning_echo_for_provider()
# covers the already-built api_messages path. Refs #45655.
existing = source_msg.get("reasoning_content")
if isinstance(existing, str):
if not needs_thinking_pad:
api_msg.pop("reasoning_content", None)
elif existing == "":
api_msg["reasoning_content"] = " "
else:
api_msg["reasoning_content"] = existing
return
# 2. Cross-provider poisoned history (#15748): on DeepSeek/Kimi,
# if the source turn has tool_calls AND a 'reasoning' field but no
# 'reasoning_content' key, the 'reasoning' text was written by a
# prior provider (e.g. MiniMax) — DeepSeek's own _build_assistant_message
# pins reasoning_content at creation time for tool-call turns, so the
# shape (reasoning set, reasoning_content absent, tool_calls present)
# is unreachable from same-provider DeepSeek history after this fix.
# Inject a single space to satisfy the API without leaking another
# provider's chain of thought to DeepSeek/Kimi. Space (not "")
# because DeepSeek V4 Pro rejects empty-string reasoning_content
# in thinking mode (refs #17341).
normalized_reasoning = source_msg.get("reasoning")
if (
needs_thinking_pad
and source_msg.get("tool_calls")
and isinstance(normalized_reasoning, str)
and normalized_reasoning
):
api_msg["reasoning_content"] = " "
return
# 3. Healthy session: promote 'reasoning' field to 'reasoning_content'
# for providers that use the internal 'reasoning' key.
# This must happen before the unconditional empty-string fallback so
# genuine reasoning content is not overwritten (#15812 regression in
# PR #15478). Only promote for providers that enforce echo-back —
# strict providers reject the field (refs #45655).
if isinstance(normalized_reasoning, str) and normalized_reasoning:
if needs_thinking_pad:
api_msg["reasoning_content"] = normalized_reasoning
else:
api_msg.pop("reasoning_content", None)
return
# 4. DeepSeek / Kimi thinking mode: all assistant messages need
# reasoning_content. Inject a single space to satisfy the provider's
# requirement when no explicit reasoning content is present. Covers
# both tool-call turns (already-poisoned history with no reasoning
# at all) and plain text turns. Space (not "") because DeepSeek V4
# Pro tightened validation and rejects empty string with HTTP 400
# ("The reasoning content in the thinking mode must be passed back
# to the API"). Refs #17341.
if needs_thinking_pad:
api_msg["reasoning_content"] = " "
return
# 5. reasoning_content was present but not a string (e.g. None after
# context compaction). Don't pass null to the API.
api_msg.pop("reasoning_content", None)
apply_reasoning_content_policy(
source_msg, api_msg, agent._needs_thinking_reasoning_pad()
)
def reapply_reasoning_echo_for_provider(agent, api_messages: list) -> int:
@ -3445,25 +3373,11 @@ def reapply_reasoning_echo_for_provider(agent, api_messages: list) -> int:
Returns the number of assistant turns whose reasoning_content was added or
removed.
"""
needs_pad = agent._needs_thinking_reasoning_pad()
changed = 0
for api_msg in api_messages:
if api_msg.get("role") != "assistant":
continue
if needs_pad:
if api_msg.get("reasoning_content"):
continue
copy_reasoning_content_for_api(agent, api_msg, api_msg)
if api_msg.get("reasoning_content"):
changed += 1
else:
# Strict provider — strip any stale reasoning_content pad left
# over from a reasoning primary so the fallback request doesn't
# 400/422 on it.
if "reasoning_content" in api_msg:
api_msg.pop("reasoning_content", None)
changed += 1
return changed
from agent.message_sanitization import reapply_reasoning_echo
return reapply_reasoning_echo(
api_messages, agent._needs_thinking_reasoning_pad()
)
def _iter_httpx_pool_objects(http_client: Any):

9
agent/codex_responses_adapter.py
View file

@ -18,6 +18,7 @@ import uuid
from types import SimpleNamespace
from typing import Any, Dict, List, Optional
from agent.message_sanitization import deterministic_call_id
from agent.prompt_builder import DEFAULT_AGENT_IDENTITY
logger = logging.getLogger(__name__)
@ -182,13 +183,13 @@ def _summarize_user_message_for_log(content: Any, *, sep: str = " ") -> str:
def _deterministic_call_id(fn_name: str, arguments: str, index: int = 0) -> str:
"""Generate a deterministic call_id from tool call content.
Used as a fallback when the API doesn't provide a call_id.
Thin wrapper over the single policy owner
``agent.message_sanitization.deterministic_call_id`` (audit F4) — kept
as a module-level name because run_agent and tests import it from here.
Deterministic IDs prevent cache invalidation — random UUIDs would
make every API call's prefix unique, breaking OpenAI's prompt cache.
"""
seed = f"{fn_name}:{arguments}:{index}"
digest = hashlib.sha256(seed.encode("utf-8", errors="replace")).hexdigest()[:12]
return f"call_{digest}"
return deterministic_call_id(fn_name, arguments, index)
def _clamp_responses_call_id(call_id: str) -> str:

375
agent/message_sanitization.py
View file

@ -14,6 +14,7 @@ re-exports from ``run_agent`` remain in place so existing imports
from __future__ import annotations
import hashlib
import json
import logging
import re
@ -474,4 +475,378 @@ __all__ = [
"_sanitize_tools_non_ascii",
"_strip_images_from_messages",
"_sanitize_structure_non_ascii",
# call_id policy owners (F4 consolidation)
"deterministic_call_id",
"coalesce_tool_call_id",
"uniquify_tool_call_ids",
# reasoning_content policy owners (F4 consolidation)
"reasoning_echo_family",
"matches_reasoning_echo_family",
"needs_reasoning_echo",
"apply_reasoning_content_policy",
"reapply_reasoning_echo",
]
# ---------------------------------------------------------------------------
# call_id policy — single owner (audit F4, incident chain I4)
# ---------------------------------------------------------------------------
#
# Three forked policy sites converged here:
# * agent/codex_responses_adapter.py `_deterministic_call_id` — hash
# synthesis when a provider omits call_id (fa3ab2ffd0 → e45f2b39e2).
# * run_agent.AIAgent._get_tool_call_id_static — `call_id or id`
# coalescing for dicts and SDK objects.
# * run_agent.AIAgent._uniquify_tool_call_ids — duplicate-id repair with
# deterministic `_d<n>` suffixes (#58327 loss class).
#
# NOT consolidated (different scheme on purpose):
# agent/transports/codex_event_projector._deterministic_call_id maps codex
# app-server ITEM ids (`codex_<type>_<item_id>`), not chat tool-call
# content; merging the two would change ids and invalidate prompt caches.
#
# HARD INVARIANT: everything here must stay deterministic (never uuid4) and
# byte-identical for existing inputs — these ids feed prompt-cache prefixes.
def deterministic_call_id(fn_name: str, arguments: str, index: int = 0) -> str:
"""Generate a deterministic call_id from tool call content.
Used as a fallback when the API doesn't provide a call_id.
Deterministic IDs prevent cache invalidation — random UUIDs would
make every API call's prefix unique, breaking OpenAI's prompt cache.
"""
seed = f"{fn_name}:{arguments}:{index}"
digest = hashlib.sha256(seed.encode("utf-8", errors="replace")).hexdigest()[:12]
return f"call_{digest}"
def coalesce_tool_call_id(tc: Any) -> str:
"""Extract the effective call ID from a tool_call entry (dict or object).
Single owner for the ``call_id or id`` coalescing rule: Codex Responses
tool calls carry ``call_id`` (authoritative pairing key), Chat
Completions ones carry ``id`` only. Returns ``""`` when neither is set.
"""
if isinstance(tc, dict):
return (tc.get("call_id", "") or tc.get("id", "") or "").strip()
return (getattr(tc, "call_id", "") or getattr(tc, "id", "") or "").strip()
def uniquify_tool_call_ids(tool_calls: list) -> list:
"""Ensure every tool call in a single assistant turn has a distinct id.
Some models/providers reuse one call id across different calls in a
single batch (observed with native Kimi Responses replays, Ollama-
compatible endpoints, and degraded models at long context; same bug
class as openclaw/openclaw#110518 / #110956). Duplicate ids are lossy
downstream: the pre-API sanitizer keeps only the first call/result
pair per id (#58327), so the later call's result silently vanishes
from every replayed payload, and strict providers (Anthropic
tool_use, DeepSeek) reject duplicate ids outright.
The first occurrence keeps its id; later collisions get a
deterministic ``<id>_d<n>`` suffix — never a random UUID, which would
break prompt-cache prefix stability across replays. Mutates the
entries in place (SDK models / SimpleNamespace / dicts) and returns
the same list. Blank/missing ids are left for the deterministic
fallback in ``build_assistant_message``.
"""
seen: set = set()
for tc in tool_calls or []:
# Same coalescing rule as ``coalesce_tool_call_id`` but tolerant of
# non-string ids (degraded models can emit ints/None here).
if isinstance(tc, dict):
raw = tc.get("call_id") or tc.get("id") or ""
else:
raw = getattr(tc, "call_id", None) or getattr(tc, "id", None) or ""
raw = raw.strip() if isinstance(raw, str) else ""
if not raw:
continue
# Composite Responses ids ("call_x|fc_y") collide on the call
# half — that's the pairing key providers enforce per turn.
cid = raw.split("|", 1)[0]
if not cid:
continue
if cid not in seen:
seen.add(cid)
continue
n = 2
new_id = f"{cid}_d{n}"
while new_id in seen:
n += 1
new_id = f"{cid}_d{n}"
seen.add(new_id)
def _renamed(value):
# Preserve a composite id's response-item half so the
# provider's real fc_/item id survives the rename.
if isinstance(value, str) and "|" in value:
return f"{new_id}|{value.split('|', 1)[1]}"
return new_id
try:
if isinstance(tc, dict):
if tc.get("id"):
tc["id"] = _renamed(tc["id"])
else:
tc["id"] = new_id
if tc.get("call_id"):
tc["call_id"] = new_id
else:
tc.id = _renamed(getattr(tc, "id", None))
if getattr(tc, "call_id", None):
tc.call_id = new_id
except Exception:
logger.warning(
"Could not uniquify duplicate tool call id %s", cid
)
continue
_fn = tc.get("function") if isinstance(tc, dict) else getattr(tc, "function", None)
_fn_name = (_fn.get("name") if isinstance(_fn, dict) else getattr(_fn, "name", None)) or "?"
logger.warning(
"Model reused tool call id %s within one turn; renamed the "
"duplicate to %s (tool=%s) to keep call/result pairing "
"lossless.", cid, new_id, _fn_name,
)
return tool_calls
# ---------------------------------------------------------------------------
# reasoning_content policy — single owner (audit F4)
# ---------------------------------------------------------------------------
#
# The strip-vs-repad decision was previously forked across the wire files in
# separate incident commits (2b3a4f0af8 strip for strict providers,
# b5495db701 re-pad for require-side, 94b3131be7/9a9f8a6d99 kimi pad). The
# POLICY — which provider direction gets which treatment — lives here as one
# rule table + apply functions; adapters keep only SYNTAX mapping (e.g.
# anthropic_adapter turning reasoning_content into a thinking block).
#
# Direction table:
# require-side (echo-back enforced; replays 400 without the field):
# kimi — provider kimi-coding/kimi-coding-cn, or host api.kimi.com /
# moonshot.ai / moonshot.cn. Host-driven on purpose:
# aggregators re-exporting kimi models reject the echo.
# deepseek — provider "deepseek", model contains "deepseek", or host
# api.deepseek.com (#15250; V4 rejects empty-string pads,
# hence the " " single-space pad, #17341).
# mimo — provider "xiaomi", model contains "mimo", or host
# *.xiaomimimo.com.
# strict side (field rejected with 400/422 "Extra inputs are not
# permitted"): everyone else — Mistral, Cerebras, Groq, SambaNova, …
# (#45655). Strip the key entirely, even a single-space pad.
_REASONING_ECHO_RULES: tuple = (
# (family, exact providers (raw), exact providers (lowered),
# model substrings (lowered), base_url hosts)
("kimi", frozenset({"kimi-coding", "kimi-coding-cn"}), frozenset(), (),
("api.kimi.com", "moonshot.ai", "moonshot.cn")),
("deepseek", frozenset(), frozenset({"deepseek"}), ("deepseek",),
("api.deepseek.com",)),
("mimo", frozenset(), frozenset({"xiaomi"}), ("mimo",),
("api.xiaomimimo.com", "xiaomimimo.com")),
)
def _family_rule(family: str) -> tuple:
for rule in _REASONING_ECHO_RULES:
if rule[0] == family:
return rule
raise KeyError(family)
def matches_reasoning_echo_family(
family: str, provider: Any, model: Any, base_url: Any
) -> bool:
"""True when (provider, model, base_url) matches one echo-back family.
Families can overlap (e.g. a deepseek-named model pointed at a kimi
host); this membership test is independent per family so per-family
predicates keep their original semantics.
"""
from utils import base_url_host_matches
_, raw_providers, lowered_providers, model_subs, hosts = _family_rule(family)
provider_lower = (provider or "").lower()
model_lower = (model or "").lower()
if provider in raw_providers or provider_lower in lowered_providers:
return True
if any(sub in model_lower for sub in model_subs):
return True
return any(base_url_host_matches(base_url, host) for host in hosts)
def reasoning_echo_family(provider: Any, model: Any, base_url: Any) -> "str | None":
"""Classify the provider direction for the reasoning_content echo policy.
Returns ``"kimi"``, ``"deepseek"``, or ``"mimo"`` (first match in table
order) when the target endpoint enforces reasoning_content echo-back on
assistant turns, else ``None`` (strict/indifferent side — the field must
be stripped).
"""
for rule in _REASONING_ECHO_RULES:
if matches_reasoning_echo_family(rule[0], provider, model, base_url):
return rule[0]
return None
def needs_reasoning_echo(provider: Any, model: Any, base_url: Any) -> bool:
"""True when the endpoint requires reasoning_content echo-back."""
return reasoning_echo_family(provider, model, base_url) is not None
def apply_reasoning_content_policy(
source_msg: dict, api_msg: dict, needs_thinking_pad: bool
) -> None:
"""Copy provider-facing reasoning fields onto an API replay message.
``needs_thinking_pad`` is the require-side flag (see
``needs_reasoning_echo`` / the agent's cached
``_needs_thinking_reasoning_pad``). Mutates ``api_msg`` in place.
"""
if source_msg.get("role") != "assistant":
return
# 1. Explicit reasoning_content already set.
#
# When the active provider enforces the thinking-mode echo-back
# (DeepSeek / Kimi / MiMo), preserve it verbatim — that includes their
# own space-placeholder written at creation time and any valid reasoning
# from the same provider. Sessions persisted BEFORE #17341 have
# empty-string placeholders pinned at creation time; DeepSeek V4 Pro
# rejects those with HTTP 400, so upgrade "" → " " on replay.
#
# When the active provider does NOT enforce echo-back, strip the field
# entirely. Strict OpenAI-compatible providers (Mistral, Cerebras, Groq,
# SambaNova, …) reject ANY reasoning_content key in input messages with
# HTTP 400/422 ("Extra inputs are not permitted"), even an empty string
# or a single-space pad. This is the cross-provider fallback case: a
# reasoning primary (DeepSeek/Kimi/MiMo) pads history with " ", then a
# fallback to a strict provider replays that pad and 422s. Stripping
# here covers the rebuild path; ``reapply_reasoning_echo`` covers the
# already-built api_messages path. Refs #45655.
existing = source_msg.get("reasoning_content")
if isinstance(existing, str):
if not needs_thinking_pad:
api_msg.pop("reasoning_content", None)
elif existing == "":
api_msg["reasoning_content"] = " "
else:
api_msg["reasoning_content"] = existing
return
# 2. Cross-provider poisoned history (#15748): on DeepSeek/Kimi,
# if the source turn has tool_calls AND a 'reasoning' field but no
# 'reasoning_content' key, the 'reasoning' text was written by a
# prior provider (e.g. MiniMax) — DeepSeek's own _build_assistant_message
# pins reasoning_content at creation time for tool-call turns, so the
# shape (reasoning set, reasoning_content absent, tool_calls present)
# is unreachable from same-provider DeepSeek history after this fix.
# Inject a single space to satisfy the API without leaking another
# provider's chain of thought to DeepSeek/Kimi. Space (not "")
# because DeepSeek V4 Pro rejects empty-string reasoning_content
# in thinking mode (refs #17341).
normalized_reasoning = source_msg.get("reasoning")
if (
needs_thinking_pad
and source_msg.get("tool_calls")
and isinstance(normalized_reasoning, str)
and normalized_reasoning
):
api_msg["reasoning_content"] = " "
return
# 3. Healthy session: promote 'reasoning' field to 'reasoning_content'
# for providers that use the internal 'reasoning' key.
# This must happen before the unconditional empty-string fallback so
# genuine reasoning content is not overwritten (#15812 regression in
# PR #15478). Only promote for providers that enforce echo-back —
# strict providers reject the field (refs #45655).
if isinstance(normalized_reasoning, str) and normalized_reasoning:
if needs_thinking_pad:
api_msg["reasoning_content"] = normalized_reasoning
else:
api_msg.pop("reasoning_content", None)
return
# 4. DeepSeek / Kimi thinking mode: all assistant messages need
# reasoning_content. Inject a single space to satisfy the provider's
# requirement when no explicit reasoning content is present. Covers
# both tool-call turns (already-poisoned history with no reasoning
# at all) and plain text turns. Space (not "") because DeepSeek V4
# Pro tightened validation and rejects empty string with HTTP 400
# ("The reasoning content in the thinking mode must be passed back
# to the API"). Refs #17341.
if needs_thinking_pad:
api_msg["reasoning_content"] = " "
return
# 5. reasoning_content was present but not a string (e.g. None after
# context compaction). Don't pass null to the API.
api_msg.pop("reasoning_content", None)
def reapply_reasoning_echo(api_messages: list, needs_thinking_pad: bool) -> int:
"""Re-pad (or strip) assistant turns' reasoning_content for the active provider.
``api_messages`` is built once, before the retry loop, while the *primary*
provider is active. A mid-conversation fallback can then switch providers,
so the reasoning fields baked into ``api_messages`` are shaped for the
*prior* provider and must be reconciled against the *current* one:
* Switching TO a require-side provider (DeepSeek / Kimi / MiMo thinking
mode): assistant turns built when the prior provider did NOT need the
echo-back go out without ``reasoning_content`` and the new provider
rejects them with HTTP 400 ("The reasoning_content in the thinking mode
must be passed back"). Re-apply the pad.
* Switching TO a strict provider that rejects the field (Mistral,
Cerebras, Groq, SambaNova, …): assistant turns built under a reasoning
primary carry a ``reasoning_content`` pad (often a single space ``" "``),
and the strict provider rejects it with HTTP 400/422 ("Extra inputs are
not permitted"). Strip the field. This is the exact cross-provider
fallback bug from #45655 — a DeepSeek primary pads history with ``" "``,
the request falls back to Mistral, and Mistral 422s on the stale pad.
Calling this immediately before building the request kwargs reconciles the
fields against the *current* provider. It is idempotent and safe to call
every iteration; it covers every fallback path.
Returns the number of assistant turns whose reasoning_content was added or
removed.
"""
changed = 0
for api_msg in api_messages:
if api_msg.get("role") != "assistant":
continue
if needs_thinking_pad:
if api_msg.get("reasoning_content"):
continue
apply_reasoning_content_policy(api_msg, api_msg, needs_thinking_pad)
if api_msg.get("reasoning_content"):
changed += 1
else:
# Strict provider — strip any stale reasoning_content pad left
# over from a reasoning primary so the fallback request doesn't
# 400/422 on it.
if "reasoning_content" in api_msg:
api_msg.pop("reasoning_content", None)
changed += 1
return changed
# ---------------------------------------------------------------------------
# Image / multimodal parts — evaluated, NOT consolidated (verdict: syntax)
# ---------------------------------------------------------------------------
#
# The per-adapter image handling is format-specific SYNTAX, not shared policy:
# * anthropic_adapter (~1817): data-URL → Anthropic `source: {type: base64}`
# block mapping — Anthropic wire shape only.
# * codex_responses_adapter (~113/165/812): chat `image_url` parts →
# Responses `input_image` items and image counting for log summaries —
# Responses wire shape only.
# * transports/chat_completions: pass-through (native format).
# The one genuinely shared image POLICY — removing images when a server
# rejects them while preserving tool_call_id pairing — already has a single
# owner here: ``_strip_images_from_messages`` above.

122
run_agent.py
View file

@ -181,6 +181,8 @@ from agent.message_sanitization import ( # noqa: F401
_sanitize_tools_non_ascii,
_strip_images_from_messages,
_sanitize_structure_non_ascii,
coalesce_tool_call_id as _sanitize_coalesce_tool_call_id,
uniquify_tool_call_ids as _sanitize_uniquify_tool_call_ids,
)
from agent.codex_responses_adapter import (
_derive_responses_function_call_id as _codex_derive_responses_function_call_id,
@ -4172,10 +4174,12 @@ class AIAgent:
@staticmethod
def _get_tool_call_id_static(tc) -> str:
"""Extract call ID from a tool_call entry (dict or object)."""
if isinstance(tc, dict):
return (tc.get("call_id", "") or tc.get("id", "") or "").strip()
return (getattr(tc, "call_id", "") or getattr(tc, "id", "") or "").strip()
"""Extract call ID from a tool_call entry (dict or object).
Forwarder — policy owner is
``agent.message_sanitization.coalesce_tool_call_id`` (audit F4).
"""
return _sanitize_coalesce_tool_call_id(tc)
@staticmethod
def _get_tool_call_name_static(tc) -> str:
@ -4336,78 +4340,13 @@ class AIAgent:
def _uniquify_tool_call_ids(tool_calls: list) -> list:
"""Ensure every tool call in a single assistant turn has a distinct id.
Some models/providers reuse one call id across different calls in a
single batch (observed with native Kimi Responses replays, Ollama-
compatible endpoints, and degraded models at long context; same bug
class as openclaw/openclaw#110518 / #110956). Duplicate ids are lossy
downstream: the pre-API sanitizer keeps only the first call/result
pair per id (#58327), so the later call's result silently vanishes
from every replayed payload, and strict providers (Anthropic
tool_use, DeepSeek) reject duplicate ids outright.
The first occurrence keeps its id; later collisions get a
deterministic ``<id>_d<n>`` suffix — never a random UUID, which would
break prompt-cache prefix stability across replays. Mutates the
entries in place (SDK models / SimpleNamespace / dicts) and returns
the same list. Blank/missing ids are left for the deterministic
fallback in ``build_assistant_message``.
Forwarder — policy owner is
``agent.message_sanitization.uniquify_tool_call_ids`` (audit F4).
First occurrence keeps its id; later collisions get a deterministic
``<id>_d<n>`` suffix (never uuid4 — prompt-cache prefix stability).
Mutates entries in place and returns the same list.
"""
seen: set = set()
for tc in tool_calls or []:
if isinstance(tc, dict):
raw = tc.get("call_id") or tc.get("id") or ""
else:
raw = getattr(tc, "call_id", None) or getattr(tc, "id", None) or ""
raw = raw.strip() if isinstance(raw, str) else ""
if not raw:
continue
# Composite Responses ids ("call_x|fc_y") collide on the call
# half — that's the pairing key providers enforce per turn.
cid = raw.split("|", 1)[0]
if not cid:
continue
if cid not in seen:
seen.add(cid)
continue
n = 2
new_id = f"{cid}_d{n}"
while new_id in seen:
n += 1
new_id = f"{cid}_d{n}"
seen.add(new_id)
def _renamed(value):
# Preserve a composite id's response-item half so the
# provider's real fc_/item id survives the rename.
if isinstance(value, str) and "|" in value:
return f"{new_id}|{value.split('|', 1)[1]}"
return new_id
try:
if isinstance(tc, dict):
if tc.get("id"):
tc["id"] = _renamed(tc["id"])
else:
tc["id"] = new_id
if tc.get("call_id"):
tc["call_id"] = new_id
else:
tc.id = _renamed(getattr(tc, "id", None))
if getattr(tc, "call_id", None):
tc.call_id = new_id
except Exception:
logger.warning(
"Could not uniquify duplicate tool call id %s", cid
)
continue
_fn = tc.get("function") if isinstance(tc, dict) else getattr(tc, "function", None)
_fn_name = (_fn.get("name") if isinstance(_fn, dict) else getattr(_fn, "name", None)) or "?"
logger.warning(
"Model reused tool call id %s within one turn; renamed the "
"duplicate to %s (tool=%s) to keep call/result pairing "
"lossless.", cid, new_id, _fn_name,
)
return tool_calls
return _sanitize_uniquify_tool_call_ids(tool_calls)
def _repair_tool_call(self, tool_name: str) -> str | None:
"""Forwarder — see ``agent.agent_runtime_helpers.repair_tool_call``."""
@ -6686,12 +6625,12 @@ class AIAgent:
protocol and reject ``reasoning_content`` echoes. We only enable the
kimi-reasoning replay when the request actually targets a
kimi/moonshot endpoint or the dedicated kimi-coding provider.
Rule table owner: ``agent.message_sanitization.reasoning_echo_family``.
"""
return (
self.provider in {"kimi-coding", "kimi-coding-cn"}
or base_url_host_matches(self.base_url, "api.kimi.com")
or base_url_host_matches(self.base_url, "moonshot.ai")
or base_url_host_matches(self.base_url, "moonshot.cn")
from agent.message_sanitization import matches_reasoning_echo_family
return matches_reasoning_echo_family(
"kimi", self.provider, None, self.base_url
)
def _needs_deepseek_tool_reasoning(self) -> bool:
@ -6700,13 +6639,12 @@ class AIAgent:
DeepSeek V4 thinking mode requires ``reasoning_content`` on every
assistant tool-call turn; omitting it causes HTTP 400 when the
message is replayed in a subsequent API request (#15250).
Rule table owner: ``agent.message_sanitization.reasoning_echo_family``.
"""
provider = (self.provider or "").lower()
model = (self.model or "").lower()
return (
provider == "deepseek"
or "deepseek" in model
or base_url_host_matches(self.base_url, "api.deepseek.com")
from agent.message_sanitization import matches_reasoning_echo_family
return matches_reasoning_echo_family(
"deepseek", (self.provider or "").lower(), self.model, self.base_url
)
def _needs_mimo_tool_reasoning(self) -> bool:
@ -6715,14 +6653,12 @@ class AIAgent:
MiMo thinking mode requires ``reasoning_content`` on every assistant
tool-call message when replaying history; omitting it causes HTTP 400.
Refs: https://platform.xiaomimimo.com/docs/zh-CN/usage-guide/passing-back-reasoning_content
Rule table owner: ``agent.message_sanitization.reasoning_echo_family``.
"""
provider = (self.provider or "").lower()
model = (self.model or "").lower()
return (
provider == "xiaomi"
or "mimo" in model
or base_url_host_matches(self.base_url, "api.xiaomimimo.com")
or base_url_host_matches(self.base_url, "xiaomimimo.com")
from agent.message_sanitization import matches_reasoning_echo_family
return matches_reasoning_echo_family(
"mimo", (self.provider or "").lower(), self.model, self.base_url
)
def _copy_reasoning_content_for_api(self, source_msg: dict, api_msg: dict) -> None:

296
tests/agent/test_message_sanitization_policy.py Normal file
View file

@ -0,0 +1,296 @@
"""Tests for the single-owner call_id + reasoning_content policies.
Audit F4 consolidation: agent/message_sanitization.py now owns the
deterministic call_id synthesis, call_id coalescing/dedup, and the
reasoning_content strip-vs-repad provider-direction policy. These tests pin
the owner functions' behavior (including byte-exact hash outputs — they feed
prompt-cache keys) and verify the legacy entry points still delegate here.
"""
from types import SimpleNamespace
import pytest
from agent.message_sanitization import (
apply_reasoning_content_policy,
coalesce_tool_call_id,
deterministic_call_id,
matches_reasoning_echo_family,
needs_reasoning_echo,
reapply_reasoning_echo,
reasoning_echo_family,
uniquify_tool_call_ids,
)
# ---------------------------------------------------------------------------
# deterministic_call_id — byte-exact (prompt-cache keys)
# ---------------------------------------------------------------------------
class TestDeterministicCallId:
def test_known_hash_outputs_are_stable(self):
# Golden values: sha256(f"{fn}:{args}:{index}")[:12] prefixed call_.
# Any change here invalidates users' prompt caches — do NOT update
# these expectations without a migration plan.
assert deterministic_call_id("terminal", '{"command":"ls"}', 0) == \
"call_40ccaef54d02"
assert deterministic_call_id("terminal", '{"command":"ls"}', 1) == \
"call_567cb168d22d"
assert deterministic_call_id("", "", 0) == "call_feda901d71ea"
def test_deterministic_across_calls(self):
a = deterministic_call_id("web_search", '{"q":"x"}', 3)
b = deterministic_call_id("web_search", '{"q":"x"}', 3)
assert a == b
assert a.startswith("call_")
assert len(a) == len("call_") + 12
def test_index_disambiguates(self):
assert deterministic_call_id("t", "{}", 0) != deterministic_call_id("t", "{}", 1)
def test_surrogates_do_not_crash(self):
out = deterministic_call_id("t", "bad \ud800 arg", 0)
assert out.startswith("call_")
def test_codex_adapter_wrapper_delegates(self):
from agent.codex_responses_adapter import _deterministic_call_id
assert _deterministic_call_id("terminal", '{"command":"ls"}', 0) == \
deterministic_call_id("terminal", '{"command":"ls"}', 0)
def test_run_agent_static_delegates(self):
from run_agent import AIAgent
assert AIAgent._deterministic_call_id("terminal", '{"command":"ls"}', 0) == \
deterministic_call_id("terminal", '{"command":"ls"}', 0)
# ---------------------------------------------------------------------------
# coalesce_tool_call_id
# ---------------------------------------------------------------------------
class TestCoalesceToolCallId:
def test_dict_call_id_wins_over_id(self):
assert coalesce_tool_call_id({"call_id": "c", "id": "i"}) == "c"
def test_dict_falls_back_to_id_and_strips(self):
assert coalesce_tool_call_id({"id": " i "}) == "i"
assert coalesce_tool_call_id({"call_id": "", "id": "i2"}) == "i2"
def test_dict_empty(self):
assert coalesce_tool_call_id({}) == ""
def test_object_forms(self):
assert coalesce_tool_call_id(SimpleNamespace(call_id="c", id="i")) == "c"
assert coalesce_tool_call_id(SimpleNamespace(call_id=None, id=" i ")) == "i"
assert coalesce_tool_call_id(SimpleNamespace(call_id=None, id=None)) == ""
def test_run_agent_static_delegates(self):
from run_agent import AIAgent
tc = {"call_id": "c9", "id": "i9"}
assert AIAgent._get_tool_call_id_static(tc) == coalesce_tool_call_id(tc)
# ---------------------------------------------------------------------------
# uniquify_tool_call_ids
# ---------------------------------------------------------------------------
class TestUniquifyToolCallIds:
def test_no_duplicates_untouched(self):
tcs = [
{"id": "a", "function": {"name": "f", "arguments": "{}"}},
{"id": "b", "function": {"name": "g", "arguments": "{}"}},
]
out = uniquify_tool_call_ids(tcs)
assert out is tcs
assert [tc["id"] for tc in out] == ["a", "b"]
def test_duplicate_gets_deterministic_suffix(self):
tcs = [
{"id": "x", "call_id": "x", "function": {"name": "f", "arguments": "{}"}},
{"id": "x", "call_id": "x", "function": {"name": "g", "arguments": "{}"}},
{"id": "x", "function": {"name": "h", "arguments": "{}"}},
]
uniquify_tool_call_ids(tcs)
assert tcs[0]["id"] == "x"
assert tcs[1]["id"] == "x_d2"
assert tcs[1]["call_id"] == "x_d2"
assert tcs[2]["id"] == "x_d3"
def test_composite_id_collides_on_call_half_and_preserves_item_half(self):
tcs = [
{"id": "call_y|fc_1", "function": {"name": "f", "arguments": "{}"}},
{"id": "call_y|fc_2", "function": {"name": "g", "arguments": "{}"}},
]
uniquify_tool_call_ids(tcs)
assert tcs[0]["id"] == "call_y|fc_1"
assert tcs[1]["id"] == "call_y_d2|fc_2"
def test_suffix_collision_advances_counter(self):
tcs = [
{"id": "z", "function": {"name": "a", "arguments": "{}"}},
{"id": "z_d2", "function": {"name": "b", "arguments": "{}"}},
{"id": "z", "function": {"name": "c", "arguments": "{}"}},
]
uniquify_tool_call_ids(tcs)
assert tcs[2]["id"] == "z_d3"
def test_blank_and_non_string_ids_skipped(self):
tcs = [
{"id": "", "function": {"name": "a", "arguments": "{}"}},
{"id": None, "function": {"name": "b", "arguments": "{}"}},
SimpleNamespace(id=42, call_id=None, function=None),
]
uniquify_tool_call_ids(tcs)
assert tcs[0]["id"] == ""
assert tcs[1]["id"] is None
def test_namespace_objects_mutated(self):
tcs = [
SimpleNamespace(id="n", call_id="n",
function=SimpleNamespace(name="a", arguments="{}")),
SimpleNamespace(id="n", call_id="n",
function=SimpleNamespace(name="b", arguments="{}")),
]
uniquify_tool_call_ids(tcs)
assert tcs[1].id == "n_d2"
assert tcs[1].call_id == "n_d2"
def test_empty_and_none_inputs(self):
assert uniquify_tool_call_ids([]) == []
assert uniquify_tool_call_ids(None) is None
# ---------------------------------------------------------------------------
# reasoning_echo_family — the provider-direction table
# ---------------------------------------------------------------------------
class TestReasoningEchoFamily:
@pytest.mark.parametrize("provider,model,base_url,family", [
("kimi-coding", None, "https://x", "kimi"),
("kimi-coding-cn", None, "https://x", "kimi"),
("custom", None, "https://api.kimi.com/v1", "kimi"),
("custom", None, "https://api.moonshot.ai/v1", "kimi"),
("custom", None, "https://api.moonshot.cn/v1", "kimi"),
("deepseek", "whatever", "https://x", "deepseek"),
("DeepSeek", "whatever", "https://x", "deepseek"),
("openrouter", "deepseek/deepseek-v3", "https://openrouter.ai", "deepseek"),
("custom", None, "https://api.deepseek.com", "deepseek"),
("xiaomi", None, "https://x", "mimo"),
("custom", "MiMo-7B", "https://x", "mimo"),
("custom", None, "https://api.xiaomimimo.com/v1", "mimo"),
("openai", "gpt-5", "https://api.openai.com/v1", None),
("mistral", "mistral-large", "https://api.mistral.ai/v1", None),
(None, None, None, None),
])
def test_table(self, provider, model, base_url, family):
assert reasoning_echo_family(provider, model, base_url) == family
assert needs_reasoning_echo(provider, model, base_url) is (family is not None)
def test_kimi_provider_match_is_exact_not_lowered(self):
# Original predicate compared the raw provider string against the
# kimi-coding set; keep that semantic.
assert matches_reasoning_echo_family("kimi", "KIMI-CODING", None, "https://x") is False
def test_membership_is_per_family(self):
# A deepseek model pointed at a kimi host matches both families
# independently (the per-family predicates on AIAgent rely on this).
assert matches_reasoning_echo_family(
"kimi", "custom", "deepseek-chat", "https://api.kimi.com") is True
assert matches_reasoning_echo_family(
"deepseek", "custom", "deepseek-chat", "https://api.kimi.com") is True
def test_unknown_family_raises(self):
with pytest.raises(KeyError):
matches_reasoning_echo_family("nope", "p", "m", "https://x")
# ---------------------------------------------------------------------------
# apply_reasoning_content_policy
# ---------------------------------------------------------------------------
class TestApplyReasoningContentPolicy:
def test_non_assistant_untouched(self):
api = {"role": "user", "content": "u", "reasoning_content": "keep"}
apply_reasoning_content_policy(
{"role": "user", "content": "u", "reasoning_content": "keep"}, api, True)
assert api["reasoning_content"] == "keep"
def test_require_side_preserves_existing(self):
api = {"role": "assistant", "content": "x"}
apply_reasoning_content_policy(
{"role": "assistant", "content": "x", "reasoning_content": "thoughts"},
api, True)
assert api["reasoning_content"] == "thoughts"
def test_require_side_upgrades_empty_string_to_space(self):
api = {"role": "assistant", "content": "x", "reasoning_content": ""}
apply_reasoning_content_policy(
{"role": "assistant", "content": "x", "reasoning_content": ""}, api, True)
assert api["reasoning_content"] == " "
def test_strict_side_strips_existing(self):
api = {"role": "assistant", "content": "x", "reasoning_content": " "}
apply_reasoning_content_policy(
{"role": "assistant", "content": "x", "reasoning_content": " "}, api, False)
assert "reasoning_content" not in api
def test_cross_provider_poisoned_history_pads_with_space(self):
src = {"role": "assistant", "content": "x", "reasoning": "other-provider CoT",
"tool_calls": [{"id": "c", "function": {"name": "t", "arguments": "{}"}}]}
api = {"role": "assistant", "content": "x"}
apply_reasoning_content_policy(src, api, True)
assert api["reasoning_content"] == " " # pad, never the foreign CoT
def test_reasoning_promoted_only_on_require_side(self):
src = {"role": "assistant", "content": "x", "reasoning": "healthy"}
api = {"role": "assistant", "content": "x"}
apply_reasoning_content_policy(src, api, True)
assert api["reasoning_content"] == "healthy"
api2 = {"role": "assistant", "content": "x", "reasoning_content": "stale"}
apply_reasoning_content_policy(src, api2, False)
assert "reasoning_content" not in api2
def test_require_side_pads_bare_assistant_turn(self):
api = {"role": "assistant", "content": "x"}
apply_reasoning_content_policy({"role": "assistant", "content": "x"}, api, True)
assert api["reasoning_content"] == " "
def test_non_string_reasoning_content_removed(self):
api = {"role": "assistant", "content": "x", "reasoning_content": None}
apply_reasoning_content_policy(
{"role": "assistant", "content": "x", "reasoning_content": None}, api, False)
assert "reasoning_content" not in api
# ---------------------------------------------------------------------------
# reapply_reasoning_echo
# ---------------------------------------------------------------------------
class TestReapplyReasoningEcho:
MSGS = [
{"role": "assistant", "content": "a1", "reasoning_content": " "},
{"role": "assistant", "content": "a2"},
{"role": "user", "content": "u"},
{"role": "tool", "content": "t", "tool_call_id": "c"},
]
def test_require_side_pads_missing_only(self):
import copy
msgs = copy.deepcopy(self.MSGS)
assert reapply_reasoning_echo(msgs, True) == 1
assert msgs[0]["reasoning_content"] == " " # untouched
assert msgs[1]["reasoning_content"] == " " # padded
assert "reasoning_content" not in msgs[2]
def test_strict_side_strips_all(self):
import copy
msgs = copy.deepcopy(self.MSGS)
assert reapply_reasoning_echo(msgs, False) == 1
assert all("reasoning_content" not in m for m in msgs)
def test_idempotent(self):
import copy
msgs = copy.deepcopy(self.MSGS)
reapply_reasoning_echo(msgs, True)
assert reapply_reasoning_echo(msgs, True) == 0
reapply_reasoning_echo(msgs, False)
assert reapply_reasoning_echo(msgs, False) == 0