test(e2e): replay a real tool-search assistant turn back to Bedrock Invoke

The tool_search x bedrock_invoke cell only ever probed the first turn, so
nothing in the suite has sent a server_tool_use block back to a provider.
Every turn of a real Claude Code session after the first carries the
server_tool_use and tool_search_tool_result blocks the previous turn
produced, and that path was uncovered.

Adds probe_tool_search_multiturn, which takes the real assistant turn
back, answers any client-side tool_use with the id the model actually
emitted, and replays the whole thing as history with the tools still
declared. The assertion refuses to go green unless both server-tool
blocks made it into the replayed history, so a first turn truncated at
max_tokens reads as a failure instead of a vacuous pass.

The replay assertion's red paths never run in a green cell, so they get
markerless harness tests of their own alongside the existing
_builder_unit_tests tree.

No production code.

tests/e2e/claude_code/_probe_unit_tests/test_http_probe.py

@@ -0,0 +1,106 @@
+"""Unit tests for the tool-search replay assertion in `http_probe`.
+
+Markerless harness tests: they exercise probe plumbing over hand-built
+`Result` values, not a product feature, so they run without a proxy and carry
+no `e2e` marker.
+
+The red paths are what these are for. A live cell only ever executes the green
+one, so a broken diagnostic in the failure branch would sit undetected until
+the day the provider actually rejects the history, which is the day the
+diagnostic has to be right.
+"""
+
+from __future__ import annotations
+
+from e2e_http import Result, Success, UnknownApiError
+from models import (
+    AnthropicContentBlock,
+    AnthropicMessagesResponse,
+    AnthropicToolResultTurn,
+    ChatMessage,
+)
+
+from claude_code.http_probe import (
+    ToolSearchReplay,
+    _replay_history,
+    assert_tool_search_replay_shape,
+)
+
+_REJECTED: Result[AnthropicMessagesResponse] = UnknownApiError(
+    status_code=400,
+    body="server_tool_use blocks are not supported",
+)
+_ACCEPTED: Result[AnthropicMessagesResponse] = Success(
+    status_code=200,
+    data=AnthropicMessagesResponse(content=[AnthropicContentBlock(type="text", text="done")]),
+)
+
+
+def _replay(block_types: tuple[str, ...], second_turn: Result[AnthropicMessagesResponse]) -> ToolSearchReplay:
+    answer = AnthropicMessagesResponse(
+        content=[AnthropicContentBlock(type=block_type, id="srvtoolu_01") for block_type in block_types]
+    )
+    return ToolSearchReplay(
+        first_turn=Success(status_code=200, data=answer),
+        history=_replay_history(answer),
+        second_turn=second_turn,
+    )
+
+
+def test_accepts_a_replayed_server_tool_pair() -> None:
+    replay = _replay(("text", "server_tool_use", "tool_search_tool_result"), _ACCEPTED)
+    assert assert_tool_search_replay_shape(replay) is None
+
+
+def test_reports_the_status_when_the_replayed_history_is_rejected() -> None:
+    replay = _replay(("server_tool_use", "tool_search_tool_result"), _REJECTED)
+    error = assert_tool_search_replay_shape(replay)
+    assert error is not None
+    assert "status 400" in error
+    assert "server_tool_use" in error
+
+
+def test_a_turn_truncated_before_the_result_block_is_not_a_pass() -> None:
+    replay = _replay(("server_tool_use",), _ACCEPTED)
+    error = assert_tool_search_replay_shape(replay)
+    assert error is not None
+    assert "tool_search_tool_result" in error
+
+
+def test_a_history_with_no_server_tool_block_is_not_a_pass() -> None:
+    replay = _replay(("text",), _ACCEPTED)
+    error = assert_tool_search_replay_shape(replay)
+    assert error is not None
+    assert "server_tool_use" in error
+
+
+def test_a_failed_first_turn_is_reported_as_the_first_turn() -> None:
+    replay = ToolSearchReplay(first_turn=_REJECTED, history=(), second_turn=None)
+    error = assert_tool_search_replay_shape(replay)
+    assert error is not None
+    assert error.startswith("first turn: ")
+
+
+def test_a_pending_tool_use_is_answered_with_the_id_the_model_returned() -> None:
+    answer = AnthropicMessagesResponse(
+        content=[
+            AnthropicContentBlock(type="server_tool_use", id="srvtoolu_01"),
+            AnthropicContentBlock(type="tool_search_tool_result", id=None),
+            AnthropicContentBlock(type="tool_use", id="toolu_99"),
+        ]
+    )
+    last_turn = _replay_history(answer)[-1]
+    assert isinstance(last_turn, AnthropicToolResultTurn)
+    assert [block.tool_use_id for block in last_turn.content] == ["toolu_99"]
+
+
+def test_a_turn_with_no_pending_tool_use_gets_a_plain_follow_up() -> None:
+    answer = AnthropicMessagesResponse(
+        content=[
+            AnthropicContentBlock(type="server_tool_use", id="srvtoolu_01"),
+            AnthropicContentBlock(type="tool_search_tool_result"),
+        ]
+    )
+    last_turn = _replay_history(answer)[-1]
+    assert isinstance(last_turn, ChatMessage)
+    assert last_turn.role == "user"

