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
This commit is contained in:
ryan-crabbe-berri 2026-11-07 14:53:07 -08:00 • committed by GitHub
parent b603e84c02
commit fa254b9346
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 138 additions and 4 deletions

View file

@ -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,
)

View file

@ -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:

View file

@ -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 "