fix(logging): classify allm_passthrough_route as async to prevent duplicate success callbacks (#32265)

* fix(logging): classify allm_passthrough_route as async to prevent duplicate success callbacks

Async passthrough requests set kwargs["allm_passthrough_route"]=True but that
flag is never propagated into litellm_params, and _is_sync_litellm_request only
checks acompletion/aresponses/aembedding/aimage_generation/atranscription.
Every async passthrough is misclassified as sync, which trips the CustomLogger
sync branch in success_handler and fires log_success_event in addition to the
async worker's async_log_success_event, causing 2-3 duplicate LangSmith runs
per Bedrock passthrough request

Propagate allm_passthrough_route through get_litellm_params and teach the
classifier about it. /chat/completions and other non-passthrough paths are
untouched

* test(passthrough): assert allm_passthrough_route flag propagates end-to-end

Integration-level guard on top of the unit tests in test_litellm_logging.py:
verifies that when kwargs["allm_passthrough_route"]=True enters
llm_passthrough_route, the flag survives get_litellm_params(**kwargs), lands
in the logging object's litellm_params, and _is_sync_litellm_request reads
the request as async

---------

Co-authored-by: yucheng <yucheng@yuchengs-MBP.localdomain>

litellm/litellm_core_utils/get_litellm_params.py

@@ -76,6 +76,7 @@ def get_litellm_params(
     proxy_server_request=None,
     acompletion=None,
     aembedding=None,
+    allm_passthrough_route=None,
     preset_cache_key=None,
     no_log=None,
     input_cost_per_second=None,
@@ -118,6 +119,7 @@ def get_litellm_params(
     # Build base dict with explicit parameters (always included)
     litellm_params = {
         "acompletion": acompletion,
+        "allm_passthrough_route": allm_passthrough_route,
         "api_key": api_key,
         "force_timeout": force_timeout,
         "logger_fn": logger_fn,

litellm/litellm_core_utils/litellm_logging.py

@@ -1530,6 +1530,7 @@ class Logging(LiteLLMLoggingBaseClass):
             and litellm_params.get(CallTypes.aembedding.value, False) is not True
             and litellm_params.get(CallTypes.aimage_generation.value, False) is not True
             and litellm_params.get(CallTypes.atranscription.value, False) is not True
+            and litellm_params.get(CallTypes.allm_passthrough_route.value, False) is not True
         )
 
     def _is_assembled_stream_success(self, result=None) -> bool:

litellm/passthrough/main.py

@@ -171,7 +171,6 @@ def llm_passthrough_route(
     api_key: Optional[str] = None,
     request_query_params: Optional[dict] = None,
     request_headers: Optional[dict] = None,
-    allm_passthrough_route: bool = False,
     content: Optional[Any] = None,
     data: Optional[dict] = None,
     files: Optional[RequestFiles] = None,
@@ -198,7 +197,7 @@ def llm_passthrough_route(
     from litellm.types.utils import LlmProviders
     from litellm.utils import ProviderConfigManager
 
-    _is_async = allm_passthrough_route
+    _is_async = bool(kwargs.get("allm_passthrough_route", False))
 
     litellm_logging_obj = cast("LiteLLMLoggingObj", kwargs.get("litellm_logging_obj"))
 

tests/test_litellm/litellm_core_utils/test_litellm_logging.py

@@ -704,7 +704,9 @@ async def test_logging_non_streaming_request():
         litellm.callbacks = original_callbacks
 
 
-@pytest.mark.parametrize("async_flag", ["acompletion", "aresponses"])
+@pytest.mark.parametrize(
+    "async_flag", ["acompletion", "aresponses", "allm_passthrough_route"]
+)
 def test_success_handler_skips_sync_callbacks_for_async_requests(
     logging_obj, async_flag
 ):
@@ -792,6 +794,21 @@ def test_success_handler_runs_sync_callbacks_for_sync_requests(logging_obj, call
 def test_is_sync_litellm_request():
     assert LitellmLogging._is_sync_litellm_request({}) is True
     assert LitellmLogging._is_sync_litellm_request({"acompletion": True}) is False
+    assert (
+        LitellmLogging._is_sync_litellm_request({"allm_passthrough_route": True})
+        is False
+    )
+
+
+def test_get_litellm_params_propagates_allm_passthrough_route():
+    """`allm_passthrough_route=True` set on kwargs by the async passthrough entrypoint
+    must land in `litellm_params` so `_is_sync_litellm_request` sees it and the
+    request is classified as async. Regression guard for LIT-4192."""
+    from litellm.litellm_core_utils.get_litellm_params import get_litellm_params
+
+    params = get_litellm_params(allm_passthrough_route=True)
+    assert params.get("allm_passthrough_route") is True
+    assert LitellmLogging._is_sync_litellm_request(params) is False
 
 
 @pytest.mark.asyncio

tests/test_litellm/passthrough/test_passthrough_main.py

@@ -726,3 +726,78 @@ async def test_allm_passthrough_route_429_streaming_raises():
 
     assert exc_info.value.response.status_code == 429
     assert len(chunks) == 0, "No chunks should be yielded before the 429 raises"
+
+
+def test_llm_passthrough_route_propagates_allm_passthrough_route_to_logging_obj():
+    """
+    Regression guard for LIT-4192: `allm_passthrough_route` sets
+    `kwargs["allm_passthrough_route"] = True` on the async entrypoint, and the
+    inner `llm_passthrough_route` must let that flag flow through
+    `get_litellm_params(**kwargs)` and land in the logging object's
+    `litellm_params`. Without that, `_is_sync_litellm_request` misclassifies
+    the request as sync and fires duplicate success callbacks.
+    """
+    import asyncio
+
+    from litellm.litellm_core_utils.litellm_logging import Logging as LitellmLogging
+
+    client = HTTPHandler()
+
+    mock_provider_config = MagicMock()
+    mock_provider_config.get_complete_url.return_value = (
+        httpx.URL("https://bedrock-runtime.us-east-1.amazonaws.com/model/foo/converse"),
+        "https://bedrock-runtime.us-east-1.amazonaws.com",
+    )
+    mock_provider_config.get_api_key.return_value = "fake-key"
+    mock_provider_config.validate_environment.return_value = {}
+    mock_provider_config.sign_request.return_value = ({}, None)
+    mock_provider_config.is_streaming_request.return_value = False
+
+    captured_litellm_params: dict = {}
+
+    def _capture_update_env(*args, **kwargs):
+        captured_litellm_params.clear()
+        captured_litellm_params.update(kwargs.get("litellm_params") or {})
+
+    mock_logging_obj = MagicMock()
+    mock_logging_obj.update_environment_variables.side_effect = _capture_update_env
+
+    with (
+        patch(
+            "litellm.utils.ProviderConfigManager.get_provider_passthrough_config",
+            return_value=mock_provider_config,
+        ),
+        patch(
+            "litellm.litellm_core_utils.get_llm_provider_logic.get_llm_provider",
+            return_value=(
+                "bedrock/foo",
+                "bedrock",
+                "fake-key",
+                "https://bedrock-runtime.us-east-1.amazonaws.com",
+            ),
+        ),
+        patch.object(
+            client.client,
+            "send",
+            return_value=MagicMock(status_code=200, json=lambda: {}),
+        ),
+        patch.object(client.client, "build_request"),
+    ):
+        result = llm_passthrough_route(
+            model="bedrock/foo",
+            endpoint="model/foo/converse",
+            method="POST",
+            custom_llm_provider="bedrock",
+            api_base="https://bedrock-runtime.us-east-1.amazonaws.com",
+            api_key="fake-key",
+            json={"messages": []},
+            client=client,
+            litellm_logging_obj=mock_logging_obj,
+            allm_passthrough_route=True,
+        )
+
+    if asyncio.iscoroutine(result):
+        result.close()
+
+    assert captured_litellm_params.get("allm_passthrough_route") is True
+    assert LitellmLogging._is_sync_litellm_request(captured_litellm_params) is False