fix(anthropic): keep Claude tool turns streaming and Claude 5.5 thinking valid

providers/anthropic.py

@@ -5,6 +5,9 @@
 from typing import Any
 
 HAIKU_5_5_MODEL_ID = "claude-haiku-5-5"
+ADAPTIVE_THINKING_MODEL_IDS = frozenset(
+    {HAIKU_5_5_MODEL_ID, "claude-sonnet-5-5", "claude-opus-5-5"}
+)
 ANTHROPIC_MODEL_IDS = (HAIKU_5_5_MODEL_ID,)
 ANTHROPIC_REQUEST_HEADER_WHITELIST = {
     "anthropic-beta": "Anthropic-Beta",

services/protocol_translation/anthropic_continuation.py

@@ -142,7 +142,9 @@ def observe(self, event):
                 if not isinstance(value, str):
                     raise ValueError("Invalid Anthropic tool arguments")
                 self.arguments[index] = self.arguments.get(index, "") + value
-        elif index in self.arguments:
+        elif self.arguments.get(index):
+            # Empty fragments carry no JSON; retain the input from block start.
+            # Nonempty malformed JSON must still fail instead of losing arguments.
             self.blocks[index]["input"] = json.loads(self.arguments[index])
         self.bytes += len(json.dumps(event, ensure_ascii=False).encode())
         if self.bytes > MAX_CONTINUATION_BYTES:

services/protocol_translation/anthropic_request_policy.py

@@ -3,7 +3,7 @@
 from collections.abc import Mapping
 from typing import Any
 
-from providers.anthropic import HAIKU_5_5_MODEL_ID
+from providers.anthropic import ADAPTIVE_THINKING_MODEL_IDS
 
 MIN_THINKING_BUDGET = 1024
 THINKING_BUDGETS = {
@@ -15,10 +15,21 @@
     "max": 32768,
 }
 EFFORTS = frozenset({"none", "minimal", "low", "medium", "high", "xhigh", "max"})
+# How each Claude 5.5 model turns thinking off, from the API's 400s on 2026-10-26: Haiku accepts
+# "disabled", Sonnet only "between_tools", and Opus cannot turn it off, so there it stays adaptive.
+THINKING_OFF = {
+    "claude-haiku-5-5": {"type": "disabled"},
+    "claude-sonnet-5-5": {"type": "between_tools"},
+}
+
+
+def _thinking_off(model: Any) -> dict:
+    return dict(THINKING_OFF.get(model, {"type": "adaptive"}))
 
 
 def apply_thinking(body: dict, payload: Mapping[str, Any]) -> None:
-    if body.get("model") == HAIKU_5_5_MODEL_ID:
+    model = body.get("model")
+    if isinstance(model, str) and model in ADAPTIVE_THINKING_MODEL_IDS:
         _apply_adaptive_thinking(body, payload)
         return
     thinking = payload.get("thinking")
@@ -64,7 +75,7 @@ def apply_thinking(body: dict, payload: Mapping[str, Any]) -> None:
 
 
 def _apply_adaptive_thinking(body: dict, payload: Mapping[str, Any]) -> None:
-    # Haiku 5.5 rejects manual budgets and most client sampling defaults.
+    # Claude 5.5 models reject manual budgets and most client sampling defaults.
     for field in ("temperature", "top_p", "top_k"):
         body.pop(field, None)
     output_config = payload.get("output_config")
@@ -80,13 +91,15 @@ def _apply_adaptive_thinking(body: dict, payload: Mapping[str, Any]) -> None:
         body["thinking"].pop("budget_tokens", None)
         if thinking["type"] == "enabled":
             body["thinking"]["type"] = "adaptive"
+        elif thinking["type"] == "disabled":
+            body["thinking"] = _thinking_off(body.get("model"))
     effort = payload.get("reasoning_effort")
     if effort is None and isinstance(payload.get("reasoning"), Mapping):
         effort = payload["reasoning"].get("effort")
     if effort not in EFFORTS:
         return
     if effort == "none":
-        body["thinking"] = {"type": "disabled"}
+        body["thinking"] = _thinking_off(body.get("model"))
         effort = "low"
     elif "thinking" not in body:
         body["thinking"] = {"type": "adaptive"}

services/protocol_translation/anthropic_stream.py

@@ -155,9 +155,9 @@ def chunk(
             if (
                 block.get("kind") == "tool"
                 and not block["streamed"]
-                and block.get("input")
+                and isinstance(block.get("input"), Mapping)
             ):
-                # Some compatible servers send the whole input at block start.
+                # With no argument fragments, preserve the start input, even {}.
                 yield from chunk(
                     {
                         "tool_calls": [

services/protocol_translation/common.py

@@ -9,6 +9,7 @@
 from __future__ import annotations
 
 import json
+import logging
 import re
 import time
 import uuid
@@ -18,6 +19,8 @@
 
 from streaming.sse import SSEEvent, iter_sse_events
 
+logger = logging.getLogger(__name__)
+
 CHAT = "chat"
 RESPONSES = "responses"
 MESSAGES = "messages"
@@ -481,7 +484,8 @@ def guarded_items(items: Iterable[ChatItem]) -> Iterator[ChatItem]:
     """End with a failure, not an exception, when an upstream event has a bad shape."""
     try:
         yield from items
-    except (TypeError, AttributeError, KeyError, IndexError, ValueError):
+    except (TypeError, AttributeError, KeyError, IndexError, ValueError) as error:
+        logger.warning("Upstream stream translation failed type=%s", type(error).__name__)
         yield StreamFailure("The upstream stream could not be translated.")
 
 

tests/outbound_inventory.json

@@ -47,7 +47,7 @@
   "providers/anthropic.py": {
     "kind": "non-provider",
     "reason": "Runtime module pinned independently of HTTP client spelling; network attempts require runtime admission.",
-    "sha256": "775f4b5241684e69c80a62b23cd61a9b41de7e39e5931e2a4ea83c027e95a8fe"
+    "sha256": "3e389860fc6d38fa82f67dc2326261aeb4e3fa4fecca586d648189b85579a6cb"
   },
   "providers/base.py": {
     "kind": "non-provider",
@@ -1097,17 +1097,17 @@
   "services/protocol_translation/anthropic_continuation.py": {
     "kind": "non-provider",
     "reason": "Runtime module pinned independently of HTTP client spelling; network attempts require runtime admission.",
-    "sha256": "876a709fcf9661f9508d6c6e566362901c5133880a80b46c98ac67b678dff7bb"
+    "sha256": "4c963cca96a891cd909675d15e4c2ea4b48b9e9cf29aa281f84ebd548a6856ed"
   },
   "services/protocol_translation/anthropic_request_policy.py": {
     "kind": "non-provider",
     "reason": "Runtime module pinned independently of HTTP client spelling; network attempts require runtime admission.",
-    "sha256": "61bb336865809714a61f7af0cc620ee618ab805f441744697e8c5758d6d0e062"
+    "sha256": "a31b4e907e811f53316bb4a45a4c575e1ba03efe83c4d572d0021443e36c0a36"
   },
   "services/protocol_translation/anthropic_stream.py": {
     "kind": "non-provider",
     "reason": "Runtime module pinned independently of HTTP client spelling; network attempts require runtime admission.",
-    "sha256": "b2709c2e9b6d1abd6996ef8ec9be59ab6cb1cbc291f3f4002c419edfa5c916f0"
+    "sha256": "6aece70e3943ec2aa3f4eaeebc3c421b3009e13299af3d7f4406b3e35ffb219b"
   },
   "services/protocol_translation/anthropic_usage.py": {
     "kind": "non-provider",
@@ -1117,7 +1117,7 @@
   "services/protocol_translation/common.py": {
     "kind": "non-provider",
     "reason": "Runtime module pinned independently of HTTP client spelling; network attempts require runtime admission.",
-    "sha256": "8e7824816c68a0ac6c5a9d0d0db7ee56561896f967e6ab8638e2f4ff546f9483"
+    "sha256": "65c88feb3b60523eb8ca0ede1f0ebfaa94972eda37cc42bd5aebc0088db4d75f"
   },
   "services/protocol_translation/responses.py": {
     "kind": "non-provider",

tests/test_anthropic_parallel_stream.py

@@ -0,0 +1,205 @@
+"""Signed parallel tool turns, including tools without argument fragments."""
+
+import json
+import logging
+
+import pytest
+
+from services.protocol_translation import (
+    CHAT,
+    MESSAGES,
+    translate_request,
+    translate_stream,
+)
+from services.protocol_translation.anthropic import messages_sse_items
+from services.protocol_translation.common import (
+    StreamFailure,
+    aggregate_chat_items,
+    guarded_items,
+)
+from streaming.sse import iter_sse_events
+
+CONTENT = [
+    {"type": "thinking", "thinking": "", "signature": "opaque-signature"},
+    {
+        "type": "tool_use",
+        "id": "toolu_tokyo",
+        "name": "get_current_time",
+        "input": {"timezone": "Asia/Tokyo"},
+    },
+    {
+        "type": "tool_use",
+        "id": "toolu_london",
+        "name": "get_current_time",
+        "input": {"timezone": "Europe/London"},
+    },
+    {
+        "type": "tool_use",
+        "id": "toolu_reminders",
+        "name": "list_reminders",
+        "input": {},
+    },
+]
+
+
+def parallel_frames(empty_fragments, interleaved=False):
+    events = [
+        {
+            "type": "message_start",
+            "message": {
+                "id": "msg_parallel",
+                "model": "claude-haiku-5-5",
+                "usage": {"input_tokens": 5, "output_tokens": 0},
+            },
+        },
+        {
+            "type": "content_block_start",
+            "index": 0,
+            "content_block": {"type": "thinking", "thinking": ""},
+        },
+        {
+            "type": "content_block_delta",
+            "index": 0,
+            "delta": {"type": "signature_delta", "signature": "opaque-signature"},
+        },
+    ]
+    if not interleaved:
+        events.append({"type": "content_block_stop", "index": 0})
+    for index, block in enumerate(CONTENT[1:], 1):
+        events.append(
+            {
+                "type": "content_block_start",
+                "index": index,
+                "content_block": {**block, "input": {}},
+            }
+        )
+        fragments = (
+            [json.dumps(block["input"])[:8], json.dumps(block["input"])[8:]]
+            if block["input"]
+            else empty_fragments
+        )
+        events.extend(
+            {
+                "type": "content_block_delta",
+                "index": index,
+                "delta": {"type": "input_json_delta", "partial_json": fragment},
+            }
+            for fragment in fragments
+        )
+        if not interleaved:
+            events.append({"type": "content_block_stop", "index": index})
+    if interleaved:
+        events.extend(
+            {"type": "content_block_stop", "index": index} for index in (2, 0, 3, 1)
+        )
+    events.extend(
+        [
+            {
+                "type": "message_delta",
+                "delta": {"stop_reason": "tool_use"},
+                "usage": {"output_tokens": 8},
+            },
+            {"type": "message_stop"},
+        ]
+    )
+    return [
+        f"event: {event['type']}\ndata: {json.dumps(event)}\n\n".encode()
+        for event in events
+    ]
+
+
+@pytest.mark.parametrize("interleaved", [False, True])
+@pytest.mark.parametrize("empty_fragments", [[], [""], ["", ""], ["{}"]])
+def test_signed_parallel_tools_stream_to_chat(empty_fragments, interleaved):
+    frames = parallel_frames(empty_fragments, interleaved)
+    # Exercise the unguarded translator too, so failures expose their real type.
+    items = list(messages_sse_items(frames))
+    assert not any(isinstance(item, StreamFailure) for item in items)
+    output = list(translate_stream(frames, MESSAGES, CHAT))
+    events = list(iter_sse_events(output))
+    assert events[-1].is_done
+    chunks = [
+        json.loads(event.data) for event in events if event.data and not event.is_done
+    ]
+    assert not any("error" in chunk for chunk in chunks)
+    starts = [
+        call
+        for chunk in chunks
+        for call in chunk["choices"][0]["delta"].get("tool_calls", [])
+        if "id" in call
+    ]
+    assert [(call["index"], call["id"]) for call in starts] == [
+        (index, block["id"]) for index, block in enumerate(CONTENT[1:])
+    ]
+    completion = aggregate_chat_items(chunks)
+    assert completion["choices"][0]["finish_reason"] == "tool_calls"
+    assert completion["usage"] == {
+        "prompt_tokens": 5,
+        "completion_tokens": 8,
+        "total_tokens": 13,
+    }
+    assistant = completion["choices"][0]["message"]
+    calls = assistant["tool_calls"]
+    assert [json.loads(call["function"]["arguments"]) for call in calls] == [
+        block["input"] for block in CONTENT[1:]
+    ]
+    assert calls[0]["extra_content"]["anthropic"]["content"] == CONTENT
+    assert all("extra_content" not in call for call in calls[1:])
+    request = {
+        "model": "claude-haiku-5-5",
+        "messages": [
+            {"role": "user", "content": "Times and reminders?"},
+            assistant,
+            *[
+                {"role": "tool", "tool_call_id": call["id"], "content": "ok"}
+                for call in calls
+            ],
+        ],
+    }
+    assert (
+        translate_request(request, CHAT, MESSAGES)["messages"][1]["content"] == CONTENT
+    )
+
+
+@pytest.mark.parametrize("fragments", [["{"], [" "]])
+def test_nonempty_invalid_arguments_still_fail_closed(fragments, caplog):
+    with caplog.at_level(logging.WARNING):
+        output = list(translate_stream(parallel_frames(fragments), MESSAGES, CHAT))
+    events = list(iter_sse_events(output))
+    chunks = [
+        json.loads(event.data) for event in events if event.data and not event.is_done
+    ]
+    assert events[-1].is_done
+    assert (
+        chunks[-1]["error"]["message"] == "The upstream stream could not be translated."
+    )
+    assert not any(
+        choice.get("finish_reason")
+        for chunk in chunks
+        for choice in chunk.get("choices", [])
+    )
+    assert "type=JSONDecodeError" in caplog.text
+
+
+@pytest.mark.parametrize(
+    "error_type", [TypeError, AttributeError, KeyError, IndexError, ValueError]
+)
+def test_stream_guard_logs_only_exception_type(error_type, caplog):
+    def broken_items():
+        yield {"choices": []}
+        raise error_type("private upstream content")
+
+    with caplog.at_level(logging.WARNING):
+        items = list(guarded_items(broken_items()))
+    assert items == [
+        {"choices": []},
+        StreamFailure("The upstream stream could not be translated."),
+    ]
+    record = caplog.records[-1]
+    assert (
+        record.getMessage()
+        == f"Upstream stream translation failed type={error_type.__name__}"
+    )
+    assert record.exc_info is None
+    assert record.stack_info is None
+    assert "private upstream content" not in caplog.text