fix(proxy): report a bridged stream failure to the failure hook once and skip the estimate for background responses

litellm/litellm_core_utils/litellm_logging.py

@@ -796,6 +796,7 @@ class Logging(LiteLLMLoggingBaseClass):
         self._async_success_scheduled: bool = False
         self._sync_success_scheduled: bool = False
         self.on_detached_stream_failure: Callable[[Exception], Awaitable[None]] | None = None
+        self._stream_failure_claimed: bool = False
         self.shadow_eval_request_snapshot: GuardrailRequestSnapshot | None = None
 
     def copy_for_new_call(self, litellm_call_id: str, start_time: datetime.datetime) -> "Logging":
@@ -2409,6 +2410,12 @@ class Logging(LiteLLMLoggingBaseClass):
                 },
             )
 
+    def claim_stream_failure_report(self) -> bool:
+        if self._stream_failure_claimed:
+            return False
+        self._stream_failure_claimed = True
+        return True
+
     def record_assembled_response_for_failure(self, assembled: ModelResponse) -> None:
         """Bill a fully streamed response on the failure log when a post-call hook rejects it."""
         usage: Final = getattr(assembled, "usage", None)

litellm/proxy/common_request_processing.py

@@ -3437,6 +3437,8 @@ class ProxyBaseLLMRequestProcessing:
         request_data: Final = self.data
 
         async def _on_detached_stream_failure(exc: Exception) -> None:
