fix(anthropic_adapter): preserve first chunk on content-block transitions

The `/v1/messages` endpoint's AnthropicStreamWrapper silently dropped the
trigger chunk of every new content block on a detected block transition
(e.g. text -> thinking, thinking -> text, text -> tool_use). The code
emitted `content_block_stop` -> `content_block_start` and then returned
without re-enqueuing `processed_chunk`, relying on an incorrect assumption
that `content_block_start` already carries the trigger chunk's content.
In practice, `_translate_streaming_openai_chunk_to_anthropic_content_block()`
returns an empty `TextBlock(text="")` for text transitions, so the first
characters of every new text/thinking block were lost on the wire.

Symptoms for Bedrock Converse reasoning providers (MiniMax, Kimi, Claude
extended thinking) using `/v1/messages` + `stream=true`:
- Responses start mid-sentence (leading characters missing).
- If the model emits the text as a single Bedrock chunk, the text block
  is streamed with zero `content_block_delta` events, causing clients
  like Claude Code CLI to see an empty response.
- Non-streaming on the same deployment returns the full text correctly.

Fix: after emitting the synthetic `content_block_stop` + `content_block_start`
pair, re-enqueue `processed_chunk` when it is a non-empty
`content_block_delta`. A small `_trigger_delta_has_content` helper inspects
the standard Anthropic delta variants (`text`, `thinking`, `partial_json`,
`signature`) so that empty tool_use openers (whose tool name is already
carried by `content_block_start`) are intentionally skipped.

Applied to both the sync (`__next__`) and async (`__anext__`) paths.

Tests:
- Add `test_first_chunk_on_block_transition.py` with a focused regression
  test exercising a text -> thinking -> text sequence (the exact Bedrock
  Converse reasoning-model pattern). The new tests fail on the unpatched
  `main` and pass after the fix, for both sync and async iterators.
- Update the existing `test_parallel_tool_calls.py::test_anthropic_stream_wrapper_interleaved_tool_calls_and_text`
  expected sequence: two `content_block_delta` events for the interleaved
  text chunks were previously missing because the wrapper dropped them;
  the test now also asserts the text content is preserved verbatim.

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

@@ -14,6 +14,31 @@ if TYPE_CHECKING:
     from litellm.types.utils import ModelResponseStream
 
 
