fix(bedrock): surface a converse-stream 200 that decodes to no events as a 502 instead of an empty turn (#43213)

* fix(bedrock): surface a converse-stream 200 that decodes to no events as a 502 instead of an empty turn

* fix(bedrock): quote the body head only when a stream decoded no events

The leftover-bytes error keeps the byte and event counts, the content type and the request id but no longer quotes the first bytes of a stream that already decoded events, since that head is the start of a healthy stream and can hold model output. The anthropic_messages empty-stream warning no longer prints the request's model name.

---------

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

litellm/litellm_core_utils/litellm_logging.py

@@ -4213,6 +4213,9 @@ class Logging(LiteLLMLoggingBaseClass):
                 json_mode=False,
                 litellm_params={},
             )
+        elif result is None:
+            verbose_logger.warning("LiteLLM: the anthropic_messages stream assembled no response, logging an empty one")
+            return litellm.ModelResponse(model=self.model)
         else:
             from litellm.types.llms.anthropic import AnthropicResponse
 

litellm/llms/bedrock/chat/converse_handler.py

@@ -69,7 +69,9 @@ def make_sync_call(
         completion_stream: Any = MockResponseIterator(model_response=model_response, json_mode=json_mode)
     else:
         decoder: Final = AWSEventStreamDecoder(model=model, json_mode=json_mode)
-        completion_stream = decoder.iter_bytes(response.iter_bytes(chunk_size=stream_chunk_size))
+        completion_stream = decoder.iter_bytes(
+            response.iter_bytes(chunk_size=stream_chunk_size), response_headers=response.headers
+        )
 
     # LOGGING
     logging_obj.post_call(

litellm/llms/bedrock/chat/invoke_handler.py

@@ -1,6 +1,6 @@
 import types
-from collections.abc import AsyncIterator, Iterator
-from typing import Final, cast
+from collections.abc import AsyncIterator, Iterator, Mapping
+from typing import TYPE_CHECKING, Final, cast
 
 import httpx
 from pydantic import TypeAdapter
@@ -51,7 +51,11 @@ from ..common_utils import (
 bedrock_tool_name_mappings: Final[InMemoryCache] = InMemoryCache(max_size_in_memory=50, default_ttl=600)
 from litellm.llms.bedrock.chat.converse_transformation import AmazonConverseConfig
 
+if TYPE_CHECKING:
+    from botocore.eventstream import EventStreamMessage
+
 converse_config: Final = AmazonConverseConfig()
+_STREAM_HEAD_BYTES: Final = 200
 NOVA_INVOKE_STREAM_EVENT_TYPES: Final = (
     "messageStart",
     "contentBlockStart",
@@ -162,6 +166,22 @@ class AmazonCohereChatConfig:
         return optional_params
 
 
+def _stream_decoder(
+    bedrock_invoke_provider: litellm.BEDROCK_INVOKE_PROVIDERS_LITERAL | None,
+    *,
+    model: str,
+    json_mode: bool | None,
+    sync_stream: bool,
+) -> "AWSEventStreamDecoder":
+    if bedrock_invoke_provider == "anthropic":
+        return AmazonAnthropicClaudeStreamDecoder(model=model, sync_stream=sync_stream, json_mode=json_mode)
+    if bedrock_invoke_provider == "deepseek_r1":
+        return AmazonDeepSeekR1StreamDecoder(model=model, sync_stream=sync_stream)
+    if bedrock_invoke_provider == "moonshot":
+        return AmazonOpenAICompatibleStreamDecoder(model=model, sync_stream=sync_stream)
+    return AWSEventStreamDecoder(model=model, json_mode=json_mode)
+
+
 async def make_call(
     client: AsyncHTTPHandler | None,
     api_base: str,
@@ -218,28 +238,13 @@ async def make_call(
             completion_stream: MockResponseIterator | AsyncIterator[GChunk | ModelResponseStream | dict] = (
                 MockResponseIterator(model_response=model_response, json_mode=json_mode)
             )
-        elif bedrock_invoke_provider == "anthropic":
-            decoder: AWSEventStreamDecoder = AmazonAnthropicClaudeStreamDecoder(
-                model=model,
-                sync_stream=False,
-                json_mode=json_mode,
-            )
-            completion_stream = decoder.aiter_bytes(response.aiter_bytes(chunk_size=stream_chunk_size))
-        elif bedrock_invoke_provider == "deepseek_r1":
-            decoder = AmazonDeepSeekR1StreamDecoder(
-                model=model,
-                sync_stream=False,
-            )
-            completion_stream = decoder.aiter_bytes(response.aiter_bytes(chunk_size=stream_chunk_size))
-        elif bedrock_invoke_provider == "moonshot":
-            decoder = AmazonOpenAICompatibleStreamDecoder(
-                model=model,
-                sync_stream=False,
-            )
-            completion_stream = decoder.aiter_bytes(response.aiter_bytes(chunk_size=stream_chunk_size))
         else:
-            decoder = AWSEventStreamDecoder(model=model, json_mode=json_mode)
-            completion_stream = decoder.aiter_bytes(response.aiter_bytes(chunk_size=stream_chunk_size))
+            decoder: Final = _stream_decoder(
+                bedrock_invoke_provider, model=model, json_mode=json_mode, sync_stream=False
+            )
+            completion_stream = decoder.aiter_bytes(
+                response.aiter_bytes(chunk_size=stream_chunk_size), response_headers=response.headers
+            )
 
         # LOGGING
         logging_obj.post_call(
@@ -322,28 +327,13 @@ def make_sync_call(
             completion_stream: MockResponseIterator | Iterator[GChunk | ModelResponseStream | dict] = (
                 MockResponseIterator(model_response=model_response, json_mode=json_mode)
             )
-        elif bedrock_invoke_provider == "anthropic":
-            decoder: AWSEventStreamDecoder = AmazonAnthropicClaudeStreamDecoder(
-                model=model,
-                sync_stream=True,
-                json_mode=json_mode,
-            )
-            completion_stream = decoder.iter_bytes(response.iter_bytes(chunk_size=stream_chunk_size))
-        elif bedrock_invoke_provider == "deepseek_r1":
-            decoder = AmazonDeepSeekR1StreamDecoder(
-                model=model,
-                sync_stream=True,
-            )
-            completion_stream = decoder.iter_bytes(response.iter_bytes(chunk_size=stream_chunk_size))
-        elif bedrock_invoke_provider == "moonshot":
-            decoder = AmazonOpenAICompatibleStreamDecoder(
-                model=model,
-                sync_stream=True,
-            )
-            completion_stream = decoder.iter_bytes(response.iter_bytes(chunk_size=stream_chunk_size))
         else:
-            decoder = AWSEventStreamDecoder(model=model, json_mode=json_mode)
-            completion_stream = decoder.iter_bytes(response.iter_bytes(chunk_size=stream_chunk_size))
+            decoder: Final = _stream_decoder(
+                bedrock_invoke_provider, model=model, json_mode=json_mode, sync_stream=True
+            )
+            completion_stream = decoder.iter_bytes(
+                response.iter_bytes(chunk_size=stream_chunk_size), response_headers=response.headers
+            )
 
         # LOGGING
         logging_obj.post_call(
@@ -370,6 +360,49 @@ def make_sync_call(
         raise BedrockError(status_code=500, message=str(e))
 
 
+def _response_header(response_headers: Mapping[str, str] | None, name: str) -> str | None:
+    return None if response_headers is None else response_headers.get(name)
+
+
+class _EventStreamTally:
+    def __init__(self) -> None:
+        self.bytes_received = 0
+        self.bytes_decoded = 0
+        self.events = 0
+        self.head = b""
+
+    def add_chunk(self, chunk: bytes) -> None:
+        self.bytes_received += len(chunk)
+        if len(self.head) < _STREAM_HEAD_BYTES:
+            self.head = (self.head + chunk)[:_STREAM_HEAD_BYTES]
+
+    def add_event(self, event: "EventStreamMessage") -> None:
+        self.events += 1
+        self.bytes_decoded += event.prelude.total_length
+
+    def undecoded_stream_error(self, response_headers: Mapping[str, str] | None) -> BedrockError | None:
+        undecoded: Final = self.bytes_received - self.bytes_decoded
+        if self.events and not undecoded:
+            return None
+        detail: Final = (
+            f"content-type={_response_header(response_headers, 'content-type')!r}, "
+            f"x-amzn-requestid={_response_header(response_headers, 'x-amzn-requestid')!r}, "
+            f"{self.bytes_received} bytes received"
+        )
+        if not self.events:
+            return BedrockError(
+                status_code=502,
+                message=(
+                    "Bedrock answered the stream with HTTP 200 but its body decoded to no events "
+                    f"({detail}, first bytes={self.head!r})"
+                ),
+            )
+        return BedrockError(
+            status_code=502,
+            message=f"Bedrock stream ended with {undecoded} undecoded bytes after {self.events} events ({detail})",
+        )
+
+
 class AWSEventStreamDecoder:
     def __init__(self, model: str, json_mode: bool | None = False) -> None:
         from botocore.parsers import EventStreamJSONParser
@@ -709,32 +742,48 @@ class AWSEventStreamDecoder:
             tool_use=None,
         )
 
-    def iter_bytes(self, iterator: Iterator[bytes]) -> Iterator[GChunk | ModelResponseStream | dict]:
+    def iter_bytes(
+        self, iterator: Iterator[bytes], *, response_headers: Mapping[str, str] | None = None
+    ) -> Iterator[GChunk | ModelResponseStream | dict]:
         """Given an iterator that yields lines, iterate over it & yield every event encountered"""
         from botocore.eventstream import EventStreamBuffer
 
         event_stream_buffer: Final = EventStreamBuffer()
+        tally: Final = _EventStreamTally()
         for chunk in iterator:
             event_stream_buffer.add_data(chunk)
+            tally.add_chunk(chunk)
             for event in event_stream_buffer:
+                tally.add_event(event)
                 message = self._parse_message_from_event(event)
                 if message:
                     # sse_event = ServerSentEvent(data=message, event="completion")
                     _data = json.loads(message)
                     yield self._chunk_parser(chunk_data=_data)
+        undecoded_stream_error: Final = tally.undecoded_stream_error(response_headers)
+        if undecoded_stream_error is not None:
+            raise undecoded_stream_error
 
-    async def aiter_bytes(self, iterator: AsyncIterator[bytes]) -> AsyncIterator[GChunk | ModelResponseStream | dict]:
+    async def aiter_bytes(
+        self, iterator: AsyncIterator[bytes], *, response_headers: Mapping[str, str] | None = None
+    ) -> AsyncIterator[GChunk | ModelResponseStream | dict]:
         """Given an async iterator that yields lines, iterate over it & yield every event encountered"""
         from botocore.eventstream import EventStreamBuffer
 
         event_stream_buffer: Final = EventStreamBuffer()
+        tally: Final = _EventStreamTally()
         async for chunk in iterator:
             event_stream_buffer.add_data(chunk)
+            tally.add_chunk(chunk)
             for event in event_stream_buffer:
+                tally.add_event(event)
                 message = self._parse_message_from_event(event)
                 if message:
                     _data = json.loads(message)
                     yield self._chunk_parser(chunk_data=_data)
+        undecoded_stream_error: Final = tally.undecoded_stream_error(response_headers)
+        if undecoded_stream_error is not None:
+            raise undecoded_stream_error
 
     def _parse_message_from_event(self, event) -> str | None:
         response_stream_shape: Final = get_bedrock_response_stream_shape()

litellm/llms/bedrock/messages/invoke_transformations/anthropic_claude3_transformation.py

@@ -770,7 +770,9 @@ class AmazonAnthropicClaudeMessagesConfig(
         aws_decoder: Final = AmazonAnthropicClaudeMessagesStreamDecoder(
             model=model,
         )
-        completion_stream: Final = aws_decoder.aiter_bytes(httpx_response.aiter_bytes())
+        completion_stream: Final = aws_decoder.aiter_bytes(
+            httpx_response.aiter_bytes(), response_headers=httpx_response.headers
+        )
         # Convert decoded Bedrock events to Server-Sent Events expected by Anthropic clients.
         return self.bedrock_sse_wrapper(
             completion_stream=completion_stream,

tests/test_litellm/litellm_core_utils/test_litellm_logging.py

@@ -5308,6 +5308,19 @@ def test_handle_anthropic_messages_response_logging_passes_model_response_throug
     assert logging_obj._handle_anthropic_messages_response_logging(result=model_response) is model_response
 
 
+def test_anthropic_messages_logged_response_tolerates_a_stream_that_assembled_nothing():
+    """A /v1/messages stream whose upstream yielded no chunks assembles to None; the spend
+    row must still land under the message id the caller was served instead of crashing."""
+    logging_obj = _anthropic_messages_logging_obj()
+    logging_obj.record_streamed_anthropic_message_id("msg_served")
+
+    result = logging_obj._anthropic_messages_logged_response(result=None)
+
+    assert isinstance(result, ModelResponse)
+    assert result.id == "msg_served"
+    assert result.model == "openai/my-local"
+
+
 def test_handle_anthropic_messages_response_logging_degrades_on_unparseable_responses_payload():
     """If the Responses translation raises (eg. empty output on an incomplete response),
     the row must still land: a minimal ModelResponse with model + usage is returned."""

tests/unit/llms/bedrock/chat/test_invoke_handler.py

@@ -1,5 +1,6 @@
 import base64
 import binascii
+import itertools
 import datetime
 import json
 import struct
@@ -14,10 +15,13 @@ import litellm
 from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
 from litellm.litellm_core_utils.streaming_handler import CustomStreamWrapper
 from litellm.llms.bedrock.chat.invoke_handler import (
+    AmazonOpenAICompatibleStreamDecoder,
     AWSEventStreamDecoder,
     make_call,
     make_sync_call,
 )
+from litellm.exceptions import MidStreamFallbackError
+from litellm.llms.bedrock.common_utils import BedrockError
 from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler, HTTPHandler
 from litellm.types.utils import ModelResponseStream
 
@@ -799,3 +803,125 @@ async def test_moonshot_invoke_async_stream_yields_openai_shaped_chunks(_aws_tes
     )
 
     _assert_moonshot_stream_content([chunk async for chunk in stream])
+
+
+def _truncated_frame() -> bytes:
+    return _bedrock_event_stream_frame(_openai_stream_chunk({"role": "assistant"}))[:-8]
+
+
+def _event_stream_headers() -> httpx.Headers:
+    return httpx.Headers({"content-type": "application/vnd.amazon.eventstream", "x-amzn-RequestId": "req-empty-1"})
+
+
+_UNDECODABLE_STREAM_BODIES: Final = (
+    pytest.param(b"", id="empty"),
+    pytest.param(b"\x00\x00\x00\x05", id="shorter-than-a-prelude"),
+    pytest.param(_truncated_frame(), id="truncated-first-message"),
+)
+
+
+def _assert_no_events_error(error: BedrockError, body: bytes) -> None:
+    assert error.status_code == 502
+    assert "HTTP 200" in error.message
+    assert "decoded to no events" in error.message
+    assert f"{len(body)} bytes received" in error.message
+    assert "application/vnd.amazon.eventstream" in error.message
+    assert "req-empty-1" in error.message
+    assert f"first bytes={body[:200]!r}" in error.message
+
+
+@pytest.mark.parametrize("body", _UNDECODABLE_STREAM_BODIES)
+def test_iter_bytes_raises_when_a_200_body_decodes_to_no_events(body: bytes) -> None:
+    decoder: Final = AWSEventStreamDecoder(model="us.moonshotai.kimi-k3")
+
+    with pytest.raises(BedrockError) as exc_info:
+        list(decoder.iter_bytes(iter([body]), response_headers=_event_stream_headers()))
+
+    _assert_no_events_error(exc_info.value, body)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("body", _UNDECODABLE_STREAM_BODIES)
+async def test_aiter_bytes_raises_when_a_200_body_decodes_to_no_events(body: bytes) -> None:
+    async def _chunks() -> AsyncIterator[bytes]:
+        yield body
+
+    decoder: Final = AWSEventStreamDecoder(model="us.moonshotai.kimi-k3")
+
+    with pytest.raises(BedrockError) as exc_info:
+        _ = [chunk async for chunk in decoder.aiter_bytes(_chunks(), response_headers=_event_stream_headers())]
+
+    _assert_no_events_error(exc_info.value, body)
+
+
+def test_iter_bytes_raises_when_the_stream_ends_mid_message() -> None:
+    decoder: Final = AmazonOpenAICompatibleStreamDecoder(model="moonshot.kimi-k2-thinking", sync_stream=True)
+    stream: Final = decoder.iter_bytes(iter([_MOONSHOT_RAW_STREAM, _truncated_frame()]))
+
+    chunks: Final = list(itertools.islice(stream, 4))
+    with pytest.raises(BedrockError) as exc_info:
+        next(stream)
+
+    _assert_moonshot_stream_content(chunks)
+    assert exc_info.value.status_code == 502
+    assert f"{len(_truncated_frame())} undecoded bytes after 4 events" in exc_info.value.message
+    assert "first bytes=" not in exc_info.value.message
+
+
+def test_iter_bytes_yields_a_complete_stream_without_raising() -> None:
+    decoder: Final = AmazonOpenAICompatibleStreamDecoder(model="moonshot.kimi-k2-thinking", sync_stream=True)
+
+    chunks: Final = list(decoder.iter_bytes(iter([_MOONSHOT_RAW_STREAM[:100], _MOONSHOT_RAW_STREAM[100:]])))
+
+    _assert_moonshot_stream_content(chunks)
+
+
+def _assert_empty_stream_surfaced_as_bad_gateway(error: MidStreamFallbackError) -> None:
+    assert error.status_code == 502
+    assert error.is_pre_first_chunk is True
+    assert isinstance(error.original_exception, litellm.BadGatewayError)
+    assert "decoded to no events" in str(error)
+    assert "req-empty-1" in str(error)
+
+
+def test_converse_stream_with_an_empty_200_body_raises_instead_of_an_empty_turn(_aws_test_credentials: None) -> None:
+    response: Final = MagicMock(status_code=200, headers=_event_stream_headers())
+    response.iter_bytes = lambda chunk_size=None: iter([b""])
+    client: Final = HTTPHandler()
+    client.post = MagicMock(return_value=response)
+
+    with pytest.raises(MidStreamFallbackError) as exc_info:
+        list(
+            litellm.completion(
+                model="bedrock/us.moonshotai.kimi-k3",
+                messages=[{"role": "user", "content": "hi"}],
+                stream=True,
+                client=client,
+            )
+        )
+
+    _assert_empty_stream_surfaced_as_bad_gateway(exc_info.value)
+
+
+@pytest.mark.asyncio
+async def test_async_converse_stream_with_an_empty_200_body_raises_instead_of_an_empty_turn(
+    _aws_test_credentials: None,
+) -> None:
+    async def _aiter_bytes(chunk_size: int | None = None) -> AsyncIterator[bytes]:
+        yield b""
+
+    response: Final = MagicMock(status_code=200, headers=_event_stream_headers())
+    response.aiter_bytes = _aiter_bytes
+    client: Final = AsyncHTTPHandler()
+    client.post = AsyncMock(return_value=response)
+
+    stream: Final = await litellm.acompletion(
+        model="bedrock/us.moonshotai.kimi-k3",
+        messages=[{"role": "user", "content": "hi"}],
+        stream=True,
+        client=client,
+    )
+    with pytest.raises(MidStreamFallbackError) as exc_info:
+        _ = [chunk async for chunk in stream]
+
+    _assert_empty_stream_surfaced_as_bad_gateway(exc_info.value)