+            if not logging_obj.claim_stream_failure_report():
+                return
             try:
                 await proxy_logging_obj.post_call_failure_hook(
                     user_api_key_dict=user_api_key_dict,
@@ -4384,6 +4386,7 @@ class ProxyBaseLLMRequestProcessing:
         client_disconnected = False
         delivered_chunk = False
         recent_tail = SSE_STREAM_START_TAIL  # rebind-ok: rolling window over the yielded bytes
+        stream_logging_obj: Final[object] = request_data.get("litellm_logging_obj")  # pyright: ignore[reportUnknownVariableType, reportUnknownMemberType]  # legacy request data is narrowed to Logging below
         guarded_stream: Final[AsyncGenerator[object, None]] = proxy_logging_obj.async_post_call_streaming_iterator_hook(
             user_api_key_dict=user_api_key_dict,
             response=response,
@@ -4459,10 +4462,18 @@ class ProxyBaseLLMRequestProcessing:
             verbose_proxy_logger.exception(
                 "litellm.proxy.proxy_server.async_data_generator(): Exception occured - %s", e
             )
-            transformed_exception: Final = await proxy_logging_obj.post_call_failure_hook(
-                user_api_key_dict=user_api_key_dict,
-                original_exception=e,
-                request_data=request_data,
+            already_reported: Final = (
+                isinstance(stream_logging_obj, LiteLLMLoggingObj)
+                and not stream_logging_obj.claim_stream_failure_report()
+            )
+            transformed_exception: Final = (
+                None
+                if already_reported
+                else await proxy_logging_obj.post_call_failure_hook(
+                    user_api_key_dict=user_api_key_dict,
+                    original_exception=e,
+                    request_data=request_data,
+                )
             )
             if transformed_exception is not None:
                 e = transformed_exception

litellm/responses/streaming_iterator.py

@@ -965,7 +965,11 @@ class BaseResponsesAPIStreamingIterator:
 
     def _record_partial_usage_for_failure(self) -> None:
         generated_text: Final = self._generated_content + self._generated_tool_arguments
-        if not generated_text or self.logging_obj.model_call_details.get("combined_usage_object") is not None:
+        if (
+            not generated_text
+            or self.request_data.get("background") is True
+            or self.logging_obj.model_call_details.get("combined_usage_object") is not None
+        ):
             return
         estimate: Final = _estimate_usage_safely(
             self.model or "",

tests/unit/proxy/test_common_request_processing.py

@@ -10669,6 +10669,81 @@ class TestDetachedStreamFailureHook:
 
         assert [call["original_exception"] for call in recorder.calls] == [failure]
 
+    @staticmethod
+    def _armed_stream(recorder: _FailureHookRecorder, iterator_hook_for):
+        from litellm.proxy._types import UserAPIKeyAuth as RealUserAPIKeyAuth
+
+        logging_obj = LiteLLMLoggingObj(
+            model="gpt-6.1-sol",
+            messages=[{"role": "user", "content": "Write a story."}],
+            stream=True,
+            call_type="anthropic_messages",
+            start_time=None,
+            litellm_call_id="call-lit6871",
+            function_id="function-lit6871",
+        )
+        user_api_key_dict = RealUserAPIKeyAuth(api_key="sk-test")
+        processing_obj = ProxyBaseLLMRequestProcessing(data={"litellm_logging_obj": logging_obj})
+        ProxyLogging._callback_capabilities_cache.clear()
+        proxy_logging_obj = ProxyLogging(user_api_key_cache=MagicMock())
+        proxy_logging_obj.post_call_failure_hook = recorder.post_call_failure_hook
+        proxy_logging_obj.async_post_call_streaming_iterator_hook = iterator_hook_for(logging_obj)
+        processing_obj._arm_detached_stream_failure_hook(
+            logging_obj=logging_obj,
+            user_api_key_dict=user_api_key_dict,
+            proxy_logging_obj=proxy_logging_obj,
+        )
+        stream = ProxyBaseLLMRequestProcessing.async_streaming_data_generator(
+            response=MagicMock(),
+            user_api_key_dict=user_api_key_dict,
+            request_data=processing_obj.data,
+            proxy_logging_obj=proxy_logging_obj,
+            serialize_chunk=lambda chunk: "data: chunk\n\n",
+            serialize_error=lambda error: "data: error\n\n",
+        )
+        return logging_obj, stream
+
+    @pytest.mark.asyncio
+    async def test_failure_reported_by_the_stream_reaches_the_failure_hook_once_when_an_iterator_hook_raises_after_it(
+        self,
+    ):
+        recorder = _FailureHookRecorder()
+        upstream_failure = RuntimeError("upstream Responses stream failed")
+        hook_failure = RuntimeError("iterator hook rejected the stream")
+
+        def iterator_hook_for(logging_obj):
+            async def iterator_hook(**_kwargs):
+                yield b"event: error\n\n"
+                await logging_obj.on_detached_stream_failure(upstream_failure)
+                raise hook_failure
+
+            return iterator_hook
+
+        _, stream = self._armed_stream(recorder, iterator_hook_for)
+
+        frames = [frame async for frame in stream]
+
+        assert frames == ["data: chunk\n\n", "data: error\n\n"]
+        assert [call["original_exception"] for call in recorder.calls] == [upstream_failure]
+
+    @pytest.mark.asyncio
+    async def test_failure_reported_by_an_iterator_hook_is_not_reported_again_when_the_stream_closes(self):
+        recorder = _FailureHookRecorder()
+        upstream_failure = RuntimeError("upstream Responses stream failed")
+        hook_failure = RuntimeError("iterator hook rejected the error frame")
+
+        async def iterator_hook(**_kwargs):
+            yield b"event: message_start\n\n"
+            raise hook_failure
+
+        logging_obj, stream = self._armed_stream(recorder, lambda _logging_obj: iterator_hook)
+
+        frames = [frame async for frame in stream]
+        await logging_obj.on_detached_stream_failure(upstream_failure)
+
+        assert frames == ["data: chunk\n\n", "data: error\n\n"]
+        assert [call["original_exception"] for call in recorder.calls] == [hook_failure]
+
 
 class TestPostCallMaskedOutputReachesDeferredLogging:
     @pytest.mark.asyncio

tests/unit/responses/test_responses_streaming_iterator.py

@@ -270,6 +270,7 @@ async def _failure_payload_after_mid_stream_timeout(
     monkeypatch: pytest.MonkeyPatch,
     request_input: str | list[dict[str, object]],
     upstream: _UpstreamThatTimesOutAfterThreeDeltas | None = None,
+    background: bool = False,
 ) -> tuple[FailureRecorder, StandardLoggingPayload]:
     upstream_response: Final = upstream or _UpstreamThatTimesOutAfterThreeDeltas()
     recorder = FailureRecorder()
@@ -289,7 +290,7 @@ async def _failure_payload_after_mid_stream_timeout(
         responses_api_provider_config=OpenAIResponsesAPIConfig(),
         logging_obj=logging_obj,
         custom_llm_provider="openai",
-        request_data={"model": "gpt-5.4-nano", "input": request_input, "stream": True},
+        request_data={"model": "gpt-5.4-nano", "input": request_input, "stream": True, "background": background},
     )
     received = _ChunkRecorder()
 
@@ -335,6 +336,17 @@ async def test_mid_stream_read_timeout_bills_the_tool_call_arguments_already_str
     assert payload["response_cost"] > 0
 
 
+@pytest.mark.asyncio
+async def test_mid_stream_failure_of_a_background_response_bills_nothing_since_retrieving_it_bills_the_whole_run(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    _, payload = await _failure_payload_after_mid_stream_timeout(monkeypatch, _SEA_PROMPT, background=True)
+
+    assert payload["status"] == "failure"
+    assert payload["completion_tokens"] == 0
+    assert payload["response_cost"] == 0
+
+
 @pytest.mark.asyncio
 async def test_mid_stream_failure_usage_estimate_never_fetches_image_dimensions(
     monkeypatch: pytest.MonkeyPatch,