From fa254b93467a35204729a1fea078b23d584ba71a Mon Sep 17 00:00:00 2001 From: ryan-crabbe-berri Date: Sat, 7 Nov 2026 14:53:07 -0800 Subject: [PATCH] fix(responses): surface mid-stream upstream errors that arrive without an event type (#50466) * fix(responses): surface mid-stream upstream errors that arrive without an event type Bedrock's native Responses route can end a stream with a bare {"error": {...}} chunk. It parsed as a typeless GenericEvent and crashed the iterator with AttributeError, so clients saw that instead of the upstream error. Such chunks are now read as error events, which raise the mapped upstream error and reach the client as response.failed. * refactor(responses): inline the untyped error chunk check to keep LIT001 at the base count --- litellm/responses/streaming_iterator.py | 11 ++- .../test_bedrock_gpt_responses_native_wire.py | 61 +++++++++++++++- .../test_streaming_iterator_error_events.py | 70 +++++++++++++++++++ 3 files changed, 138 insertions(+), 4 deletions(-) diff --git a/litellm/responses/streaming_iterator.py b/litellm/responses/streaming_iterator.py index 4273647567e..08b2c518632 100644 --- a/litellm/responses/streaming_iterator.py +++ b/litellm/responses/streaming_iterator.py @@ -486,9 +486,18 @@ class BaseResponsesAPIStreamingIterator: if _is_json_object(parsed_chunk): if self.responses_api_provider_config is None: raise ValueError("responses_api_provider_config is required to process live streaming chunks") + typed_chunk: Final = ( + parsed_chunk + if parsed_chunk.get("type") is not None or "error" not in parsed_chunk + else { + "sequence_number": SEQUENCE_NUMBER_STAMPED_ON_EMISSION, + **parsed_chunk, + "type": ResponsesAPIStreamEvents.ERROR.value, + } + ) openai_responses_api_chunk: Final = self.responses_api_provider_config.transform_streaming_response( model=self.model, - parsed_chunk=parsed_chunk, + parsed_chunk=typed_chunk, logging_obj=self.logging_obj, ) diff --git a/tests/integration/providers/test_bedrock_gpt_responses_native_wire.py b/tests/integration/providers/test_bedrock_gpt_responses_native_wire.py index 159097c80ee..a5cba28a48a 100644 --- a/tests/integration/providers/test_bedrock_gpt_responses_native_wire.py +++ b/tests/integration/providers/test_bedrock_gpt_responses_native_wire.py @@ -1,14 +1,15 @@ import base64 +import json import uuid from dataclasses import dataclass from typing import Final import openai -from integration._support.bedrock_runtime_peer import NATIVE_RESPONSES, answer, respond, target_of +from integration._support.bedrock_runtime_peer import NATIVE_RESPONSES, answer, marker_of, respond, target_of from integration._support.client import Gateway, Scenario, eventually from integration._support.database import read_rows -from integration._support.wire import Request, Wire, wire_server -from openai.types.responses import ResponseCompletedEvent, ResponseTextDeltaEvent +from integration._support.wire import Reply, Request, Wire, wire_server +from openai.types.responses import ResponseCompletedEvent, ResponseFailedEvent, ResponseTextDeltaEvent from pydantic import JsonValue, TypeAdapter from litellm.proxy.common_utils.encrypt_decrypt_utils import decrypt_if_encrypted_with @@ -17,6 +18,7 @@ GPT: Final = "us.openai.gpt-5.6-sol" TOKEN: Final = "synthetic-bedrock-bearer" SALT: Final = "sk-integration-salt" _JSON_OBJECT: Final = TypeAdapter(dict[str, JsonValue]) +UPSTREAM_OUTAGE: Final = "The server had an error while processing your request. Sorry about that!" @dataclass(frozen=True, slots=True) @@ -123,6 +125,59 @@ async def test_async_openai_sdk_responses_stream_is_served_by_the_native_respons assert _spend_row(completed.response.id) == _success_row(model) +def _stream_then_bare_error(request: Request) -> Reply: + marker: Final = marker_of(request) + created: Final[dict[str, JsonValue]] = { + "type": "response.created", + "sequence_number": 0, + "response": { + "id": f"resp_upstream_{marker}", + "object": "response", + "created_at": 1, + "status": "in_progress", + "model": GPT, + "output": [], + }, + } + delta: Final[dict[str, JsonValue]] = { + "type": "response.output_text.delta", + "sequence_number": 1, + "item_id": f"msg_{marker}", + "output_index": 0, + "content_index": 0, + "delta": answer(marker), + } + outage: Final[dict[str, JsonValue]] = { + "error": {"message": UPSTREAM_OUTAGE, "type": "server_error", "code": "service_unavailable"} + } + return Reply( + content_type="text/event-stream", + chunks=tuple(f"data: {json.dumps(event)}\n\n".encode() for event in (created, delta, outage)), + ) + + +async def test_mid_stream_upstream_error_without_an_event_type_reaches_the_client_as_response_failed( + gateway: Gateway, +) -> None: + marker: Final = uuid.uuid4().hex + with wire_server(_stream_then_bare_error) as wire, gateway.scenario() as scenario: + model: Final = _deployment(scenario, wire) + client: Final = openai.AsyncOpenAI(base_url=f"{gateway.client.base_url}/v1", api_key=gateway.key, max_retries=0) + stream: Final = await client.responses.create( + model=model, input=_prompt(marker), stream=True, extra_body={"cache": {"no-cache": True}} + ) + events: Final = [event async for event in stream] + assert [event.type for event in events] == [ + "response.created", + "response.output_text.delta", + "response.failed", + ], events + failed: Final = events[-1] + assert isinstance(failed, ResponseFailedEvent), failed + assert failed.response.error is not None and UPSTREAM_OUTAGE in failed.response.error.message, failed + _native_request(wire) + + def test_codex_agent_message_with_encrypted_content_reaches_the_native_responses_route_intact( gateway: Gateway, ) -> None: diff --git a/tests/unit/responses/test_streaming_iterator_error_events.py b/tests/unit/responses/test_streaming_iterator_error_events.py index 0acf5d9ce62..29cb0412422 100644 --- a/tests/unit/responses/test_streaming_iterator_error_events.py +++ b/tests/unit/responses/test_streaming_iterator_error_events.py @@ -959,3 +959,73 @@ async def test_responses_websocket_turn_failure_reports_the_deployment_it_was_se assert failure is not None _assert_names_the_azure_deployment(failure, reported_api_base) + + +_BEDROCK_SERVICE_UNAVAILABLE_CHUNK: Final = { + "error": { + "message": "The server had an error while processing your request. Sorry about that!", + "type": "server_error", + "code": "service_unavailable", + } +} + + +def _real_config_iterator(provider: str, model: str, events: tuple[Mapping[str, object], ...]): + payload: Final = b"".join(b"data: " + json.dumps(event).encode() + b"\n\n" for event in events) + + async def aiter_bytes(): + yield payload + + response: Final = Mock() + response.headers = {} + response.aiter_bytes = aiter_bytes + logging_obj: Final = Mock(spec=LiteLLMLoggingObj) + logging_obj.model_call_details = {"litellm_params": {}} + logging_obj.completion_start_time = None + config: Final = ProviderConfigManager.get_provider_responses_api_config( + provider=litellm.LlmProviders(provider), model=model + ) + assert config is not None + return ResponsesAPIStreamingIterator( + response=response, + model=model, + responses_api_provider_config=config, + logging_obj=logging_obj, + custom_llm_provider=provider, + ) + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("provider", "model"), + [ + ("bedrock", "us.openai.gpt-5.6-sol"), + ("openai", "gpt-5.5"), + ("volcengine", "doubao-seed-1-6"), + ("bigmodel", "glm-5"), + ], +) +async def test_untyped_mid_stream_error_chunk_raises_the_upstream_error(provider: str, model: str): + iterator: Final = _real_config_iterator( + provider, + model, + ( + { + "type": "response.output_text.delta", + "item_id": "msg_1", + "output_index": 0, + "content_index": 0, + "delta": "partial ", + "sequence_number": 1, + }, + _BEDROCK_SERVICE_UNAVAILABLE_CHUNK, + ), + ) + stream: Final = aiter(iterator) + assert (await anext(stream)).type == ResponsesAPIStreamEvents.OUTPUT_TEXT_DELTA + + with pytest.raises(MidStreamFallbackError) as exc_info: + await anext(stream) + assert "The server had an error while processing your request" in str(exc_info.value.original_exception) + assert exc_info.value.status_code >= 500 + assert exc_info.value.generated_content == "partial "