diff --git a/litellm/proxy/pass_through_endpoints/streaming_handler.py b/litellm/proxy/pass_through_endpoints/streaming_handler.py index 19d8b063dd7..dc1129f961b 100644 --- a/litellm/proxy/pass_through_endpoints/streaming_handler.py +++ b/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: diff --git a/tests/unit/proxy/hooks/test_prompt_cache_observer.py b/tests/unit/proxy/hooks/test_prompt_cache_observer.py index 82af3e9a6ef..b5fcad3add4 100644 --- a/tests/unit/proxy/hooks/test_prompt_cache_observer.py +++ b/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 { diff --git a/tests/unit/proxy/pass_through_endpoints/test_streaming_handler.py b/tests/unit/proxy/pass_through_endpoints/test_streaming_handler.py index fdd21afbc67..5016372dc7c 100644 --- a/tests/unit/proxy/pass_through_endpoints/test_streaming_handler.py +++ b/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()