+def _trigger_delta_has_content(processed_chunk: Optional[Dict[str, Any]]) -> bool:
+    """Return True when a ``content_block_delta`` carries non-empty content.
+
+    Used on content-block transitions to decide whether the trigger chunk's delta
+    should be re-emitted alongside the synthetic ``content_block_stop`` /
+    ``content_block_start`` pair. Re-emitting is required for text and thinking
+    transitions (where the trigger chunk holds the first characters of the new
+    block) and is intentionally skipped for tool_use openers, whose delta is an
+    empty ``input_json_delta`` because the tool name is already carried by
+    ``content_block_start``.
+    """
+    if not processed_chunk or processed_chunk.get("type") != "content_block_delta":
+        return False
+    delta = processed_chunk.get("delta") or {}
+    if not isinstance(delta, dict):
+        return False
+    # Anthropic delta variants all carry their content under exactly one of
+    # these fields; any non-empty value means the chunk is worth emitting.
+    for key in ("text", "thinking", "partial_json", "signature"):
+        value = delta.get(key)
+        if value:
+            return True
+    return False
+
+
 class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
     """
     - first chunk return 'message_start'
@@ -129,8 +154,21 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
 
                 if should_start_new_block and not self.sent_content_block_finish:
                     # Queue the sequence: content_block_stop -> content_block_start
-                    # The trigger chunk itself is not emitted as a delta since the
-                    # content_block_start already carries the relevant information.
+                    # (-> content_block_delta, when the trigger chunk carries
+                    # content).
+                    #
+                    # NOTE: a non-empty trigger delta MUST be re-emitted here. The
+                    # `content_block_start` event carries only the block type and an
+                    # empty body (see
+                    # `_translate_streaming_openai_chunk_to_anthropic_content_block`
+                    # which returns `TextBlock(text="")` for text transitions), so
+                    # dropping a non-empty `processed_chunk` silently loses the first
+                    # characters of every new content block for providers that stream
+                    # reasoning then text (e.g. Bedrock Converse MiniMax / Kimi /
+                    # Claude extended thinking). For tool_use openers the trigger
+                    # chunk carries only the tool name + empty arguments, so its
+                    # delta is empty and is intentionally skipped - the tool name is
+                    # already part of `content_block_start`.
                     self.chunk_queue.append(
                         {
                             "type": "content_block_stop",
@@ -144,6 +182,8 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
                             "content_block": self.current_content_block_start,
                         }
                     )
+                    if _trigger_delta_has_content(processed_chunk):
+                        self.chunk_queue.append(processed_chunk)
                     self.sent_content_block_finish = False
                     return self.chunk_queue.popleft()
 
@@ -305,8 +345,13 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
                 if not self.queued_usage_chunk:
                     if should_start_new_block and not self.sent_content_block_finish:
                         # Queue the sequence: content_block_stop -> content_block_start
-                        # The trigger chunk itself is not emitted as a delta since the
-                        # content_block_start already carries the relevant information.
+                        # (-> content_block_delta, when the trigger chunk carries
+                        # content). See the sync `__next__` path for a full
+                        # explanation - the short version is that
+                        # `content_block_start` only carries the block type and an
+                        # empty body, so a non-empty trigger delta must be emitted
+                        # separately or the first characters of every new content
+                        # block are silently dropped.
 
                         # 1. Stop current content block
                         self.chunk_queue.append(
@@ -325,6 +370,13 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
                             }
                         )
 
+                        # 3. Emit the trigger chunk's delta when it has content
+                        # (skip empty deltas such as tool_use openers whose
+                        # arguments are empty and whose name is already carried by
+                        # `content_block_start`).
+                        if _trigger_delta_has_content(processed_chunk):
+                            self.chunk_queue.append(processed_chunk)
+
                         # Reset state for new block
                         self.sent_content_block_finish = False
 

tests/test_litellm/llms/anthropic/experimental_pass_through/messages/test_first_chunk_on_block_transition.py

@@ -0,0 +1,221 @@
+"""
+Regression tests for AnthropicStreamWrapper dropping the first chunk of content
+when a new content block is started.
+
+Context:
+Before the fix, `__next__` / `__anext__` emitted the sequence
+`content_block_stop` -> `content_block_start` on a detected block transition
+(e.g. text -> thinking, thinking -> text, text -> tool_use), but the `processed_chunk`
+that actually *triggered* the transition was silently discarded. Because
+`_translate_streaming_openai_chunk_to_anthropic_content_block()` returns a block
+with an empty body for text transitions (`TextBlock(text="")`), dropping the
+trigger chunk meant the first character(s) of every new block were lost on the
+wire. For Bedrock Converse reasoning providers (MiniMax, Kimi, Claude extended
+thinking), this typically manifested as responses starting mid-sentence or, if
+the model emitted the text as a single chunk, as an empty text block with zero
+`content_block_delta` events.
+
+The fix enqueues the trigger chunk's `content_block_delta` after the
+`content_block_start` so that no content is lost.
+"""
+
+import os
+import sys
+from typing import List
+
+import pytest
+
+sys.path.insert(0, os.path.abspath("../../../../.."))
+
+from litellm.llms.anthropic.experimental_pass_through.adapters.streaming_iterator import (
+    AnthropicStreamWrapper,
+)
+from litellm.types.utils import Delta, ModelResponseStream, StreamingChoices, Usage
+
+
+def _text_chunk(text: str) -> ModelResponseStream:
+    return ModelResponseStream(
+        choices=[
+            StreamingChoices(
+                delta=Delta(content=text, role="assistant"),
+                index=0,
+                finish_reason=None,
+            )
+        ],
+    )
+
+
+def _thinking_chunk(thinking: str) -> ModelResponseStream:
+    return ModelResponseStream(
+        choices=[
+            StreamingChoices(
+                delta=Delta(
+                    content="",
+                    role="assistant",
+                    reasoning_content=thinking,
+                    thinking_blocks=[
+                        {
+                            "type": "thinking",
+                            "thinking": thinking,
+                            "signature": None,
+                        }
+                    ],
+                    provider_specific_fields={
+                        "thinking_blocks": [
+                            {
+                                "type": "thinking",
+                                "thinking": thinking,
+                                "signature": None,
+                            }
+                        ]
+                    },
+                ),
+                index=0,
+                finish_reason=None,
+            )
+        ],
+    )
+
+
+def _stop_chunk() -> ModelResponseStream:
+    return ModelResponseStream(
+        choices=[
+            StreamingChoices(
+                delta=Delta(content="", role="assistant"),
+                index=0,
+                finish_reason="stop",
+            )
+        ],
+        usage=Usage(prompt_tokens=10, completion_tokens=10, total_tokens=20),
+    )
+
+
+class _MockStreamThinkingThenText:
+    """Simulates a Bedrock Converse reasoning model stream.
+
+    The first chunk is a thinking block (which forces a text->thinking transition
+    on the wrapper's initial state) and the first text chunk forces a
+    thinking->text transition. Both transition chunks carry real content that
+    must not be dropped.
+    """
+
+    def __init__(self) -> None:
+        self.responses: List[ModelResponseStream] = [
+            _thinking_chunk("The user says hi. I should"),
+            _thinking_chunk(" greet them back."),
+            _text_chunk("안녕하세요, 저는 MiniMax입니다."),
+            _text_chunk(" 무엇을 도와드릴까요?"),
+            _stop_chunk(),
+        ]
+        self.index = 0
+
+    def __iter__(self):
+        return self
+
+    def __next__(self):
+        if self.index >= len(self.responses):
+            raise StopIteration
+        response = self.responses[self.index]
+        self.index += 1
+        return response
+
+    def __aiter__(self):
+        return self
+
+    async def __anext__(self):
+        if self.index >= len(self.responses):
+            raise StopAsyncIteration
+        response = self.responses[self.index]
+        self.index += 1
+        return response
+
+
+def _collect_deltas(chunks: List[dict]) -> dict:
+    """Aggregate text_delta / thinking_delta content from a stream of wrapper chunks."""
+    text = ""
+    thinking = ""
+    for chunk in chunks:
+        if chunk.get("type") != "content_block_delta":
+            continue
+        delta = chunk.get("delta", {}) or {}
+        if delta.get("type") == "text_delta":
+            text += delta.get("text", "") or ""
+        elif delta.get("type") == "thinking_delta":
+            thinking += delta.get("thinking", "") or ""
+    return {"text": text, "thinking": thinking}
+
+
+EXPECTED_THINKING = "The user says hi. I should greet them back."
+EXPECTED_TEXT = "안녕하세요, 저는 MiniMax입니다. 무엇을 도와드릴까요?"
+
+
+def test_sync_first_chunk_preserved_on_block_transitions():
+    """Sync path: the first chunk of each new content block must not be dropped."""
+    wrapper = AnthropicStreamWrapper(
+        completion_stream=_MockStreamThinkingThenText(),
+        model="bedrock/converse/minimax.minimax-m2.5",
+    )
+
+    chunks = list(wrapper)
+    aggregated = _collect_deltas(chunks)
+
+    assert aggregated["thinking"] == EXPECTED_THINKING, (
+        "Thinking content was dropped on the text->thinking transition. "
+        f"Expected {EXPECTED_THINKING!r}, got {aggregated['thinking']!r}"
+    )
+    assert aggregated["text"] == EXPECTED_TEXT, (
+        "Text content was dropped on the thinking->text transition. "
+        f"Expected {EXPECTED_TEXT!r}, got {aggregated['text']!r}"
+    )
+
+
+@pytest.mark.asyncio
+async def test_async_first_chunk_preserved_on_block_transitions():
+    """Async path: same guarantee as the sync path."""
+    wrapper = AnthropicStreamWrapper(
+        completion_stream=_MockStreamThinkingThenText(),
+        model="bedrock/converse/minimax.minimax-m2.5",
+    )
+
+    chunks: List[dict] = []
+    async for chunk in wrapper:
+        chunks.append(chunk)
+
+    aggregated = _collect_deltas(chunks)
+
+    assert aggregated["thinking"] == EXPECTED_THINKING
+    assert aggregated["text"] == EXPECTED_TEXT
+
+
+def test_sync_block_transition_event_ordering():
+    """On a block transition the wrapper must emit
+    content_block_stop -> content_block_start -> content_block_delta in that order,
+    with the delta carrying the trigger chunk's content."""
+    wrapper = AnthropicStreamWrapper(
+        completion_stream=_MockStreamThinkingThenText(),
+        model="bedrock/converse/minimax.minimax-m2.5",
+    )
+
+    chunks = list(wrapper)
+    types = [c.get("type") for c in chunks]
+
+    # Find the first text->thinking transition: content_block_stop then start then delta.
+    # The wrapper emits an initial content_block_start(index=0,text) before the first
+    # real chunk, so the first stop we see is the index=0 text block being closed.
+    first_stop = types.index("content_block_stop")
+    assert types[first_stop + 1] == "content_block_start"
+    assert (
+        types[first_stop + 2] == "content_block_delta"
+    ), "First trigger chunk's delta must immediately follow content_block_start."
+
+    # And the delta must carry the *first* trigger chunk's content verbatim -
+    # i.e. the text that would otherwise have been dropped. This is the
+    # specific regression this test guards against.
+    delta_chunk = chunks[first_stop + 2]
+    inner_delta = delta_chunk.get("delta", {}) or {}
+    assert inner_delta.get("type") == "thinking_delta"
+    assert inner_delta.get("thinking") == "The user says hi. I should", (
+        "Expected the first thinking chunk's content to be emitted as the first "
+        "delta after content_block_start, but got "
+        f"{inner_delta.get('thinking')!r}"
+    )

tests/test_litellm/llms/anthropic/experimental_pass_through/messages/test_parallel_tool_calls.py

@@ -279,6 +279,7 @@ def test_anthropic_stream_wrapper_interleaved_tool_calls_and_text():
         "content_block_delta",  # "NY"}
         "content_block_stop",  # End of first tool_use content block
         "content_block_start",  # "The weather is nice today"
+        "content_block_delta",  # "The weather is nice today." (trigger chunk)
         "content_block_stop",
         "content_block_start",  # Start of second tool_use content block
         "content_block_delta",  # {"city":
@@ -289,6 +290,7 @@ def test_anthropic_stream_wrapper_interleaved_tool_calls_and_text():
         "content_block_delta",  # " CHI"}
         "content_block_stop",  # End of third tool_use content block
         "content_block_start",  # "The weather is not so nice today"
+        "content_block_delta",  # "The weather is not so nice today." (trigger chunk)
         "content_block_stop",
         "message_delta",  # Stop reason with merged usage
         "message_stop",  # Final message stop
@@ -296,6 +298,21 @@ def test_anthropic_stream_wrapper_interleaved_tool_calls_and_text():
 
     assert expected_types == chunk_types
 
+    # Verify the content of the text content blocks is preserved across the
+    # transition (previously the trigger chunk was dropped, losing the first
+    # characters of every new content block - see
+    # test_first_chunk_on_block_transition.py for a dedicated regression test).
+    text_deltas = [
+        (chunk.get("delta") or {}).get("text", "")
+        for chunk in chunks
+        if chunk.get("type") == "content_block_delta"
+        and (chunk.get("delta") or {}).get("type") == "text_delta"
+    ]
+    assert text_deltas == [
+        "The weather is nice today.",
+        "The weather is not so nice today.",
+    ]
+
     get_weather_calls = 0
 
     for chunk in chunks: