fix: preserve pass_through call_type during proxy error logging

_handle_logging_proxy_only_error was overwriting call_type from
"pass_through_endpoint" to "acompletion" when request_data contained
"messages" (e.g. Claude Code CLI requests). This defeated the dedup
guard in failure_handler, causing duplicate logs to persist.

Skip the call_type overwrite when the logging object already has
call_type == pass_through_endpoint.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: Harshit28j

litellm/proxy/utils.py

@@ -1843,20 +1843,26 @@ async def _handle_logging_proxy_only_error(
             )
 
             input: Union[list, str, dict] = ""
+            _is_pass_through = (
+                litellm_logging_obj.call_type == CallTypes.pass_through.value
+            )
             if "messages" in request_data and isinstance(
                 request_data["messages"], list
             ):
                 input = request_data["messages"]
                 litellm_logging_obj.model_call_details["messages"] = input
-                litellm_logging_obj.call_type = CallTypes.acompletion.value
+                if not _is_pass_through:
+                    litellm_logging_obj.call_type = CallTypes.acompletion.value
             elif "prompt" in request_data and isinstance(request_data["prompt"], str):
                 input = request_data["prompt"]
                 litellm_logging_obj.model_call_details["prompt"] = input
-                litellm_logging_obj.call_type = CallTypes.atext_completion.value
+                if not _is_pass_through:
+                    litellm_logging_obj.call_type = CallTypes.atext_completion.value
             elif "input" in request_data and isinstance(request_data["input"], list):
                 input = request_data["input"]
                 litellm_logging_obj.model_call_details["input"] = input
-                litellm_logging_obj.call_type = CallTypes.aembedding.value
+                if not _is_pass_through:
+                    litellm_logging_obj.call_type = CallTypes.aembedding.value
             litellm_logging_obj.pre_call(
                 input=input,
                 api_key="",

tests/test_litellm/proxy/pass_through_endpoints/test_pass_through_endpoints.py

@@ -345,6 +345,79 @@ def test_non_pass_through_failure_calls_custom_logger_callback():
         litellm.failure_callback = original_failure_callback
 
 
+@pytest.mark.asyncio
+async def test_handle_logging_proxy_only_error_preserves_pass_through_call_type():
+    """
+    Test that _handle_logging_proxy_only_error does NOT overwrite call_type
+    from "pass_through_endpoint" to "acompletion" when request_data contains
+    "messages". This ensures the dedup guard in failure_handler works even for
+    Claude Code CLI requests forwarded via pass-through with a chat body.
+    """
+    from unittest.mock import AsyncMock, MagicMock, patch
+
+    from litellm.integrations.custom_logger import CustomLogger
+    from litellm.litellm_core_utils.litellm_logging import Logging
+    from litellm.proxy._types import UserAPIKeyAuth
+
+    # Create a logging_obj with pass_through call_type (as pass-through endpoints do)
+    logging_obj = Logging(
+        model="claude-sonnet-4-20250514",
+        messages=[{"role": "user", "content": "hello"}],
+        stream=False,
+        call_type="pass_through_endpoint",
+        start_time=None,
+        litellm_call_id="test-call-id",
+        function_id="test",
+    )
+
+    # request_data with "messages" — simulates a Claude Code CLI chat request
+    request_data = {
+        "model": "claude-sonnet-4-20250514",
+        "messages": [{"role": "user", "content": "hello"}],
+        "litellm_logging_obj": logging_obj,
+        "call_type": "pass_through_endpoint",
+    }
+
+    mock_user_api_key_dict = MagicMock(spec=UserAPIKeyAuth)
+
+    # Mock the proxy_logging_obj to get access to _handle_logging_proxy_only_error
+    from litellm.proxy.utils import ProxyLogging
+
+    proxy_logging = ProxyLogging(user_api_key_cache=MagicMock())
+
+    # Track what happens to call_type through the flow
+    original_async_failure = logging_obj.async_failure_handler
+    call_types_seen = []
+
+    async def tracking_async_failure(**kwargs):
+        call_types_seen.append(("async", logging_obj.call_type))
+        # Don't actually run handlers — just track
+        return
+
+    logging_obj.async_failure_handler = tracking_async_failure
+
+    original_sync_failure = logging_obj.failure_handler
+
+    def tracking_sync_failure(*args, **kwargs):
+        call_types_seen.append(("sync", logging_obj.call_type))
+        return
+
+    logging_obj.failure_handler = tracking_sync_failure
+
+    await proxy_logging._handle_logging_proxy_only_error(
+        request_data=request_data,
+        user_api_key_dict=mock_user_api_key_dict,
+        route="/v1/chat/completions",
+        original_exception=Exception("test error"),
+    )
+
+    # call_type should still be "pass_through_endpoint", NOT "acompletion"
+    assert logging_obj.call_type == "pass_through_endpoint", (
+        f"call_type was mutated to '{logging_obj.call_type}', "
+        "expected 'pass_through_endpoint'"
+    )
+
+
 def test_is_langfuse_route():
     """
     Test that the is_langfuse_route method correctly identifies Langfuse routes