Merge 670c1fb9ba into bb4f7211d7

litellm/proxy/pass_through_endpoints/streaming_handler.py

@@ -12,6 +12,7 @@ from litellm.litellm_core_utils.asyncify import asyncify
 from litellm.litellm_core_utils.core_helpers import bind_budget_reservation_to_callbacks
 from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
 from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER
+from litellm.llms.anthropic.common_utils import AnthropicError
 from litellm.proxy._types import PassThroughEndpointLoggingResultValues
 from litellm.proxy.common_request_processing import ProxyBaseLLMRequestProcessing
 from litellm.proxy.common_utils.sse_keepalive import split_complete_sse_frames
@@ -297,6 +298,7 @@ class PassThroughStreamingHandler:
         from litellm.llms.anthropic.pass_through.messages.streaming_iterator import (
             _is_message_stop_chunk,  # pyright: ignore[reportPrivateUsage]  # both native stream paths share terminal-event detection
             _is_provider_error_chunk,  # pyright: ignore[reportPrivateUsage]  # provider errors must not become cache evidence
+            parse_anthropic_error_event,
         )
 
         # Transport reads can split event names and JSON payloads. Recognize terminal
@@ -312,6 +314,22 @@ class PassThroughStreamingHandler:
             and _is_message_stop_chunk(complete_frames)
             and not _is_provider_error_chunk(complete_frames)
         )
+        provider_error: Final = parse_anthropic_error_event(complete_frames) if complete_frames else None
+        if endpoint_type == EndpointType.ANTHROPIC and provider_error is not None:
+            _, error_message, error_status_code = provider_error
+            await PassThroughStreamingHandler.schedule_stream_failure_logging(
+                litellm_logging_obj=litellm_logging_obj,
+                endpoint_type=endpoint_type,
+                request_body=request_body,  # pyright: ignore[reportUnknownArgumentType]  # request_body is an untyped dict in this signature
+                raw_bytes=raw_bytes,
+                exception=AnthropicError(status_code=error_status_code, message=error_message),
+                stream_context=PassThroughStreamContext(
+                    passthrough_success_handler_obj=passthrough_success_handler_obj,
+                    url_route=url_route,
+                    start_time=start_time,
+                ),
+            )
+            return
         try:
             # TinyFish billing is owned by the detached poller; the $0 fallback below is only for streams with no run_id
             if endpoint_type == EndpointType.TINYFISH:

tests/unit/proxy/hooks/test_prompt_cache_observer.py

@@ -204,6 +204,9 @@ class RecordingObserver(PromptCacheObserver):
         await super().async_log_success_event(kwargs, response_obj, start_time, end_time)
         self.finished.set()
 
+    async def async_log_failure_event(self, kwargs, response_obj, start_time, end_time):
+        self.finished.set()
+
 
 def native_response():
     return {

tests/unit/proxy/pass_through_endpoints/test_streaming_handler.py

@@ -2,7 +2,7 @@ import json
 import logging
 from collections.abc import Iterator
 from datetime import datetime
-from unittest.mock import MagicMock
+from unittest.mock import AsyncMock, MagicMock
 
 import pytest
 
@@ -235,3 +235,71 @@ async def test_failed_anthropic_stream_records_partial_usage_off_the_event_loop(
     partial_usage = logging_obj.record_partial_usage_for_failure.call_args.kwargs["usage"]
     assert partial_usage.completion_tokens > 100_000
     assert_loop_stayed_free(took, lags)
+
+
+CLAUDE_MODEL = "claude-fable-5"
+
+
+def _anthropic_error_frame(error_type: str, message: str) -> bytes:
+    payload = {"type": "error", "error": {"type": error_type, "message": message}}
+    return f"event: error\ndata: {json.dumps(payload)}\n\n".encode()
+
+
+def _anthropic_error_logging_obj() -> MagicMock:
+    logging_obj = MagicMock(spec=LiteLLMLoggingObj)
+    logging_obj.model_call_details = {"model": CLAUDE_MODEL, "stream": True}
+    logging_obj.optional_params = {}
+    logging_obj.litellm_params = {}
+    logging_obj.litellm_call_id = "test-call-id"
+    logging_obj.get_router_model_id.return_value = None
+    logging_obj.dispatch_success_handlers = AsyncMock()
+    logging_obj.dispatch_failure_handlers = AsyncMock()
+    return logging_obj
+
+
+async def _route_anthropic_stream(logging_obj: MagicMock, raw_bytes: list[bytes]) -> None:
+    await PassThroughStreamingHandler._route_streaming_logging_to_handler(
+        litellm_logging_obj=logging_obj,
+        passthrough_success_handler_obj=PassThroughEndpointLogging(),
+        url_route="/anthropic/v1/messages",
+        request_body={"model": CLAUDE_MODEL, "stream": True},
+        endpoint_type=EndpointType.ANTHROPIC,
+        start_time=datetime.now(),
+        raw_bytes=raw_bytes,
+        end_time=datetime.now(),
+        model=CLAUDE_MODEL,
+    )
+    await GLOBAL_LOGGING_WORKER.flush()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "error_type, expected_status",
+    [("api_error", 500), ("overloaded_error", 503), ("rate_limit_error", 429)],
+)
+async def test_in_band_error_frame_is_logged_as_failure_not_success(error_type: str, expected_status: int):
+    logging_obj = _anthropic_error_logging_obj()
+
+    await _route_anthropic_stream(
+        logging_obj,
+        [*_interrupted_anthropic_stream(CLAUDE_MODEL, "partial answer"), _anthropic_error_frame(error_type, "boom")],
+    )
+
+    logging_obj.dispatch_success_handlers.assert_not_awaited()
+    logging_obj.dispatch_failure_handlers.assert_awaited_once()
+    exception = logging_obj.dispatch_failure_handlers.await_args.args[0]
+    assert exception.status_code == expected_status
+    assert "boom" in str(exception)
+    partial_usage = logging_obj.record_partial_usage_for_failure.call_args.kwargs["usage"]
+    assert partial_usage.prompt_tokens == 29
+    assert partial_usage.completion_tokens > 0
+
+
+@pytest.mark.asyncio
+async def test_stream_without_error_frame_still_logs_success():
+    logging_obj = _anthropic_error_logging_obj()
+
+    await _route_anthropic_stream(logging_obj, _interrupted_anthropic_stream(CLAUDE_MODEL, "partial answer"))
+
+    logging_obj.dispatch_success_handlers.assert_awaited_once()
+    logging_obj.dispatch_failure_handlers.assert_not_awaited()