fix(streaming): log the mapped exception class when a stream fails after it opens

A stream that fails after it opens handed its failure callbacks the raw
provider exception (e.g. AnthropicError), while the caller got the mapped
InternalServerError inside MidStreamFallbackError. error_information.error_class
and llm_provider in the logged payload therefore disagreed with the non-stream
and stream-open paths. CustomStreamWrapper now maps once and hands that same
exception to the failure callbacks and to the fallback wrapper.

Fixes #49940

Co-authored-by: Sandiyo Christan <sandichrist6@gmail.com>

litellm/litellm_core_utils/streaming_handler.py

@@ -2415,9 +2415,10 @@ class CustomStreamWrapper:
                 error: Final = self._stream_eof_error()
                 eof_traceback: Final = traceback.format_exc()
                 self._record_partial_usage_for_failure()
-                threading.Thread(target=self.logging_obj.failure_handler, args=(error, eof_traceback)).start()
+                public_error: Final = self._public_stream_exception(error)
+                threading.Thread(target=self.logging_obj.failure_handler, args=(public_error, eof_traceback)).start()
                 self.close()
-                self._handle_stream_fallback_error(error)
+                self._handle_stream_fallback_error(error, public_error)
             if self.sent_last_chunk is True:
                 complete_streaming_response: Final = self._build_complete_streaming_response()
                 response = self._new_usage_chunk()
@@ -2509,10 +2510,11 @@ class CustomStreamWrapper:
                 return processed_chunk
         except Exception as e:
             traceback_exception: Final = traceback.format_exc()
+            public_exception: Final = self._public_stream_exception(e)
             # LOG FAILURE - handle streaming failure logging in the _next_ object, remove `handle_failure` once it's deprecated
-            threading.Thread(target=self.logging_obj.failure_handler, args=(e, traceback_exception)).start()
+            threading.Thread(target=self.logging_obj.failure_handler, args=(public_exception, traceback_exception)).start()
             self.close()
-            self._handle_stream_fallback_error(e)
+            self._handle_stream_fallback_error(e, public_exception)
 
     def fetch_sync_stream(self):
         if self.completion_stream is None and self.make_call is not None:
@@ -2629,9 +2631,10 @@ class CustomStreamWrapper:
             traceback_exception = traceback.format_exc()
             ## ADD DEBUG INFORMATION - E.G. LITELLM REQUEST TIMEOUT
             traceback_exception += f"\nLiteLLM Default Request Timeout - {litellm.request_timeout}"
-            self._dispatch_stream_failure(e, traceback_exception)
+            public_exception: Final = self._public_stream_exception(e)
+            self._dispatch_stream_failure(public_exception, traceback_exception)
             await self.aclose()
-            self._handle_stream_fallback_error(e)
+            self._handle_stream_fallback_error(e, public_exception)
         except (httpx.ReadError, httpx.RemoteProtocolError) as e:
             if self.received_finish_reason is None:
                 await self.aclose()
@@ -2732,8 +2735,9 @@ class CustomStreamWrapper:
             return processed_chunk
 
     def _log_stream_failure_and_raise(self, e: Exception) -> NoReturn:
-        self._dispatch_stream_failure(e, traceback.format_exc())
-        self._handle_stream_fallback_error(e)
+        public_exception: Final = self._public_stream_exception(e)
+        self._dispatch_stream_failure(public_exception, traceback.format_exc())
+        self._handle_stream_fallback_error(e, public_exception)
 
     def _dispatch_stream_failure(self, e: Exception, traceback_exception: str) -> None:
         self._record_partial_usage_for_failure()
@@ -2855,11 +2859,33 @@ class CustomStreamWrapper:
             extra_kwargs=with_deployment_api_base(self.logging_obj.model_call_details.get("litellm_params")),
         )
 
-    def _handle_stream_fallback_error(self, e: Exception) -> "NoReturn":
+    def _public_stream_exception(self, e: Exception) -> Exception:
+        """
+        The exception class the caller ends up with for a stream failure.
+
+        Failure callbacks receive this same exception, so the error_class they
+        log is the mapped OpenAI-compatible one and never the provider's
+        internal class (e.g. AnthropicError). Some providers' mappers log a
+        debug diagnostic while mapping, so call this before restoring the
+        consumer's correlation context.
+        """
+        from litellm.exceptions import MidStreamFallbackError
+
+        if isinstance(e, MidStreamFallbackError):
+            return e if e.original_exception is None else e.original_exception
+        if isinstance(e, OpenAIError):
+            return e
+        try:
+            return self._map_stream_exception(e)
+        except Exception as mapping_error:
+            return mapping_error
+
+    def _handle_stream_fallback_error(self, e: Exception, public_exception: Exception | None = None) -> "NoReturn":
         """
         Common error handling for both __next__ and __anext__.
 
-        Maps the raw exception to an OpenAI-compatible type, then decides
+        Maps the raw exception to an OpenAI-compatible type (or reuses the
+        already mapped public_exception the failure callbacks got), then decides
         whether to raise it directly (non-retriable 4xx) or wrap it in
         MidStreamFallbackError so the Router can trigger a fallback.
 
@@ -2897,13 +2923,7 @@ class CustomStreamWrapper:
         # restore the consumer's outer context only after this completes, so
         # that diagnostic log line still carries the failing stream's own
         # trace_id/session_id instead of the consumer's (or an empty one).
-        if isinstance(e, OpenAIError):
-            mapped_exception: Exception = e
-        else:
-            try:
-                mapped_exception = self._map_stream_exception(e)
-            except Exception as mapping_error:
-                mapped_exception = mapping_error
+        mapped_exception: Final = self._public_stream_exception(e) if public_exception is None else public_exception
         self._restore_consumer_correlation_context()
 
         def _normalize_status_code(exc: Exception) -> int | None:

tests/integration/observability/test_anthropic_overloaded_failure_signals.py

@@ -11,7 +11,7 @@ from urllib.parse import unquote
 import httpx
 import pytest
 import yaml
-from integration._support.anthropic_sse import error_body
+from integration._support.anthropic_sse import error_body, error_frame, message_start, stream_reply
 from integration._support.client import (
     JSON_OBJECT,
     Gateway,
@@ -39,6 +39,8 @@ _TRACED_CHAT: Final = "audit-signals-traced-chat"
 _DATADOG_MESSAGES_ANTHROPIC: Final = "audit-signals-datadog-messages-anthropic"
 _DATADOG_MESSAGES_BEDROCK: Final = "audit-signals-datadog-messages-bedrock"
 _DATADOG_CHAT_ANTHROPIC: Final = "audit-signals-datadog-chat-anthropic"
+_DATADOG_MID_STREAM_CHAT: Final = "audit-signals-datadog-mid-stream-chat"
+_MID_STREAM_PROMPT: Final = "signals-mid-stream-"
 _GUARDED_MESSAGES: Final = "audit-signals-guarded-messages"
 _GUARDED_CHAT: Final = "audit-signals-guarded-chat"
 _AKTO_DOWN: Final = "audit-signals-akto-down"
@@ -47,6 +49,7 @@ _LATE_PASS_THROUGH_BACKEND: Final = "claude-late-pass-through"
 _KEEPALIVE_SECONDS: Final = 0.2
 _LATE_ANSWER_SECONDS: Final = 1.0
 _OVERLOADED: Final = Reply(status=529, body=error_body(529, "Overloaded"))
+_OVERLOADED_MID_STREAM: Final = stream_reply((message_start("msg_scripted", _MODEL), error_frame(529, "Overloaded")))
 _SERVER: Final = 2
 
 AuditConfigWriter = Callable[[Path, Mapping[str, JsonValue]], Path]
@@ -66,6 +69,8 @@ def _respond(request: Request) -> Reply:
         return Reply(status=503, body=_BEDROCK_503)
     if json.loads(request.body).get("model") == _LATE_PASS_THROUGH_BACKEND:
         time.sleep(_LATE_ANSWER_SECONDS)
+    if _MID_STREAM_PROMPT in request.body.decode():
+        return _OVERLOADED_MID_STREAM
     return _OVERLOADED
 
 
@@ -92,6 +97,7 @@ def _config(directory: Path, tracing: Path, wire: Wire) -> Path:
         (_DATADOG_MESSAGES_ANTHROPIC, anthropic),
         (_DATADOG_MESSAGES_BEDROCK, bedrock),
         (_DATADOG_CHAT_ANTHROPIC, anthropic),
+        (_DATADOG_MID_STREAM_CHAT, anthropic),
         (_GUARDED_MESSAGES, anthropic),
         (_GUARDED_CHAT, anthropic),
     )
@@ -329,6 +335,56 @@ def test_the_datadog_proxy_failure_log_carries_the_answered_status_and_the_llm_f
         assert _datadog_status_codes(rig, alias, 2) == (("llm", deployment_status), ("proxy", str(answered)))
 
 
+def _datadog_llm_failures(rig: _Rig, aliases: Sequence[str]) -> Mapping[str, Mapping[str, JsonValue]]:
+    batches: Final[list[Request]] = []  # mutable-ok: drain() consumes the queue, later polls must keep earlier batches
+
+    def delivered() -> Mapping[str, Mapping[str, JsonValue]]:
+        batches.extend(rig.datadog.drain())
+        return {
+            alias: object_value(message["error_information"])
+            for message in _datadog_messages(batches)
+            if "user_api_key_dict" not in message
+            for alias in aliases
+            if _keyed_by(message, alias)
+        }
+
+    return eventually(delivered, lambda found: len(found) >= len(aliases), seconds=40, return_last_on_timeout=True)
+
+
+def test_the_datadog_llm_failure_log_of_a_stream_that_fails_after_it_opens_matches_the_one_of_a_stream_that_never_opens(
+    rig: _Rig,
+) -> None:
+    route: Final = "/v1/chat/completions"
+    model: Final = _DATADOG_MID_STREAM_CHAT
+    rig.wire.drain()
+    rig.datadog.drain()
+    never_opened: Final = f"audit-signals-datadog-never-opened-{uuid.uuid4().hex}"
+    failed_open: Final = f"audit-signals-datadog-failed-open-{uuid.uuid4().hex}"
+    with rig.proxy.scenario() as scenario:
+        opening: Final = _send(rig, route, model, key=scenario.key(key_alias=never_opened, models=[model]))
+        assert opening.status_code >= 500, opening.text
+        mid_stream: Final = _send(
+            rig,
+            route,
+            model,
+            key=scenario.key(key_alias=failed_open, models=[model]),
+            extra={"stream": True, "messages": [{"role": "user", "content": f"{_MID_STREAM_PROMPT}{uuid.uuid4().hex}"}]},
+        )
+        assert "Overloaded" in mid_stream.text, mid_stream.text
+        assert len(rig.wire.drain()) == 2
+        failures: Final = _datadog_llm_failures(rig, (never_opened, failed_open))
+    assert set(failures) == {never_opened, failed_open}, json.dumps(failures)
+    control: Final = failures[never_opened]
+    observed: Final = failures[failed_open]
+    assert control["error_class"] == "InternalServerError", json.dumps(control)
+    assert (observed["error_class"], observed["llm_provider"], observed["error_code"]) == (
+        control["error_class"],
+        control["llm_provider"],
+        control["error_code"],
+    ), json.dumps((observed, control))
+    assert "Overloaded" in string_value(observed["error_message"]), json.dumps(observed)
+
+
 @pytest.mark.parametrize(
     ("route", "model", "answered"),
     (

tests/unit/litellm_core_utils/test_streaming_handler.py

@@ -9649,3 +9649,83 @@ def test_chunk_creator_usage_only_chunk_with_dict_usage_keeps_details(
     assert usage.prompt_tokens_details.cached_tokens == 80
     assert usage.completion_tokens_details is not None
     assert usage.completion_tokens_details.reasoning_tokens == 5
+
+
+class _RecordingLogging(Logging):
+    """Logging whose failure callbacks record the exception they were handed."""
+
+    def __init__(self) -> None:
+        super().__init__(
+            model="claude-sonnet-4-5",
+            messages=[{"role": "user", "content": "Hey"}],
+            stream=True,
+            call_type="completion",
+            start_time=datetime.datetime.now(),
+            litellm_call_id="12345",
+            function_id="1245",
+        )
+        self.logged_failures: list[Exception] = []  # mutable-ok: test double accumulating callback arguments
+
+    def failure_handler(self, exception: Exception, traceback_exception: str, *args: object, **kwargs: object) -> None:
+        self.logged_failures.append(exception)
+
+    async def dispatch_attempt_failure_handlers(self, exception: Exception, traceback_exception: str) -> None:
+        self.logged_failures.append(exception)
+
+
+def _assert_failure_callback_got_the_public_exception(logging: _RecordingLogging, raised: MidStreamFallbackError) -> None:
+    deadline: Final = time.monotonic() + 5
+    while not logging.logged_failures and time.monotonic() < deadline:
+        time.sleep(0.01)
+    assert len(logging.logged_failures) == 1, logging.logged_failures
+    logged: Final = logging.logged_failures[0]
+    assert type(logged) is litellm.InternalServerError, logged
+    assert type(logged) is type(raised.original_exception), (logged, raised.original_exception)
+    assert getattr(logged, "llm_provider", None) == "anthropic", logged
+    assert "Overloaded" in str(logged), logged
+
+
+def test_sync_stream_failure_callback_gets_the_public_exception_class_not_the_provider_one():
+    """Regression test for https://github.com/BerriAI/litellm/issues/49940"""
+    from litellm.llms.anthropic.common_utils import AnthropicError
+
+    def overloaded_mid_stream() -> Iterator[str]:
+        raise AnthropicError(status_code=500, message="Overloaded")
+        yield ""  # pyright: ignore[reportUnreachable]  # makes this a generator
+
+    logging: Final = _RecordingLogging()
+    wrapper: Final = CustomStreamWrapper(
+        completion_stream=overloaded_mid_stream(),
+        model="claude-sonnet-4-5",
+        logging_obj=logging,
+        custom_llm_provider="anthropic",
+    )
+
+    with pytest.raises(MidStreamFallbackError) as excinfo:
+        next(wrapper)
+
+    _assert_failure_callback_got_the_public_exception(logging, excinfo.value)
+
+
+@pytest.mark.asyncio
+async def test_async_stream_failure_callback_gets_the_public_exception_class_not_the_provider_one():
+    """Regression test for https://github.com/BerriAI/litellm/issues/49940"""
+    from litellm.llms.anthropic.common_utils import AnthropicError
+
+    async def overloaded_mid_stream() -> AsyncIterator[str]:
+        raise AnthropicError(status_code=500, message="Overloaded")
+        yield ""  # pyright: ignore[reportUnreachable]  # makes this an async generator
+
+    logging: Final = _RecordingLogging()
+    wrapper: Final = CustomStreamWrapper(
+        completion_stream=overloaded_mid_stream(),
+        model="claude-sonnet-4-5",
+        logging_obj=logging,
+        custom_llm_provider="anthropic",
+    )
+
+    with pytest.raises(MidStreamFallbackError) as excinfo:
+        await wrapper.__anext__()
+    await asyncio.sleep(0)
+
+    _assert_failure_callback_got_the_public_exception(logging, excinfo.value)