diff --git a/litellm/llms/anthropic/pass_through/responses_adapters/streaming_iterator.py b/litellm/llms/anthropic/pass_through/responses_adapters/streaming_iterator.py index cc6f4da3403..03b3ceb6653 100644 --- a/litellm/llms/anthropic/pass_through/responses_adapters/streaming_iterator.py +++ b/litellm/llms/anthropic/pass_through/responses_adapters/streaming_iterator.py @@ -3,7 +3,7 @@ import asyncio import json from collections import deque -from collections.abc import AsyncIterator, Iterator, Mapping +from collections.abc import AsyncIterator, Awaitable, Callable, Iterator, Mapping from typing import TYPE_CHECKING, Any, Final from pydantic import BaseModel, ConfigDict, field_validator @@ -11,7 +11,7 @@ from pydantic import BaseModel, ConfigDict, field_validator from litellm import verbose_logger from litellm._logging import redact_internal_details_from_client_message from litellm._uuid import uuid -from litellm.exceptions import MidStreamFallbackError +from litellm.exceptions import APIError, MidStreamFallbackError from litellm.litellm_core_utils.prompt_templates.common_utils import ( encrypted_reasoning_signature, ) @@ -122,9 +122,11 @@ class AnthropicResponsesStreamWrapper: ) -> None: self.responses_stream = responses_stream self.model = model + self.litellm_logging_obj = litellm_logging_obj self._message_id: str = f"msg_{uuid.uuid4()}" if litellm_logging_obj is not None: litellm_logging_obj.record_streamed_anthropic_message_id(self._message_id) + self._unreported_failure: Exception | None = None self._current_block_index: int = -1 # Map item_id -> content_block_index so we can stop the right block later self._item_id_to_block_index: dict[str, int] = {} @@ -430,30 +432,58 @@ class AnthropicResponsesStreamWrapper: self._sent_message_stop = True return - def _fail_stream(self, status_code: int, message: str) -> None: + def _fail_stream(self, status_code: int, message: str, failure: Exception | None = None) -> None: self._stream_failed = True + self._unreported_failure = failure or APIError( + status_code=status_code, + message=message, + llm_provider=getattr(self.responses_stream, "custom_llm_provider", None), + model=self.model, + ) self._chunk_queue.append(_anthropic_error_chunk(status_code, message)) + async def _pop_chunk(self) -> dict[str, object]: + if self._unreported_failure is not None: + failure: Final = self._unreported_failure + self._unreported_failure = None + await self._report_failure(failure) + return self._chunk_queue.popleft() + + async def _report_failure(self, failure: Exception) -> None: + on_failure: Final[Callable[[Exception], Awaitable[None]] | None] = getattr( + self.litellm_logging_obj, "_on_detached_stream_failure", None + ) + if on_failure is None: + return + try: + await on_failure(failure) + except Exception as hook_failure: # noqa: BLE001 # a failing proxy hook must not hide the error event from the client + verbose_logger.warning( + "AnthropicResponsesStreamWrapper: stream failure hook raised: %s(%s)", + type(hook_failure).__name__, + hook_failure, + ) + def __aiter__(self) -> "AnthropicResponsesStreamWrapper": return self async def __anext__(self) -> dict[str, object]: if self._chunk_queue: - return self._chunk_queue.popleft() + return await self._pop_chunk() if self._stream_failed: raise StopAsyncIteration if not self._sent_message_start: self._sent_message_start = True self._chunk_queue.append(self._make_message_start()) - return self._chunk_queue.popleft() + return await self._pop_chunk() try: if hasattr(self.responses_stream, "__aiter__"): async for event in self.responses_stream: self._process_event(event) if self._chunk_queue: - return self._chunk_queue.popleft() + return await self._pop_chunk() else: if self._sync_responses_iterator is None: self._sync_responses_iterator = iter(self.responses_stream) @@ -462,14 +492,14 @@ class AnthropicResponsesStreamWrapper: while (event := await asyncio.to_thread(next, sync_iterator, missing)) is not missing: self._process_event(event) if self._chunk_queue: - return self._chunk_queue.popleft() + return await self._pop_chunk() except StopAsyncIteration: pass except Exception as e: # noqa: BLE001 # every upstream failure becomes a client error event verbose_logger.exception( "AnthropicResponsesStreamWrapper: upstream Responses stream for %s failed", self.model ) - self._fail_stream(*_failure_status_and_message(e)) + self._fail_stream(*_failure_status_and_message(e), failure=_original_failure(e)) if not self._chunk_queue and not self._sent_message_stop and not self._stream_failed: verbose_logger.error( @@ -479,7 +509,7 @@ class AnthropicResponsesStreamWrapper: self._fail_stream(500, INCOMPLETE_STREAM_ERROR_MESSAGE) if self._chunk_queue: - return self._chunk_queue.popleft() + return await self._pop_chunk() raise StopAsyncIteration diff --git a/litellm/responses/streaming_iterator.py b/litellm/responses/streaming_iterator.py index 3522a43e0de..321bfd546e7 100644 --- a/litellm/responses/streaming_iterator.py +++ b/litellm/responses/streaming_iterator.py @@ -892,7 +892,9 @@ class BaseResponsesAPIStreamingIterator: ) if estimate is None: return - usage: Final = ResponseAPILoggingUtils._transform_response_api_usage_to_chat_usage(estimate) + usage: Final = ResponseAPILoggingUtils._transform_response_api_usage_to_chat_usage( # pyright: ignore[reportPrivateUsage] # same shared transform the success path above uses + estimate + ) self.logging_obj.record_assembled_response_for_failure(ModelResponse(model=self.model, usage=usage)) def _handle_failure(self, exception: Exception): diff --git a/tests/unit/llms/anthropic/pass_through/responses_adapters/test_responses_adapters_streaming_iterator.py b/tests/unit/llms/anthropic/pass_through/responses_adapters/test_responses_adapters_streaming_iterator.py index e1dded214bf..227d2966e53 100644 --- a/tests/unit/llms/anthropic/pass_through/responses_adapters/test_responses_adapters_streaming_iterator.py +++ b/tests/unit/llms/anthropic/pass_through/responses_adapters/test_responses_adapters_streaming_iterator.py @@ -7,6 +7,8 @@ import asyncio import json import os import sys +import time +from collections.abc import Awaitable, Callable from types import SimpleNamespace import pytest @@ -15,6 +17,7 @@ sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../. import litellm from litellm.exceptions import MidStreamFallbackError +from litellm.litellm_core_utils.litellm_logging import Logging as LitellmLogging from litellm.litellm_core_utils.prompt_templates.common_utils import ( encrypted_reasoning_signature, ) @@ -622,3 +625,98 @@ class TestUpstreamFailureEndsStreamWithErrorEvent: chunks = _collect(iter(())) assert [chunk["type"] for chunk in chunks] == ["message_start", "error"] assert chunks[1]["error"] == {"type": "api_error", "message": INCOMPLETE_STREAM_ERROR_MESSAGE} + + +class _FailureHookRecorder: + def __init__(self) -> None: + self.received: tuple[Exception, ...] = () + + async def __call__(self, failure: Exception) -> None: + self.received = (*self.received, failure) + + +def _logging_obj_with_failure_hook(on_failure: Callable[[Exception], Awaitable[None]]) -> LitellmLogging: + logging_obj = LitellmLogging( + model="m", + messages=[{"role": "user", "content": "hi"}], + stream=True, + call_type="anthropic_messages", + start_time=time.time(), + litellm_call_id="lit-6871-test", + function_id="lit-6871-test", + ) + logging_obj._on_detached_stream_failure = on_failure + return logging_obj + + +def _collect_with_logging(stream, logging_obj: LitellmLogging) -> list: + async def _run() -> list: + wrapper = AnthropicResponsesStreamWrapper(responses_stream=stream, model="m", litellm_logging_obj=logging_obj) + return [chunk async for chunk in wrapper] + + return asyncio.run(_run()) + + +class TestUpstreamFailureReachesTheProxyFailureHook: + """The ``error`` event alone leaves the proxy unaware the stream failed, so the wrapper also hands + the failure to the hook the proxy arms on the logging object, the one that writes the failure + spend row and releases the budget reservation.""" + + def test_raised_upstream_failure_reaches_the_hook_unwrapped_before_the_error_event(self): + rate_limit = litellm.RateLimitError(message="You have no credits remaining.", llm_provider="openai", model="m") + wrapped = MidStreamFallbackError( + message=str(rate_limit), + model="m", + llm_provider="openai", + original_exception=rate_limit, + is_pre_first_chunk=False, + ) + recorder = _FailureHookRecorder() + + async def _gen(): + yield {"type": "response.created"} + yield {"type": "response.output_item.added", "item": {"type": "message", "id": "msg_1"}} + yield {"type": "response.output_text.delta", "item_id": "msg_1", "delta": "Hi"} + raise wrapped + + chunks = _collect_with_logging(_gen(), _logging_obj_with_failure_hook(recorder)) + assert [chunk["type"] for chunk in chunks] == [ + "message_start", + "content_block_start", + "content_block_delta", + "error", + ] + assert recorder.received == (rate_limit,) + + def test_response_failed_event_reaches_the_hook_as_an_exception_carrying_the_upstream_status(self): + recorder = _FailureHookRecorder() + failed = SimpleNamespace( + status="failed", + output=[], + usage=None, + error={"code": "rate_limit_exceeded", "message": "Rate limit reached for gpt-5.5, try again in 20s."}, + ) + + async def _gen(): + yield {"type": "response.created"} + yield {"type": "response.failed", "response": failed} + + chunks = _collect_with_logging(_gen(), _logging_obj_with_failure_hook(recorder)) + assert chunks[-1]["type"] == "error" + (failure,) = recorder.received + assert failure.status_code == 429 + assert "Rate limit reached for gpt-5.5, try again in 20s." in failure.message + + def test_a_raising_hook_never_hides_the_error_event_from_the_client(self): + async def _exploding_hook(failure: Exception) -> None: + raise RuntimeError("spend writer is down") + + logging_obj = _logging_obj_with_failure_hook(_exploding_hook) + + async def _gen(): + yield {"type": "response.created"} + raise ConnectionResetError("Response payload is not completed") + + chunks = _collect_with_logging(_gen(), logging_obj) + assert [chunk["type"] for chunk in chunks] == ["message_start", "error"] + assert chunks[-1]["error"] == {"type": "api_error", "message": "Response payload is not completed"} diff --git a/tests/unit/responses/test_responses_streaming_iterator.py b/tests/unit/responses/test_responses_streaming_iterator.py index 4aace64aaaa..16b125e56d4 100644 --- a/tests/unit/responses/test_responses_streaming_iterator.py +++ b/tests/unit/responses/test_responses_streaming_iterator.py @@ -164,14 +164,14 @@ async def test_sync_callbacks_run_only_after_async_handler_completes(recording_e class FailureRecorder(CustomLogger): def __init__(self): super().__init__() - self.failure_payloads: list = [] - self.success_payloads: list = [] + self.failure_payloads: tuple = () + self.success_payloads: tuple = () async def async_log_failure_event(self, kwargs, response_obj, start_time, end_time): - self.failure_payloads.append(kwargs["standard_logging_object"]) + self.failure_payloads = (*self.failure_payloads, kwargs["standard_logging_object"]) async def async_log_success_event(self, kwargs, response_obj, start_time, end_time): - self.success_payloads.append(kwargs["standard_logging_object"]) + self.success_payloads = (*self.success_payloads, kwargs["standard_logging_object"]) class _UpstreamThatTimesOutAfterThreeDeltas: @@ -230,4 +230,4 @@ async def test_mid_stream_read_timeout_logs_failure_with_partial_usage_and_no_su assert payload["prompt_tokens"] > 0 assert payload["completion_tokens"] > 0 assert payload["response_cost"] > 0 - assert recorder.success_payloads == [] + assert recorder.success_payloads == ()