Skip to content

fix(logging): dedupe async CustomLogger hooks with an explicit sweep flag #33577

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
yucheng-berri merged 1 commit into from
Jul 16, 2026
Merged

fix(logging): dedupe async CustomLogger hooks with an explicit sweep flag #33577

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
46 changes: 32 additions & 14 deletions litellm/litellm_core_utils/litellm_logging.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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,
)

Expand Down Expand Up @@ -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
Expand All @@ -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[
Expand Down Expand Up @@ -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(
Expand Down Expand Up @@ -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:
Expand Down
6 changes: 1 addition & 5 deletions litellm/utils.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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
Expand Down
68 changes: 25 additions & 43 deletions tests/test_litellm/litellm_core_utils/test_litellm_logging.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand All @@ -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 = {}

Expand Down Expand Up @@ -852,25 +845,25 @@ 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()


@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):
pass

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 = {}

Expand All @@ -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()
Expand Down Expand Up @@ -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 = {}

Expand All @@ -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"]

Expand All @@ -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"]

Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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 = []

Expand Down Expand Up @@ -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):
Expand All @@ -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}],
Expand All @@ -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
Expand All @@ -1174,12 +1161,7 @@ async def fake_outer_async(model: str, messages=None, **kwargs):
litellm.success_callback = original_success

@greptile-apps greptile-apps Bot Jul 16, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Test assertion weakened — delivery channel no longer verified

The previous assertion was assert events == ["async"], which confirmed the delivery arrived via the async hook. This is replaced by assert len(events) == 1, which only confirms count. The PR description acknowledges the behavioral change (nested async-to-sync delivers via sync hook now), so the new assertion is technically correct for the new behavior. However, the change means that a future regression where the delivery silently flips back to the async hook (or swaps channel unexpectedly) would not be caught. Consider asserting the specific channel, e.g. assert events == ["sync"], so the delivery path is locked.

Rule Used: What: Flag any modifications to existing tests and... (source)

Note: If this suggestion doesn't match your team's coding style, reply to this and let me know. I'll remember it for next time!

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):
Expand Down
Loading