fix(passthrough): relay upstream error streams without waiting for the log preview and keep passthrough normalization

litellm/constants.py

@@ -1529,6 +1529,11 @@
 MCP_TOOL_NAME_PREFIX: Final = "mcp_tool"
 MAXIMUM_TRACEBACK_LINES_TO_LOG: Final = int(os.getenv("MAXIMUM_TRACEBACK_LINES_TO_LOG", 100))
 PASSTHROUGH_UPSTREAM_ERROR_BODY_MAX_LOG_CHARS: Final = 4096
+PASSTHROUGH_UPSTREAM_ERROR_REPORT_TASK_NAME: Final = "passthrough-upstream-error-report"
+PASSTHROUGH_UPSTREAM_ERROR_REPORT_SHUTDOWN_WAIT_SECONDS: Final = 10.0
+PASSTHROUGH_UPSTREAM_ERROR_REDACT_MARGIN_CHARS: Final = 256
+PASSTHROUGH_UPSTREAM_ERROR_BODY_MAX_SCAN_BYTES: Final = 65536
+SSE_LATE_RELAY_BACKGROUND_WAIT_SECONDS: Final = 10.0
 
 # Headers to control callbacks
 X_LITELLM_DISABLE_CALLBACKS: Final = "x-litellm-disable-callbacks"

litellm/litellm_core_utils/error_normalization.py

@@ -59,8 +59,9 @@ class _HasProxyErrorType(Protocol):
     type: str
 
 
+_UPSTREAM_PASSTHROUGH_PATTERN: Final = re.compile(r"upstream passthrough request failed", re.IGNORECASE)
+
 _MESSAGE_PATTERNS: Final[tuple[tuple[re.Pattern[str], str], ...]] = (
-    (re.compile(r"upstream passthrough request failed", re.IGNORECASE), UPSTREAM_PASSTHROUGH),
     (
         re.compile(r"budget has been exceeded|max budget|crossed budget", re.IGNORECASE),
         BUDGET_EXCEEDED,
@@ -192,7 +193,11 @@ def normalize_error(exc: Exception | None, status_code: str, message: str) -> st
     if by_proxy_type is not None:
         return by_proxy_type
     by_message: Final = (
-        BUDGET_EXCEEDED if _exceeded_before_budget(message) else _classify_by_message(message, _MESSAGE_PATTERNS)
+        UPSTREAM_PASSTHROUGH
+        if _UPSTREAM_PASSTHROUGH_PATTERN.search(message)
+        else BUDGET_EXCEEDED
+        if _exceeded_before_budget(message)
+        else _classify_by_message(message, _MESSAGE_PATTERNS)
     )
     if by_message is not None:
         return by_message

litellm/proxy/common_request_processing.py

@@ -42,6 +42,7 @@
     MAX_PAYLOAD_SIZE_FOR_DEBUG_LOG,
     NON_INFERENCE_CALL_TYPES,
     RETURN_RAW_MODEL_NAME_METADATA_KEY,
+    SSE_LATE_RELAY_BACKGROUND_WAIT_SECONDS,
     STREAM_SSE_DATA_PREFIX,
     STREAM_SSE_KEEPALIVE_PING_BYTES,
     UNSAFE_PROXY_RESPONSE_HEADERS,
@@ -1141,27 +1142,39 @@
     return resolve_ttft_keepalive_interval(deployments, litellm.sse_keepalive_ping_interval_seconds)
 
 
-async def _aclose_late_response(produced: Response) -> None:
+async def _aclose_late_response(
+    produced: Response, *, background_wait_seconds: float = SSE_LATE_RELAY_BACKGROUND_WAIT_SECONDS
+) -> None:
     """Run the cleanup Starlette would have run, for a response it never called.
 
     Closing an already-closed async generator is a no-op, so this is safe to call
     from both the relay's own teardown and the outer one.
     """
-    if not isinstance(produced, StreamingResponse):
-        return
-    targets: Final = (
-        (produced.body_iterator, produced.upstream_generator)
-        if isinstance(produced, _UpstreamClosingStreamingResponse)
-        else (produced.body_iterator,)
-    )
-    for target in targets:
-        aclose = getattr(target, "aclose", None)
-        if aclose is None:
-            continue
-        try:
-            await aclose()
-        except BaseException as exc:  # noqa: BLE001  # teardown must not mask why the stream ended
-            verbose_proxy_logger.debug("error closing relayed streaming generator: %s", exc)
+    if isinstance(produced, StreamingResponse):
+        targets: Final = (
+            (produced.body_iterator, produced.upstream_generator)
+            if isinstance(produced, _UpstreamClosingStreamingResponse)
+            else (produced.body_iterator,)
+        )
+        for target in targets:
+            aclose = getattr(target, "aclose", None)
+            if aclose is None:
+                continue
+            try:
+                await aclose()
+            except BaseException as exc:  # noqa: BLE001  # teardown must not mask why the stream ended
+                verbose_proxy_logger.debug("error closing relayed streaming generator: %s", exc)
+    if produced.background is not None:
+        with anyio.move_on_after(background_wait_seconds) as scope:
+            try:
+                await produced.background()
+            except BaseException as exc:  # noqa: BLE001  # teardown must not mask why the stream ended
+                verbose_proxy_logger.debug("error running relayed response background task: %s", exc)
+        if scope.cancel_called:
+            verbose_proxy_logger.warning(
+                "relayed response background task still running after %.0fs, leaving it to shutdown",
+                background_wait_seconds,
+            )
 
 
 async def _relay_late_response(produced: Response) -> AsyncGenerator[bytes, None]: