From 327d4e20e1275b835c7915f26229e8aa50eebbce Mon Sep 17 00:00:00 2001 From: Rodrigo-Palma Date: Fri, 3 Jul 2026 09:00:35 -0300 Subject: [PATCH] fix(logging): stop dropping streaming /v1/messages success logs Two silent-loss paths for anthropic_messages streaming success logs: - get_standard_logging_object_payload swallows exceptions and returns None; loggers gated on standard_logging_object (e.g. s3_v2) then skip the event while spend tracking still succeeds via the response_cost fallback, so requests disappear from success logs with no error. Rebuild the payload from an empty response object when construction fails for an assembled streaming response, mirroring the failure-path fallback. Fixes #32019. - _handle_streaming_logging scheduled end-of-stream logging with an unreferenced asyncio.create_task, which can be garbage-collected before it runs. Enqueue on the logging worker instead, matching the fix applied to pass-through routes in #31485. --- litellm/litellm_core_utils/litellm_logging.py | 23 +++++ .../test_litellm_logging_slo_fallback.py | 94 +++++++++++++++++++ .../test_streaming_iterator_logging.py | 37 ++++++++ 3 files changed, 154 insertions(+) create mode 100644 tests/unit/litellm_core_utils/test_litellm_logging_slo_fallback.py create mode 100644 tests/unit/llms/anthropic/pass_through/messages/test_streaming_iterator_logging.py diff --git a/litellm/litellm_core_utils/litellm_logging.py b/litellm/litellm_core_utils/litellm_logging.py index 154893b6c21..37a203cac0e 100644 --- a/litellm/litellm_core_utils/litellm_logging.py +++ b/litellm/litellm_core_utils/litellm_logging.py @@ -3346,6 +3346,29 @@ class Logging(LiteLLMLoggingBaseClass): "LiteLLM.LoggingError: [Non-Blocking] Exception building the standard logging payload " "for a streaming response; callbacks still run without it" ) + if self.model_call_details.get("standard_logging_object") is None: + # get_standard_logging_object_payload swallows exceptions and + # returns None, and the except branch above leaves the key + # unset. Loggers gated on standard_logging_object (e.g. s3_v2) + # would then silently drop the request while spend tracking + # still succeeds via the response_cost fallback. Rebuild from an + # empty response object instead, mirroring the failure-path + # fallback. See #32019. + verbose_logger.error( + "standard_logging_object build failed for streaming call_id=%s call_type=%s; " + "rebuilding a degraded payload so success loggers still fire", + self.litellm_call_id, + self.call_type, + ) + try: + self.model_call_details["standard_logging_object"] = self._build_standard_logging_payload( + {}, start_time, end_time + ) + except Exception: # noqa: BLE001 # same rule as above: never block later callbacks + verbose_logger.exception( + "LiteLLM.LoggingError: [Non-Blocking] Exception rebuilding the degraded standard " + "logging payload for a streaming response" + ) # print standard logging payload if (standard_logging_payload := self.model_call_details.get("standard_logging_object")) is not None: diff --git a/tests/unit/litellm_core_utils/test_litellm_logging_slo_fallback.py b/tests/unit/litellm_core_utils/test_litellm_logging_slo_fallback.py new file mode 100644 index 00000000000..60ad485fd1e --- /dev/null +++ b/tests/unit/litellm_core_utils/test_litellm_logging_slo_fallback.py @@ -0,0 +1,94 @@ +""" +Regression tests for https://github.com/BerriAI/litellm/issues/32019. + +Streaming /v1/messages responses were silently dropped by loggers gated on +``standard_logging_object`` (e.g. ``s3_v2``) whenever StandardLoggingPayload +construction failed: ``get_standard_logging_object_payload`` swallows the +exception and returns ``None``, spend tracking still succeeds via the +``response_cost`` fallback, and no error surfaces to the caller. +""" + +from datetime import datetime + +import pytest + +import litellm +from litellm.litellm_core_utils.litellm_logging import Logging +from litellm.types.utils import ModelResponse + +DEGRADED_PAYLOAD = {"trace_id": "degraded-fallback"} + + +@pytest.mark.asyncio +async def test_streaming_slo_build_failure_falls_back_to_degraded_payload(monkeypatch): + """When the SLP build returns None for an assembled streaming response, + async_success_handler must rebuild it from an empty response object so + SLP-gated loggers still record the request.""" + logging_obj = Logging( + model="claude-3-5-sonnet-20240620", + messages=[{"role": "user", "content": "hi"}], + stream=True, + call_type="anthropic_messages", + start_time=datetime.now(), + litellm_call_id="test-32019", + function_id="test-32019", + ) + build_calls = [] + + def fake_build(self, init_response_obj, start_time, end_time): + build_calls.append(init_response_obj) + if init_response_obj == {}: + return DEGRADED_PAYLOAD + return None # simulates get_standard_logging_object_payload swallowing an exception + + monkeypatch.setattr(Logging, "_build_standard_logging_payload", fake_build) + monkeypatch.setattr(litellm, "_async_success_callback", []) + monkeypatch.setattr(litellm, "success_callback", []) + + result = ModelResponse(model="claude-3-5-sonnet-20240620") + + await logging_obj.async_success_handler( + result, + start_time=datetime.now(), + end_time=datetime.now(), + cache_hit=False, + ) + + assert logging_obj.model_call_details["standard_logging_object"] == DEGRADED_PAYLOAD + assert build_calls == [result, {}] + + +@pytest.mark.asyncio +async def test_streaming_slo_build_success_does_not_rebuild(monkeypatch): + """The degraded rebuild must not run when the SLP builds normally.""" + logging_obj = Logging( + model="claude-3-5-sonnet-20240620", + messages=[{"role": "user", "content": "hi"}], + stream=True, + call_type="anthropic_messages", + start_time=datetime.now(), + litellm_call_id="test-32019-ok", + function_id="test-32019-ok", + ) + build_calls = [] + healthy_payload = {"trace_id": "healthy"} + + def fake_build(self, init_response_obj, start_time, end_time): + build_calls.append(init_response_obj) + return healthy_payload + + monkeypatch.setattr(Logging, "_build_standard_logging_payload", fake_build) + monkeypatch.setattr(litellm, "_async_success_callback", []) + monkeypatch.setattr(litellm, "success_callback", []) + + result = ModelResponse(model="claude-3-5-sonnet-20240620") + + await logging_obj.async_success_handler( + result, + start_time=datetime.now(), + end_time=datetime.now(), + cache_hit=False, + ) + + assert logging_obj.model_call_details["standard_logging_object"] == healthy_payload + assert len(build_calls) == 1 diff --git a/tests/unit/llms/anthropic/pass_through/messages/test_streaming_iterator_logging.py b/tests/unit/llms/anthropic/pass_through/messages/test_streaming_iterator_logging.py new file mode 100644 index 00000000000..287e09099ac --- /dev/null +++ b/tests/unit/llms/anthropic/pass_through/messages/test_streaming_iterator_logging.py @@ -0,0 +1,37 @@ +""" +Regression test for https://github.com/BerriAI/litellm/issues/32019. + +End-of-stream logging for /v1/messages must be enqueued on the logging +worker instead of a bare ``asyncio.create_task``: an unreferenced task can +be garbage-collected before it runs, dropping the success log entirely +(same class of bug fixed for pass-through routes in #31485). +""" + +from unittest.mock import MagicMock + +import pytest + +from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER +from litellm.llms.anthropic.experimental_pass_through.messages.streaming_iterator import ( + BaseAnthropicMessagesStreamingIterator, +) + + +@pytest.mark.asyncio +async def test_handle_streaming_logging_enqueues_on_logging_worker(monkeypatch): + enqueued = [] + + def fake_enqueue(async_coroutine): + enqueued.append(async_coroutine) + async_coroutine.close() + + monkeypatch.setattr(GLOBAL_LOGGING_WORKER, "ensure_initialized_and_enqueue", fake_enqueue) + + iterator = BaseAnthropicMessagesStreamingIterator( + litellm_logging_obj=MagicMock(), + request_body={"model": "claude-3-5-sonnet-20240620", "stream": True}, + ) + + await iterator._handle_streaming_logging(collected_chunks=[b"data: {}\n\n"]) + + assert len(enqueued) == 1