From f8758dcaf89bc5c5f8608011cad86da56f6e1218 Mon Sep 17 00:00:00 2001
From: teknium1 <127238744+teknium1@users.noreply.github.com>
Date: Wed, 29 Jul 2026 12:07:40 -0700
Subject: [PATCH] refactor(agent): single-owner call_id + reasoning_content
 sanitization policies (wire-parity verified)

---
 agent/agent_runtime_helpers.py                | 114 +-----
 agent/codex_responses_adapter.py              |   9 +-
 agent/message_sanitization.py                 | 375 ++++++++++++++++++
 run_agent.py                                  | 122 ++----
 .../agent/test_message_sanitization_policy.py | 296 ++++++++++++++
 5 files changed, 719 insertions(+), 197 deletions(-)
 create mode 100644 tests/agent/test_message_sanitization_policy.py

diff --git a/agent/agent_runtime_helpers.py b/agent/agent_runtime_helpers.py
index 5505d707562..1b592c20698 100644
--- a/agent/agent_runtime_helpers.py
+++ b/agent/agent_runtime_helpers.py
@@ -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):
diff --git a/agent/codex_responses_adapter.py b/agent/codex_responses_adapter.py
index edff776536e..23708eca972 100644
--- a/agent/codex_responses_adapter.py
+++ b/agent/codex_responses_adapter.py
@@ -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:
diff --git a/agent/message_sanitization.py b/agent/message_sanitization.py
index 29a4b8691ae..dc4df3dd270 100644
--- a/agent/message_sanitization.py
+++ b/agent/message_sanitization.py
@@ -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.
diff --git a/run_agent.py b/run_agent.py
index 8a32aa62e4b..5d9af78815c 100644
--- a/run_agent.py
+++ b/run_agent.py
@@ -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:
diff --git a/tests/agent/test_message_sanitization_policy.py b/tests/agent/test_message_sanitization_policy.py
new file mode 100644
index 00000000000..b363b73ff2a
--- /dev/null
+++ b/tests/agent/test_message_sanitization_policy.py
@@ -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