fix: reduce proxy overhead for large base64 payloads (#21594)

* fix aviation safety topic filter: remove overly broad exceptions, add cockpit access block words

* fix airline brand protection filter: add identifier words, competitor/ops block words, tighten exceptions

* add constants for large payload handling and detailed timing

* add base64 truncation for logging payloads

* use shallow copy for messages, track copy and callback timing

* add callback duration and detailed timing to response metadata

* add callback duration header, size-gate debug logging, detailed timing headers

* add tests for callback timing, base64 truncation, and detailed timing

* fix code quality: extract helpers, fix regex, clean up imports

* rewrite _truncate_base64_in_value iteratively to satisfy recursive detector

litellm/constants.py

@@ -49,6 +49,19 @@ DEFAULT_REPLICATE_POLLING_DELAY_SECONDS = int(
 )
 DEFAULT_IMAGE_TOKEN_COUNT = int(os.getenv("DEFAULT_IMAGE_TOKEN_COUNT", 250))
 
+# Maximum number of base64 characters to keep in logging payloads.
+# Data URIs exceeding this are replaced with a size placeholder.
+# Set to 0 to disable truncation.
+MAX_BASE64_LENGTH_FOR_LOGGING = int(
+    os.getenv("MAX_BASE64_LENGTH_FOR_LOGGING", 64)
+)
+
+# When true, adds detailed per-phase timing breakdown headers to responses.
+# Headers: x-litellm-timing-{pre-processing,llm-api,post-processing,message-copy}-ms
+LITELLM_DETAILED_TIMING = (
+    os.getenv("LITELLM_DETAILED_TIMING", "false").lower() == "true"
+)
+
 # Model cost map validation constants
 MODEL_COST_MAP_MIN_MODEL_COUNT = int(
     os.getenv("MODEL_COST_MAP_MIN_MODEL_COUNT", 50)
@@ -1475,6 +1488,12 @@ MICROSOFT_USER_LAST_NAME_ATTRIBUTE = str(
     os.getenv("MICROSOFT_USER_LAST_NAME_ATTRIBUTE", "surname")
 )
 
+# Maximum payload size (in bytes) to fully serialize for DEBUG logging.
+# Payloads larger than this are truncated to avoid multi-second json.dumps blocking the response.
+MAX_PAYLOAD_SIZE_FOR_DEBUG_LOG = int(
+    os.getenv("MAX_PAYLOAD_SIZE_FOR_DEBUG_LOG", 102400)
+)  # 100 KB
+
 # Policy template enrichment
 MAX_COMPETITOR_NAMES = int(os.getenv("MAX_COMPETITOR_NAMES", 100))
 COMPETITOR_LLM_TEMPERATURE = float(os.getenv("COMPETITOR_LLM_TEMPERATURE", 0.3))

litellm/litellm_core_utils/litellm_logging.py

@@ -64,6 +64,7 @@ from litellm.litellm_core_utils.get_litellm_params import get_litellm_params
 from litellm.litellm_core_utils.llm_cost_calc.tool_call_cost_tracking import (
     StandardBuiltInToolCostTracking,
 )
+from litellm.litellm_core_utils.logging_utils import truncate_base64_in_messages
 from litellm.litellm_core_utils.model_param_helper import ModelParamHelper
 from litellm.litellm_core_utils.redact_messages import (
     redact_message_input_output_from_custom_logger,
@@ -334,7 +335,12 @@ class Logging(LiteLLMLoggingBaseClass):
                 messages = new_messages
 
         self.model = model
-        self.messages = copy.deepcopy(messages) if messages is not None else None
+        # Shallow copy of the outer list only (inner message dicts are shared).
+        # Safe because the logging layer does not mutate individual message dicts.
+        _copy_start = time.time()
+        self.messages = copy.copy(messages) if messages is not None else None
+        self.message_copy_duration_ms: float = (time.time() - _copy_start) * 1000
+        self.callback_duration_ms: float = 0.0
         self.stream = stream
         self.start_time = start_time  # log the call start time
         self.call_type = call_type
@@ -1629,15 +1635,26 @@ class Logging(LiteLLMLoggingBaseClass):
 
         self.model_call_details[
             "standard_logging_object"
-        ] = get_standard_logging_object_payload(
+        ] = self._build_standard_logging_payload(
+            logging_result, start_time, end_time
+        )
+
+    def _build_standard_logging_payload(
+        self, init_response_obj: Any, start_time: Any, end_time: Any
+    ) -> Any:
+        """Build StandardLoggingPayload and accumulate its construction time."""
+        _start = time.time()
+        payload = get_standard_logging_object_payload(
             kwargs=self.model_call_details,
-            init_response_obj=logging_result,
+            init_response_obj=init_response_obj,
             start_time=start_time,
             end_time=end_time,
             logging_obj=self,
             status="success",
             standard_built_in_tools_params=self.standard_built_in_tools_params,
         )
+        self.callback_duration_ms += (time.time() - _start) * 1000
+        return payload
 
     def _transform_usage_objects(self, result):
         if isinstance(result, ResponsesAPIResponse):
@@ -1732,14 +1749,8 @@ class Logging(LiteLLMLoggingBaseClass):
                 elif isinstance(result, dict) or isinstance(result, list):
                     self.model_call_details[
                         "standard_logging_object"
-                    ] = get_standard_logging_object_payload(
-                        kwargs=self.model_call_details,
-                        init_response_obj=result,
-                        start_time=start_time,
-                        end_time=end_time,
-                        logging_obj=self,
-                        status="success",
-                        standard_built_in_tools_params=self.standard_built_in_tools_params,
+                    ] = self._build_standard_logging_payload(
+                        result, start_time, end_time
                     )
             elif standard_logging_object is not None:
                 self.model_call_details[
@@ -1911,14 +1922,8 @@ class Logging(LiteLLMLoggingBaseClass):
                 ## STANDARDIZED LOGGING PAYLOAD
                 self.model_call_details[
                     "standard_logging_object"
-                ] = get_standard_logging_object_payload(
-                    kwargs=self.model_call_details,
-                    init_response_obj=complete_streaming_response,
-                    start_time=start_time,
-                    end_time=end_time,
-                    logging_obj=self,
-                    status="success",
-                    standard_built_in_tools_params=self.standard_built_in_tools_params,
+                ] = self._build_standard_logging_payload(
+                    complete_streaming_response, start_time, end_time
                 )
                 if (
                     standard_logging_payload := self.model_call_details.get(
@@ -2435,14 +2440,8 @@ class Logging(LiteLLMLoggingBaseClass):
             ## STANDARDIZED LOGGING PAYLOAD
             self.model_call_details[
                 "standard_logging_object"
-            ] = get_standard_logging_object_payload(
-                kwargs=self.model_call_details,
-                init_response_obj=complete_streaming_response,
-                start_time=start_time,
-                end_time=end_time,
-                logging_obj=self,
-                status="success",
-                standard_built_in_tools_params=self.standard_built_in_tools_params,
+            ] = self._build_standard_logging_payload(
+                complete_streaming_response, start_time, end_time
             )
 
             # print standard logging payload
@@ -2465,14 +2464,8 @@ class Logging(LiteLLMLoggingBaseClass):
             ## STANDARDIZED LOGGING PAYLOAD
             self.model_call_details[
                 "standard_logging_object"
-            ] = get_standard_logging_object_payload(
-                kwargs=self.model_call_details,
-                init_response_obj=result,
-                start_time=start_time,
-                end_time=end_time,
-                logging_obj=self,
-                status="success",
-                standard_built_in_tools_params=self.standard_built_in_tools_params,
+            ] = self._build_standard_logging_payload(
+                result, start_time, end_time
             )
 
             # print standard logging payload
@@ -5213,8 +5206,10 @@ def get_standard_logging_object_payload(
             model_id=_model_id,
             requester_ip_address=clean_metadata.get("requester_ip_address", None),
             user_agent=clean_metadata.get("user_agent", None),
-            messages=StandardLoggingPayloadSetup.append_system_prompt_messages(
-                kwargs=kwargs, messages=kwargs.get("messages")
+            messages=truncate_base64_in_messages(
+                StandardLoggingPayloadSetup.append_system_prompt_messages(
+                    kwargs=kwargs, messages=kwargs.get("messages")
+                )
             ),
             response=final_response_obj,
             model_parameters=ModelParamHelper.get_standard_logging_model_parameters(

litellm/litellm_core_utils/llm_response_utils/response_metadata.py

@@ -1,6 +1,7 @@
 import datetime
 from typing import Any, Optional, Union
 
+from litellm.constants import LITELLM_DETAILED_TIMING
 from litellm.litellm_core_utils.core_helpers import process_response_headers
 from litellm.litellm_core_utils.llm_response_utils.get_api_base import get_api_base
 from litellm.litellm_core_utils.logging_utils import LiteLLMLoggingObject
@@ -108,7 +109,18 @@ class ResponseMetadata:
             )
 
         #########################################################
-        # 3. Add duration for reading from cache
+        # 3. Add callback processing duration
+        #########################################################
+        callback_duration_ms = getattr(logging_obj, "callback_duration_ms", None)
+        if callback_duration_ms is not None:
+            self._update_hidden_params(
+                {
+                    "callback_duration_ms": round(callback_duration_ms, 4),
+                }
+            )
+
+        #########################################################
+        # 4. Add duration for reading from cache
         # In this case overhead from litellm is the difference between the cache read duration and the total response time
         #########################################################
         if (
@@ -128,6 +140,31 @@ class ResponseMetadata:
                 }
             )
 
+        #########################################################
+        # 5. Detailed per-phase timing (opt-in via env var)
+        #########################################################
+        if LITELLM_DETAILED_TIMING and llm_api_duration_ms is not None:
+            detailed: dict = {
+                "timing_llm_api_ms": round(llm_api_duration_ms, 4),
+            }
+
+            # message copy time from Logging.__init__()
+            msg_copy_ms = getattr(logging_obj, "message_copy_duration_ms", None)
+            if msg_copy_ms is not None:
+                detailed["timing_message_copy_ms"] = round(msg_copy_ms, 4)
+
+            # pre-processing = time from request start to LLM API call start
+            api_call_start = logging_obj.model_call_details.get("api_call_start_time")
+            if api_call_start is not None and start_time is not None:
+                pre_ms = (api_call_start - start_time).total_seconds() * 1000
+                detailed["timing_pre_processing_ms"] = round(pre_ms, 4)
+
+                # post-processing = total - pre - llm_api
+                post_ms = total_response_time_ms - pre_ms - llm_api_duration_ms
+                detailed["timing_post_processing_ms"] = round(max(post_ms, 0), 4)
+
+            self._update_hidden_params(detailed)
+
     def apply(self) -> None:
         """Apply metadata to the response object"""
         if hasattr(self.result, "_hidden_params"):

litellm/litellm_core_utils/logging_utils.py

@@ -1,11 +1,13 @@
 import asyncio
 import functools
 import inspect
+import re
 import time
 from datetime import datetime
 from typing import TYPE_CHECKING, Any, List, Optional, Union
 
 from litellm._logging import verbose_logger
+from litellm.constants import MAX_BASE64_LENGTH_FOR_LOGGING
 from litellm.types.utils import (
     ModelResponse,
     ModelResponseStream,
@@ -34,6 +36,110 @@ import litellm
 Helper utils used for logging callbacks
 """
 
+_BYTES_PER_KIB = 1024
+_BYTES_PER_MIB = 1024 * 1024
+
+# Regex matching data-URI base64 content: "data:<mime>;base64,<payload>"
+# Captures: group(1)=mime_type, group(2)=base64_payload
+_DATA_URI_RE = re.compile(r"data:([^;]+);base64,([A-Za-z0-9+/=]+)")
+
+# Maximum nesting depth for _truncate_base64_in_value to guard against
+# pathological payloads. OpenAI message format is typically 3-4 levels deep.
+_MAX_TRUNCATION_DEPTH = 20
+
+
+def _format_base64_size(num_chars: int) -> str:
+    """Return a human-readable byte-size estimate from a base64 character count."""
+    num_bytes = num_chars * 3 / 4
+    if num_bytes >= _BYTES_PER_MIB:
+        return f"{num_bytes / _BYTES_PER_MIB:.2f}MB"
+    if num_bytes >= _BYTES_PER_KIB:
+        return f"{num_bytes / _BYTES_PER_KIB:.1f}KB"
+    return f"{int(num_bytes)}B"
+
+
+def _base64_data_uri_replacer(match: re.Match) -> str:
+    """Replace a single base64 data-URI match with a size placeholder if too long."""
+    mime_type = match.group(1)
+    payload = match.group(2)
+    if len(payload) <= MAX_BASE64_LENGTH_FOR_LOGGING:
+        return match.group(0)
+    size_str = _format_base64_size(len(payload))
+    return f"data:{mime_type};base64,[base64_data truncated: {size_str}]"
+
+
+def _truncate_base64_in_string(value: str) -> str:
+    """Replace long base64 data-URI payloads in a string with a size placeholder."""
+    if MAX_BASE64_LENGTH_FOR_LOGGING <= 0:
+        return value
+    return _DATA_URI_RE.sub(_base64_data_uri_replacer, value)
+
+
+def _truncate_base64_in_value(value: Any) -> Any:
+    """Iteratively truncate base64 data URIs in a JSON-like value (str/list/dict).
+
+    Uses an explicit stack instead of recursion to satisfy the project's
+    recursive-function detector and avoid stack-overflow on deep payloads.
+    """
+    # Stack entries: (source_value, depth, parent_container, key_or_index)
+    # We mutate *copies* of dicts/lists in-place via parent references.
+    if isinstance(value, str):
+        return _truncate_base64_in_string(value)
+    if not isinstance(value, (dict, list)):
+        return value
+
+    # Shallow-copy the root so we don't mutate the caller's data.
+    root = {k: v for k, v in value.items()} if isinstance(value, dict) else list(value)
+    stack: list = [(root, 0)]
+
+    while stack:
+        container, depth = stack.pop()
+        if depth > _MAX_TRUNCATION_DEPTH:
+            continue
+        if isinstance(container, dict):
+            for k, v in container.items():
+                if isinstance(v, str):
+                    container[k] = _truncate_base64_in_string(v)
+                elif isinstance(v, dict):
+                    copy = {ck: cv for ck, cv in v.items()}
+                    container[k] = copy
+                    stack.append((copy, depth + 1))
+                elif isinstance(v, list):
+                    copy = list(v)
+                    container[k] = copy
+                    stack.append((copy, depth + 1))
+        elif isinstance(container, list):
+            for i, v in enumerate(container):
+                if isinstance(v, str):
+                    container[i] = _truncate_base64_in_string(v)
+                elif isinstance(v, dict):
+                    copy = {ck: cv for ck, cv in v.items()}
+                    container[i] = copy
+                    stack.append((copy, depth + 1))
+                elif isinstance(v, list):
+                    copy = list(v)
+                    container[i] = copy
+                    stack.append((copy, depth + 1))
+
+    return root
+
+
+def truncate_base64_in_messages(
+    messages: Optional[Union[str, list, dict]],
+) -> Optional[Union[str, list, dict]]:
+    """
+    Return a copy of *messages* with long base64 data-URI payloads replaced
+    by human-readable size placeholders.
+    """
+    if messages is None or MAX_BASE64_LENGTH_FOR_LOGGING <= 0:
+        return messages
+    try:
+        return _truncate_base64_in_value(messages)
+    except Exception as e:
+        verbose_logger.debug("Failed to truncate base64 in messages: %s", e)
+        return messages
+
+
 # Global service logger instance to avoid recreating it
 _service_logger = None
 

litellm/proxy/common_request_processing.py

@@ -24,6 +24,8 @@ from litellm._logging import verbose_proxy_logger
 from litellm._uuid import uuid
 from litellm.constants import (
     DD_TRACER_STREAMING_CHUNK_YIELD_RESOURCE,
+    LITELLM_DETAILED_TIMING,
+    MAX_PAYLOAD_SIZE_FOR_DEBUG_LOG,
     STREAM_SSE_DATA_PREFIX,
 )
 from litellm.litellm_core_utils.dd_tracing import tracer
@@ -434,6 +436,19 @@ class ProxyBaseLLMRequestProcessing:
             "x-litellm-overhead-duration-ms": str(
                 hidden_params.get("litellm_overhead_time_ms", None)
             ),
+            "x-litellm-callback-duration-ms": str(
+                hidden_params.get("callback_duration_ms", None)
+            ),
+            **(
+                {
+                    "x-litellm-timing-pre-processing-ms": str(hidden_params.get("timing_pre_processing_ms", None)),
+                    "x-litellm-timing-llm-api-ms": str(hidden_params.get("timing_llm_api_ms", None)),
+                    "x-litellm-timing-post-processing-ms": str(hidden_params.get("timing_post_processing_ms", None)),
+                    "x-litellm-timing-message-copy-ms": str(hidden_params.get("timing_message_copy_ms", None)),
+                }
+                if LITELLM_DETAILED_TIMING
+                else {}
+            ),
             "x-litellm-fastest_response_batch_completion": (
                 str(fastest_response_batch_completion)
                 if fastest_response_batch_completion is not None
@@ -685,6 +700,24 @@ class ProxyBaseLLMRequestProcessing:
             model_id = model_info.get("id", "") or ""
         return model_id
 
+    def _debug_log_request_payload(self) -> None:
+        """Log request payload at DEBUG level, truncating if too large."""
+        if not verbose_proxy_logger.isEnabledFor(logging.DEBUG):
+            return
+        _payload_str = json.dumps(self.data, default=str)
+        if len(_payload_str) > MAX_PAYLOAD_SIZE_FOR_DEBUG_LOG:
+            verbose_proxy_logger.debug(
+                "Request received by LiteLLM: payload too large to log (%d bytes, limit %d). Keys: %s",
+                len(_payload_str),
+                MAX_PAYLOAD_SIZE_FOR_DEBUG_LOG,
+                list(self.data.keys()) if isinstance(self.data, dict) else type(self.data).__name__,
+            )
+        else:
+            verbose_proxy_logger.debug(
+                "Request received by LiteLLM:\n%s",
+                json.dumps(self.data, indent=4, default=str),
+            )
+
     async def base_process_llm_request(
         self,
         request: Request,
@@ -769,12 +802,7 @@ class ProxyBaseLLMRequestProcessing:
         requested_model_from_client: Optional[str] = (
             self.data.get("model") if isinstance(self.data.get("model"), str) else None
         )
-        if verbose_proxy_logger.isEnabledFor(logging.DEBUG):
-            verbose_proxy_logger.debug(
-                "Request received by LiteLLM:\n{}".format(
-                    json.dumps(self.data, indent=4, default=str)
-                ),
-            )
+        self._debug_log_request_payload()
 
         self.data, logging_obj = await self.common_processing_pre_call_logic(
             request=request,

tests/test_litellm/litellm_core_utils/llm_response_utils/test_response_metadata.py

@@ -0,0 +1,264 @@
+"""
+Tests for litellm.litellm_core_utils.llm_response_utils.response_metadata
+
+Covers the callback_duration_ms timing metric that flows from the Logging object
+through _hidden_params to the x-litellm-callback-duration-ms response header.
+"""
+
+import datetime
+from unittest.mock import MagicMock
+
+import litellm.litellm_core_utils.llm_response_utils.response_metadata as response_metadata_mod
+import litellm.proxy.common_request_processing as common_request_processing_mod
+from litellm.litellm_core_utils.litellm_logging import Logging
+from litellm.litellm_core_utils.llm_response_utils.response_metadata import (
+    ResponseMetadata,
+    update_response_metadata,
+)
+from litellm.proxy._types import UserAPIKeyAuth
+from litellm.proxy.common_request_processing import ProxyBaseLLMRequestProcessing
+from litellm.types.utils import ModelResponse
+
+
+class TestCallbackDurationMs:
+    """Tests for the callback_duration_ms metric in ResponseMetadata."""
+
+    def _make_logging_obj(self, callback_duration_ms=None, llm_api_duration_ms=None):
+        """Build a minimal mock logging object."""
+        logging_obj = MagicMock()
+        logging_obj.model_call_details = {}
+        if llm_api_duration_ms is not None:
+            logging_obj.model_call_details["llm_api_duration_ms"] = llm_api_duration_ms
+        logging_obj.caching_details = None
+        if callback_duration_ms is not None:
+            logging_obj.callback_duration_ms = callback_duration_ms
+        else:
+            # Simulate a Logging object that has no callback_duration_ms
+            del logging_obj.callback_duration_ms
+        return logging_obj
+
+    def test_callback_duration_ms_set_in_hidden_params(self):
+        """When logging_obj has callback_duration_ms, it should appear in _hidden_params."""
+        result = ModelResponse()
+        logging_obj = self._make_logging_obj(callback_duration_ms=12.3456)
+
+        metadata = ResponseMetadata(result)
+        start = datetime.datetime(2025, 1, 1, 0, 0, 0)
+        end = datetime.datetime(2025, 1, 1, 0, 0, 1)
+        metadata.set_timing_metrics(start, end, logging_obj)
+        metadata.apply()
+
+        hidden = result._hidden_params
+        assert hidden.get("callback_duration_ms") == 12.3456
+
+    def test_callback_duration_ms_absent_when_not_on_logging_obj(self):
+        """When logging_obj lacks callback_duration_ms, hidden_params should not have it."""
+        result = ModelResponse()
+        logging_obj = self._make_logging_obj(callback_duration_ms=None)
+
+        metadata = ResponseMetadata(result)
+        start = datetime.datetime(2025, 1, 1, 0, 0, 0)
+        end = datetime.datetime(2025, 1, 1, 0, 0, 1)
+        metadata.set_timing_metrics(start, end, logging_obj)
+        metadata.apply()
+
+        hidden = result._hidden_params
+        assert hidden.get("callback_duration_ms") is None
+
+    def test_update_response_metadata_includes_callback_duration(self):
+        """End-to-end: update_response_metadata should propagate callback_duration_ms."""
+        result = ModelResponse()
+        logging_obj = self._make_logging_obj(
+            callback_duration_ms=5.5, llm_api_duration_ms=800.0
+        )
+        logging_obj._response_cost_calculator = MagicMock(return_value=0.001)
+        logging_obj.litellm_call_id = "test-call-id"
+
+        start = datetime.datetime(2025, 1, 1, 0, 0, 0)
+        end = datetime.datetime(2025, 1, 1, 0, 0, 1)
+
+        update_response_metadata(
+            result=result,
+            logging_obj=logging_obj,
+            model="gpt-4",
+            kwargs={},
+            start_time=start,
+            end_time=end,
+        )
+
+        hidden = result._hidden_params
+        assert hidden.get("callback_duration_ms") == 5.5
+        # overhead should also be set
+        assert hidden.get("litellm_overhead_time_ms") is not None
+
+
+class TestCallbackDurationInCustomHeaders:
+    """Test that callback_duration_ms flows into get_custom_headers."""
+
+    def test_header_present_when_callback_duration_in_hidden_params(self):
+        user_api_key_dict = UserAPIKeyAuth(api_key="sk-test")
+        hidden_params = {
+            "_response_ms": 1000.0,
+            "litellm_overhead_time_ms": 50.0,
+            "callback_duration_ms": 7.25,
+        }
+
+        headers = ProxyBaseLLMRequestProcessing.get_custom_headers(
+            user_api_key_dict=user_api_key_dict,
+            hidden_params=hidden_params,
+        )
+
+        assert "x-litellm-callback-duration-ms" in headers
+        assert headers["x-litellm-callback-duration-ms"] == "7.25"
+
+    def test_header_absent_when_no_callback_duration(self):
+        user_api_key_dict = UserAPIKeyAuth(api_key="sk-test")
+        hidden_params = {
+            "_response_ms": 1000.0,
+        }
+
+        headers = ProxyBaseLLMRequestProcessing.get_custom_headers(
+            user_api_key_dict=user_api_key_dict,
+            hidden_params=hidden_params,
+        )
+
+        # Should be excluded because value is "None" which is in exclude_values
+        assert "x-litellm-callback-duration-ms" not in headers
+
+
+class TestDetailedTiming:
+    """Tests for detailed per-phase timing headers behind LITELLM_DETAILED_TIMING."""
+
+    def _make_logging_obj(
+        self,
+        llm_api_duration_ms=500.0,
+        message_copy_duration_ms=2.5,
+        api_call_start_time=None,
+    ):
+        logging_obj = MagicMock()
+        logging_obj.model_call_details = {
+            "llm_api_duration_ms": llm_api_duration_ms,
+        }
+        if api_call_start_time is not None:
+            logging_obj.model_call_details["api_call_start_time"] = api_call_start_time
+        logging_obj.caching_details = None
+        logging_obj.callback_duration_ms = 1.0
+        logging_obj.message_copy_duration_ms = message_copy_duration_ms
+        return logging_obj
+
+    def test_detailed_timing_headers_present_when_enabled(self, monkeypatch):
+        """When LITELLM_DETAILED_TIMING is true, detailed timing keys appear in hidden_params."""
+        monkeypatch.setattr(response_metadata_mod, "LITELLM_DETAILED_TIMING", True)
+
+        result = ModelResponse()
+        start = datetime.datetime(2025, 1, 1, 0, 0, 0)
+        api_call_start = datetime.datetime(2025, 1, 1, 0, 0, 0, 20000)  # +20ms
+        end = datetime.datetime(2025, 1, 1, 0, 0, 0, 530000)  # +530ms total
+
+        logging_obj = self._make_logging_obj(
+            llm_api_duration_ms=500.0,
+            message_copy_duration_ms=2.5,
+            api_call_start_time=api_call_start,
+        )
+
+        metadata = ResponseMetadata(result)
+        metadata.set_timing_metrics(start, end, logging_obj)
+        metadata.apply()
+
+        hidden = result._hidden_params
+        assert hidden.get("timing_llm_api_ms") == 500.0
+        assert hidden.get("timing_message_copy_ms") == 2.5
+        assert hidden.get("timing_pre_processing_ms") == 20.0
+        assert hidden.get("timing_post_processing_ms") == 10.0  # 530 - 20 - 500
+
+    def test_detailed_timing_absent_when_disabled(self, monkeypatch):
+        """When LITELLM_DETAILED_TIMING is false, no detailed timing keys."""
+        monkeypatch.setattr(response_metadata_mod, "LITELLM_DETAILED_TIMING", False)
+
+        result = ModelResponse()
+        start = datetime.datetime(2025, 1, 1, 0, 0, 0)
+        end = datetime.datetime(2025, 1, 1, 0, 0, 1)
+        logging_obj = self._make_logging_obj()
+
+        metadata = ResponseMetadata(result)
+        metadata.set_timing_metrics(start, end, logging_obj)
+        metadata.apply()
+
+        hidden = result._hidden_params
+        assert hidden.get("timing_llm_api_ms") is None
+        assert hidden.get("timing_pre_processing_ms") is None
+
+    def test_detailed_timing_headers_in_custom_headers(self, monkeypatch):
+        """When LITELLM_DETAILED_TIMING is true, headers flow to get_custom_headers."""
+        monkeypatch.setattr(common_request_processing_mod, "LITELLM_DETAILED_TIMING", True)
+
+        user_api_key_dict = UserAPIKeyAuth(api_key="sk-test")
+        hidden_params = {
+            "_response_ms": 530.0,
+            "timing_llm_api_ms": 500.0,
+            "timing_pre_processing_ms": 20.0,
+            "timing_post_processing_ms": 10.0,
+            "timing_message_copy_ms": 2.5,
+        }
+
+        headers = ProxyBaseLLMRequestProcessing.get_custom_headers(
+            user_api_key_dict=user_api_key_dict,
+            hidden_params=hidden_params,
+        )
+
+        assert headers["x-litellm-timing-llm-api-ms"] == "500.0"
+        assert headers["x-litellm-timing-pre-processing-ms"] == "20.0"
+        assert headers["x-litellm-timing-post-processing-ms"] == "10.0"
+        assert headers["x-litellm-timing-message-copy-ms"] == "2.5"
+
+    def test_detailed_timing_headers_absent_when_disabled(self, monkeypatch):
+        """When LITELLM_DETAILED_TIMING is false, no timing headers emitted."""
+        monkeypatch.setattr(common_request_processing_mod, "LITELLM_DETAILED_TIMING", False)
+
+        user_api_key_dict = UserAPIKeyAuth(api_key="sk-test")
+        hidden_params = {
+            "_response_ms": 530.0,
+            "timing_llm_api_ms": 500.0,
+        }
+
+        headers = ProxyBaseLLMRequestProcessing.get_custom_headers(
+            user_api_key_dict=user_api_key_dict,
+            hidden_params=hidden_params,
+        )
+
+        assert "x-litellm-timing-llm-api-ms" not in headers
+        assert "x-litellm-timing-pre-processing-ms" not in headers
+
+
+class TestLoggingInitCallbackDuration:
+    """Test that Logging.__init__ tracks deep copy time in callback_duration_ms."""
+
+    def test_logging_init_sets_callback_duration_ms(self):
+        obj = Logging(
+            model="gpt-4",
+            messages=[{"role": "user", "content": "hello " * 100}],
+            stream=False,
+            call_type="acompletion",
+            start_time=datetime.datetime.now(),
+            litellm_call_id="test-123",
+            function_id="func-123",
+        )
+
+        # callback_duration_ms should be set and non-negative
+        assert hasattr(obj, "callback_duration_ms")
+        assert obj.callback_duration_ms >= 0
+
+    def test_logging_init_callback_duration_zero_for_none_messages(self):
+        obj = Logging(
+            model="gpt-4",
+            messages=None,
+            stream=False,
+            call_type="acompletion",
+            start_time=datetime.datetime.now(),
+            litellm_call_id="test-456",
+            function_id="func-456",
+        )
+
+        # Should still be set (deep copy of None is essentially a no-op)
+        assert hasattr(obj, "callback_duration_ms")
+        assert obj.callback_duration_ms >= 0

tests/test_litellm/litellm_core_utils/test_logging_utils.py

@@ -0,0 +1,156 @@
+"""
+Tests for litellm.litellm_core_utils.logging_utils — base64 truncation helpers.
+"""
+
+import pytest
+
+from litellm.litellm_core_utils.logging_utils import (
+    _format_base64_size,
+    _truncate_base64_in_string,
+    truncate_base64_in_messages,
+)
+
+# ---------------------------------------------------------------------------
+# _format_base64_size
+# ---------------------------------------------------------------------------
+
+
+class TestFormatBase64Size:
+    def test_bytes_range(self):
+        assert _format_base64_size(4) == "3B"
+
+    def test_kb_range(self):
+        # 2000 base64 chars ~ 1500 bytes ~ 1.5KB
+        assert "KB" in _format_base64_size(2000)
+
+    def test_mb_range(self):
+        # 2_000_000 base64 chars ~ 1.5MB
+        result = _format_base64_size(2_000_000)
+        assert "MB" in result
+
+
+# ---------------------------------------------------------------------------
+# _truncate_base64_in_string
+# ---------------------------------------------------------------------------
+
+
+class TestTruncateBase64InString:
+    def test_short_data_uri_not_truncated(self):
+        uri = "data:image/png;base64,AAAA"
+        assert _truncate_base64_in_string(uri) == uri
+
+    def test_long_data_uri_truncated(self):
+        payload = "A" * 200
+        uri = f"data:application/pdf;base64,{payload}"
+        result = _truncate_base64_in_string(uri)
+        assert "base64_data truncated" in result
+        assert "application/pdf" in result
+        assert payload not in result
+
+    def test_multiple_data_uris(self):
+        payload = "B" * 200
+        text = f"first: data:image/png;base64,{payload} second: data:image/jpeg;base64,{payload}"
+        result = _truncate_base64_in_string(text)
+        assert result.count("base64_data truncated") == 2
+
+    def test_no_data_uri(self):
+        text = "hello world, no base64 here"
+        assert _truncate_base64_in_string(text) == text
+
+
+# ---------------------------------------------------------------------------
+# truncate_base64_in_messages
+# ---------------------------------------------------------------------------
+
+
+class TestTruncateBase64InMessages:
+    def test_none_input(self):
+        assert truncate_base64_in_messages(None) is None
+
+    def test_string_messages(self):
+        payload = "C" * 200
+        msg = f"Look at data:image/png;base64,{payload}"
+        result = truncate_base64_in_messages(msg)
+        assert isinstance(result, str)
+        assert "base64_data truncated" in result
+
+    def test_openai_vision_format(self):
+        """Typical OpenAI multimodal message with image_url containing base64."""
+        payload = "D" * 500
+        messages = [
+            {
+                "role": "user",
+                "content": [
+                    {"type": "text", "text": "What is in this image?"},
+                    {
+                        "type": "image_url",
+                        "image_url": {
+                            "url": f"data:image/png;base64,{payload}",
+                            "detail": "auto",
+                        },
+                    },
+                ],
+            }
+        ]
+        result = truncate_base64_in_messages(messages)
+        # Original must not be mutated
+        assert payload in messages[0]["content"][1]["image_url"]["url"]
+        # Result should be truncated
+        url = result[0]["content"][1]["image_url"]["url"]
+        assert "base64_data truncated" in url
+        assert payload not in url
+        # Non-base64 parts preserved
+        assert result[0]["content"][0]["text"] == "What is in this image?"
+
+    def test_multiple_images(self):
+        """Two base64 images in one message."""
+        payload1 = "E" * 300
+        payload2 = "F" * 400
+        messages = [
+            {
+                "role": "user",
+                "content": [
+                    {
+                        "type": "image_url",
+                        "image_url": {"url": f"data:image/png;base64,{payload1}"},
+                    },
+                    {
+                        "type": "image_url",
+                        "image_url": {"url": f"data:application/pdf;base64,{payload2}"},
+                    },
+                ],
+            }
+        ]
+        result = truncate_base64_in_messages(messages)
+        for part in result[0]["content"]:
+            assert "base64_data truncated" in part["image_url"]["url"]
+
+    def test_does_not_mutate_original(self):
+        payload = "G" * 200
+        messages = [{"role": "user", "content": f"data:image/png;base64,{payload}"}]
+        truncate_base64_in_messages(messages)
+        # Original unchanged
+        assert payload in messages[0]["content"]
+
+    def test_dict_messages(self):
+        payload = "H" * 200
+        messages = {"prompt": f"data:image/png;base64,{payload}"}
+        result = truncate_base64_in_messages(messages)
+        assert "base64_data truncated" in result["prompt"]
+
+    def test_preserves_short_base64(self):
+        """Short base64 under threshold should not be truncated."""
+        short = "AAAA"
+        messages = [
+            {
+                "role": "user",
+                "content": [
+                    {
+                        "type": "image_url",
+                        "image_url": {"url": f"data:image/png;base64,{short}"},
+                    }
+                ],
+            }
+        ]
+        result = truncate_base64_in_messages(messages)
+        assert result[0]["content"][0]["image_url"]["url"] == f"data:image/png;base64,{short}"