diff --git a/agent/anthropic_endpoints.py b/agent/anthropic_endpoints.py index 237a562176..d8a7152f3e 100644 --- a/agent/anthropic_endpoints.py +++ b/agent/anthropic_endpoints.py @@ -157,3 +157,29 @@ def _is_azure_anthropic_endpoint(base_url: str | None) -> bool: host_padded = f".{(parsed.hostname or '').lower().rstrip('.')}." is_azure_host = ".services.ai.azure." in host_padded or ".openai.azure." in host_padded return is_azure_host and "/anthropic" in (parsed.path or "").lower() + + +def _is_bedrock_runtime_endpoint(base_url: str | None) -> bool: + """Amazon Bedrock's runtime host (``bedrock-runtime[-fips]..amazonaws.com[.cn]``, and VPC + endpoint names carrying the same label), which the AnthropicBedrock SDK path binds as its base URL.""" + host = base_url_hostname(base_url or "") + return any(label.startswith("bedrock-runtime") for label in host.split(".")) and ( + base_url_host_matches(host, "amazonaws.com") or base_url_host_matches(host, "amazonaws.com.cn") + ) + + +def _is_vertex_ai_endpoint(base_url: str | None) -> bool: + """Vertex AI hosts: global ``aiplatform.googleapis.com`` and regional ``-aiplatform.googleapis.com``.""" + host = base_url_hostname(base_url or "") + return host == "aiplatform.googleapis.com" or host.endswith("-aiplatform.googleapis.com") + + +def _is_claude_platform_endpoint(base_url: str | None) -> bool: + """Cloud platforms serving Claude under Anthropic's own thinking signatures: Bedrock, Vertex AI and + Azure AI Foundry ``/anthropic``. Anthropic documents ``signature`` values as compatible across the + Claude API, Bedrock and Google Cloud, so these replay signed thinking like the direct API.""" + return ( + _is_bedrock_runtime_endpoint(base_url) + or _is_vertex_ai_endpoint(base_url) + or _is_azure_anthropic_endpoint(base_url) + ) diff --git a/agent/anthropic_message_convert.py b/agent/anthropic_message_convert.py index 7e6d6c66dd..8a0526ab09 100644 --- a/agent/anthropic_message_convert.py +++ b/agent/anthropic_message_convert.py @@ -11,7 +11,7 @@ import re from typing import Any, Dict, List, Optional, Tuple from agent.image_eviction_policy import outbound_image_retire_count -from agent.anthropic_thinking_policy import anthropic_thinking_route, model_preserves_prior_thinking +from agent.anthropic_thinking_policy import anthropic_thinking_route, route_replays_prior_thinking logger = logging.getLogger(__name__) @@ -570,20 +570,33 @@ def _keep_valid_thinking(content: list[Any], signature_dead: bool) -> list[Any]: return new_content +def _current_tool_loop_start(result: list[dict[str, Any]]) -> int: + """Index after the last user message that is not purely tool results: every assistant turn from + there on belongs to the in-flight tool loop, whose thinking Anthropic requires back.""" + for i in range(len(result) - 1, -1, -1): + m = result[i] + if m.get("role") != "user": + continue + content = m.get("content") + if not (isinstance(content, list) and content and all(_block_type(b) == "tool_result" for b in content)): + return i + 1 + return 0 + + def _manage_thinking_signatures(result: list[dict[str, Any]], base_url: str | None, model: str | None) -> None: """Strip or preserve thinking blocks per endpoint. Mutates ``result`` in place. Anthropic signs thinking blocks against the full turn; any upstream mutation invalidates them - (400 "Invalid signature in thinking block"). Native preserved-thinking models keep valid signed - blocks on every assistant turn; older Claude models retain the established latest-turn-only - policy. Signatures are proprietary: third-party endpoints strip all thinking. Kimi replays as-is; - DeepSeek needs unsigned blocks round-tripped but rejects signed ones. Nous Portal proxies Claude - with sticky sessions and validates the same signatures, so it takes the native path despite not - being anthropic.com. + (400 "Invalid signature in thinking block"). Signed routes (Anthropic, Nous Portal, Bedrock, + Vertex, Foundry; MiniMax for its own blocks) keep valid signed blocks on every assistant turn + when the model keeps prior thinking, else on every turn of the in-flight tool loop (the API + requires those back and strips older turns itself; cutting at the latest assistant message + broke the loop's earlier steps). Unknown relays cannot verify signatures and strip all; Kimi + replays as-is; DeepSeek needs unsigned blocks round-tripped but rejects signed ones. """ route = anthropic_thinking_route(base_url, model) - last_assistant_idx = next((i for i in range(len(result) - 1, -1, -1) if result[i].get("role") == "assistant"), None) - preserve_prior = model_preserves_prior_thinking(model) + loop_start = _current_tool_loop_start(result) + preserve_prior = route_replays_prior_thinking(route, model) for idx, m in _assistant_block_lists(result): if route == "kimi": pass # shared cleanup below still strips cache markers + the flag @@ -594,7 +607,7 @@ def _manage_thinking_signatures(result: list[dict[str, Any]], base_url: str | No if _block_type(b) not in _THINKING_TYPES or not (b.get("signature") or b.get("data")) ] m["content"] = new_content or [_text_block("(empty)")] - elif route == "third_party" or (idx != last_assistant_idx and not preserve_prior): + elif route == "third_party" or (idx < loop_start and not preserve_prior): m["content"] = _strip_thinking(m["content"]) or [_text_block("(thinking elided)")] else: new_content = _keep_valid_thinking(m["content"], bool(m.get("_thinking_signature_invalidated"))) @@ -716,8 +729,8 @@ def convert_messages_to_anthropic( ) -> tuple[Optional[Any], list[dict]]: """Convert OpenAI-format messages to Anthropic format -> ``(system, messages)``. System is extracted into its own param (a string, or a block list when cache_control is present). - ``base_url``/``model`` drive thinking-signature policy — third-party endpoints strip signatures - (proprietary, they 400 on them); Kimi-family endpoints/models keep unsigned + ``base_url``/``model`` drive thinking-signature policy — unknown third-party relays strip + signatures (they cannot verify them); Kimi-family endpoints/models keep unsigned reasoning_content-derived blocks, which Kimi requires even when empty.""" system = None result: list[dict[str, Any]] = [] diff --git a/agent/anthropic_thinking_policy.py b/agent/anthropic_thinking_policy.py index 9772dfd93c..ee3ff79738 100644 --- a/agent/anthropic_thinking_policy.py +++ b/agent/anthropic_thinking_policy.py @@ -1,14 +1,15 @@ """Shared Anthropic prior-turn thinking retention policy. -Anthropic preserves and bills prior assistant thinking on Opus 4.5+ and -Sonnet 4.6+ (plus the newer Fable/Mythos families). Older models accept -replayed blocks but strip them server-side. Keep the model capability in -one place so message conversion and context accounting cannot diverge. +Anthropic preserves and bills prior assistant thinking on Opus 4.5+, Sonnet +4.6+ and Haiku 5.5+ (plus the newer Fable/Mythos families). Older models +accept replayed blocks but strip them server-side. Keep the model capability +and the per-endpoint replay contract in one place so message conversion and +context accounting cannot diverge. Unknown or future Claude ids default to keep: replaying a block the API strips costs nothing, while stripping on a model that keeps it rewrites the cached prefix on every call. Only the known last-turn-only generations -(Haiku, Claude 3, Opus < 4.5, Sonnet < 4.6) strip. Non-Claude ids never keep. +(Claude 3, Haiku < 5.5, Opus < 4.5, Sonnet < 4.6) strip. Non-Claude ids never keep. """ from __future__ import annotations @@ -17,8 +18,10 @@ import re from typing import Any from agent.anthropic_endpoints import ( + _is_claude_platform_endpoint, _is_deepseek_anthropic_endpoint, _is_kimi_family_endpoint, + _is_minimax_anthropic_endpoint, _is_nous_portal_endpoint, _is_third_party_anthropic_endpoint, _model_name_is_deepseek_thinking, @@ -28,10 +31,12 @@ from agent.anthropic_endpoints import ( _CLAUDE_VERSION_RE = re.compile( # Semantic minors are short version components. Snapshot dates such as # claude-opus-4-20250514 must remain 4.0 rather than becoming 4.20250514. - r"claude[-_.](opus|sonnet|fable|mythos)[-_.](\d+)(?:[-_.](\d{1,2})(?=$|[-_.]))?", + r"claude[-_.](opus|sonnet|haiku|fable|mythos)[-_.](\d+)(?:[-_.](\d{1,2})(?=$|[-_.]))?", re.IGNORECASE, ) -_LAST_TURN_ONLY_RE = re.compile(r"haiku|claude[-_.]?3(?!\d)", re.IGNORECASE) +# Claude 3.x ids put the family after the version (claude-3-5-haiku); 4+ ids are version-gated below. +_LAST_TURN_ONLY_RE = re.compile(r"claude[-_.]?3(?!\d)", re.IGNORECASE) +_KEEP_ALL_FROM = {"opus": (4, 5), "sonnet": (4, 6), "haiku": (5, 5)} def _claude_family_version(model: Any) -> tuple[str, tuple[int, int]] | None: @@ -54,21 +59,44 @@ def model_preserves_prior_thinking(model: Any) -> bool: if parsed is None: return True family, version = parsed - return version >= {"opus": (4, 5), "sonnet": (4, 6)}.get(family, (5, 0)) + return version >= _KEEP_ALL_FROM.get(family, (5, 0)) + + +# Routes whose thinking blocks carry signatures the upstream verifies: they replay signed blocks, and a +# signature 400 there is healed by anthropic_thinking_replay's durable suppression. +SIGNED_REPLAY_ROUTES = frozenset({"native", "minimax"}) + + +def _is_off_anthropic(base_url: Any) -> bool: + return _is_third_party_anthropic_endpoint(base_url) and not _is_nous_portal_endpoint(base_url) + + +# First match wins; an unmatched relay host is ``third_party``. DeepSeek models never sign, so they keep +# the unsigned-only contract on any host but Anthropic's own, Claude platforms included. +_ROUTE_RULES = ( + ("kimi", lambda url, model: _is_kimi_family_endpoint(url, model)), + ("deepseek", lambda url, model: _is_deepseek_anthropic_endpoint(url) + or (_is_off_anthropic(url) and _model_name_is_deepseek_thinking(model))), + ("native", lambda url, model: not _is_off_anthropic(url) or _is_claude_platform_endpoint(url)), + ("minimax", lambda url, model: _is_minimax_anthropic_endpoint(url)), +) def anthropic_thinking_route(base_url: Any, model: Any) -> str: """Which thinking-replay contract an Anthropic Messages request follows: ``kimi`` (replay as-is), - ``deepseek`` (unsigned only), ``third_party`` (strip all) or ``native`` (Anthropic-signed blocks: - direct Anthropic and Nous Portal). The converter and signature-rejection state share this.""" - is_third_party = _is_third_party_anthropic_endpoint(base_url) and not _is_nous_portal_endpoint(base_url) - if _is_kimi_family_endpoint(base_url, model): - return "kimi" - if _is_deepseek_anthropic_endpoint(base_url) or (is_third_party and _model_name_is_deepseek_thinking(model)): - return "deepseek" - return "third_party" if is_third_party else "native" + ``deepseek`` (unsigned only), ``native`` (Anthropic-signed blocks: the direct API, Nous Portal, + Bedrock, Vertex AI, Azure AI Foundry), ``minimax`` (its own signed blocks, returned unchanged on + every turn as its docs require) or ``third_party`` (an unknown relay that cannot verify Anthropic + signatures: strip all). The converter, accounting and signature-rejection state share this.""" + return next((route for route, matches in _ROUTE_RULES if matches(base_url, model)), "third_party") + + +def route_replays_prior_thinking(route: str, model: Any) -> bool: + """Whether ``route`` replays signed thinking on assistant turns before the in-flight tool loop.""" + return route == "minimax" or (route == "native" and model_preserves_prior_thinking(model)) def native_anthropic_preserves_prior_thinking(base_url: Any, model: Any) -> bool: - """True for direct Anthropic/Nous Portal routes whose model retains old thinking.""" - return anthropic_thinking_route(base_url, model) == "native" and model_preserves_prior_thinking(model) + """True when the route replays (and the model keeps) signed thinking from earlier turns, so stale + thinking reaches the wire and accounting must charge the signed carriers.""" + return route_replays_prior_thinking(anthropic_thinking_route(base_url, model), model) diff --git a/agent/anthropic_thinking_replay.py b/agent/anthropic_thinking_replay.py index 8740f06d49..ab9a66766c 100644 --- a/agent/anthropic_thinking_replay.py +++ b/agent/anthropic_thinking_replay.py @@ -4,7 +4,7 @@ Canonical history stays untouched. We persist only fingerprints of rejected opaq signature/data values, then filter those blocks from each rebuilt request copy. This is needed beyond the immediate retry because context selection and process resume can rebuild from canonical history later. The state belongs to one session (and is carried onto its -compression continuation); it applies only where Anthropic signs the blocks. +compression continuation); it applies only where the upstream signs the blocks. The fingerprinting is deliberately coarse: one signature 400 marks every signed block in the rejected request, so the session falls back to stripping that history (one cache miss, then @@ -43,13 +43,13 @@ def _fingerprint(block: Any) -> str | None: def tracks_rejected_thinking(agent: Any) -> bool: - """Native Anthropic signatures only: Kimi, DeepSeek and third-party routes keep their own replay - contract and the one-request ``reasoning_details`` repair.""" - from agent.anthropic_thinking_policy import anthropic_thinking_route + """Signed-replay routes only (Anthropic and its cloud platforms, MiniMax): Kimi, DeepSeek and + third-party routes keep their own replay contract and the one-request ``reasoning_details`` repair.""" + from agent.anthropic_thinking_policy import SIGNED_REPLAY_ROUTES, anthropic_thinking_route return getattr(agent, "api_mode", None) == "anthropic_messages" and anthropic_thinking_route( getattr(agent, "base_url", None), getattr(agent, "model", None) - ) == "native" + ) in SIGNED_REPLAY_ROUTES def rejected_thinking_fingerprints(session_db: Any, session_id: Any) -> set[str]: diff --git a/agent/transports/anthropic.py b/agent/transports/anthropic.py index 885d07dc47..a0dca6f080 100644 --- a/agent/transports/anthropic.py +++ b/agent/transports/anthropic.py @@ -43,9 +43,9 @@ class AnthropicTransport(ProviderTransport): return "anthropic_messages" def convert_messages(self, messages: list[dict[str, Any]], **kwargs) -> Any: - """Convert OpenAI messages to an Anthropic (system, messages) tuple; ``base_url`` affects thinking-signature handling.""" + """Convert OpenAI messages to an Anthropic (system, messages) tuple; ``base_url`` and ``model`` pick the thinking-replay policy.""" from agent.anthropic_message_convert import convert_messages_to_anthropic - return convert_messages_to_anthropic(messages, base_url=kwargs.get("base_url")) + return convert_messages_to_anthropic(messages, base_url=kwargs.get("base_url"), model=kwargs.get("model")) def convert_tools(self, tools: list[dict[str, Any]]) -> Any: """Convert OpenAI tool schemas to Anthropic input_schema format.""" diff --git a/tests/agent/test_anthropic_signed_platform_replay.py b/tests/agent/test_anthropic_signed_platform_replay.py new file mode 100644 index 0000000000..f00de76076 --- /dev/null +++ b/tests/agent/test_anthropic_signed_platform_replay.py @@ -0,0 +1,98 @@ +"""Signed thinking reaches the wire wherever the upstream can verify it, and only there. + +Anthropic documents ``signature`` values as valid across the Claude API, Bedrock and Google Cloud; +Foundry serves the same Claude; MiniMax's Anthropic API asks for its own thinking blocks back +unchanged. Unknown relays cannot verify them and keep stripping; DeepSeek and Kimi keep their own +contracts. Within the in-flight tool loop every model gets its thinking back (required by the API); +earlier turns only where the model keeps prior thinking. +""" + +from __future__ import annotations + +import copy + +import pytest + +from agent.anthropic_thinking_policy import native_anthropic_preserves_prior_thinking +from agent.anthropic_thinking_replay import tracks_rejected_thinking +from agent.message_sanitization import stale_thinking_reaches_wire +from agent.transports.anthropic import AnthropicTransport + +BEDROCK = "https://bedrock-runtime.us-east-1.amazonaws.com" +VERTEX = "https://us-east5-aiplatform.googleapis.com/v1/projects/p/locations/us-east5/publishers/anthropic/models" +FOUNDRY = "https://my-res.services.ai.azure.com/anthropic" +MINIMAX = "https://api.minimax.io/anthropic" +MINIMAX_CN = "https://api.minimaxi.com/anthropic" +RELAY = "https://relay.example.com/anthropic" +ALL, LOOP, NONE = ["sig_prior", "sig_loop1", "sig_loop2"], ["sig_loop1", "sig_loop2"], [] + +_CASES = { + # Claude platforms: preserving models replay every turn, last-turn-only models the whole tool loop. + "bedrock_opus": (BEDROCK, "us.anthropic.claude-opus-4-8-v1:0", ALL, True), + "bedrock_sonnet45": (BEDROCK, "us.anthropic.claude-sonnet-4-5-20250929-v1:0", LOOP, True), + "bedrock_fips": ("https://bedrock-runtime-fips.us-gov-west-1.amazonaws.com", "anthropic.claude-opus-4-8", ALL, True), + "vertex": (VERTEX, "claude-opus-4-8", ALL, True), + "foundry": (FOUNDRY, "claude-opus-4-8", ALL, True), + "minimax": (MINIMAX, "MiniMax-M2.7", ALL, True), + "minimax_cn": (MINIMAX_CN, "MiniMax-M3", ALL, True), + # Direct API: Haiku 5.5 keeps all turns; older generations keep the current tool loop. + "native_haiku55": (None, "claude-haiku-5-5", ALL, True), + "native_haiku45": (None, "claude-haiku-4-5", LOOP, True), + "native_sonnet45": (None, "claude-sonnet-4-5", LOOP, True), + "native_opus48": (None, "claude-opus-4-8", ALL, True), + # Unchanged contracts. + "unknown_relay": (RELAY, "claude-opus-4-8", NONE, False), + "bedrock_lookalike_host": ("https://bedrock-runtime.us-east-1.amazonaws.com.evil.test", "claude-opus-4-8", NONE, False), + "bedrock_lookalike_path": ("https://evil.test/bedrock-runtime.us-east-1.amazonaws.com", "claude-opus-4-8", NONE, False), + "deepseek": ("https://api.deepseek.com/anthropic", "deepseek-v4-pro", NONE, False), + "deepseek_on_bedrock_host": (BEDROCK, "deepseek-v4-pro", NONE, False), + "kimi": ("https://api.kimi.com/coding", "kimi-k2.5", ALL, False), +} + + +def _signed(sig): + return {"type": "thinking", "thinking": f"thought {sig}", "signature": sig} + + +def _loop_step(n): + call = {"id": f"t{n}", "type": "function", "function": {"name": "read_file", "arguments": "{}"}} + return [ + {"role": "assistant", "content": "", "tool_calls": [call], "reasoning_details": [_signed(f"sig_loop{n}")], + "anthropic_content_blocks": [_signed(f"sig_loop{n}"), {"type": "tool_use", "id": f"t{n}", "name": "read_file", "input": {}}]}, + {"role": "tool", "tool_call_id": f"t{n}", "content": "ok"}, + ] + + +def _history(): + return [ + {"role": "user", "content": "Q1"}, + {"role": "assistant", "content": "A1", "reasoning_details": [_signed("sig_prior")]}, + {"role": "user", "content": "Q2"}, + *_loop_step(1), + *_loop_step(2), + ] + + +@pytest.mark.parametrize("case", list(_CASES)) +def test_signed_thinking_replays_exactly_where_the_route_can_verify_it(case): + """Invariant: the transport entry point sends the route's signed blocks in their captured order, + the signature-400 self-heal covers every signed route, accounting charges prior-turn thinking + exactly when it is on the wire, and stored history is never mutated.""" + base_url, model, expected, signed_route = _CASES[case] + history = _history() + before = copy.deepcopy(history) + + _system, wire = AnthropicTransport().convert_messages(history, base_url=base_url, model=model) + + blocks = [b for m in wire if m["role"] == "assistant" for b in m["content"]] + assert [b.get("signature") for b in blocks if b.get("type") == "thinking"] == expected + for step in (1, 2): # interleaved order: each step's thinking precedes its tool_use + types = [b["type"] for b in blocks if b.get("signature") == f"sig_loop{step}" or b.get("id") == f"t{step}"] + assert types == (["thinking", "tool_use"] if f"sig_loop{step}" in expected else ["tool_use"]) + agent = type("A", (), {"api_mode": "anthropic_messages", "base_url": base_url, "model": model})() + assert tracks_rejected_thinking(agent) is signed_route + # Kimi replays as-is through its own reasoning_content echo; signed routes are priced by the carriers. + assert native_anthropic_preserves_prior_thinking(base_url, model) is (signed_route and "sig_prior" in expected) + if signed_route: + assert stale_thinking_reaches_wire("anthropic_messages", "custom", model, base_url) is ("sig_prior" in expected) + assert history == before diff --git a/tests/agent/test_nous_portal_anthropic_wire.py b/tests/agent/test_nous_portal_anthropic_wire.py index 93ee08bfbd..9aeb5982c9 100644 --- a/tests/agent/test_nous_portal_anthropic_wire.py +++ b/tests/agent/test_nous_portal_anthropic_wire.py @@ -411,13 +411,13 @@ class TestPortalThinkingReplay: self._assert_thinking_kept(STAGING_URL) def test_other_third_party_gateways_still_strip_thinking(self): - """The Portal carve-out must not leak into MiniMax-style proxies.""" + """The Portal carve-out must not leak into unknown Anthropic-compatible relays.""" from agent.anthropic_message_convert import convert_messages_to_anthropic _system, converted = convert_messages_to_anthropic( self._messages(), - base_url="https://api.minimax.io/anthropic", - model="MiniMax-M2.7", + base_url="https://relay.example.com/anthropic", + model="claude-opus-4-8", ) assistant = next(m for m in converted if m["role"] == "assistant") thinking = [