fix(responses): backport mid-stream content_policy_violation fallback routing to stable/1.100.x

litellm/responses/streaming_iterator.py

@@ -176,6 +176,13 @@ def _status_code_for_error_fields(error_type: str | None, error_code: str | None
     )
 
 
+def _mid_stream_fallback_eligible(mapped_exception: Exception) -> bool:
+    if isinstance(mapped_exception, litellm.ContentPolicyViolationError):
+        return True
+    status_code: Final = getattr(mapped_exception, "status_code", None)
+    return not isinstance(status_code, int) or status_code >= 500 or status_code == 429
+
+
 class BaseResponsesAPIStreamingIterator:
     """
     Base class for streaming iterators that process responses from the Responses API.
@@ -464,15 +471,8 @@ def _handle_logging_failed_response(self):
             getattr(self.completed_response, "response", None) if self.completed_response else None
         )
         error_info: Final = getattr(response_obj, "error", None) if response_obj else None
-        error_message, error_type, error_code = _error_event_fields(error_info)
         self._record_failed_response_usage(response_obj)
-        exception: Final = litellm.APIError(
-            status_code=_status_code_for_error_fields(error_type, error_code),
-            message=error_message,
-            llm_provider=self.custom_llm_provider or "",
-            model=self.model or "",
-        )
-        self._handle_failure(exception)
+        self._handle_failure(self._map_error_event_exception(error_info))
 
     def _record_failed_response_usage(self, response_obj: ResponsesAPIResponse | None) -> None:
         if response_obj is None or self.logging_obj is None:
@@ -494,6 +494,28 @@ def _record_failed_response_usage(self, response_obj: ResponsesAPIResponse | Non
             self.logging_obj._response_cost_calculator(result=response_obj) or 0.0
         )
 
+    def _map_error_event_exception(self, error_obj: object) -> Exception:
+        from litellm.llms.base_llm.chat.transformation import BaseLLMException
+
+        error_message, error_type, error_code = _error_event_fields(error_obj)
+        status_code: Final = _status_code_for_error_fields(error_type, error_code)
+        error_body: Final = {"message": error_message, "type": error_type, "code": error_code}
+        provider_exception: Final = BaseLLMException(
+            status_code=status_code,
+            message=f"Error code: {status_code} - {{'error': {error_body}}}",
+            body=error_body,
+        )
+        try:
+            return litellm.exception_type(
+                model=self.model or "",
+                custom_llm_provider=self.custom_llm_provider or "",
+                original_exception=provider_exception,
+                completion_kwargs={},
+                extra_kwargs={},
+            )
+        except Exception as mapped_exception:
+            return mapped_exception
+
     def _maybe_raise_for_error_event(self, result: object) -> None:
         chunk_type: Final = getattr(result, "type", None)
         if chunk_type not in ("error", "response.failed"):
@@ -505,15 +527,8 @@ def _maybe_raise_for_error_event(self, result: object) -> None:
             else getattr(result, "error", None)
         )
 
-        error_message, error_type, error_code = _error_event_fields(error_obj)
-        status_code: Final = _status_code_for_error_fields(error_type, error_code)
-        mapped_exception: Final = litellm.APIError(
-            status_code=status_code,
-            message=error_message,
-            llm_provider=self.custom_llm_provider or "",
-            model=self.model or "",
-        )
-        if 400 <= status_code < 500 and status_code != 429:
+        mapped_exception: Final = self._map_error_event_exception(error_obj)
+        if not _mid_stream_fallback_eligible(mapped_exception):
             raise mapped_exception
         raise MidStreamFallbackError(
             message=str(mapped_exception),

litellm/router.py

@@ -2926,8 +2926,15 @@ async def stream_with_fallbacks():
                         kwargs=initial_kwargs,
                         metadata_variable_name="litellm_metadata",
                     )
+                    # The content-policy dispatch branch matches on the trigger's own type, so a refusal's
+                    # MidStreamFallbackError envelope is unwrapped here or the wrong fallback list is consulted.
+                    fallback_trigger: Final[Exception] = (
+                        e.original_exception
+                        if isinstance(e.original_exception, litellm.ContentPolicyViolationError)
+                        else e
+                    )
                     fallback_response = await self.async_function_with_fallbacks_common_utils(
-                        e=e,
+                        e=fallback_trigger,
                         disable_fallbacks=False,
                         fallbacks=fallbacks,
                         context_window_fallbacks=context_window_fallbacks,

tests/llm_responses_api_testing/test_openai_responses_api.py

@@ -1627,9 +1627,10 @@ async def test_openai_responses_api_token_limit_error():
 
     Parsing the in-stream ErrorEvent must not raise
     "pydantic_core._pydantic_core.ValidationError: 3 validation errors for ErrorEvent".
-    The iterator now surfaces the event as litellm.APIError with status 400
-    (invalid_request_error is a non-retriable client error, so no
-    MidStreamFallbackError wrapping) carrying the provider's message.
+    The iterator routes the event through litellm.exception_type, so it surfaces as
+    the typed 400 client error the non-streaming path raises (litellm.BadRequestError)
+    carrying the provider's message. invalid_request_error is a non-retriable client
+    error, so there is no MidStreamFallbackError wrapping.
     """
     litellm._turn_on_debug()
 
