From a7696a2bc2fe18b18e8617a3231ebc6a01c04e68 Mon Sep 17 00:00:00 2001 From: paveln83 <349044532+paveln83@users.noreply.github.com> Date: Tue, 10 Nov 2026 13:29:32 -0800 Subject: [PATCH] fix(proxy): send the streamed Responses fallback metadata as a typed Responses event (#50804) * fix(proxy): send the streamed Responses fallback metadata as a typed Responses event With expose_fallback_errors_to_caller on and include_fallback_errors set, a streamed /v1/responses call that fell back sent its fallback metadata as a chat.completion.chunk with no type, so the OpenAI SDK read it as an audio delta event with type None. On the Responses route the frame is now {"type": "litellm.fallback_metadata", "litellm_fallback": {...}} and goes through the same Responses framing as every other event, so it gets an event: line whenever its neighbours do. The chat route keeps its chat chunk unchanged * test(proxy): stop pinning the Responses stream terminator in the fallback metadata test Main ends Responses streams without data: [DONE] since #47980, so the test reads the event frames whether or not a terminator follows them, and passes both before and after that change merges * fix(proxy): keep the top-level model on the Responses fallback metadata event * fix(proxy): keep the fallback metadata id on the Responses event and omit an empty model * fix(proxy): leave the top-level model off the Responses fallback event when raw model names are on The complexity router's return_raw_model_name leaves every other event unstamped, so the fallback event now matches. A new test streams a raw upstream frame before the event to cover how main names it. * chore: push again after GitHub dropped the previous push event * fix(proxy): send the Responses fallback event without an event line so older OpenAI SDKs read it --------- Co-authored-by: mateo-berri <277851410+mateo-berri@users.noreply.github.com> --- litellm/proxy/common_request_processing.py | 2 +- litellm/proxy/proxy_server.py | 32 +++-- .../test_include_fallback_errors_wire.py | 39 +++++- .../proxy_server/test_streaming_helpers.py | 123 ++++++++++++++++++ 4 files changed, 184 insertions(+), 12 deletions(-) diff --git a/litellm/proxy/common_request_processing.py b/litellm/proxy/common_request_processing.py index 9e2b49ef692..bfa559f86c5 100644 --- a/litellm/proxy/common_request_processing.py +++ b/litellm/proxy/common_request_processing.py @@ -381,7 +381,7 @@ def resolve_litellm_call_id(client_call_id: str | None) -> str: return str(uuid.uuid4()) -def should_return_raw_model_name(request_data: dict[str, object]) -> bool: +def should_return_raw_model_name(request_data: Mapping[str, object]) -> bool: return any( isinstance(metadata, dict) and metadata.get(RETURN_RAW_MODEL_NAME_METADATA_KEY) is True for metadata in (request_data.get("metadata"), request_data.get("litellm_metadata")) diff --git a/litellm/proxy/proxy_server.py b/litellm/proxy/proxy_server.py index f69d9bd4724..be90b028cf4 100644 --- a/litellm/proxy/proxy_server.py +++ b/litellm/proxy/proxy_server.py @@ -10303,23 +10303,34 @@ def _get_streaming_fallback_metadata( return True, None, fallback_errors +_FALLBACK_METADATA_ID: Final = "litellm-fallback-metadata" +_RESPONSES_FALLBACK_METADATA_EVENT_TYPE: Final = "litellm.fallback_metadata" + + def _format_fallback_metadata_sse_event( *, fallback_model: str | None, fallback_errors: list[dict[str, object]], + responses_stream: bool, + return_raw_model_name: bool, ) -> str: - import time - + litellm_fallback: Final = {"fallback_model": fallback_model, "errors": fallback_errors} + if responses_stream: + stamped_model: Final = None if return_raw_model_name else fallback_model + event: Final = { + "type": _RESPONSES_FALLBACK_METADATA_EVENT_TYPE, + "id": _FALLBACK_METADATA_ID, + **({"model": stamped_model} if stamped_model else {}), + "litellm_fallback": litellm_fallback, + } + return f"data: {json.dumps(event)}\n\n" payload: Final = { - "id": "litellm-fallback-metadata", + "id": _FALLBACK_METADATA_ID, "object": "chat.completion.chunk", "created": int(time.time()), "model": fallback_model or "", "choices": [], - "litellm_fallback": { - "fallback_model": fallback_model, - "errors": fallback_errors, - }, + "litellm_fallback": litellm_fallback, } return f"data: {json.dumps(payload)}\n\n" @@ -10328,7 +10339,7 @@ def _restamp_streaming_chunk_model( *, chunk: object, requested_model_from_client: str, - request_data: dict, + request_data: Mapping[str, object], model_mismatch_logged: bool, fallback_was_attempted: bool = False, fallback_model_from_metadata: str | None = None, @@ -10906,6 +10917,7 @@ async def async_data_generator( model_mismatch_logged = False fallback_metadata_event_sent = False include_fallback_errors: Final = _should_include_fallback_errors(request_data) + return_raw_model_name: Final = should_return_raw_model_name(request_data) # Use a running string instead of list + join to avoid O(n^2) overhead. # Previously "".join(str_so_far_parts) was called every chunk, re-joining # the entire accumulated response. String += is O(n) amortized total. @@ -11003,6 +11015,8 @@ async def async_data_generator( yield _format_fallback_metadata_sse_event( fallback_model=fallback_model_from_metadata, fallback_errors=fallback_errors, + responses_stream=error_state is not None, + return_raw_model_name=return_raw_model_name, ) fallback_metadata_event_sent = True continue @@ -11059,6 +11073,8 @@ async def async_data_generator( yield _format_fallback_metadata_sse_event( fallback_model=fallback_model_from_metadata, fallback_errors=fallback_errors, + responses_stream=error_state is not None, + return_raw_model_name=return_raw_model_name, ) fallback_metadata_event_sent = True diff --git a/tests/integration/routing/test_include_fallback_errors_wire.py b/tests/integration/routing/test_include_fallback_errors_wire.py index 2f48bc43a22..a41fefd5ce1 100644 --- a/tests/integration/routing/test_include_fallback_errors_wire.py +++ b/tests/integration/routing/test_include_fallback_errors_wire.py @@ -16,7 +16,14 @@ from urllib.parse import unquote import httpx import pytest import yaml -from integration._support.anthropic_sse import error_body, message_json, message_stream, parse_sse, stream_reply +from integration._support.anthropic_sse import ( + SseEvent, + error_body, + message_json, + message_stream, + parse_sse, + stream_reply, +) from integration._support.client import Gateway, eventually, gateway_from_environment from integration._support.database import read_rows from integration._support.openai_wire import chat_reply, responses_reply @@ -34,6 +41,7 @@ _API_KEY: Final = "synthetic-fallback-errors-key" _ROUTER_ONLY_KEYS: Final = ("include_fallback_errors", "silent_model") _ANSWER: Final = "answered by" _UNAUTHORIZED_MESSAGE: Final = "scripted 401: the primary key was revoked" +_RESPONSES_FALLBACK_METADATA_EVENT: Final = "litellm.fallback_metadata" _NONCE: Final = re.compile(r"nonce=([0-9a-f]{32})") _JSON: Final = TypeAdapter(dict[str, JsonValue]) _ERRORS: Final = TypeAdapter(list[dict[str, JsonValue]]) @@ -445,11 +453,25 @@ def exposed(tmp_path_factory: pytest.TempPathFactory) -> Iterator[_Exposed]: yield _Exposed(proxy, wire) +def _mention_the_outage(errors: list[dict[str, JsonValue]]) -> bool: + return len(errors) == 1 and _UNAUTHORIZED_MESSAGE in str(errors[0]["message"]) + + def _errors_mention_the_outage(errors_header: str | None) -> bool: if errors_header is None: return False - errors: Final = _ERRORS.validate_json(errors_header) - return len(errors) == 1 and _UNAUTHORIZED_MESSAGE in str(errors[0]["message"]) + return _mention_the_outage(_ERRORS.validate_json(errors_header)) + + +def _fallback_reports(events: tuple[SseEvent, ...]) -> tuple[SseEvent, ...]: + return tuple(event for event in events if "litellm_fallback" in event.data) + + +def _reports_the_outage_served_by_the_backup(report: SseEvent) -> bool: + fallback: Final = _JSON.validate_python(report.data["litellm_fallback"]) + return fallback["fallback_model"] == _EXPOSED_OPENAI_BACKUP and _mention_the_outage( + _ERRORS.validate_python(fallback["errors"]) + ) @pytest.mark.parametrize("stream", _STREAMING) @@ -473,6 +495,10 @@ def test_exposed_chat_fallback_reports_the_errors_without_putting_the_flag_on_th if not stream: assert observed.attempted == "1", observed assert _errors_mention_the_outage(observed.errors_header), observed + return + reports: Final = _fallback_reports(parse_sse(observed.text)) + assert tuple(report.data.get("object") for report in reports) == ("chat.completion.chunk",), observed + assert _reports_the_outage_served_by_the_backup(reports[0]), observed def test_exposed_chat_without_a_fallback_keeps_the_flag_off_the_wire_and_the_errors_header_off( @@ -561,6 +587,13 @@ def test_exposed_responses_fallback_keeps_the_flag_off_the_wire(exposed: _Expose assert observed.served_id == _identity("responses", _EXPOSED_OPENAI_BACKUP, sent.nonce), observed assert observed.hit == (_EXPOSED_OPENAI_PRIMARY, _EXPOSED_OPENAI_BACKUP), observed assert observed.leaks == (), observed + if not stream: + return + events: Final = parse_sse(observed.text) + reports: Final = _fallback_reports(events) + assert tuple(report.data.get("type") for report in reports) == (_RESPONSES_FALLBACK_METADATA_EVENT,), observed + assert _reports_the_outage_served_by_the_backup(reports[0]), observed + assert all("type" in event.data for event in events if event.data), observed @pytest.mark.parametrize("stream", _STREAMING) diff --git a/tests/unit/proxy/proxy_server/test_streaming_helpers.py b/tests/unit/proxy/proxy_server/test_streaming_helpers.py index fdf58ec2fd7..4136a4acb16 100644 --- a/tests/unit/proxy/proxy_server/test_streaming_helpers.py +++ b/tests/unit/proxy/proxy_server/test_streaming_helpers.py @@ -874,6 +874,8 @@ def test_format_fallback_metadata_sse_event(): event = _format_fallback_metadata_sse_event( fallback_model="fallback-model", fallback_errors=fallback_errors, + responses_stream=False, + return_raw_model_name=False, ) assert isinstance(event, str) @@ -1469,6 +1471,127 @@ async def test_async_data_generator_emits_fallback_error_metadata_event(monkeypa assert isinstance(metadata_payload["created"], int) +@pytest.mark.asyncio +async def test_chat_stream_sends_fallback_metadata_in_place_of_a_stripped_usage_artifact( + monkeypatch: pytest.MonkeyPatch, +) -> None: + _patch_logging_flags(monkeypatch) + monkeypatch.setitem(ps.general_settings, "expose_fallback_errors_to_caller", True) + fallback_errors: Final = [ + {"message": "litellm.RateLimitError: upstream limited request", "type": "RateLimitError", "code": "429"} + ] + usage_artifact: Final = ModelResponseStream( + id="chatcmpl-test", choices=[], created=0, model="openai/internal-fallback", object="chat.completion.chunk" + ) + upstream: Final = _FakeStream( + [usage_artifact, _simple_chunk(model="openai/internal-fallback", content="hello")], + hidden_params={ + "additional_headers": { + "x-litellm-attempted-fallbacks": 1, + "x-litellm-model-group": "fallback-model", + "x-litellm-fallback-errors": json.dumps(fallback_errors), + } + }, + ) + + frames: Final = tuple( + [ + frame.decode() if isinstance(frame, bytes) else frame + async for frame in async_data_generator( + response=upstream, + user_api_key_dict=_user_auth(), + request_data={ + "model": "primary-model", + "include_fallback_errors": True, + "_litellm_strip_stream_usage": True, + }, + ) + ] + ) + payloads: Final = tuple(json.loads(frame.removeprefix("data: ")) for frame in frames[:-1]) + + assert frames[-1] == "data: [DONE]\n\n" + assert len(payloads) == 2 + assert payloads[0]["object"] == "chat.completion.chunk" + assert payloads[0]["choices"] == [] + assert payloads[0]["litellm_fallback"] == {"fallback_model": "fallback-model", "errors": fallback_errors} + assert payloads[1]["choices"][0]["delta"]["content"] == "hello" + + +@pytest.mark.asyncio +@pytest.mark.parametrize("fallback_model", ["fallback-model", None]) +@pytest.mark.parametrize("return_raw_model_name", [False, True]) +async def test_responses_stream_sends_fallback_metadata_as_a_typed_event_without_an_event_line( + monkeypatch: pytest.MonkeyPatch, + fallback_model: str | None, + return_raw_model_name: bool, +) -> None: + _patch_logging_flags(monkeypatch) + monkeypatch.setitem(ps.general_settings, "expose_fallback_errors_to_caller", True) + fallback_errors: Final = [ + { + "message": "litellm.AuthenticationError: Incorrect API key provided", + "type": "AuthenticationError", + "param": None, + "code": "401", + } + ] + response: Final = ResponsesAPIResponse(id="resp_fallback", created_at=1, model="fallback-model", output=[]) + upstream: Final = _FakeStream( + [ + ResponseCreatedEvent.model_validate( + {"type": "response.created", "sequence_number": 0, "response": response} + ), + ResponseCompletedEvent.model_validate( + {"type": "response.completed", "sequence_number": 1, "response": response} + ), + ], + hidden_params={ + "additional_headers": { + "x-litellm-attempted-fallbacks": 1, + **({"x-litellm-model-group": fallback_model} if fallback_model else {}), + "x-litellm-fallback-errors": json.dumps(fallback_errors), + } + }, + ) + + frames: Final = tuple( + [ + frame.decode() if isinstance(frame, bytes) else frame + async for frame in async_data_generator( + response=upstream, + user_api_key_dict=_user_auth(), + request_data={ + "model": "primary-model", + "include_fallback_errors": True, + "metadata": {RETURN_RAW_MODEL_NAME_METADATA_KEY: return_raw_model_name}, + }, + responses_stream_errors=True, + ) + ] + ) + event_frames: Final = tuple(frame for frame in frames if frame != "data: [DONE]\n\n") + payloads: Final = tuple( + json.loads(next(line.removeprefix("data: ") for line in frame.splitlines() if line.startswith("data: "))) + for frame in event_frames + ) + + assert tuple(payload.get("type") for payload in payloads) == ( + "response.created", + "litellm.fallback_metadata", + "response.completed", + ) + stamped_model: Final = None if return_raw_model_name else fallback_model + assert payloads[1] == { + "type": "litellm.fallback_metadata", + "id": "litellm-fallback-metadata", + **({"model": stamped_model} if stamped_model else {}), + "litellm_fallback": {"fallback_model": fallback_model, "errors": fallback_errors}, + } + assert {payload.get("model") for payload in payloads} == {stamped_model} + assert event_frames[1] == f"data: {json.dumps(payloads[1])}\n\n" + + @pytest.mark.asyncio async def test_async_data_generator_skips_fallback_error_event_without_opt_in( monkeypatch,