tests/e2e/claude_code/http_probe.py

@@ -28,6 +28,7 @@ the upstream, or LiteLLM 500 on a transformation bug).
 
 from __future__ import annotations
 
+from dataclasses import dataclass
 from typing import TYPE_CHECKING
 
 from pydantic import BaseModel
@@ -42,10 +43,14 @@ from e2e_http import (
     ValidationError,
 )
 from models import (
+    AnthropicAssistantTurn,
     AnthropicCustomTool,
+    AnthropicMessage,
     AnthropicMessagesBody,
     AnthropicMessagesResponse,
     AnthropicTool,
+    AnthropicToolResultBlock,
+    AnthropicToolResultTurn,
     AnthropicToolSearchTool,
     ChatMessage,
     CountTokensBody,
@@ -132,6 +137,7 @@ def probe_tool_search(
     client: ProxyClient,
     api_key: str,
     model: str,
+    max_tokens: int = 64,
     rate_limiter: RateLimiter | None = None,
 ) -> Result[AnthropicMessagesResponse]:
     """POST to `/v1/messages` with a `tool_search_tool_regex_20251119` tool
@@ -155,13 +161,115 @@ def probe_tool_search(
         api_key,
         AnthropicMessagesBody(
             model=model,
-            max_tokens=64,
+            max_tokens=max_tokens,
             messages=[ChatMessage(role="user", content=_TOOL_SEARCH_PROMPT)],
             tools=list(_TOOL_SEARCH_TOOLS),
         ),
     )
 
 
+_TOOL_SEARCH_FOLLOW_UP = "Thanks. Now reply with the word 'done'."
+_TOOL_RESULT_STUB = "3"
+# A `server_tool_use` block and the `tool_search_tool_result` answering it are
+# one indivisible pair: replaying the request without its result is malformed
+# Anthropic and 400s on any provider. 64 output tokens is not enough room for
+# both, so the turn we replay is generated with a budget that fits the whole
+# discovery round trip.
+_REPLAY_SOURCE_MAX_TOKENS = 1024
+_REPLAYED_SERVER_BLOCKS = frozenset({"server_tool_use", "tool_search_tool_result"})
+
+
+@dataclass(frozen=True, slots=True)
+class ToolSearchReplay:
+    """Both turns of the multi-turn probe plus the history the second turn
+    carried, so a failing cell can report which turn broke and what was on the
+    wire when it did."""
+
+    first_turn: Result[AnthropicMessagesResponse]
+    history: tuple[AnthropicMessage, ...]
+    second_turn: Result[AnthropicMessagesResponse] | None
+
+
+def _replayed_server_block_types(history: tuple[AnthropicMessage, ...]) -> frozenset[str]:
+    return frozenset(
+        block.type
+        for turn in history
+        if isinstance(turn, AnthropicAssistantTurn)
+        for block in turn.content
+        if block.type in _REPLAYED_SERVER_BLOCKS
+    )
+
+
+def _replay_history(answer: AnthropicMessagesResponse) -> tuple[AnthropicMessage, ...]:
+    """Turn a real first-turn answer into a well-formed two-turn history.
+
+    Every client-side `tool_use` the model emitted gets a `tool_result` keyed on
+    the id the model actually returned; a turn with none gets a plain follow-up
+    instead. An unanswered `tool_use`, or a `tool_result` pointing at an invented
+    id, is malformed Anthropic and 400s on any provider, which would make this
+    probe measure our own request rather than the provider's handling of the
+    replayed server-tool blocks."""
+    blocks = tuple(answer.content or ())
+    pending = tuple(block.id for block in blocks if block.type == "tool_use" and block.id is not None)
+    reply: AnthropicMessage = (
+        AnthropicToolResultTurn(
+            content=[
+                AnthropicToolResultBlock(tool_use_id=tool_use_id, content=_TOOL_RESULT_STUB)
+                for tool_use_id in pending
+            ]
+        )
+        if pending
+        else ChatMessage(role="user", content=_TOOL_SEARCH_FOLLOW_UP)
+    )
+    return (
+        ChatMessage(role="user", content=_TOOL_SEARCH_PROMPT),
+        AnthropicAssistantTurn(content=list(blocks)),
+        reply,
+    )
+
+
+def probe_tool_search_multiturn(
+    *,
+    client: ProxyClient,
+    api_key: str,
+    model: str,
+    rate_limiter: RateLimiter | None = None,
+) -> ToolSearchReplay:
+    """Run `probe_tool_search`, then send the real assistant turn back as
+    history with the same tools still declared.
+
+    The first turn only proves the proxy attaches the tool-search beta header on
+    the way out. Nothing proves the provider accepts the `server_tool_use` and
+    `tool_search_tool_result` blocks it produced when they come back in
+    `messages`, which is every turn of a real Claude Code session after the
+    first."""
+    first_turn = probe_tool_search(
+        client=client,
+        api_key=api_key,
+        model=model,
+        max_tokens=_REPLAY_SOURCE_MAX_TOKENS,
+        rate_limiter=rate_limiter,
+    )
+    if not isinstance(first_turn, Success):
+        return ToolSearchReplay(first_turn=first_turn, history=(), second_turn=None)
+
+    history = _replay_history(first_turn.data)
+    _acquire(model, rate_limiter)
+    return ToolSearchReplay(
+        first_turn=first_turn,
+        history=history,
+        second_turn=client.messages(
+            api_key,
+            AnthropicMessagesBody(
+                model=model,
+                max_tokens=64,
+                messages=list(history),
+                tools=list(_TOOL_SEARCH_TOOLS),
+            ),
+        ),
+    )
+
+
 def _failure_diagnostic[R: BaseModel](result: Result[R], route: str) -> str:
     """Map a non-success `Result` to a one-line diagnostic. The `status 429`
     wording is load-bearing: the compat conftest classifies a rate-limited cell
@@ -207,6 +315,40 @@ def assert_tool_search_shape(result: Result[AnthropicMessagesResponse]) -> str |
             return _failure_diagnostic(result, "/v1/messages")
 
 
+def assert_tool_search_replay_shape(replay: ToolSearchReplay) -> str | None:
+    """Return None on success, else describe the first violation.
+
+    Acceptance criteria:
+
+      1. The first turn succeeded, on the same terms as `assert_tool_search_shape`.
+      2. That turn produced a complete `server_tool_use` / `tool_search_tool_result`
+         pair to replay. Without both the second turn carries either an ordinary
+         text history or a half-finished tool call, and the cell would report on
+         our own request rather than on the provider's handling of server-tool
+         blocks in history.
+      3. The provider accepted the history containing those blocks.
+    """
+    first_error = assert_tool_search_shape(replay.first_turn)
+    if first_error is not None:
+        return f"first turn: {first_error}"
+
+    replayed = _replayed_server_block_types(replay.history)
+    missing = _REPLAYED_SERVER_BLOCKS - replayed
+    if missing:
+        return (
+            f"first turn returned no {' or '.join(sorted(missing))} block to replay, so the history "
+            "proves nothing about server-tool handling; a turn truncated at max_tokens looks like this"
+        )
+
+    if replay.second_turn is None:
+        return "second turn was never sent"
+
+    second_error = assert_tool_search_shape(replay.second_turn)
+    if second_error is not None:
+        return f"history replaying {sorted(replayed)} rejected: {second_error}"
+    return None
+
+
 def assert_count_tokens_shape(result: Result[CountTokensResponse]) -> str | None:
     """Return None on success, or an error string describing the first violation.
 

tests/e2e/claude_code/tool_search/test_bedrock_invoke.py

@@ -1,12 +1,17 @@
 """tool_search x Bedrock (Invoke).
 
-HTTP-probe row. Sends a single `/v1/messages` request whose `tools`
-array includes a `tool_search_tool_regex_20251119` discovery tool, and
+HTTP-probe row. Sends a `/v1/messages` request whose `tools` array
+includes a `tool_search_tool_regex_20251119` discovery tool, and
 asserts the proxy round-trips it to the upstream without a 400. This
 verifies LiteLLM's tool-search beta-header translation
 (`advanced-tool-use-2025-11-20` for Anthropic-shape providers,
 `tool-search-tool-2025-10-19` for Vertex/Bedrock) survives end-to-end.
 
+A second probe then replays that turn's answer as history, which is
+what every turn of a real session after the first looks like: the
+first turn only exercises the outbound header, and the blocks the
+model sends back have to be accepted on the way in too.
+
 The (feature, provider) for this cell is inferred from the file path by
 `tests/e2e/claude_code/conftest.py`:
 
@@ -47,8 +52,10 @@ import pytest
 
 from claude_code._env import require_proxy_client
 from claude_code.http_probe import (
+    assert_tool_search_replay_shape,
     assert_tool_search_shape,
     probe_tool_search,
+    probe_tool_search_multiturn,
 )
 
 
@@ -80,3 +87,31 @@ def test_tool_search_bedrock_invoke(compat_result):
 
     if failures:
         pytest.fail("; ".join(failures), pytrace=False)
+
+
+@pytest.mark.covers("llm.messages.bedrock_invoke.tool_search_history.nonstream.works")
+def test_tool_search_history_bedrock_invoke(compat_result):
+    """Send the tool-search request, take the real assistant turn back, and
+    replay it as history with the tools still declared.
+
+    Every turn of a real Claude Code session after the first carries the
+    `server_tool_use` and `tool_search_tool_result` blocks the previous turn
+    produced. The single-turn probe above never sends them, so it cannot see a
+    provider or a transformation that accepts tool_search on the way out and
+    rejects the blocks it gets back."""
+    client, api_key = require_proxy_client(compat_result)
+
+    failures = []
+    for model in BEDROCK_INVOKE_MODELS:
+        replay = probe_tool_search_multiturn(client=client, api_key=api_key, model=model)
+        shape_error = assert_tool_search_replay_shape(replay)
+        if shape_error is not None:
+            error = f"[{model}] tool_search history replay failed: {shape_error}"
+            compat_result.add({"status": "fail", "error": error})
+            failures.append(error)
+            continue
+
+        compat_result.add({"status": "pass"})
+
+    if failures:
+        pytest.fail("; ".join(failures), pytrace=False)

tests/e2e/coverage_registry/llm_claude_code_compat.yaml

@@ -8,7 +8,8 @@
 #   route      : anthropic | azure_foundry | bedrock_converse | bedrock_invoke | vertex
 #   capability : basic | tool_use | vision | thinking | prompt_cache_5m | prompt_cache_1h
 #              | structured_output | pdf_input | long_context_1m
-#              | thinking_with_tool_use | tool_search | count_tokens | web_search
+#              | thinking_with_tool_use | tool_search | tool_search_history | count_tokens
+#              | web_search
 #   streaming  : stream | nonstream
 
 # ---- basic / non-streaming ----
@@ -94,6 +95,7 @@
 - {id: llm.messages.bedrock_converse.tool_search.nonstream.works, module: llm, tier: P1, subject_endpoint: messages, route: bedrock_converse, capability: tool_search, streaming: nonstream, assertions: [works], source: "claude_code compat matrix", rationale: "tool_search discovery tool over Bedrock Converse"}
 - {id: llm.messages.bedrock_invoke.tool_search.nonstream.works, module: llm, tier: P1, subject_endpoint: messages, route: bedrock_invoke, capability: tool_search, streaming: nonstream, assertions: [works], source: "claude_code compat matrix", rationale: "tool_search discovery tool over Bedrock Invoke"}
 - {id: llm.messages.vertex.tool_search.nonstream.works, module: llm, tier: P1, subject_endpoint: messages, route: vertex, capability: tool_search, streaming: nonstream, assertions: [works], source: "claude_code compat matrix", rationale: "tool_search discovery tool over Vertex AI"}
+- {id: llm.messages.bedrock_invoke.tool_search_history.nonstream.works, module: llm, tier: P1, subject_endpoint: messages, route: bedrock_invoke, capability: tool_search_history, streaming: nonstream, assertions: [works], source: "claude_code compat matrix", rationale: "A real server_tool_use / tool_search_tool_result pair replayed as history over Bedrock Invoke"}
 
 # ---- count_tokens ----
 - {id: llm.messages.anthropic.count_tokens.nonstream.works, module: llm, tier: P1, subject_endpoint: messages, route: anthropic, capability: count_tokens, streaming: nonstream, assertions: [works], source: "claude_code compat matrix", rationale: "/v1/messages/count_tokens over Anthropic direct"}

tests/e2e/coverage_registry/schema.py

@@ -76,6 +76,7 @@ LlmCapability = Literal[
     "thinking",
     "thinking_with_tool_use",
     "tool_search",
+    "tool_search_history",
     "tool_use",
     "vision",
     "web_search",

tests/e2e/models.py

@@ -384,9 +384,45 @@ class AnthropicCustomTool(BaseModel):
 type AnthropicTool = AnthropicToolSearchTool | AnthropicWebSearchTool | AnthropicCustomTool
 
 
+class AnthropicContentBlock(BaseModel):
+    """One block of a `content` array. Only the fields a test reads are
+    declared; `extra="allow"` keeps the rest (a `server_tool_use` block's
+    `input`, a `tool_search_tool_result` block's nested `content`) so an
+    assistant turn read off the wire can be replayed into history verbatim
+    instead of being silently flattened to its text."""
+
+    model_config = ConfigDict(extra="allow")
+    type: str | None = None
+    text: str | None = None
+    id: str | None = None
+
+
+class AnthropicToolResultBlock(BaseModel):
+    """The user-turn answer to a client-side `tool_use`. `tool_use_id` must be
+    the id the model actually emitted; an invented one is rejected by
+    Anthropic's own schema validator, which Bedrock inherits."""
+
+    type: Literal["tool_result"] = "tool_result"
+    tool_use_id: str
+    content: str
+
+
+class AnthropicAssistantTurn(BaseModel):
+    role: Literal["assistant"] = "assistant"
+    content: list[AnthropicContentBlock]
+
+
+class AnthropicToolResultTurn(BaseModel):
+    role: Literal["user"] = "user"
+    content: list[AnthropicToolResultBlock]
+
+
+type AnthropicMessage = ChatMessage | AnthropicAssistantTurn | AnthropicToolResultTurn
+
+
 class AnthropicMessagesBody(BaseModel):
     model: str
-    messages: list[ChatMessage]
+    messages: list[AnthropicMessage]
     max_tokens: int
     stream: bool | None = None
     tools: list[AnthropicTool] | None = None
@@ -401,11 +437,6 @@ class CountTokensBody(BaseModel):
     messages: list[ChatMessage]
 
 
-class AnthropicContentBlock(BaseModel):
-    type: str | None = None
-    text: str | None = None
-
-
 class AnthropicMessagesResponse(BaseModel):
     """A /v1/messages answer. `content` is the Anthropic-native passthrough
     shape; `choices` is the OpenAI-normalized shape LiteLLM emits for some