From 966ff65fec6b815012ef1dfc701d26038c098814 Mon Sep 17 00:00:00 2001 From: yuneng-jiang Date: Fri, 17 Jul 2026 17:26:33 -0700 Subject: [PATCH] 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> --- .../responses_adapters/streaming_iterator.py | 5 +- ...t_responses_adapters_streaming_iterator.py | 59 ++++++++++++++++++- 2 files changed, 59 insertions(+), 5 deletions(-) diff --git a/litellm/llms/anthropic/experimental_pass_through/responses_adapters/streaming_iterator.py b/litellm/llms/anthropic/experimental_pass_through/responses_adapters/streaming_iterator.py index 0d02b4fa969..4fd49a35417 100644 --- a/litellm/llms/anthropic/experimental_pass_through/responses_adapters/streaming_iterator.py +++ b/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 ---- diff --git a/tests/test_litellm/llms/anthropic/experimental_pass_through/responses_adapters/test_responses_adapters_streaming_iterator.py b/tests/test_litellm/llms/anthropic/experimental_pass_through/responses_adapters/test_responses_adapters_streaming_iterator.py index 450f69fb87c..9b5197d9028 100644 --- a/tests/test_litellm/llms/anthropic/experimental_pass_through/responses_adapters/test_responses_adapters_streaming_iterator.py +++ b/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."""