@@ -1644,7 +1645,7 @@ async def _drain():
         async for event in response:
             print(event)
 
-    with pytest.raises(litellm.APIError) as exc_info:
+    with pytest.raises(litellm.BadRequestError) as exc_info:
         await _drain()
 
     assert exc_info.value.status_code == 400

tests/router_unit_tests/test_router_aresponses_streaming_fallback.py

@@ -372,7 +372,7 @@ async def __anext__(self):
     raised = mock_fallback.await_args.kwargs["e"]
     assert isinstance(raised, MidStreamFallbackError)
     assert raised.status_code == 429
-    assert isinstance(raised.original_exception, litellm.APIError)
+    assert isinstance(raised.original_exception, litellm.RateLimitError)
     assert raised.original_exception.status_code == 429
     assert mock_fallback.await_args.kwargs["kwargs"]["input"] == "original question"
 

tests/test_litellm/responses/test_streaming_iterator_error_events.py

@@ -1,9 +1,14 @@
 """
 Regression: in-stream error events (type="error", type="response.failed") must
 raise instead of being returned as benign chunks, mirroring chat streaming
-semantics (_handle_stream_fallback_error): non-retriable 4xx (except 429)
-raise litellm.APIError directly; 429 and 5xx are wrapped in
-MidStreamFallbackError so the Router's mid-stream fallback machinery fires.
+semantics (_handle_stream_fallback_error). The event's code, type and status go
+through litellm.exception_type, so each event raises the same typed exception
+the non-streaming path raises for that provider error: non-retriable 4xx
+(except 429) raise that typed exception directly, so a context-length event
+surfaces as ContextWindowExceededError(400) with no MidStreamFallbackError
+wrapping, while 429, 5xx and ContentPolicyViolationError are wrapped in
+MidStreamFallbackError so the Router's mid-stream fallback machinery fires and
+its content_policy_fallbacks dispatch sees the trigger it matches on.
 
 Status mapping must consider both the OpenAI error `type` (e.g.
 "invalid_request_error") and `code` (e.g. "invalid_prompt",
@@ -15,6 +20,7 @@
 """
 
 import json
+from importlib import import_module
 from unittest.mock import Mock, patch
 
 import pytest
