diff --git a/litellm/litellm_core_utils/litellm_logging.py b/litellm/litellm_core_utils/litellm_logging.py index fcc521cf6cf..a425603d921 100644 --- a/litellm/litellm_core_utils/litellm_logging.py +++ b/litellm/litellm_core_utils/litellm_logging.py @@ -7394,6 +7394,15 @@ class StandardLoggingPayloadSetup: # ProxyException uses .code, LiteLLM exceptions use .status_code, # httpx.HTTPStatusError exposes status only as .response.status_code. # Stringified for Prisma JSON compatibility. + # If wrapped in MidStreamFallbackError (or similar wrapper carrying .original_exception), + # unwrap to the carried provider/mapped exception so error_class, status_code, and provider + # reflect the underlying failure rather than the internal wrapper. + current_exc: Exception | None = original_exception + while getattr(current_exc, "original_exception", None) is not None: + current_exc = getattr(current_exc, "original_exception") + if current_exc is not None: + original_exception = current_exc + error_code_attr: Final[object] = getattr(original_exception, "code", None) if error_code_attr is not None and str(error_code_attr) not in ("", "None"): error_status: str = str(error_code_attr) diff --git a/litellm/litellm_core_utils/streaming_handler.py b/litellm/litellm_core_utils/streaming_handler.py index 6839e11f8c7..00bd325bee9 100644 --- a/litellm/litellm_core_utils/streaming_handler.py +++ b/litellm/litellm_core_utils/streaming_handler.py @@ -2508,10 +2508,11 @@ class CustomStreamWrapper: return processed_chunk except Exception as e: traceback_exception: Final = traceback.format_exc() + mapped_exception: Exception = self._get_mapped_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=(mapped_exception, traceback_exception)).start() self.close() - self._handle_stream_fallback_error(e) + self._handle_stream_fallback_error(e, mapped_exception=mapped_exception) def fetch_sync_stream(self): if self.completion_stream is None and self.make_call is not None: @@ -2628,9 +2629,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) + mapped_exception = self._get_mapped_stream_exception(e) + self._dispatch_stream_failure(mapped_exception, traceback_exception) await self.aclose() - self._handle_stream_fallback_error(e) + self._handle_stream_fallback_error(e, mapped_exception=mapped_exception) except (httpx.ReadError, httpx.RemoteProtocolError) as e: if self.received_finish_reason is None: await self.aclose() @@ -2731,8 +2733,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) + mapped_exception = self._get_mapped_stream_exception(e) + self._dispatch_stream_failure(mapped_exception, traceback.format_exc()) + self._handle_stream_fallback_error(e, mapped_exception=mapped_exception) def _dispatch_stream_failure(self, e: Exception, traceback_exception: str) -> None: self._record_partial_usage_for_failure() @@ -2854,7 +2857,19 @@ 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 _get_mapped_stream_exception(self, e: Exception) -> Exception: + from litellm.exceptions import MidStreamFallbackError + + if isinstance(e, MidStreamFallbackError): + return e.original_exception if e.original_exception is not None else e + 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, mapped_exception: Exception | None = None) -> "NoReturn": """ Common error handling for both __next__ and __anext__. @@ -2890,19 +2905,20 @@ class CustomStreamWrapper: emitted_disqualifying_content=self._emitted_disqualifying_content or e.emitted_disqualifying_content, ) - # Map to OpenAI exception format. Some providers' mappers (e.g. + # Map to OpenAI exception format if not already mapped. Some providers' mappers (e.g. # _map_anthropic_exception, _map_aleph_alpha_exception) synchronously # log a debug diagnostic (the raw status code) as part of mapping - # 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 + if mapped_exception is None: + if isinstance(e, OpenAIError): + mapped_exception = e + else: + try: + mapped_exception = self._map_stream_exception(e) + except Exception as mapping_error: + mapped_exception = mapping_error self._restore_consumer_correlation_context() def _normalize_status_code(exc: Exception) -> int | None: diff --git a/tests/unit/test_stream_failure_logging_error_class.py b/tests/unit/test_stream_failure_logging_error_class.py new file mode 100644 index 00000000000..22eba363397 --- /dev/null +++ b/tests/unit/test_stream_failure_logging_error_class.py @@ -0,0 +1,134 @@ +import asyncio +import threading +import time +from http.server import BaseHTTPRequestHandler, HTTPServer + +import pytest + +import litellm +from litellm.exceptions import InternalServerError, MidStreamFallbackError +from litellm.integrations.custom_logger import CustomLogger +from litellm.litellm_core_utils.litellm_logging import StandardLoggingPayloadSetup + + +class _MockSSEHandler(BaseHTTPRequestHandler): + def do_POST(self): + self.rfile.read(int(self.headers.get("content-length", 0))) + self.send_response(200) + self.send_header("content-type", "text/event-stream") + self.end_headers() + sse_payload = ( + 'event: error\ndata: {"type": "error", "error": {"type": "overloaded_error", "message": "Overloaded"}}\n\n' + ) + self.wfile.write(sse_payload.encode()) + + def log_message(self, *args): + pass + + +@pytest.fixture +def mock_sse_server(): + server = HTTPServer(("127.0.0.1", 0), _MockSSEHandler) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + yield server + server.shutdown() + + +def test_get_error_information_unwraps_mid_stream_fallback_error(): + inner = InternalServerError( + message="Anthropic overloaded", + response=None, + llm_provider="anthropic", + model="claude-sonnet-4-5", + ) + wrapper = MidStreamFallbackError( + message="MidStreamFallbackError: overloaded", + original_exception=inner, + model="claude-sonnet-4-5", + llm_provider="anthropic", + ) + + info = StandardLoggingPayloadSetup.get_error_information(wrapper) + assert info["error_class"] == "InternalServerError" + assert info["error_code"] == "500" + assert info["llm_provider"] == "anthropic" + + +def test_stream_failure_callback_logs_mapped_error_class_sync(mock_sse_server): + logged = [] + + class _SyncFailLogger(CustomLogger): + def log_failure_event(self, kwargs, response_obj, start_time, end_time): + std_obj = kwargs.get("standard_logging_object", {}) + if "error_information" in std_obj: + logged.append(std_obj["error_information"]) + + original_callbacks = list(litellm.callbacks) + litellm.callbacks = [_SyncFailLogger()] + try: + with pytest.raises(Exception) as exc_info: + for _ in litellm.completion( + model="anthropic/claude-sonnet-4-5", + api_key="sk-mock-key", + api_base=f"http://127.0.0.1:{mock_sse_server.server_address[1]}", + messages=[{"role": "user", "content": "hi"}], + stream=True, + ): + pass + + # Underlying exception raised/wrapped should be InternalServerError + raised = getattr(exc_info.value, "original_exception", None) or exc_info.value + assert isinstance(raised, InternalServerError) + + for _ in range(50): + if logged: + break + time.sleep(0.05) + + assert len(logged) > 0 + info = logged[0] + assert info["error_class"] == "InternalServerError" + assert str(info["error_code"]) == "500" + finally: + litellm.callbacks = original_callbacks + + +@pytest.mark.asyncio +async def test_stream_failure_callback_logs_mapped_error_class_async(mock_sse_server): + logged = [] + + class _AsyncFailLogger(CustomLogger): + async def async_log_failure_event(self, kwargs, response_obj, start_time, end_time): + std_obj = kwargs.get("standard_logging_object", {}) + if "error_information" in std_obj: + logged.append(std_obj["error_information"]) + + original_callbacks = list(litellm.callbacks) + litellm.callbacks = [_AsyncFailLogger()] + try: + with pytest.raises(Exception) as exc_info: + stream = await litellm.acompletion( + model="anthropic/claude-sonnet-4-5", + api_key="sk-mock-key", + api_base=f"http://127.0.0.1:{mock_sse_server.server_address[1]}", + messages=[{"role": "user", "content": "hi"}], + stream=True, + ) + async for _ in stream: + pass + + raised = getattr(exc_info.value, "original_exception", None) or exc_info.value + assert isinstance(raised, InternalServerError) + + for _ in range(50): + if logged: + break + await asyncio.sleep(0.05) + + assert len(logged) > 0 + info = logged[0] + assert info["error_class"] == "InternalServerError" + assert str(info["error_code"]) == "500" + finally: + litellm.callbacks = original_callbacks