From f89f20e078946c051b25009c21a089ace753bae2 Mon Sep 17 00:00:00 2001 From: yucheng Date: Thu, 16 Jul 2026 19:38:50 +0000 Subject: [PATCH] fix(logging): dedupe async CustomLogger hooks with explicit run_custom_logger_hooks sweep flag The async success/failure paths run the sync handler only to service legacy sync-only callbacks (langfuse, s3, sync-only CustomLoggers), so thread run_custom_logger_hooks=False from the two async sync-sweep sites (dispatch_success_handlers, handle_sync_success_callbacks_for_async_calls) and the async wrapper failure path. That suppresses the sync CustomLogger hook and the standard/openmeter emitters the async handler already ran, closing the anthropic_messages double-log without depending on the a* request classifier Replaces the is_async_entrypoint wrapper stamp, which required mutable per-request state and broke duck-typed logging stubs in CI Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- litellm/litellm_core_utils/litellm_logging.py | 46 +++++++++---- litellm/utils.py | 6 +- .../test_litellm_logging.py | 68 +++++++------------ 3 files changed, 58 insertions(+), 62 deletions(-) diff --git a/litellm/litellm_core_utils/litellm_logging.py b/litellm/litellm_core_utils/litellm_logging.py index 2c2440ab078..673a8811ed4 100644 --- a/litellm/litellm_core_utils/litellm_logging.py +++ b/litellm/litellm_core_utils/litellm_logging.py @@ -367,7 +367,6 @@ def __init__( self.stream = stream self.start_time = start_time # log the call start time self.call_type = call_type - self.is_async_entrypoint: Optional[bool] = None self.litellm_call_id = litellm_call_id self.litellm_trace_id: str = litellm_trace_id if litellm_trace_id else str(uuid.uuid4()) self.function_id = function_id @@ -1544,15 +1543,9 @@ async def _response_cost_calculator_async( ) -> Optional[float]: return self._response_cost_calculator(result=result, cache_hit=cache_hit) - def _is_sync_litellm_request(self, litellm_params: dict) -> bool: - """True for sync SDK entrypoints (``completion``), false for async (``acompletion``, etc.). - - ``is_async_entrypoint`` is stamped by the ``@client`` wrapper that ran the request - and is authoritative; the ``a*`` flag heuristic covers logging objects constructed - outside ``@client`` (proxy passthrough endpoints, realtime, MCP). - """ - if self.is_async_entrypoint is not None: - return not self.is_async_entrypoint + @staticmethod + def _is_sync_litellm_request(litellm_params: dict) -> bool: + """True for sync SDK entrypoints (``completion``), false for async (``acompletion``, etc.).""" return ( litellm_params.get(CallTypes.acompletion.value, False) is not True and litellm_params.get(CallTypes.aresponses.value, False) is not True @@ -1632,6 +1625,7 @@ async def dispatch_success_handlers( start_time=start_time, end_time=end_time, cache_hit=cache_hit, + run_custom_logger_hooks=False, **kwargs, ) @@ -1985,7 +1979,15 @@ async def async_flush_passthrough_collected_chunks( await self.async_success_handler(result=complete_streaming_response) return - def success_handler(self, result=None, start_time=None, end_time=None, cache_hit=None, **kwargs): + def success_handler( + self, + result=None, + start_time=None, + end_time=None, + cache_hit=None, + run_custom_logger_hooks: Optional[bool] = None, + **kwargs, + ): verbose_logger.debug(f"Logging Details LiteLLM-Success Call: Cache_hit={cache_hit}") if not self.should_run_logging(event_type="sync_success"): # prevent double logging return @@ -1997,7 +1999,11 @@ def success_handler(self, result=None, start_time=None, end_time=None, cache_hit standard_logging_object=kwargs.get("standard_logging_object", None), ) litellm_params = self.model_call_details.get("litellm_params", {}) - is_sync_request = self._is_sync_litellm_request(litellm_params) + is_sync_request = ( + run_custom_logger_hooks + if run_custom_logger_hooks is not None + else self._is_sync_litellm_request(litellm_params) + ) try: ## BUILD COMPLETE STREAMED RESPONSE complete_streaming_response: Optional[ @@ -2783,12 +2789,23 @@ async def special_failure_handlers(self, exception: Exception): kwargs=self.model_call_details, ) # type: ignore - def failure_handler(self, exception, traceback_exception, start_time=None, end_time=None): + def failure_handler( + self, + exception, + traceback_exception, + start_time=None, + end_time=None, + run_custom_logger_hooks: Optional[bool] = None, + ): verbose_logger.debug(f"Logging Details LiteLLM-Failure Call: {litellm.failure_callback}") if not self.should_run_logging(event_type="sync_failure"): # prevent double logging return litellm_params = self.model_call_details.get("litellm_params", {}) - is_sync_request = self._is_sync_litellm_request(litellm_params) + is_sync_request = ( + run_custom_logger_hooks + if run_custom_logger_hooks is not None + else self._is_sync_litellm_request(litellm_params) + ) try: start_time, end_time = self._failure_handler_helper_fn( @@ -3084,6 +3101,7 @@ def handle_sync_success_callbacks_for_async_calls( start_time, end_time, cache_hit, + run_custom_logger_hooks=False, ) def _should_run_sync_callbacks_for_async_calls(self) -> bool: diff --git a/litellm/utils.py b/litellm/utils.py index ed38bb0ea2c..737e4da3e4a 100644 --- a/litellm/utils.py +++ b/litellm/utils.py @@ -1324,8 +1324,6 @@ def wrapper(*args, **kwargs): # Type assertion: logging_obj is guaranteed to be non-None after function_setup assert logging_obj is not None, "logging_obj should not be None after function_setup" - if getattr(logging_obj, "is_async_entrypoint", None) is None: - logging_obj.is_async_entrypoint = False ## LOAD CREDENTIALS load_credentials_from_list(kwargs) @@ -1607,8 +1605,6 @@ async def wrapper_async(*args, **kwargs): # Type assertion: logging_obj is guaranteed to be non-None after function_setup assert logging_obj is not None, "logging_obj should not be None after function_setup" - if getattr(logging_obj, "is_async_entrypoint", None) is None: - logging_obj.is_async_entrypoint = True modified_kwargs = await async_pre_call_deployment_hook(kwargs, call_type) if modified_kwargs is not None: @@ -1804,7 +1800,7 @@ def _enqueue_deferred_logging() -> None: if logging_obj and not _is_litellm_internal_call: try: logging_obj.failure_handler( - e, traceback_exception, start_time, end_time + e, traceback_exception, start_time, end_time, run_custom_logger_hooks=False ) # DO NOT MAKE THREADED - router retry fallback relies on this! except Exception as e: raise e 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 56eadc15fee..599135d4f89 100644 --- a/tests/test_litellm/litellm_core_utils/test_litellm_logging.py +++ b/tests/test_litellm/litellm_core_utils/test_litellm_logging.py @@ -799,22 +799,16 @@ class DummyLogger(CustomLogger): def test_is_sync_litellm_request(logging_obj): - assert logging_obj.is_async_entrypoint is None assert logging_obj._is_sync_litellm_request({}) is True assert logging_obj._is_sync_litellm_request({"acompletion": True}) is False assert logging_obj._is_sync_litellm_request({"allm_passthrough_route": True}) is False - logging_obj.is_async_entrypoint = True - assert logging_obj._is_sync_litellm_request({}) is False - - logging_obj.is_async_entrypoint = False - assert logging_obj._is_sync_litellm_request({"acompletion": True}) is True - @pytest.mark.asyncio async def test_anthropic_messages_success_logs_custom_logger_exactly_once(logging_obj): - """A request stamped async by the @client wrapper must reach a CustomLogger exactly - once, via the async hook only; the sync success_handler skips CustomLogger hooks. + """On an async request the sync success_handler runs only to service legacy sync-only + callbacks; when the async sync-sweep passes run_custom_logger_hooks=False it must skip + the CustomLogger hook so a both-hook logger fires exactly once, via the async hook. Regression guard for LIT-4447.""" from litellm.integrations.custom_logger import CustomLogger @@ -823,7 +817,6 @@ class DummyLogger(CustomLogger): logging_obj.stream = False logging_obj.call_type = "anthropic_messages" - logging_obj.is_async_entrypoint = True logging_obj.model_call_details["litellm_params"] = {} logging_obj.litellm_params = {} @@ -852,7 +845,7 @@ class DummyLogger(CustomLogger): ), ): await logging_obj.async_success_handler(result=model_response) - logging_obj.success_handler(result=model_response) + logging_obj.success_handler(result=model_response, run_custom_logger_hooks=False) mock_async_log.assert_awaited_once() mock_sync_log.assert_not_called() @@ -860,9 +853,10 @@ class DummyLogger(CustomLogger): @pytest.mark.asyncio async def test_anthropic_messages_failure_logs_custom_logger_exactly_once(logging_obj): - """A request stamped async by the @client wrapper must reach a CustomLogger exactly - once on failure, via the async hook only; the sync failure_handler skips CustomLogger - hooks. Regression guard for LIT-4447.""" + """On an async request the sync failure_handler runs only to service legacy sync-only + callbacks; when the async sync-sweep passes run_custom_logger_hooks=False it must skip + the CustomLogger hook so a both-hook logger fires exactly once, via the async hook. + Regression guard for LIT-4447.""" from litellm.integrations.custom_logger import CustomLogger class DummyLogger(CustomLogger): @@ -870,7 +864,6 @@ class DummyLogger(CustomLogger): logging_obj.stream = False logging_obj.call_type = "anthropic_messages" - logging_obj.is_async_entrypoint = True logging_obj.model_call_details["litellm_params"] = {} logging_obj.litellm_params = {} @@ -887,7 +880,7 @@ class DummyLogger(CustomLogger): ), ): await logging_obj.async_failure_handler(exception, "traceback") - logging_obj.failure_handler(exception, "traceback") + logging_obj.failure_handler(exception, "traceback", run_custom_logger_hooks=False) mock_async_log.assert_awaited_once() mock_sync_log.assert_not_called() @@ -951,7 +944,6 @@ def log_failure_event(self, kwargs, response_obj, start_time, end_time): logging_obj.stream = False logging_obj.call_type = "anthropic_messages" - logging_obj.is_async_entrypoint = True logging_obj.model_call_details["litellm_params"] = {} logging_obj.litellm_params = {} @@ -976,7 +968,7 @@ def log_failure_event(self, kwargs, response_obj, start_time, end_time): return_value=[sync_only_logger], ): await logging_obj.async_success_handler(result=model_response) - logging_obj.success_handler(result=model_response) + logging_obj.success_handler(result=model_response, run_custom_logger_hooks=False) assert events == ["sync_success"] @@ -989,7 +981,7 @@ def log_failure_event(self, kwargs, response_obj, start_time, end_time): return_value=[sync_only_logger], ): await logging_obj.async_failure_handler(exception, "traceback") - logging_obj.failure_handler(exception, "traceback") + logging_obj.failure_handler(exception, "traceback", run_custom_logger_hooks=False) assert events == ["sync_failure"] @@ -1038,7 +1030,6 @@ def log_success_event(self, kwargs, response_obj, start_time, end_time): logging_obj.stream = False logging_obj.call_type = "anthropic_messages" - logging_obj.is_async_entrypoint = True logging_obj.model_call_details["litellm_params"] = {} logging_obj.litellm_params = {} logging_obj.dynamic_success_callbacks = None @@ -1072,12 +1063,13 @@ def log_success_event(self, kwargs, response_obj, start_time, end_time): @pytest.mark.asyncio -async def test_async_client_entrypoint_stamps_and_dedupes_flagless_call_type(): +async def test_async_client_entrypoint_dedupes_flagless_call_type(): """End-to-end through the real @client wrapper: litellm.anthropic_messages sets no - `a*` flag in litellm_params, so only the wrapper-stamped is_async_entrypoint marks - the request async. With the executor sync dispatch open (a surviving plain-callable - callback, same mechanism as a legacy string callback), the CustomLogger must fire - via the async hook exactly once. Regression guard for LIT-4447 and LIT-4475.""" + `a*` flag in litellm_params, yet the async path fires async_log_success_event and then + runs the sync sweep with run_custom_logger_hooks=False. With the executor sync dispatch + open (a surviving plain-callable callback, same mechanism as a legacy string callback), + a both-hook CustomLogger must fire via the async hook exactly once. Regression guard + for LIT-4447.""" events = [] @@ -1120,17 +1112,15 @@ def _gate_opener(kwargs, completion_response, start_time, end_time): @pytest.mark.asyncio -async def test_client_wrapper_stamp_is_first_wins_across_nested_calls(): - """An async entrypoint that internally invokes a sync @client function with the - shared logging object (e.g. agenerate_content delegating to generate_content) must - keep is_async_entrypoint=True; the inner sync wrapper must not overwrite the - entrypoint's stamp, and the nested request must reach a CustomLogger exactly once, - via the async hook. Regression guard for LIT-4475 (gemini /generate_content kept - double-logging because the inner sync wrapper flipped the bit back to sync).""" +async def test_nested_async_to_sync_client_dedupes_custom_logger(): + """An async entrypoint that internally invokes a sync @client function on the shared + logging object (e.g. agenerate_content delegating to generate_content) must reach a + both-hook CustomLogger exactly once, not twice. Regression guard for LIT-4475 (gemini + /generate_content double-logged because the inner sync wrapper also ran the CustomLogger + hook).""" from litellm.litellm_core_utils.thread_pool_executor import executor from litellm.utils import client - observed = {} events = [] class Rec(CustomLogger): @@ -1145,7 +1135,6 @@ def _gate_opener(kwargs, completion_response, start_time, end_time): @client def fake_inner_sync(model: str, messages=None, **kwargs): - observed["inner_bit"] = kwargs["litellm_logging_obj"].is_async_entrypoint return ModelResponse( model=model, choices=[{"message": {"role": "assistant", "content": "ok"}, "finish_reason": "stop", "index": 0}], @@ -1154,9 +1143,7 @@ def fake_inner_sync(model: str, messages=None, **kwargs): @client async def fake_outer_async(model: str, messages=None, **kwargs): - result = fake_inner_sync(model=model, messages=messages, **kwargs) - observed["outer_bit_after_inner"] = kwargs["litellm_logging_obj"].is_async_entrypoint - return result + return fake_inner_sync(model=model, messages=messages, **kwargs) rec = Rec() original_success = litellm.success_callback @@ -1174,12 +1161,7 @@ async def fake_outer_async(model: str, messages=None, **kwargs): litellm.success_callback = original_success litellm._async_success_callback = original_async - assert observed == {"inner_bit": True, "outer_bit_after_inner": True} - assert events == ["async"] - - fake_inner_sync(model="gpt-4.1-mini", messages=[{"role": "user", "content": "hi"}]) - - assert observed["inner_bit"] is False + assert len(events) == 1 def test_get_litellm_params_propagates_allm_passthrough_route(logging_obj):