@@ -65,28 +71,28 @@ def test_maybe_raise_for_error_event_wraps_unknown_error_in_mid_stream_fallback(
     with pytest.raises(MidStreamFallbackError) as exc_info:
         iterator._maybe_raise_for_error_event(chunk)
     assert exc_info.value.status_code == 500
-    assert isinstance(exc_info.value.original_exception, litellm.APIError)
+    assert isinstance(exc_info.value.original_exception, litellm.InternalServerError)
     assert exc_info.value.original_exception.status_code == 500
 
 
 def test_maybe_raise_for_error_event_maps_rate_limit_code_to_429_mid_stream_fallback():
-    """429 is retriable: it must be wrapped so the Router can fall back, carrying the mapped APIError."""
+    """429 is retriable: it must be wrapped so the Router can fall back, carrying the mapped RateLimitError."""
     iterator = _make_iterator()
     chunk = _make_error_chunk("tokens", "rate_limit_exceeded", "Too many requests")
     with pytest.raises(MidStreamFallbackError) as exc_info:
         iterator._maybe_raise_for_error_event(chunk)
     assert exc_info.value.status_code == 429
     assert exc_info.value.generated_content == ""
     assert exc_info.value.is_pre_first_chunk is True
-    assert isinstance(exc_info.value.original_exception, litellm.APIError)
+    assert isinstance(exc_info.value.original_exception, litellm.RateLimitError)
     assert exc_info.value.original_exception.status_code == 429
 
 
 def test_maybe_raise_for_error_event_maps_invalid_request_type_to_400():
-    """Client errors classified via the `type` field must raise APIError directly (no fallback)."""
+    """Client errors classified via the `type` field must raise BadRequestError directly (no fallback)."""
     iterator = _make_iterator()
     chunk = _make_error_chunk("invalid_request_error", "invalid_prompt", "bad request")
-    with pytest.raises(litellm.APIError) as exc_info:
+    with pytest.raises(litellm.BadRequestError) as exc_info:
         iterator._maybe_raise_for_error_event(chunk)
     assert exc_info.value.status_code == 400
     assert not isinstance(exc_info.value, MidStreamFallbackError)
@@ -98,12 +104,86 @@ def test_maybe_raise_for_error_event_maps_context_length_code_to_400():
     chunk = Mock()
     chunk.type = "error"
     chunk.error = {"code": "context_length_exceeded", "message": "too long"}
-    with pytest.raises(litellm.APIError) as exc_info:
+    with pytest.raises(litellm.BadRequestError) as exc_info:
         iterator._maybe_raise_for_error_event(chunk)
     assert exc_info.value.status_code == 400
     assert not isinstance(exc_info.value, MidStreamFallbackError)
 
 
+def test_maybe_raise_for_error_event_raises_context_window_exceeded_directly():
+    """A context-length error event maps to ContextWindowExceededError exactly like the non-streaming
+    path and, being a non-retriable client error, is raised directly rather than wrapped for mid-stream
+    fallback, preserving the direct-SDK 400 contract from issue #15785."""
+    iterator = _make_iterator()
+    chunk = _make_error_chunk(
+        "invalid_request_error",
+        "context_length_exceeded",
+        "This model's maximum context length is 128000 tokens. However, your messages resulted in 130000 tokens.",
+    )
+    with pytest.raises(litellm.ContextWindowExceededError) as exc_info:
+        iterator._maybe_raise_for_error_event(chunk)
+    assert exc_info.value.status_code == 400
+    assert not isinstance(exc_info.value, MidStreamFallbackError)
+    assert "maximum context length" in str(exc_info.value)
+
+
+CONTENT_POLICY_MESSAGE = "This content was flagged for possible cybersecurity risk. The response was halted mid-stream."
+
+
+@pytest.mark.parametrize("custom_llm_provider", ["openai", "azure"])
+def test_maybe_raise_for_error_event_wraps_content_policy_violation_for_content_policy_fallbacks(
+    custom_llm_provider: str,
+):
+    """Regression: a content_policy_violation error event used to raise a bare APIError, so the Router's
+    content_policy_fallbacks never fired. It must map to ContentPolicyViolationError (the same exception the
+    non-streaming path raises) and be wrapped so the Router's mid-stream fallback catches it."""
+    iterator = _make_iterator()
+    iterator.custom_llm_provider = custom_llm_provider
+    chunk = _make_error_chunk("invalid_request_error", "content_policy_violation", CONTENT_POLICY_MESSAGE)
+    with pytest.raises(MidStreamFallbackError) as exc_info:
+        iterator._maybe_raise_for_error_event(chunk)
+    assert isinstance(exc_info.value.original_exception, litellm.ContentPolicyViolationError)
+    assert exc_info.value.original_exception.status_code == 400
+    assert exc_info.value.status_code == 400
+    assert exc_info.value.is_pre_first_chunk is True
+    assert CONTENT_POLICY_MESSAGE in str(exc_info.value.original_exception)
+
+
+def test_maybe_raise_for_response_failed_event_wraps_content_policy_violation():
+    iterator = _make_iterator()
+    chunk = _make_failed_chunk(
+        {"type": "invalid_request_error", "code": "content_policy_violation", "message": CONTENT_POLICY_MESSAGE}
+    )
+    with pytest.raises(MidStreamFallbackError) as exc_info:
+        iterator._maybe_raise_for_error_event(chunk)
+    assert isinstance(exc_info.value.original_exception, litellm.ContentPolicyViolationError)
+
+
+@pytest.mark.parametrize(
+    "error_type,error_code,expected_exception",
+    [
+        ("invalid_request_error", "content_policy_violation", litellm.ContentPolicyViolationError),
+        ("tokens", "rate_limit_exceeded", litellm.RateLimitError),
+        ("invalid_request_error", "insufficient_quota", litellm.RateLimitError),
+        ("server_error", "internal_error", litellm.InternalServerError),
+        ("invalid_request_error", "invalid_prompt", litellm.BadRequestError),
+        ("invalid_request_error", "model_not_found", litellm.NotFoundError),
+        ("server_error", "vector_store_timeout", litellm.Timeout),
+    ],
+)
+def test_error_event_raises_the_same_typed_exception_as_the_non_streaming_path(
+    error_type: str, error_code: str, expected_exception: type[Exception]
+):
+    iterator = _make_iterator()
+    chunk = _make_error_chunk(error_type, error_code, "provider message")
+    with pytest.raises((MidStreamFallbackError, expected_exception)) as exc_info:
+        iterator._maybe_raise_for_error_event(chunk)
+    raised = exc_info.value
+    typed_exception = raised.original_exception if isinstance(raised, MidStreamFallbackError) else raised
+    assert type(typed_exception) is expected_exception
+    assert "provider message" in str(typed_exception)
+
+
 def test_maybe_raise_for_error_event_maps_insufficient_quota_to_429():
     """OpenAI returns HTTP 429 for insufficient_quota; it must not map to 400 even though its type
     is invalid_request_error-adjacent, and it must be wrapped for fallback."""
@@ -112,6 +192,7 @@ def test_maybe_raise_for_error_event_maps_insufficient_quota_to_429():
     with pytest.raises(MidStreamFallbackError) as exc_info:
         iterator._maybe_raise_for_error_event(chunk)
     assert exc_info.value.status_code == 429
+    assert isinstance(exc_info.value.original_exception, litellm.RateLimitError)
 
 
 def test_maybe_raise_for_error_event_passes_through_normal_chunk():
@@ -185,10 +266,40 @@ async def test_async_iterator_raises_mid_stream_fallback_on_rate_limit_error_eve
     assert exc_info.value.status_code == 429
     assert exc_info.value.is_pre_first_chunk is True
     assert exc_info.value.generated_content == ""
-    assert isinstance(exc_info.value.original_exception, litellm.APIError)
+    assert isinstance(exc_info.value.original_exception, litellm.RateLimitError)
     assert exc_info.value.original_exception.status_code == 429
 
 
+@pytest.mark.asyncio
+async def test_async_iterator_content_policy_violation_after_first_chunk_carries_generated_content():
+    """The customer's case: text streams, then the provider halts the stream with a
+    content_policy_violation error event. The iterator must surface ContentPolicyViolationError
+    inside MidStreamFallbackError, together with the text already streamed."""
+    iterator = _make_async_iterator_with_events(
+        [
+            {"type": "response.output_text.delta", "delta": "partial "},
+            {
+                "type": "error",
+                "error": {
+                    "type": "invalid_request_error",
+                    "code": "content_policy_violation",
+                    "message": CONTENT_POLICY_MESSAGE,
+                },
+            },
+        ]
+    )
+
+    stream = aiter(iterator)
+    first_chunk = await anext(stream)
+    assert first_chunk is not None
+
+    with pytest.raises(MidStreamFallbackError) as exc_info:
+        await anext(stream)
+    assert isinstance(exc_info.value.original_exception, litellm.ContentPolicyViolationError)
+    assert exc_info.value.is_pre_first_chunk is False
+    assert exc_info.value.generated_content == "partial "
+
+
 @pytest.mark.asyncio
 async def test_async_iterator_error_after_first_chunk_carries_generated_content():
     """An error after streamed output must expose the accumulated text so the router's
@@ -204,14 +315,13 @@ async def test_async_iterator_error_after_first_chunk_carries_generated_content(
         ]
     )
 
-    chunks = []
-    async def _drain():
-        async for chunk in iterator:
-            chunks.append(chunk)
+    stream = aiter(iterator)
+    first_chunk = await anext(stream)
+    second_chunk = await anext(stream)
+    assert first_chunk is not None and second_chunk is not None
 
     with pytest.raises(MidStreamFallbackError) as exc_info:
-        await _drain()
-    assert len(chunks) == 2
+        await anext(stream)
     assert exc_info.value.status_code == 500
     assert exc_info.value.is_pre_first_chunk is False
     assert exc_info.value.generated_content == "hello world"
@@ -264,7 +374,7 @@ def test_handle_logging_failed_response_maps_rate_limit_to_429():
     ):
         iterator._handle_logging_failed_response()
     logged_exception = mock_run_async.call_args.kwargs["exception"]
-    assert isinstance(logged_exception, litellm.APIError)
+    assert isinstance(logged_exception, litellm.RateLimitError)
     assert logged_exception.status_code == 429
     assert "throttled" in str(logged_exception)
 
@@ -281,8 +391,26 @@ def test_handle_logging_failed_response_maps_type_field_to_400():
     ):
         iterator._handle_logging_failed_response()
     logged_exception = mock_run_async.call_args.kwargs["exception"]
-    assert isinstance(logged_exception, litellm.APIError)
+    assert isinstance(logged_exception, litellm.BadRequestError)
+    assert logged_exception.status_code == 400
+
+
+def test_handle_logging_failed_response_logs_content_policy_violation():
+    """Failure logging must record the same typed exception the stream raises, so logging
+    integrations see a content policy violation instead of a generic APIError."""
+    iterator = _make_iterator()
+    iterator.completed_response = _make_failed_chunk(
+        {"type": "invalid_request_error", "code": "content_policy_violation", "message": CONTENT_POLICY_MESSAGE}
+    )
+    with (
+        patch.object(import_module("litellm.responses.streaming_iterator"), "run_async_function") as mock_run_async,
+        patch.object(import_module("litellm.responses.streaming_iterator"), "executor"),
+    ):
+        iterator._handle_logging_failed_response()
+    logged_exception = mock_run_async.call_args.kwargs["exception"]
+    assert isinstance(logged_exception, litellm.ContentPolicyViolationError)
     assert logged_exception.status_code == 400
+    assert CONTENT_POLICY_MESSAGE in str(logged_exception)
 
 
 def test_handle_logging_failed_response_records_usage_and_cost():
@@ -356,7 +484,7 @@ def test_sync_iterator_raises_mid_stream_fallback_on_rate_limit_error_event():
         for _ in iterator:
             pass
     assert exc_info.value.status_code == 429
-    assert isinstance(exc_info.value.original_exception, litellm.APIError)
+    assert isinstance(exc_info.value.original_exception, litellm.RateLimitError)
 
 
 def test_every_openai_sdk_response_error_code_has_explicit_status_mapping():
@@ -412,7 +540,7 @@ def test_maybe_raise_for_response_failed_event_maps_image_code_to_400():
     chunk = Mock()
     chunk.type = "response.failed"
     chunk.response = mock_response_obj
-    with pytest.raises(litellm.APIError) as exc_info:
+    with pytest.raises(litellm.BadRequestError) as exc_info:
         iterator._maybe_raise_for_error_event(chunk)
     assert exc_info.value.status_code == 400
     assert not isinstance(exc_info.value, MidStreamFallbackError)