fix(anthropic): emit message_start once in Responses stream adapter (#32667) (#33793)

* fix(anthropic): emit message_start once in Responses stream adapter

* test(anthropic): cover response.created message_start guard branch

Co-authored-by: Mateo Wang <277851410+mateo-berri@users.noreply.github.com>
Co-authored-by: Napuh <55241721+Napuh@users.noreply.github.com>

litellm/llms/anthropic/experimental_pass_through/responses_adapters/streaming_iterator.py

@@ -75,8 +75,9 @@ class AnthropicResponsesStreamWrapper:
 
         # ---- message_start ----
         if event_type == "response.created":
-            self._sent_message_start = True
-            self._chunk_queue.append(self._make_message_start())
+            if not self._sent_message_start:
+                self._sent_message_start = True
+                self._chunk_queue.append(self._make_message_start())
             return
 
         # ---- content_block_start for a new output message item ----

tests/test_litellm/llms/anthropic/experimental_pass_through/responses_adapters/test_responses_adapters_streaming_iterator.py

@@ -3,12 +3,11 @@ Tests for AnthropicResponsesStreamWrapper
 (litellm/llms/anthropic/experimental_pass_through/responses_adapters/streaming_iterator.py)
 """
 
+import asyncio
 import os
 import sys
 
-sys.path.insert(
-    0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../../../../../.."))
-)
+sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../../../../../..")))
 
 from litellm.llms.anthropic.experimental_pass_through.responses_adapters.streaming_iterator import (
     AnthropicResponsesStreamWrapper,
@@ -22,6 +21,60 @@ def _process_all(events: list) -> list:
     return list(wrapper._chunk_queue)
 
 
+def _drain_async(events: list) -> list:
+    async def _gen():
+        for event in events:
+            yield event
+
+    async def _run() -> list:
+        wrapper = AnthropicResponsesStreamWrapper(responses_stream=_gen(), model="m")
+        return [chunk async for chunk in wrapper]
+
+    return asyncio.run(_run())
+
+
+class TestMessageStartEmittedExactlyOnce:
+    """The ``__anext__`` fallback emits ``message_start`` before consuming the
+    stream, so ``_process_event`` must not emit a second one when
+    ``response.created`` later arrives. Two ``message_start`` events (byte
+    identical, same id) break strict Anthropic SDK clients (e.g. Claude Code)
+    with 'Content block is not a thinking block' once thinking blocks follow."""
+
+    def test_response_created_does_not_duplicate_message_start(self):
+        chunks = _drain_async(
+            [
+                {"type": "response.created"},
+                {"type": "response.output_text.delta", "item_id": "m1", "delta": "hi"},
+            ]
+        )
+        message_starts = [c for c in chunks if c["type"] == "message_start"]
+        assert len(message_starts) == 1
+
+    def test_message_start_is_first_event(self):
+        chunks = _drain_async([{"type": "response.created"}])
+        assert chunks[0]["type"] == "message_start"
+
+
+class TestProcessEventResponseCreatedGuard:
+    """``_process_event`` must emit ``message_start`` exactly once even if
+    ``response.created`` arrives more than once. The guard mirrors the
+    ``__anext__`` fallback's ``_sent_message_start`` flag, so a direct caller
+    and the async fallback can never double-emit. This also exercises the
+    guard's emit-branch, which the async path never reaches because the
+    fallback sets the flag before the upstream stream is consumed."""
+
+    def test_first_response_created_emits_message_start(self):
+        chunks = _process_all([{"type": "response.created"}])
+        assert len(chunks) == 1
+        assert chunks[0]["type"] == "message_start"
+        assert chunks[0]["message"]["model"] == "m"
+
+    def test_second_response_created_is_skipped(self):
+        chunks = _process_all([{"type": "response.created"}, {"type": "response.created"}])
+        message_starts = [c for c in chunks if c["type"] == "message_start"]
+        assert len(message_starts) == 1
+
+
 class TestProcessEventTextDeltaWithoutOutputItemAdded:
     """Streams that skip response.output_item.added (e.g. LMStudio) must still
     open a text block before any delta and never emit index -1."""