From b6fcd03848d9245b415a7fb5a92ce82d7770da7b Mon Sep 17 00:00:00 2001 From: "devin-ai-integration[bot]" <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Fri, 25 Sep 2026 15:18:03 -0700 Subject: [PATCH] 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 | 3 + litellm/llms/bedrock/chat/converse_handler.py | 4 +- litellm/llms/bedrock/chat/invoke_handler.py | 141 ++++++++++++------ .../anthropic_claude3_transformation.py | 4 +- .../test_litellm_logging.py | 13 ++ .../llms/bedrock/chat/test_invoke_handler.py | 126 ++++++++++++++++ 6 files changed, 243 insertions(+), 48 deletions(-) diff --git a/litellm/litellm_core_utils/litellm_logging.py b/litellm/litellm_core_utils/litellm_logging.py index d8182140a17..54ed9171dc8 100644 --- a/litellm/litellm_core_utils/litellm_logging.py +++ b/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 diff --git a/litellm/llms/bedrock/chat/converse_handler.py b/litellm/llms/bedrock/chat/converse_handler.py index bd358805743..65a34f72167 100644 --- a/litellm/llms/bedrock/chat/converse_handler.py +++ b/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( diff --git a/litellm/llms/bedrock/chat/invoke_handler.py b/litellm/llms/bedrock/chat/invoke_handler.py index c7b4018b80b..93804e20041 100644 --- a/litellm/llms/bedrock/chat/invoke_handler.py +++ b/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() diff --git a/litellm/llms/bedrock/messages/invoke_transformations/anthropic_claude3_transformation.py b/litellm/llms/bedrock/messages/invoke_transformations/anthropic_claude3_transformation.py index 14bd2bee6cf..cefc8afed25 100644 --- a/litellm/llms/bedrock/messages/invoke_transformations/anthropic_claude3_transformation.py +++ b/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, diff --git a/tests/test_litellm/litellm_core_utils/test_litellm_logging.py b/tests/test_litellm/litellm_core_utils/test_litellm_logging.py index c4829ced9f3..166eeb53f5f 100644 --- a/tests/test_litellm/litellm_core_utils/test_litellm_logging.py +++ b/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.""" diff --git a/tests/unit/llms/bedrock/chat/test_invoke_handler.py b/tests/unit/llms/bedrock/chat/test_invoke_handler.py index 466e9b4fda8..ed8b7023977 100644 --- a/tests/unit/llms/bedrock/chat/test_invoke_handler.py +++ b/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)