From 4a6e1db6d392b853ebcf42612939fe03e50b4ac8 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Sun, 8 Nov 2026 01:23:27 +0000 Subject: [PATCH] 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 --- .../litellm_core_utils/streaming_handler.py | 54 +++++++++---- ...st_anthropic_overloaded_failure_signals.py | 58 +++++++++++++- .../test_streaming_handler.py | 80 +++++++++++++++++++ 3 files changed, 174 insertions(+), 18 deletions(-) diff --git a/litellm/litellm_core_utils/streaming_handler.py b/litellm/litellm_core_utils/streaming_handler.py index f7bd86f3a0c..9c2d729863f 100644 --- a/litellm/litellm_core_utils/streaming_handler.py +++ b/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: diff --git a/tests/integration/observability/test_anthropic_overloaded_failure_signals.py b/tests/integration/observability/test_anthropic_overloaded_failure_signals.py index e86ed98fae7..4a53f3b8e5c 100644 --- a/tests/integration/observability/test_anthropic_overloaded_failure_signals.py +++ b/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"), ( diff --git a/tests/unit/litellm_core_utils/test_streaming_handler.py b/tests/unit/litellm_core_utils/test_streaming_handler.py index df50ae9dc44..d729674288a 100644 --- a/tests/unit/litellm_core_utils/test_streaming_handler.py +++ b/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)