Anthropic thinking block signature management in multi-turn

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #35859 fix(anthropic): demote dead thinking signature when orphan-strip mutates the latest turn

  merged to main

  files: agent/anthropic_adapter.py, scripts/release.py, tests/agent/test_anthropic_adapter.py

• #43943 fix(anthropic): preserve interleaved thinking/tool_use block order on replay (#17861)

  merged to main

  files: agent/anthropic_adapter.py, agent/chat_completion_helpers.py, agent/transports/anthropic.py, agent/transports/types.py, scripts/release.py, tests/agent/test_anthropic_output_field_leak.py, tests/agent/test_anthropic_thinking_block_order.py

  Show the diff

  • agent/anthropic_adapter.py

    @@ -1571,6 +1571,15 @@ def _convert_content_part_to_anthropic(part: Any) -> Optional[Dict[str, Any]]:
     
         if ptype == "input_text":
             block: Dict[str, Any] = {"type": "text", "text": part.get("text", "")}
    +    elif ptype == "text":
    +        # A stored Anthropic text block. Rebuild from whitelisted fields only —
    +        # SDK response text blocks carry output-only siblings (parsed_output,
    +        # citations=None) that the Messages INPUT schema rejects with HTTP 400
    +        # "Extra inputs are not permitted". Do NOT dict(part) it verbatim.
    +        block = {"type": "text", "text": part.get("text", "")}
    +        cits = part.get("citations")
    +        if isinstance(cits, list) and cits:
    +            block["citations"] = cits
         elif ptype in {"image_url", "input_image"}:
             image_value = part.get("image_url", {})
             url = image_value.get("url", "") if isinstance(image_value, dict) else str(image_value or "")
    @@ -1685,6 +1694,58 @@ def _content_parts_to_anthropic_blocks(parts: Any) -> List[Dict[str, Any]]:
         return out
     
     
    +def _sanitize_replay_block(b: Dict[str, Any]) -> Optional[Dict[str, Any]]:
    +    """Strip output-only fields from a stored Anthropic content block so it is
    +    valid as REQUEST input on replay.
    +
    +    The SDK response objects carry output-only attributes that the Messages
    +    *input* schema forbids ("Extra inputs are not permitted"): text blocks get
    +    ``parsed_output``/``citations`` (when null), tool_use blocks get ``caller``,
    +    etc. ``normalize_response`` captured blocks verbatim via ``_to_plain_data``,
    +    so these leak back as input on the next turn → HTTP 400.
    +
    +    Whitelist per type (NOT a blacklist) so future SDK output-only fields can't
    +    reintroduce the bug. Returns a clean block, or None to drop it.
    +    """
    +    if not isinstance(b, dict):
    +        return None
    +    btype = b.get("type")
    +    if btype == "text":
    +        out: Dict[str, Any] = {"type": "text", "text": b.get("text", "")}
    +        # citations is input-valid ONLY when it's a non-empty list; the SDK
    +        # emits citations=None on responses, which the input schema rejects.
    +        cits = b.get("citations")
    +        if isinstance(cits, list) and cits:
    +            out["citations"] = cits
    +        if isinstance(b.get("cache_control"), dict):
    +            out["cache_control"] = b["cache_control"]
    +        return out
    +    if btype == "thinking":
    +        out = {"type": "thinking", "thinking": b.get("thinking", "")}
    +        if b.get("signature"):
    +            out["signature"] = b["signature"]
    +        return out
    +    if btype == "redacted_thinking":
    +        # Only valid with its data payload; drop if missing.
    +        return {"type": "redacted_thinking", "data": b["data"]} if b.get("data") else None
    +    if btype == "tool_use":
    +        out = {
    +            "type": "tool_use",
    +            "id": _sanitize_tool_id(b.get("id", "")),
    +            "name": b.get("name", ""),
    +            "input": b.get("input", {}),
    +        }
    +        if isinstance(b.get("cache_control"), dict):
    +            out["cache_control"] = b["cache_control"]
    +        return out
    +    if btype == "image":
    +        src = b.get("source")
    +        return {"type": "image", "source": src} if isinstance(src, dict) else None
    +    # Unknown/unsupported block type on the input path — drop rather than risk
    +    # another "Extra inputs are not permitted".
    +    return None
    +
    +
     def _convert_assistant_message(m: Dict[str, Any]) -> Dict[str, Any]:
         """Convert an assistant message to Anthropic content blocks.
     
    @@ -1692,6 +1753,55 @@ def _convert_assistant_message(m: Dict[str, Any]) -> Dict[str, Any]:
         reasoning_content injection for Kimi/DeepSeek endpoints.
         """
         content = m.get("content", "")
    +    # Anthropic interleaved-thinking fast path: when this turn carries a
    +    # verbatim, order-preserving block list (set by normalize_response only
    +    # for turns that interleave SIGNED thinking with tool_use), replay it.
    +    # Each block is run through _sanitize_replay_block to strip output-only
    +    # SDK fields (parsed_output, caller, citations=None, …) that the Messages
    +    # INPUT schema forbids — replaying them verbatim caused HTTP 400 "Extra
    +    # inputs are not permitted" (text.parsed_output). Block ORDER is preserved
    +    # (the reason this channel exists); only forbidden sibling fields are
    +    # dropped, leaving thinking signatures and tool_use id/name/input intact.
    +    ordered_blocks = m.get("anthropic_content_blocks")
    +    if isinstance(ordered_blocks, list) and ordered_blocks:
    +        # Re-source each tool_use input from the stored tool_calls map rather
    +        # than the captured block. The ordered-blocks list captures tool_use
    +        # input from the RAW API response (normalize_response), which is NOT
    +        # credential-redacted; tool_calls[].function.arguments IS redacted at
    +        # storage time (build_assistant_message, #19798). Replaying the raw
    +        # block input would resurrect a secret the model inlined into a tool
    +        # call (e.g. terminal(command="curl -H 'Authorization: Bearer sk-...'")
    +        # onto the wire, even though the same value is redacted everywhere else
    +        # in history. Keying by sanitized tool id preserves interleave order
    +        # (the reason this channel exists) while swapping in the redacted
    +        # input. Adapted from #36071 (replay-time tool-input re-sourcing).
    +        redacted_input_by_id: Dict[str, Any] = {}
    +        for tc in m.get("tool_calls", []) or []:
    +            if not isinstance(tc, dict):
    +                continue
    +            fn = tc.get("function", {}) or {}
    +            raw_args = fn.get("arguments", "{}")
    +            try:
    +                parsed_args = json.loads(raw_args) if isinstance(raw_args, str) else raw_args
    +            except (json.JSONDecodeError, ValueError):
    +                parsed_args = {}
    +            redacted_input_by_id[_sanitize_tool_id(tc.get("id", ""))] = parsed_args
    +        replayed: List[Dict[str, Any]] = []
    +        for b in ordered_blocks:
    +            clean = _sanitize_replay_block(b)
    +            if clean is None:
    +                continue
    +            if clean.get("type") == "tool_use":
    +                # Override raw (un-redacted) input with the redacted copy when
    +                # we have one for this id; fall back to the sanitized block
    +                # input only if the tool_call is missing (shape mismatch).
    +                redacted = redacted_input_by_id.get(clean.get("id", ""))
    +                if redacted is not None:
    +                    clean["input"] = redacted
    +            replayed.append(clean)
    +        if replayed:
    +            return {"role": "assistant", "content": replayed}
    +
         blocks = _extract_preserved_thinking_blocks(m)
         if content:
             if isinstance(content, list):

  • agent/chat_completion_helpers.py

    @@ -952,6 +952,18 @@ def build_assistant_message(agent, assistant_message, finish_reason: str) -> dic
             if preserved:
                 msg["reasoning_details"] = preserved
     
    +    # Anthropic interleaved-thinking replay: when a turn interleaves signed
    +    # thinking blocks with tool_use, the parallel reasoning_details +
    +    # tool_calls fields lose the cross-type ordering, and reconstruction
    +    # front-loads thinking — reordering signed blocks and triggering HTTP 400
    +    # ("thinking ... blocks in the latest assistant message cannot be
    +    # modified"). Carry the verbatim ordered block list so the adapter can
    +    # replay the latest assistant message unchanged. See
    +    # agent/transports/anthropic.py and agent/anthropic_adapter.py.
    +    ordered_blocks = getattr(assistant_message, "anthropic_content_blocks", None)
    +    if ordered_blocks:
    +        msg["anthropic_content_blocks"] = ordered_blocks
    +
         # Codex Responses API: preserve encrypted reasoning items for
         # multi-turn continuity. These get replayed as input on the next turn.
         codex_items = getattr(assistant_message, "codex_reasoning_items", None)

  • agent/transports/anthropic.py

    @@ -84,7 +84,7 @@ def normalize_response(self, response: Any, **kwargs) -> NormalizedResponse:
             to OpenAI finish_reason, and collects reasoning_details in provider_data.
             """
             import json
    -        from agent.anthropic_adapter import _to_plain_data
    +        from agent.anthropic_adapter import _to_plain_data, _sanitize_replay_block
             from agent.transports.types import ToolCall
     
             strip_tool_prefix = kwargs.get("strip_tool_prefix", False)
    @@ -94,14 +94,40 @@ def normalize_response(self, response: Any, **kwargs) -> NormalizedResponse:
             reasoning_parts = []
             reasoning_details = []
             tool_calls = []
    +        # Verbatim, order-preserving copy of every content block in the turn.
    +        # Anthropic signs each thinking block against the turn content that
    +        # PRECEDES it at its position; when a turn interleaves thinking and
    +        # tool_use (adaptive/interleaved thinking, Claude 4.6+), the parallel
    +        # reasoning_details + tool_calls lists below lose that cross-type
    +        # ordering. Replaying the latest assistant message in the wrong order
    +        # invalidates the signatures -> HTTP 400 "thinking ... blocks in the
    +        # latest assistant message cannot be modified". Preserve the exact
    +        # block sequence here so the adapter can replay it unchanged. See
    +        # tests/agent/test_anthropic_thinking_block_order.py.
    +        ordered_blocks = []
     
             for block in response.content:
    +            block_dict = _to_plain_data(block)
    +            clean_block = None
    +            if isinstance(block_dict, dict):
    +                # Sanitize at capture so output-only SDK fields (parsed_output,
    +                # caller, citations=None, …) never persist to state.db and leak
    +                # back as request input on replay → HTTP 400 "Extra inputs are
    +                # not permitted". Defence-in-depth with the replay-side sanitize.
    +                clean_block = _sanitize_replay_block(block_dict)
    +                if clean_block is not None:
    +                    ordered_blocks.append(clean_block)
                 if block.type == "text":
                     text_parts.append(block.text)
    -            elif block.type == "thinking":
    -                reasoning_parts.append(block.thinking)
    -                block_dict = _to_plain_data(block)
    -                if isinstance(block_dict, dict):
    +            elif block.type in ("thinking", "redacted_thinking"):
    +                if block.type == "thinking":
    +                    reasoning_parts.append(block.thinking)
    +                # Use the sanitized block (clean_block) for reasoning_details too,
    +                # since _extract_preserved_thinking_blocks replays these on the
    +                # non-ordered path. Falls back to raw only if sanitize dropped it.
    +                if isinstance(clean_block, dict):
    +                    reasoning_details.append(clean_block)
    +                elif isinstance(block_dict, dict):
                         reasoning_details.append(block_dict)
                 elif block.type == "tool_use":
                     name = block.name
    @@ -130,6 +156,23 @@ def normalize_response(self, response: Any, **kwargs) -> NormalizedResponse:
             provider_data = {}
             if reasoning_details:
                 provider_data["reasoning_details"] = reasoning_details
    +        # Only worth carrying the ordered-blocks channel when the turn
    +        # actually interleaves signed thinking with tool_use — that's the
    +        # only shape the parallel lists reconstruct incorrectly. A turn that
    +        # is purely text, or thinking-then-tools with a single leading
    +        # thinking block, replays correctly without it.
    +        _has_signed_thinking = any(
    +            isinstance(b, dict)
    +            and b.get("type") in ("thinking", "redacted_thinking")
    +            and (b.get("signature") or b.get("data"))
    +            for b in ordered_blocks
    +        )
    +        _has_tool_use = any(
    +            isinstance(b, dict) and b.get("type") == "tool_use"
    +            for b in ordered_blocks
    +        )
    +        if _has_signed_thinking and _has_tool_use:
    +            provider_data["anthropic_content_blocks"] = ordered_blocks
     
             return NormalizedResponse(
                 content="\n".join(text_parts) if text_parts else None,

  • agent/transports/types.py

    @@ -121,6 +121,18 @@ def reasoning_details(self):
             pd = self.provider_data or {}
             return pd.get("reasoning_details")
     
    +    @property
    +    def anthropic_content_blocks(self):
    +        """Verbatim, order-preserving Anthropic content blocks for a turn.
    +
    +        Present only when an Anthropic turn interleaves signed thinking with
    +        tool_use — the one shape the parallel reasoning_details + tool_calls
    +        lists reconstruct in the wrong order, invalidating thinking-block
    +        signatures on replay. See agent/transports/anthropic.py.
    +        """
    +        pd = self.provider_data or {}
    +        return pd.get("anthropic_content_blocks")
    +
         @property
         def codex_reasoning_items(self):
             pd = self.provider_data or {}

  • scripts/release.py

    @@ -944,6 +944,7 @@
         "[redacted:email]": "malaiwah",
         "[redacted:email]": "gnanam1990",
         "[redacted:email]": "0xyg3n",
    +    "[redacted:email]": "Spaceman-Spiffy",  # PR #35586 (renamed account)
         "[redacted:email]": "ianculling",  # PR #36087
         "[redacted:email]": "0xyg3n",
         "[redacted:email]": "nftpoetrist",  # PR #18982

  • tests/agent/test_anthropic_output_field_leak.py

    @@ -0,0 +1,96 @@
    +"""Regression: output-only SDK fields must not leak into Anthropic request input.
    +
    +Reproduces HTTP 400 `messages.N.content.M.text.parsed_output: Extra inputs are
    +not permitted`. Anthropic SDK response blocks carry output-only attributes
    +(text blocks: `parsed_output`, `citations=None`; tool_use blocks: `caller`)
    +that the Messages *input* schema forbids. normalize_response captured blocks
    +verbatim via _to_plain_data and replayed them as input → 400.
    +
    +Fix: whitelist input-permitted fields per block type at three points —
    +normalize_response capture, _sanitize_replay_block (ordered-blocks replay), and
    +_convert_content_part_to_anthropic (content-list replay).
    +"""
    +import sys, os
    +sys.path.insert(0, os.path.expanduser("~/.hermes/hermes-agent"))
    +
    +import pytest
    +from agent.anthropic_adapter import (
    +    _sanitize_replay_block,
    +    _convert_content_part_to_anthropic,
    +    _convert_assistant_message,
    +)
    +
    +FORBIDDEN = {"parsed_output", "caller"}
    +
    +
    +def _assert_clean(block):
    +    """No forbidden output-only key, and no null citations, anywhere."""
    +    assert isinstance(block, dict)
    +    for k in FORBIDDEN:
    +        assert k not in block, f"forbidden field {k!r} survived: {block}"
    +    if "citations" in block:
    +        assert isinstance(block["citations"], list) and block["citations"], \
    +            "citations must be a non-empty list if present (None/[] is input-invalid)"
    +
    +
    +class TestSanitizeReplayBlock:
    +    def test_text_block_strips_parsed_output_and_null_citations(self):
    +        poisoned = {"type": "text", "text": "hi", "parsed_output": None, "citations": None}
    +        out = _sanitize_replay_block(poisoned)
    +        _assert_clean(out)
    +        assert out == {"type": "text", "text": "hi"}
    +
    +    def test_tool_use_strips_caller(self):
    +        poisoned = {"type": "tool_use", "id": "toolu_1", "name": "read_file",
    +                    "input": {"path": "a"}, "caller": {"type": "agent"}}
    +        out = _sanitize_replay_block(poisoned)
    +        _assert_clean(out)
    +        assert out["name"] == "read_file" and out["input"] == {"path": "a"}
    +
    +    def test_thinking_preserves_signature(self):
    +        b = {"type": "thinking", "thinking": "x", "signature": "sig-AAA"}
    +        out = _sanitize_replay_block(b)
    +        assert out == {"type": "thinking", "thinking": "x", "signature": "sig-AAA"}
    +
    +    def test_text_keeps_real_citations(self):
    +        real = [{"type": "char_location", "cited_text": "q"}]
    +        out = _sanitize_replay_block({"type": "text", "text": "t", "citations": real})
    +        assert out["citations"] == real
    +
    +    def test_unknown_type_dropped(self):
    +        assert _sanitize_replay_block({"type": "server_tool_use", "foo": 1}) is None
    +
    +
    +class TestContentPartConversion:
    +    def test_stored_text_block_with_parsed_output_cleaned(self):
    +        # The exact content.N.text.parsed_output failure shape.
    +        part = {"type": "text", "text": "hello", "parsed_output": None, "citations": None}
    +        out = _convert_content_part_to_anthropic(part)
    +        _assert_clean(out)
    +
    +
    +class TestAssistantReplay:
    +    def test_interleaved_blocks_replayed_clean_and_ordered(self):
    +        m = {
    +            "role": "assistant",
    +            "anthropic_content_blocks": [
    +                {"type": "thinking", "thinking": "plan", "signature": "s1"},
    +                {"type": "text", "text": "doing it", "parsed_output": None, "citations": None},
    +                {"type": "tool_use", "id": "toolu_1", "name": "read_file",
    +                 "input": {"path": "a"}, "caller": {"type": "agent"}},
    +            ],
    +        }
    +        out = _convert_assistant_message(m)
    +        blocks = out["content"]
    +        # order preserved
    +        assert [b["type"] for b in blocks] == ["thinking", "text", "tool_use"]
    +        # every block clean
    +        for b in blocks:
    +            _assert_clean(b)
    +        # signature + tool fields intact
    +        assert blocks[0]["signature"] == "s1"
    +        assert blocks[2]["name"] == "read_file"
    +
    +
    +if __name__ == "__main__":
    +    raise SystemExit(pytest.main([__file__, "-v"]))

  • tests/agent/test_anthropic_thinking_block_order.py

    @@ -0,0 +1,314 @@
    +"""Regression test for the Anthropic interleaved thinking-block 400.
    +
    +Reproduces: HTTP 400 ``messages.N.content.M: thinking or redacted_thinking
    +blocks in the latest assistant message cannot be modified. These blocks must
    +remain as they were in the original response.``
    +
    +Root cause under test
    +----------------------
    +With adaptive / interleaved thinking (Claude 4.6+, e.g. Opus 4.8), a single
    +assistant turn can emit content blocks in an interleaved order::
    +
    +    thinking_1 (signed) · tool_use_1 · thinking_2 (signed) · tool_use_2
    +
    +Anthropic signs each thinking block against the turn content that precedes it
    +at its position.  ``thinking_2`` is signed with ``tool_use_1`` before it.
    +
    +``AnthropicTransport.normalize_response`` (agent/transports/anthropic.py)
    +splits the turn into two *parallel* lists — ``reasoning_details`` (thinking
    +blocks) and ``tool_calls`` (tool_use blocks) — discarding the cross-type
    +ordering.  ``run_agent`` stores those as separate fields on the assistant
    +message.  On replay, ``_convert_assistant_message`` (agent/anthropic_adapter.py)
    +rebuilds the content as ``[all thinking][text][all tool_use]``, which reorders
    +``thinking_2`` ahead of ``tool_use_1``.  The signature no longer matches its
    +original position, so Anthropic rejects the latest assistant message with the
    +400 above.
    +
    +This test asserts that an interleaved turn round-trips through
    +normalize_response -> stored message -> convert_messages_to_anthropic with its
    +block order preserved.  It FAILS on the current code (documenting the bug) and
    +should PASS once block ordering is preserved on replay.
    +"""
    +
    +import json
    +from types import SimpleNamespace
    +
    +import pytest
    +
    +from agent.transports import get_transport
    +from agent.anthropic_adapter import convert_messages_to_anthropic
    +
    +
    +def _thinking_block(text: str, signature: str) -> SimpleNamespace:
    +    """A signed Anthropic thinking block, shaped like the SDK object."""
    +    return SimpleNamespace(type="thinking", thinking=text, signature=signature)
    +
    +
    +def _tool_use_block(block_id: str, name: str, payload: dict) -> SimpleNamespace:
    +    return SimpleNamespace(type="tool_use", id=block_id, name=name, input=payload)
    +
    +
    +def _interleaved_response() -> SimpleNamespace:
    +    """An assistant turn with thinking interleaved between two tool_use blocks."""
    +    return SimpleNamespace(
    +        content=[
    +            _thinking_block("Plan: inspect file A first.", "sig-AAA"),
    +            _tool_use_block("toolu_1", "read_file", {"path": "a.py"}),
    +            _thinking_block("A looked fine; now inspect B.", "sig-BBB"),
    +            _tool_use_block("toolu_2", "read_file", {"path": "b.py"}),
    +        ],
    +        stop_reason="tool_use",
    +        usage=None,
    +    )
    +
    +
    +def _stored_assistant_message(normalized) -> dict:
    +    """Reconstruct the OpenAI-style assistant message the way run_agent stores it.
    +
    +    run_agent.py persists assistant turns as separate fields: content,
    +    reasoning_details (from provider_data), and tool_calls.  See
    +    run_agent.py L1513-1516 and hermes_state.py.
    +    """
    +    provider_data = normalized.provider_data or {}
    +    tool_calls = []
    +    for tc in (normalized.tool_calls or []):
    +        tool_calls.append({
    +            "id": tc.id,
    +            "type": "function",
    +            "function": {"name": tc.name, "arguments": tc.arguments},
    +        })
    +    msg = {
    +        "role": "assistant",
    +        "content": normalized.content or "",
    +        "reasoning_details": provider_data.get("reasoning_details"),
    +        "tool_calls": tool_calls,
    +    }
    +    # build_assistant_message lifts the verbatim ordered-block channel onto
    +    # the stored message; mirror that here.
    +    blocks = provider_data.get("anthropic_content_blocks")
    +    if blocks:
    +        msg["anthropic_content_blocks"] = blocks
    +    return msg
    +
    +
    +def _original_block_order(response) -> list:
    +    """The (type, key) sequence of the original interleaved response."""
    +    order = []
    +    for b in response.content:
    +        if b.type == "thinking":
    +            order.append(("thinking", b.signature))
    +        elif b.type == "tool_use":
    +            order.append(("tool_use", b.id))
    +    return order
    +
    +
    +def _replayed_block_order(assistant_content) -> list:
    +    order = []
    +    for b in assistant_content:
    +        if not isinstance(b, dict):
    +            continue
    +        if b.get("type") in ("thinking", "redacted_thinking"):
    +            order.append(("thinking", b.get("signature")))
    +        elif b.get("type") == "tool_use":
    +            order.append(("tool_use", b.get("id")))
    +    return order
    +
    +
    +class TestInterleavedThinkingBlockOrder:
    +    def test_normalize_response_loses_interleaving(self):
    +        """Confirm the lossy split: normalize_response stores thinking and
    +        tool_use in independent fields with no positional linkage."""
    +        transport = get_transport("anthropic_messages")
    +        normalized = transport.normalize_response(_interleaved_response())
    +
    +        # Both thinking blocks are captured...
    +        details = (normalized.provider_data or {}).get("reasoning_details")
    +        assert details is not None and len(details) == 2
    +        # ...and both tool calls...
    +        assert normalized.tool_calls is not None and len(normalized.tool_calls) == 2
    +        # ...but they live in separate fields. There is no single ordered
    +        # structure recording that thinking_2 sat between the two tool calls.
    +        # (This is the structural precondition for the reorder bug.)
    +
    +    def test_interleaved_order_preserved_on_replay(self):
    +        """The latest assistant message must replay blocks in their ORIGINAL
    +        order, or Anthropic rejects the signed thinking blocks with a 400.
    +
    +        FAILS on current code: _convert_assistant_message front-loads all
    +        thinking blocks, producing
    +            thinking_1 · thinking_2 · tool_use_1 · tool_use_2
    +        instead of the original
    +            thinking_1 · tool_use_1 · thinking_2 · tool_use_2
    +        """
    +        response = _interleaved_response()
    +        original_order = _original_block_order(response)
    +
    +        transport = get_transport("anthropic_messages")
    +        normalized = transport.normalize_response(response)
    +        assistant_msg = _stored_assistant_message(normalized)
    +
    +        # Build a minimal conversation where this assistant turn is the LATEST
    +        # assistant message (the one whose signed blocks are sent verbatim).
    +        messages = [
    +            {"role": "user", "content": "Inspect a.py and b.py."},
    +            assistant_msg,
    +            {"role": "tool", "tool_call_id": "toolu_1", "content": "a.py: ok"},
    +            {"role": "tool", "tool_call_id": "toolu_2", "content": "b.py: ok"},
    +        ]
    +
    +        _system, anthropic_messages = convert_messages_to_anthropic(
    +            messages,
    +            base_url=None,             # direct Anthropic
    +            model="claude-opus-4-8",   # adaptive thinking family
    +        )
    +
    +        # Find the (latest) assistant message in the converted output.
    +        assistant_out = [m for m in anthropic_messages if m.get("role") == "assistant"]
    +        assert assistant_out, "no assistant message in converted output"
    +        replayed_order = _replayed_block_order(assistant_out[-1]["content"])
    +
    +        assert replayed_order == original_order, (
    +            "Interleaved thinking/tool_use order was not preserved on replay.\n"
    +            f"  original: {original_order}\n"
    +            f"  replayed: {replayed_order}\n"
    +            "Anthropic signs thinking blocks against their original position; "
    +            "reordering invalidates the signature -> HTTP 400 'thinking blocks "
    +            "in the latest assistant message cannot be modified'."
    +        )
    +
    +    def test_replay_falls_back_gracefully_without_ordered_blocks(self):
    +        """Without the ordered-block channel, conversion must not crash.
    +
    +        The channel is intentionally NOT persisted to state.db (in-memory
    +        only): a session reloaded from disk after a crash loses the field
    +        and falls back to reconstruction. That replay may take one HTTP 400,
    +        which the thinking-signature recovery (#43667) absorbs by stripping
    +        reasoning_details and retrying. This test pins the fallback shape:
    +        conversion still produces a valid assistant message from the
    +        parallel reasoning_details + tool_calls fields.
    +        """
    +        response = _interleaved_response()
    +        transport = get_transport("anthropic_messages")
    +        normalized = transport.normalize_response(response)
    +        assistant_msg = _stored_assistant_message(normalized)
    +        # Simulate a disk reload: the in-memory-only channel is gone.
    +        assistant_msg.pop("anthropic_content_blocks", None)
    +
    +        messages = [
    +            assistant_msg,
    +            {"role": "tool", "tool_call_id": "toolu_1", "content": "a ok"},
    +            {"role": "tool", "tool_call_id": "toolu_2", "content": "b ok"},
    +        ]
    +        _system, anthropic_messages = convert_messages_to_anthropic(
    +            messages, base_url=None, model="claude-opus-4-8",
    +        )
    +        assistant_out = [m for m in anthropic_messages if m.get("role") == "assistant"]
    +        assert assistant_out, "no assistant message in converted output"
    +        content = assistant_out[-1]["content"]
    +        assert isinstance(content, list) and content, "fallback produced empty content"
    +        # Reconstruction keeps both tool_use blocks (answered by results).
    +        tool_ids = [b.get("id") for b in content if isinstance(b, dict) and b.get("type") == "tool_use"]
    +        assert set(tool_ids) == {"toolu_1", "toolu_2"}
    +
    +
    +class TestInterleavedReplayCredentialRedaction:
    +    """The verbatim-replay fast path must not leak un-redacted secrets.
    +
    +    anthropic_content_blocks captures each tool_use ``input`` from the RAW API
    +    response (normalize_response), which is NOT credential-redacted. The
    +    parallel tool_calls[].function.arguments IS redacted at storage time
    +    (build_assistant_message, #19798). If the fast path replays the block's raw
    +    input verbatim, a secret the model inlined into a tool call rides back onto
    +    the wire — even though it is redacted everywhere else in history. The fix
    +    re-sources tool_use input from the redacted tool_calls map by id.
    +    """
    +
    +    def test_tool_use_input_resourced_from_redacted_tool_calls(self):
    +        REDACTED = "[REDACTED_SECRET]"
    +        # Ordered channel: raw input carries the live secret (as captured from
    +        # the unredacted API response).
    +        ordered = [
    +            {"type": "thinking", "thinking": "Call the API.", "signature": "sig-AAA"},
    +            {
    +                "type": "tool_use",
    +                "id": "toolu_1",
    +                "name": "terminal",
    +                "input": {"command": "curl -H 'Authorization: Bearer sk-LIVE-SECRET-123'"},
    +            },
    +            {"type": "thinking", "thinking": "Now the second call.", "signature": "sig-BBB"},
    +            {
    +                "type": "tool_use",
    +                "id": "toolu_2",
    +                "name": "terminal",
    +                "input": {"command": "echo done"},
    +            },
    +        ]
    +        # Stored tool_calls: arguments already redacted (the #19798 path).
    +        assistant_msg = {
    +            "role": "assistant",
    +            "content": "",
    +            "reasoning_details": [b for b in ordered if b["type"] == "thinking"],
    +            "tool_calls": [
    +                {
    +                    "id": "toolu_1",
    +                    "type": "function",
    +                    "function": {
    +                        "name": "terminal",
    +                        "arguments": json.dumps(
    +                            {"command": f"curl -H 'Authorization: Bearer {REDACTED}'"}
    +                        ),
    +                    },
    +                },
    +                {
    +                    "id": "toolu_2",
    +                    "type": "function",
    +                    "function": {
    +                        "name": "terminal",
    +                        "arguments": json.dumps({"command": "echo done"}),
    +                    },
    +                },
    +            ],
    +            "anthropic_content_blocks": ordered,
    +        }
    +        messages = [
    +            {"role": "user", "content": "Hit the API twice."},
    +            assistant_msg,
    +            {"role": "tool", "tool_call_id": "toolu_1", "content": "200 OK"},
    +            {"role": "tool", "tool_call_id": "toolu_2", "content": "done"},
    +        ]
    +
    +        _system, anthropic_messages = convert_messages_to_anthropic(
    +            messages, base_url=None, model="claude-opus-4-8",
    +        )
    +        assistant_out = [m for m in anthropic_messages if m.get("role") == "assistant"]
    +        assert assistant_out, "no assistant message in converted output"
    +        blocks = assistant_out[-1]["content"]
    +
    +        tool_uses = {b["id"]: b for b in blocks if b.get("type") == "tool_use"}
    +        assert set(tool_uses) == {"toolu_1", "toolu_2"}, "tool_use blocks missing/renamed"
    +
    +        # The replayed input must be the REDACTED value, not the live secret.
    +        replayed_cmd = tool_uses["toolu_1"]["input"]["command"]
    +        assert "sk-LIVE-SECRET-123" not in replayed_cmd, (
    +            "Un-redacted secret leaked onto the wire via the verbatim-replay "
    +            "fast path. tool_use input must be re-sourced from the redacted "
    +            "tool_calls map, not the raw captured block."
    +        )
    +        assert REDACTED in replayed_cmd
    +
    +        # Interleave order is still preserved (the reason the channel exists).
    +        order = [
    +            ("thinking", b.get("signature")) if b.get("type") == "thinking"
    +            else ("tool_use", b.get("id"))
    +            for b in blocks if b.get("type") in ("thinking", "tool_use")
    +        ]
    +        assert order == [
    +            ("thinking", "sig-AAA"),
    +            ("tool_use", "toolu_1"),
    +            ("thinking", "sig-BBB"),
    +            ("tool_use", "toolu_2"),
    +        ]
    +
    +
    +if __name__ == "__main__":
    +    raise SystemExit(pytest.main([__file__, "-v"]))