diff --git a/litellm/litellm_core_utils/litellm_logging.py b/litellm/litellm_core_utils/litellm_logging.py index 5f68505de5b..caab14e4cc7 100644 --- a/litellm/litellm_core_utils/litellm_logging.py +++ b/litellm/litellm_core_utils/litellm_logging.py @@ -627,6 +627,7 @@ class Logging(LiteLLMLoggingBaseClass): # enqueue closure here instead of firing it immediately. self._defer_async_logging: bool = False self._enqueue_deferred_logging: Callable[[], None] | None = None + self._async_success_scheduled: bool = False self._on_detached_stream_failure: Callable[[Exception], Awaitable[None]] | None = None def set_response_timing_metrics(self, timing_metrics: Mapping[str, float]) -> None: @@ -2019,6 +2020,15 @@ class Logging(LiteLLMLoggingBaseClass): except Exception: return True + def claim_async_success_log(self) -> bool: + """One async success log per request. The innermost @client wrapper always exits first, + since the outer one is awaiting it, so it claims the log here and the outer wrapper gets + False and schedules nothing.""" + if self._async_success_scheduled: + return False + self._async_success_scheduled = True + return True + def has_run_logging( self, event_type: Literal["async_success", "sync_success", "async_failure", "sync_failure"], diff --git a/litellm/utils.py b/litellm/utils.py index b1b24012afd..e79e200c7bb 100644 --- a/litellm/utils.py +++ b/litellm/utils.py @@ -1253,7 +1253,7 @@ def _dispatch_success_logging( ) logging_obj._enqueue_deferred_logging = _enqueue_deferred_logging - else: + elif logging_obj.claim_async_success_log(): asyncio.create_task( _client_async_logging_helper( logging_obj=logging_obj, diff --git a/tests/test_litellm/litellm_core_utils/test_litellm_logging.py b/tests/test_litellm/litellm_core_utils/test_litellm_logging.py index a4774d3f578..58674350f65 100644 --- a/tests/test_litellm/litellm_core_utils/test_litellm_logging.py +++ b/tests/test_litellm/litellm_core_utils/test_litellm_logging.py @@ -4839,7 +4839,7 @@ def _interactions_logging_obj(stream: bool, call_type: str = "acreate"): messages=[], stream=stream, call_type=call_type, - start_time=time.time(), + start_time=datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc).timestamp(), litellm_call_id="interactions-call-id", function_id="interactions-fn-id", ) @@ -7022,3 +7022,61 @@ def test_add_dynamic_callback_registers_once_per_list_without_touching_the_calle assert logging_obj.dynamic_async_failure_callbacks == [callback] assert LitellmLogging._with_dynamic_callback(None, callback) == [callback] assert LitellmLogging._with_dynamic_callback((callback,), callback) == [callback] + + +@pytest.mark.asyncio +async def test_background_interaction_completion_logs_while_in_progress_handler_is_parked(monkeypatch): + """ + The create's in_progress success handler can still be parked on awaited work when + the settlement arrives. The completed result is the only event carrying usage and + cost, so it has to reach the success callbacks whatever the one-success-log-per-request + dedupe does, or the settlement is marked billed with no spend row behind it. + """ + from litellm.litellm_core_utils import litellm_logging + from litellm.types.interactions import InteractionsAPIResponse + + class CountingLogger(CustomLogger): + def __init__(self): + super().__init__() + self.logged_results: list[object] = [] + + async def async_log_success_event(self, kwargs, response_obj, start_time, end_time): + self.logged_results.append(response_obj) + + counting_logger: Final = CountingLogger() + monkeypatch.setattr(litellm, "_async_success_callback", [counting_logger]) + + in_progress_parked: Final = asyncio.Event() + release_in_progress: Final = asyncio.Event() + real_truncate: Final = litellm_logging.truncate_base64_in_messages_async + calls: list[int] = [] + + async def parked_then_real_truncate(messages): + calls.append(1) + if len(calls) == 1: + in_progress_parked.set() + await release_in_progress.wait() + return await real_truncate(messages) + + monkeypatch.setattr(litellm_logging, "truncate_base64_in_messages_async", parked_then_real_truncate) + + logging_obj: Final = _interactions_logging_obj(stream=False) + in_progress: Final = InteractionsAPIResponse(id="interactions/abc", model="gemini-2.5-flash", status="in_progress") + completed: Final = InteractionsAPIResponse( + id="interactions/abc", + model="gemini-2.5-flash", + status="completed", + steps=[], + usage=dict(INTERACTIONS_USAGE_BLOCK), + ) + + first: Final = asyncio.create_task(logging_obj.async_success_handler(result=in_progress)) + await in_progress_parked.wait() + completion: Final = asyncio.create_task(logging_obj.async_log_background_interaction_completion(result=completed)) + await asyncio.sleep(0) + await asyncio.sleep(0) + release_in_progress.set() + await first + await completion + + assert counting_logger.logged_results == [completed, in_progress], counting_logger.logged_results diff --git a/tests/test_litellm/test_utils.py b/tests/test_litellm/test_utils.py index 8bc4abc9efa..c589b338b1c 100644 --- a/tests/test_litellm/test_utils.py +++ b/tests/test_litellm/test_utils.py @@ -6731,3 +6731,67 @@ def test_completion_finishes_response_metadata_before_handing_the_response_to_th assert snapshot["litellm_call_id"] assert snapshot["response_cost"] is not None assert snapshot["api_base"] + + +@pytest.mark.asyncio +async def test_nested_wrapper_exits_schedule_one_async_success_log(monkeypatch: pytest.MonkeyPatch) -> None: + """Chat over the Responses bridge exits two @client wrappers with one logging object. Issue + #44500: both exits enqueued a success handler, and on a large prompt the first one yielded to + the worker-thread base64 offload before it marked ``has_logged_async_success``, so the second + passed the check too and the request was logged and billed twice. The schedule step claims the + log for the object synchronously, so only the inner provider-shaped result is ever logged.""" + from litellm.litellm_core_utils import litellm_logging + from litellm.litellm_core_utils.litellm_logging import Logging + from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER + from litellm.utils import _dispatch_success_logging + + class CountingLogger(CustomLogger): + def __init__(self) -> None: + super().__init__() + self.logged_results: list[object] = [] + + async def async_log_success_event(self, kwargs, response_obj, start_time, end_time) -> None: + if kwargs["litellm_call_id"] == "bridge-call-id": + self.logged_results.append(response_obj) + + counting_logger: Final = CountingLogger() + monkeypatch.setattr(litellm, "_async_success_callback", [counting_logger]) + real_truncate: Final = litellm_logging.truncate_base64_in_messages_async + + async def yielding_truncate(messages): + await asyncio.sleep(0) + return await real_truncate(messages) + + monkeypatch.setattr(litellm_logging, "truncate_base64_in_messages_async", yielding_truncate) + + messages: Final = [{"role": "user", "content": "hello"}] + logging_obj: Final = Logging( + model="gpt-5.6-luna", + messages=messages, + stream=False, + call_type="acompletion", + start_time=datetime(2026, 1, 1, tzinfo=timezone.utc).timestamp(), + litellm_call_id="bridge-call-id", + function_id="bridge-fn-id", + ) + logging_obj.update_environment_variables( + litellm_params={}, optional_params={}, model="gpt-5.6-luna", custom_llm_provider="openai", input=messages + ) + inner_result: Final = ModelResponse(id="inner") + outer_result: Final = ModelResponse(id="outer") + now: Final = datetime(2026, 1, 1, tzinfo=timezone.utc) + + for result in (inner_result, outer_result): + _dispatch_success_logging( + logging_obj=logging_obj, + result=result, + start_time=now, + end_time=now, + is_completion_with_fallbacks=False, + is_litellm_internal_call=False, + ) + await asyncio.sleep(0) + await asyncio.wait_for(GLOBAL_LOGGING_WORKER.flush(), timeout=10) + + assert len(counting_logger.logged_results) == 1, counting_logger.logged_results + assert counting_logger.logged_results[0] is inner_result