Merge pull request #30889 from BerriAI/litellm_backport_1_88_x_bp-188x-0620

chore(release): backport #29015, #29444, #29447, #30480, #30573 to stable/1.88.x and cut 1.88.4

litellm/integrations/anthropic_cache_control_hook.py

@@ -27,6 +27,11 @@ else:
     LiteLLMLoggingObj = Any
 
 
+# Anthropic (and Bedrock Claude) reject requests with more than 4 cache_control
+# breakpoints: "A maximum of 4 blocks with cache_control may be provided."
+MAX_CACHE_CONTROL_BLOCKS = 4
+
+
 class AnthropicCacheControlHook(CustomPromptManagement):
     def get_chat_completion_prompt(
         self,
@@ -61,16 +66,30 @@ class AnthropicCacheControlHook(CustomPromptManagement):
         processed_messages = copy.deepcopy(messages)
 
         # Separate message-level and non-message-level injection points
-        remaining_points = []
+        message_points: List[CacheControlMessageInjectionPoint] = []
+        remaining_points: List[CacheControlInjectionPoint] = []
         for point in injection_points:
             if point.get("location") == "message":
-                point = cast(CacheControlMessageInjectionPoint, point)
-                processed_messages = self._process_message_injection(
-                    point=point, messages=processed_messages
-                )
+                message_points.append(cast(CacheControlMessageInjectionPoint, point))
             else:
                 remaining_points.append(point)
 
+        # Non-message points (currently Bedrock tool_config) are handled in the
+        # provider transform, where each tool_config point appends at most one
+        # cachePoint to the tools. That block also counts toward Anthropic's
+        # limit, so reserve a slot for it here to leave room.
+        reserved_blocks = (
+            1
+            if any(p.get("location") == "tool_config" for p in remaining_points)
+            else 0
+        )
+
+        processed_messages = self._apply_message_injections(
+            points=message_points,
+            messages=processed_messages,
+            max_blocks=MAX_CACHE_CONTROL_BLOCKS - reserved_blocks,
+        )
+
         # Pass through non-message injection points for provider-specific handling
         if remaining_points:
             non_default_params["cache_control_injection_points"] = remaining_points
@@ -78,14 +97,71 @@ class AnthropicCacheControlHook(CustomPromptManagement):
         return model, processed_messages, non_default_params
 
     @staticmethod
-    def _process_message_injection(
-        point: CacheControlMessageInjectionPoint, messages: List[AllMessageValues]
+    def _apply_message_injections(
+        points: List[CacheControlMessageInjectionPoint],
+        messages: List[AllMessageValues],
+        max_blocks: int,
     ) -> List[AllMessageValues]:
-        """Process message-level cache control injection."""
-        control: ChatCompletionCachedContent = point.get(
-            "control", None
-        ) or ChatCompletionCachedContent(type="ephemeral")
+        """Apply message-level cache control injection points in order.
 
+        Anthropic allows at most ``MAX_CACHE_CONTROL_BLOCKS`` cache_control
+        breakpoints per request. Client-supplied breakpoints count toward that
+        limit, so we never inject onto a message that already carries
+        cache_control (preserving the client's TTL) and we stop injecting once
+        ``max_blocks`` is reached. Injection points are honored in config order,
+        so earlier points win when slots are scarce.
+        """
+        used_blocks = sum(
+            AnthropicCacheControlHook._count_cache_control_blocks(msg)
+            for msg in messages
+        )
+
+        limit_reached = False
+        for point in points:
+            if used_blocks >= max_blocks:
+                limit_reached = True
+                break
+
+            control: ChatCompletionCachedContent = point.get(
+                "control", None
+            ) or ChatCompletionCachedContent(type="ephemeral")
+
+            for target_index in AnthropicCacheControlHook._resolve_target_indices(
+                point=point, messages=messages
+            ):
+                if used_blocks >= max_blocks:
+                    limit_reached = True
+                    break
+
+                if AnthropicCacheControlHook._message_has_cache_control(
+                    messages[target_index]
+                ):
+                    # Client already marked this message; don't overwrite it.
+                    continue
+
+                messages[target_index] = (
+                    AnthropicCacheControlHook._safe_insert_cache_control_in_message(
+                        messages[target_index], control
+                    )
+                )
+                used_blocks += 1
+
+            if limit_reached:
+                break
+
+        if limit_reached:
+            verbose_logger.warning(
+                f"AnthropicCacheControlHook: Reached the Anthropic limit of "
+                f"{MAX_CACHE_CONTROL_BLOCKS} cache_control blocks. Skipping further injection."
+            )
+
+        return messages
+
+    @staticmethod
+    def _resolve_target_indices(
+        point: CacheControlMessageInjectionPoint, messages: List[AllMessageValues]
+    ) -> List[int]:
+        """Resolve which message indices an injection point targets."""
         _targetted_index: Optional[Union[int, str]] = point.get("index", None)
         targetted_index: Optional[int] = None
         if isinstance(_targetted_index, str):
@@ -96,36 +172,49 @@ class AnthropicCacheControlHook(CustomPromptManagement):
         else:
             targetted_index = _targetted_index
 
-        targetted_role = point.get("role", None)
-
         # Case 1: Target by specific index
         if targetted_index is not None:
             original_index = targetted_index
-            # Handle negative indices (convert to positive)
             if targetted_index < 0:
                 targetted_index += len(messages)
 
             if 0 <= targetted_index < len(messages):
-                messages[targetted_index] = (
-                    AnthropicCacheControlHook._safe_insert_cache_control_in_message(
-                        messages[targetted_index], control
-                    )
-                )
-            else:
-                verbose_logger.warning(
-                    f"AnthropicCacheControlHook: Provided index {original_index} is out of bounds for message list of length {len(messages)}. "
-                    f"Targeted index was {targetted_index}. Skipping cache control injection for this point."
-                )
+                return [targetted_index]
+
+            verbose_logger.warning(
+                f"AnthropicCacheControlHook: Provided index {original_index} is out of bounds for message list of length {len(messages)}. "
+                f"Targeted index was {targetted_index}. Skipping cache control injection for this point."
+            )
+            return []
+
         # Case 2: Target by role
-        elif targetted_role is not None:
-            for msg in messages:
-                if msg.get("role") == targetted_role:
-                    msg = (
-                        AnthropicCacheControlHook._safe_insert_cache_control_in_message(
-                            message=msg, control=control
-                        )
-                    )
-        return messages
+        targetted_role = point.get("role", None)
+        if targetted_role is not None:
+            return [
+                idx
+                for idx, msg in enumerate(messages)
+                if msg.get("role") == targetted_role
+            ]
+
+        return []
+
+    @staticmethod
+    def _count_cache_control_blocks(message: AllMessageValues) -> int:
+        """Count cache_control breakpoints on a message (message + content level)."""
+        count = 0
+        if message.get("cache_control") is not None:
+            count += 1
+        content = message.get("content")
+        if isinstance(content, list):
+            for block in content:
+                if isinstance(block, dict) and block.get("cache_control") is not None:
+                    count += 1
+        return count
+
+    @staticmethod
+    def _message_has_cache_control(message: AllMessageValues) -> bool:
+        """Return True if the message already carries any cache_control."""
+        return AnthropicCacheControlHook._count_cache_control_blocks(message) > 0
 
     @staticmethod
     def _safe_insert_cache_control_in_message(

litellm/integrations/datadog/datadog.py

@@ -41,6 +41,7 @@ from litellm.integrations.datadog.datadog_handler import (
 )
 from litellm.litellm_core_utils.dd_tracing import tracer
 from litellm.llms.custom_httpx.http_handler import (
+    MaskedHTTPStatusError,
     _get_httpx_client,
     get_async_httpx_client,
     httpxSpecialProvider,
@@ -68,6 +69,22 @@ DD_LOGGED_SUCCESS_SERVICE_TYPES = [
 ]
 
 
+def _resolve_dd_batch_size() -> int:
+    raw = os.getenv("DD_BATCH_SIZE")
+    if raw is None:
+        return DD_MAX_BATCH_SIZE
+    try:
+        value = int(raw)
+    except ValueError:
+        verbose_logger.warning(
+            "Datadog: ignoring invalid DD_BATCH_SIZE=%r, using %s",
+            raw,
+            DD_MAX_BATCH_SIZE,
+        )
+        return DD_MAX_BATCH_SIZE
+    return max(1, min(value, DD_MAX_BATCH_SIZE))
+
+
 class DataDogLogger(
     CustomBatchLogger,
     AdditionalLoggingUtils,
@@ -128,7 +145,9 @@ class DataDogLogger(
             asyncio.create_task(self.periodic_flush())
             self.flush_lock = asyncio.Lock()
             super().__init__(
-                **kwargs, flush_lock=self.flush_lock, batch_size=DD_MAX_BATCH_SIZE
+                **kwargs,
+                flush_lock=self.flush_lock,
+                batch_size=_resolve_dd_batch_size(),
             )
         except Exception as e:
             verbose_logger.exception(
@@ -339,28 +358,14 @@ class DataDogLogger(
                     "[DATADOG MOCK] Mock mode enabled - API calls will be intercepted"
                 )
 
-            response = await self.async_send_compressed_data(batch_to_send)
-            if response.status_code == 413:
-                verbose_logger.exception(DD_ERRORS.DATADOG_413_ERROR.value)
-                self.log_queue = batch_to_send + self.log_queue
-                return
-
-            response.raise_for_status()
-            if response.status_code != 202:
-                raise Exception(
-                    f"Response from datadog API status_code: {response.status_code}, text: {response.text}"
-                )
+            undelivered = await self._send_with_413_split(batch_to_send)
+            if undelivered:
+                self.log_queue = undelivered + self.log_queue
 
             if self.is_mock_mode:
                 verbose_logger.debug(
                     f"[DATADOG MOCK] Batch of {len(batch_to_send)} events successfully mocked"
                 )
-            else:
-                verbose_logger.debug(
-                    "Datadog: Response from datadog API status_code: %s, text: %s",
-                    response.status_code,
-                    response.text,
-                )
 
         except Exception as e:
             self.log_queue = batch_to_send + self.log_queue
@@ -368,6 +373,62 @@ class DataDogLogger(
                 f"Datadog Error sending batch API - {str(e)}\n{traceback.format_exc()}"
             )
 
+    async def _send_with_413_split(self, batch: List) -> List:
+        """
+        Send a batch, halving any sub-batch that 413s (payload too large) and retrying the
+        halves, since Datadog enforces a 5MB uncompressed limit per request.
+
+        A 413 surfaces as a raised MaskedHTTPStatusError (httpx raise_for_status), not a
+        returned response, so both paths are handled. A lone event that still 413s is
+        dropped to avoid wedging the queue on an undeliverable payload. Returns the events
+        that could not be delivered because of a non-413 (transient) error, so the caller
+        re-queues only those and never the events already accepted by Datadog.
+        """
+        pending: List[List] = [batch]
+        while pending:
+            chunk = pending.pop()
+            if not chunk:
+                continue
+            try:
+                response = await self.async_send_compressed_data(chunk)
+            except Exception as e:
+                if isinstance(e, MaskedHTTPStatusError) and e.status_code == 413:
+                    response = e.response
+                else:
+                    verbose_logger.exception(
+                        f"Datadog Error sending batch API - {str(e)}"
+                    )
+                    return self._undelivered(chunk, pending)
+
+            if response.status_code == 413:
+                if len(chunk) == 1:
+                    verbose_logger.error(DD_ERRORS.DATADOG_413_ERROR.value)
+                    continue
+                mid = len(chunk) // 2
+                pending.append(chunk[mid:])
+                pending.append(chunk[:mid])
+                continue
+
+            if response.status_code != 202:
+                verbose_logger.error(
+                    "Datadog: unexpected response status_code=%s, text=%s",
+                    response.status_code,
+                    response.text,
+                )
+                return self._undelivered(chunk, pending)
+
+            verbose_logger.debug(
+                "Datadog: delivered %s events, status_code=%s, text=%s",
+                len(chunk),
+                response.status_code,
+                response.text,
+            )
+        return []
+
+    @staticmethod
+    def _undelivered(chunk: List, pending: List[List]) -> List:
+        return chunk + [event for remaining in reversed(pending) for event in remaining]
+
     async def flush_queue(self):
         if self.flush_lock is None:
             return

litellm/proxy/_types.py

@@ -3695,6 +3695,11 @@ class ProxyException(Exception):
         provider_specific_fields: Optional[dict] = None,
     ):
         self.message = str(message)
+        # Populate Exception.args so str(self) returns the message.
+        # Without this, logging paths that call str(original_exception)
+        # (e.g. StandardLoggingPayloadSetup.get_error_information) record an
+        # empty error_message for ProxyException-based failures. See LIT-3094.
+        super().__init__(self.message)
         self.type = type
         self.param = param
         self.openai_code = openai_code or code

litellm/proxy/common_request_processing.py

@@ -1900,6 +1900,13 @@ class ProxyBaseLLMRequestProcessing:
         except Exception:
             pass
 
+        if isinstance(e, ProxyException):
+            e.headers = {
+                **e.headers,
+                **{k: v if isinstance(v, str) else str(v) for k, v in headers.items()},
+            }
+            raise e
+
         if isinstance(e, HTTPException):
             raw_detail = getattr(e, "detail", str(e))
             message, structured_fields = _serialize_http_exception_detail(raw_detail)

litellm/proxy/guardrails/guardrail_hooks/aim/aim.py

@@ -9,7 +9,6 @@ import json
 import os
 from typing import TYPE_CHECKING, Any, AsyncGenerator, Optional, Type, Union
 
-from fastapi import HTTPException
 from pydantic import BaseModel
 from websockets.asyncio.client import ClientConnection, connect
 
@@ -21,7 +20,7 @@ from litellm.llms.custom_httpx.http_handler import (
     get_async_httpx_client,
     httpxSpecialProvider,
 )
-from litellm.proxy._types import UserAPIKeyAuth
+from litellm.proxy._types import ProxyException, UserAPIKeyAuth
 from litellm.proxy.guardrails._content_utils import (
     apply_redacted_messages_back,
     build_inspection_messages,
@@ -129,6 +128,16 @@ class AimGuardrail(CustomGuardrail):
             verbose_proxy_logger.error(f"Aim: {action_type} action")
         return data
 
+    @staticmethod
+    def _rejection(message: str, *, openai_code: str | None = None) -> ProxyException:
+        return ProxyException(
+            message=message,
+            type="invalid_request_error",
+            param=None,
+            code=400,
+            openai_code=openai_code,
+        )
+
     def _handle_block_action(self, analysis_result: Any, required_action: Any) -> None:
         detection_message = required_action.get("detection_message", None)
         verbose_proxy_logger.info(
@@ -136,7 +145,7 @@ class AimGuardrail(CustomGuardrail):
                 policies=list(analysis_result["policy_drill_down"].keys()),
             ),
         )
-        raise HTTPException(status_code=400, detail=detection_message)
+        raise self._rejection(detection_message, openai_code="content_policy_violation")
 
     def _anonymize_request(self, res: Any, data: dict) -> dict:
         verbose_proxy_logger.info("Aim: anonymize action")
@@ -148,14 +157,11 @@ class AimGuardrail(CustomGuardrail):
         # parts from a multimodal request — degrade to block so the
         # multimodal payload is never silently rewritten.
         if has_non_string_content(data):
-            raise HTTPException(
-                status_code=400,
-                detail=(
-                    "Aim: anonymize action requested for multimodal input "
-                    "but mask-in-place would drop non-text parts. Send the "
-                    "request with plain string content to use anonymize, "
-                    "or rely on block-mode policies."
-                ),
+            raise self._rejection(
+                "Aim: anonymize action requested for multimodal input "
+                "but mask-in-place would drop non-text parts. Send the "
+                "request with plain string content to use anonymize, "
+                "or rely on block-mode policies."
             )
         redacted_messages = [
             {
@@ -287,9 +293,9 @@ class AimGuardrail(CustomGuardrail):
             if aim_output_guardrail_result and aim_output_guardrail_result.get(
                 "detection_message"
             ):
-                raise HTTPException(
-                    status_code=400,
-                    detail=aim_output_guardrail_result.get("detection_message"),
+                raise self._rejection(
+                    aim_output_guardrail_result.get("detection_message"),
+                    openai_code="content_policy_violation",
                 )
             if aim_output_guardrail_result and aim_output_guardrail_result.get(
                 "redacted_output"

litellm/proxy/utils.py

@@ -1991,7 +1991,7 @@ class ProxyLogging:
             litellm_call_id=request_data.get("litellm_call_id", ""), status="fail"
         )
         if AlertType.llm_exceptions in self.alert_types and not isinstance(
-            original_exception, HTTPException
+            original_exception, (HTTPException, ProxyException)
         ):
             """
             Just alert on LLM API exceptions. Do not alert on user errors
@@ -2095,6 +2095,7 @@ class ProxyLogging:
         e.g should only return True for:
             - Authentication Errors from user_api_key_auth
             - HTTP HTTPException (rate limit errors)
+            - ProxyException (guardrail blocks, budget / rate-limit errors)
         """
 
         #########################################################
@@ -2111,7 +2112,7 @@ class ProxyLogging:
         ):
             return False
 
-        return isinstance(original_exception, HTTPException) or (
+        return isinstance(original_exception, (HTTPException, ProxyException)) or (
             error_type == ProxyErrorTypes.auth_error
         )
 

litellm/types/utils.py

@@ -3183,6 +3183,7 @@ all_litellm_params = (
         "allowed_openai_params",
         "litellm_session_id",
         "use_litellm_proxy",
+        "use_chat_completions_api",
         "prompt_label",
         "shared_session",
         "search_tool_name",

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "litellm"
-version = "1.88.3"
+version = "1.88.4"
 description = "Library to easily interface with LLM API providers"
 readme = "README.md"
 requires-python = ">=3.10, <3.14"
@@ -264,7 +264,7 @@ source-exclude = [
 profile = "black"
 
 [tool.commitizen]
-version = "1.88.3"
+version = "1.88.4"
 version_files = [
     "pyproject.toml:^version",
 ]

tests/local_testing/test_aim_guardrails.py

@@ -6,10 +6,10 @@ import sys
 from unittest.mock import AsyncMock, patch, call
 
 import pytest
-from fastapi.exceptions import HTTPException
 from httpx import Request, Response
 
 from litellm import DualCache
+from litellm.proxy._types import ProxyException
 from litellm.proxy.guardrails.guardrail_hooks.aim.aim import (
     AimGuardrail,
     AimGuardrailMissingSecrets,
@@ -101,7 +101,7 @@ async def test_block_callback(mode: str):
         ],
     }
 
-    with pytest.raises(HTTPException, match="Jailbreak detected"):
+    with pytest.raises(ProxyException, match="Jailbreak detected") as exc_info:
         with patch(
             "litellm.llms.custom_httpx.http_handler.AsyncHTTPHandler.post",
             return_value=Response(
@@ -135,6 +135,137 @@ async def test_block_callback(mode: str):
                     call_type="completion",
                 )
 
+    exc = exc_info.value
+    assert exc.code == "400"
+    assert exc.type == "invalid_request_error"
+    assert exc.param is None
+    assert exc.openai_code == "content_policy_violation"
+
+
+@pytest.mark.asyncio
+async def test_output_block_raises_proxy_exception():
+    """An output-side block is a content-policy violation, like the input block:
+    it must surface a conformant ProxyException, not a bare HTTPException whose
+    type/param serialize as the literal string "None". Regression for LIT-3751."""
+    init_guardrails_v2(
+        all_guardrails=[
+            {
+                "guardrail_name": "gibberish-guard",
+                "litellm_params": {
+                    "guardrail": "aim",
+                    "mode": "post_call",
+                    "api_key": "hs-aim-key",
+                },
+            },
+        ],
+        config_file_path="",
+    )
+    aim_guardrails = [
+        callback for callback in litellm.callbacks if isinstance(callback, AimGuardrail)
+    ]
+    assert len(aim_guardrails) == 1
+    aim_guardrail = aim_guardrails[0]
+
+    block_on_output = Response(
+        json={
+            "analysis_result": {"policy_drill_down": {"PII": {}}},
+            "required_action": {
+                "action_type": "block_action",
+                "detection_message": "Output blocked: leaked secret",
+                "policy_name": "blocking policy",
+            },
+        },
+        status_code=200,
+        request=Request(method="POST", url="http://aim"),
+    )
+    response = ModelResponse(
+        choices=[
+            {
+                "finish_reason": "stop",
+                "index": 0,
+                "message": {"content": "here is the secret", "role": "assistant"},
+            }
+        ]
+    )
+
+    with patch(
+        "litellm.llms.custom_httpx.http_handler.AsyncHTTPHandler.post",
+        return_value=block_on_output,
+    ):
+        with pytest.raises(ProxyException, match="Output blocked") as exc_info:
+            await aim_guardrail.async_post_call_success_hook(
+                data={"messages": [{"role": "user", "content": "tell me a secret"}]},
+                response=response,
+                user_api_key_dict=UserAPIKeyAuth(),
+            )
+
+    exc = exc_info.value
+    assert exc.code == "400"
+    assert exc.type == "invalid_request_error"
+    assert exc.param is None
+    assert exc.openai_code == "content_policy_violation"
+
+
+@pytest.mark.asyncio
+async def test_anonymize_multimodal_rejection_raises_proxy_exception():
+    """Anonymize on multimodal input degrades to a 400 because mask-in-place would
+    drop non-text parts. That is a usage error, not a content-policy violation, so
+    it must raise a conformant ProxyException WITHOUT the content_policy_violation
+    code. Regression for LIT-3751."""
+    init_guardrails_v2(
+        all_guardrails=[
+            {
+                "guardrail_name": "gibberish-guard",
+                "litellm_params": {
+                    "guardrail": "aim",
+                    "mode": "pre_call",
+                    "api_key": "hs-aim-key",
+                },
+            },
+        ],
+        config_file_path="",
+    )
+    aim_guardrails = [
+        callback for callback in litellm.callbacks if isinstance(callback, AimGuardrail)
+    ]
+    assert len(aim_guardrails) == 1
+    aim_guardrail = aim_guardrails[0]
+
+    data = {
+        "messages": [
+            {
+                "role": "user",
+                "content": [
+                    {"type": "text", "text": "Hi my name is Brian"},
+                    {
+                        "type": "image_url",
+                        "image_url": {"url": "data:image/png;base64,iVBORw0KGgo="},
+                    },
+                ],
+            },
+        ],
+    }
+
+    with patch(
+        "litellm.llms.custom_httpx.http_handler.AsyncHTTPHandler.post",
+        return_value=response_with_detections,
+    ):
+        with pytest.raises(
+            ProxyException, match="anonymize action requested for multimodal"
+        ) as exc_info:
+            await aim_guardrail.async_pre_call_hook(
+                data=data,
+                cache=DualCache(),
+                user_api_key_dict=UserAPIKeyAuth(),
+                call_type="completion",
+            )
+
+    exc = exc_info.value
+    assert exc.code == "400"
+    assert exc.type == "invalid_request_error"
+    assert exc.param is None
+    assert exc.openai_code != "content_policy_violation"
+
 
 @pytest.mark.asyncio
 @pytest.mark.parametrize("mode", ["pre_call", "during_call"])

tests/test_litellm/integrations/datadog/test_datadog_logger_batching.py

@@ -1,10 +1,49 @@
 from unittest.mock import AsyncMock, Mock, patch
 
+import httpx
 import pytest
 from httpx import Request, Response
 
 from litellm.integrations.datadog.datadog import DataDogLogger
-from litellm.types.integrations.datadog import DatadogPayload
+from litellm.llms.custom_httpx.http_handler import MaskedHTTPStatusError
+from litellm.types.integrations.datadog import DD_MAX_BATCH_SIZE, DatadogPayload
+
+
+def _payloads(n):
+    return [
+        DatadogPayload(
+            ddsource="litellm",
+            ddtags="env:test",
+            hostname="host",
+            message=f'{{"event": {i}}}',
+            service="svc",
+            status="info",
+        )
+        for i in range(n)
+    ]
+
+
+def _raised_413():
+    request = Request("POST", "https://example.com")
+    response = Response(413, request=request, text="Payload Too Large")
+    return MaskedHTTPStatusError(
+        httpx.HTTPStatusError("413", request=request, response=response)
+    )
+
+
+def _make_send(max_ok, delivered, *, raise_413=True):
+    """Datadog double: 413 batches larger than max_ok, 202 (recording delivery) otherwise."""
+
+    async def _send(data):
+        request = Request("POST", "https://example.com")
+        if len(data) > max_ok:
+            if raise_413:
+                raise _raised_413()
+            return Response(413, request=request, text="Payload Too Large")
+        delivered.extend(event["message"] for event in data)
+        return Response(202, request=request, text="Accepted")
+
+    return _send
 
 
 @pytest.fixture
@@ -75,40 +114,152 @@ async def test_failure_hook_threshold_flush_uses_flush_queue(datadog_env):
 
 
 @pytest.mark.asyncio
-async def test_async_send_batch_requeues_events_on_413(datadog_env):
+async def test_413_splits_oversized_batch_and_delivers_every_event(datadog_env):
+    """A raised 413 (the real httpx path) halves the batch until each piece is accepted."""
     with patch("asyncio.create_task"):
         logger = DataDogLogger()
 
-    logger.log_queue = [
-        DatadogPayload(
-            ddsource="litellm",
-            ddtags="env:test",
-            hostname="host",
-            message=f'{{"event": {i}}}',
-            service="svc",
-            status="info",
+    logger.log_queue = _payloads(4)
+    delivered: list = []
+    logger.async_send_compressed_data = AsyncMock(side_effect=_make_send(1, delivered))
+
+    await logger.async_send_batch()
+
+    assert sorted(delivered) == [f'{{"event": {i}}}' for i in range(4)]
+    assert logger.log_queue == []
+
+
+@pytest.mark.asyncio
+async def test_413_does_not_requeue_oversized_batch(datadog_env):
+    """Regression for the infinite 413 loop: an undeliverable batch must not be re-queued."""
+    with patch("asyncio.create_task"):
+        logger = DataDogLogger()
+
+    logger.log_queue = _payloads(4)
+    logger.async_send_compressed_data = AsyncMock(side_effect=_make_send(0, []))
+
+    await logger.async_send_batch()
+    await logger.async_send_batch()
+
+    assert logger.log_queue == []
+
+
+@pytest.mark.asyncio
+async def test_413_drops_single_oversized_event(datadog_env):
+    with patch("asyncio.create_task"):
+        logger = DataDogLogger()
+
+    logger.log_queue = _payloads(1)
+    send = AsyncMock(side_effect=_make_send(0, []))
+    logger.async_send_compressed_data = send
+
+    await logger.async_send_batch()
+
+    assert send.await_count == 1
+    assert logger.log_queue == []
+
+
+@pytest.mark.asyncio
+async def test_413_returned_response_also_splits(datadog_env):
+    """Defensive path: a 413 returned (not raised) is handled the same way."""
+    with patch("asyncio.create_task"):
+        logger = DataDogLogger()
+
+    logger.log_queue = _payloads(4)
+    delivered: list = []
+    logger.async_send_compressed_data = AsyncMock(
+        side_effect=_make_send(1, delivered, raise_413=False)
+    )
+
+    await logger.async_send_batch()
+
+    assert sorted(delivered) == [f'{{"event": {i}}}' for i in range(4)]
+    assert logger.log_queue == []
+
+
+@pytest.mark.asyncio
+async def test_partial_delivery_then_transient_error_requeues_only_undelivered(
+    datadog_env,
+):
+    """A transient error after a partial split delivery must not duplicate delivered events."""
+    with patch("asyncio.create_task"):
+        logger = DataDogLogger()
+
+    logger.log_queue = _payloads(4)
+    delivered: list = []
+
+    async def _send(data):
+        messages = [event["message"] for event in data]
+        if len(data) > 2:
+            raise _raised_413()
+        if messages == ['{"event": 2}', '{"event": 3}']:
+            raise RuntimeError("transient network error")
+        delivered.extend(messages)
+        return Response(
+            202, request=Request("POST", "https://example.com"), text="Accepted"
         )
-        for i in range(2)
+
+    logger.async_send_compressed_data = AsyncMock(side_effect=_send)
+
+    await logger.async_send_batch()
+
+    assert delivered == ['{"event": 0}', '{"event": 1}']
+    assert [event["message"] for event in logger.log_queue] == [
+        '{"event": 2}',
+        '{"event": 3}',
     ]
 
+
+@pytest.mark.asyncio
+async def test_unexpected_non_202_status_requeues(datadog_env):
+    """A non-413, non-202 response is treated as undelivered and re-queued."""
+    with patch("asyncio.create_task"):
+        logger = DataDogLogger()
+
+    logger.log_queue = _payloads(2)
     logger.async_send_compressed_data = AsyncMock(
         return_value=Response(
-            413,
-            request=Request("POST", "https://example.com"),
-            text="Payload Too Large",
+            200, request=Request("POST", "https://example.com"), text="OK"
         )
     )
 
     await logger.async_send_batch()
 
-    assert logger.async_send_compressed_data.await_count == 1
-    assert len(logger.log_queue) == 2
     assert [event["message"] for event in logger.log_queue] == [
         '{"event": 0}',
         '{"event": 1}',
     ]
 
 
+@pytest.mark.parametrize(
+    "value, expected",
+    [
+        ("50", 50),
+        ("1", 1),
+        ("0", 1),
+        ("-5", 1),
+        (str(DD_MAX_BATCH_SIZE + 100), DD_MAX_BATCH_SIZE),
+        ("not_an_int", DD_MAX_BATCH_SIZE),
+    ],
+)
+def test_dd_batch_size_env_resolution(monkeypatch, value, expected):
+    monkeypatch.setenv("DD_API_KEY", "test_api_key")
+    monkeypatch.setenv("DD_SITE", "test.datadoghq.com")
+    monkeypatch.setenv("DD_BATCH_SIZE", value)
+    with patch("asyncio.create_task"):
+        logger = DataDogLogger()
+    assert logger.batch_size == expected
+
+
+def test_dd_batch_size_defaults_to_max(monkeypatch):
+    monkeypatch.setenv("DD_API_KEY", "test_api_key")
+    monkeypatch.setenv("DD_SITE", "test.datadoghq.com")
+    monkeypatch.delenv("DD_BATCH_SIZE", raising=False)
+    with patch("asyncio.create_task"):
+        logger = DataDogLogger()
+    assert logger.batch_size == DD_MAX_BATCH_SIZE
+
+
 @pytest.mark.asyncio
 async def test_async_send_batch_handles_empty_queue(datadog_env):
     with patch("asyncio.create_task"):

tests/test_litellm/integrations/test_anthropic_cache_control_hook.py

@@ -1087,3 +1087,357 @@ async def test_anthropic_cache_control_hook_string_negative_index():
                 f"Expected cachePoint in last message content, got: {last_message_content}. "
                 "String index '-1' was not parsed correctly (str.isdigit() returns False for negative strings)."
             )
+
+
+def _count_cache_control(messages: List[AllMessageValues]) -> int:
+    """Count cache_control breakpoints across messages (message + content level)."""
+    count = 0
+    for message in messages:
+        if message.get("cache_control") is not None:
+            count += 1
+        content = message.get("content")
+        if isinstance(content, list):
+            for block in content:
+                if isinstance(block, dict) and block.get("cache_control") is not None:
+                    count += 1
+    return count
+
+
+def _build_injection_points():
+    return [
+        {
+            "location": "message",
+            "role": "system",
+            "control": {"type": "ephemeral", "ttl": "1h"},
+        },
+        {
+            "location": "message",
+            "index": -1,
+            "control": {"type": "ephemeral", "ttl": "5m"},
+        },
+    ]
+
+
+def test_cache_control_hook_caps_at_four_blocks_with_client_cache_control():
+    """Regression for LIT-3667 / Anthropic 'A maximum of 4 blocks ... Found 5'.
+
+    A Hermes-style request already carries 4 client cache_control breakpoints on
+    its system messages. With both auto-inject points configured the hook must
+    NOT add a 5th breakpoint, and must NOT overwrite the client's existing
+    breakpoints (TTL must be preserved).
+    """
+    hook = AnthropicCacheControlHook()
+
+    messages: List[AllMessageValues] = [
+        {
+            "role": "system",
+            "content": [
+                {
+                    "type": "text",
+                    "text": f"System block {i}",
+                    "cache_control": {"type": "ephemeral", "ttl": "1h"},
+                }
+            ],
+        }
+        for i in range(4)
+    ]
+    messages.append({"role": "user", "content": "hello"})
+
+    _, processed, _ = hook.get_chat_completion_prompt(
+        model="bedrock/us.anthropic.claude-opus-4-6-v1:0",
+        messages=messages,
+        non_default_params={
+            "cache_control_injection_points": _build_injection_points()
+        },
+        prompt_id=None,
+        prompt_variables=None,
+        dynamic_callback_params={},
+    )
+
+    assert (
+        _count_cache_control(processed) == 4
+    ), "Hook must cap cache_control at Anthropic's limit of 4 blocks"
+
+    # Client TTL on system blocks must be preserved (not overwritten by config).
+    for i in range(4):
+        assert processed[i]["content"][-1]["cache_control"] == {
+            "type": "ephemeral",
+            "ttl": "1h",
+        }
+
+    # The last (user) message must not receive a 5th breakpoint.
+    user_message = processed[-1]
+    assert user_message.get("cache_control") is None
+    user_content = user_message.get("content")
+    if isinstance(user_content, list):
+        assert all(
+            block.get("cache_control") is None
+            for block in user_content
+            if isinstance(block, dict)
+        )
+
+
+def test_cache_control_hook_caps_at_four_blocks_without_client_cache_control():
+    """Four plain system messages + role:system + index:-1 must stay at 4 blocks.
+
+    role:system fills all four slots, so the index:-1 point is skipped.
+    """
+    hook = AnthropicCacheControlHook()
+
+    messages: List[AllMessageValues] = [
+        {"role": "system", "content": f"System {i}"} for i in range(4)
+    ]
+    messages.append({"role": "user", "content": "hello"})
+
+    _, processed, _ = hook.get_chat_completion_prompt(
+        model="bedrock/us.anthropic.claude-opus-4-6-v1:0",
+        messages=messages,
+        non_default_params={
+            "cache_control_injection_points": _build_injection_points()
+        },
+        prompt_id=None,
+        prompt_variables=None,
+        dynamic_callback_params={},
+    )
+
+    assert _count_cache_control(processed) == 4
+    # All four system messages cached; user message skipped (limit reached).
+    assert all(processed[i].get("cache_control") is not None for i in range(4))
+    assert processed[-1].get("cache_control") is None
+
+
+def test_cache_control_hook_does_not_overwrite_existing_cache_control():
+    """If a targeted message already has client cache_control, do not inject."""
+    hook = AnthropicCacheControlHook()
+
+    messages: List[AllMessageValues] = [
+        {
+            "role": "system",
+            "content": [
+                {
+                    "type": "text",
+                    "text": "Cached by client",
+                    "cache_control": {"type": "ephemeral", "ttl": "1h"},
+                }
+            ],
+        },
+        {"role": "user", "content": "hello"},
+    ]
+
+    _, processed, _ = hook.get_chat_completion_prompt(
+        model="bedrock/us.anthropic.claude-opus-4-6-v1:0",
+        messages=messages,
+        # Target the already-cached system message with a different TTL.
+        non_default_params={
+            "cache_control_injection_points": [
+                {
+                    "location": "message",
+                    "index": 0,
+                    "control": {"type": "ephemeral", "ttl": "5m"},
+                }
+            ]
+        },
+        prompt_id=None,
+        prompt_variables=None,
+        dynamic_callback_params={},
+    )
+
+    # Client's 1h TTL must be preserved, not replaced by the config's 5m.
+    assert processed[0]["content"][-1]["cache_control"] == {
+        "type": "ephemeral",
+        "ttl": "1h",
+    }
+    assert _count_cache_control(processed) == 1
+
+
+@pytest.mark.asyncio
+async def test_cache_control_hook_bedrock_payload_caps_cachepoints_at_four():
+    """End-to-end: outgoing Bedrock payload must not exceed 4 cachePoint blocks.
+
+    Reproduces the customer report where 4 client cache_control system blocks
+    plus auto-inject produced 5 cachePoint blocks and Bedrock returned 400.
+    """
+    with patch.dict(
+        os.environ,
+        {
+            "AWS_ACCESS_KEY_ID": "fake_access_key_id",
+            "AWS_SECRET_ACCESS_KEY": "fake_secret_access_key",
+            "AWS_REGION_NAME": "us-east-1",
+        },
+    ):
+        litellm.callbacks = [AnthropicCacheControlHook()]
+
+        mock_response = MagicMock()
+        mock_response.json.return_value = {
+            "output": {"message": {"role": "assistant", "content": "ok"}},
+            "stopReason": "end_turn",
+            "usage": {"inputTokens": 100, "outputTokens": 4, "totalTokens": 104},
+        }
+        mock_response.status_code = 200
+
+        client = AsyncHTTPHandler()
+        with patch.object(client, "post", return_value=mock_response) as mock_post:
+            messages = [
+                {
+                    "role": "system",
+                    "content": [
+                        {
+                            "type": "text",
+                            "text": f"System block {i}",
+                            "cache_control": {"type": "ephemeral", "ttl": "1h"},
+                        }
+                    ],
+                }
+                for i in range(4)
+            ]
+            messages.append({"role": "user", "content": "hello"})
+
+            await litellm.acompletion(
+                model="bedrock/us.anthropic.claude-opus-4-6-v1:0",
+                messages=messages,
+                max_tokens=32,
+                cache_control_injection_points=_build_injection_points(),
+                client=client,
+            )
+
+            request_body = json.loads(mock_post.call_args.kwargs["data"])
+
+            cache_points = sum(
+                1
+                for block in request_body.get("system", [])
+                if isinstance(block, dict) and "cachePoint" in block
+            )
+            for msg in request_body.get("messages", []):
+                content = msg.get("content", [])
+                if isinstance(content, list):
+                    cache_points += sum(
+                        1
+                        for block in content
+                        if isinstance(block, dict) and "cachePoint" in block
+                    )
+
+            assert cache_points <= 4, (
+                f"Bedrock payload exceeded Anthropic's 4 cache_control block limit: "
+                f"found {cache_points} cachePoint blocks"
+            )
+
+
+def test_cache_control_hook_reserves_slot_for_tool_config_point():
+    """A tool_config injection point consumes one of the 4 slots downstream.
+
+    With role:system targeting 4 system messages plus a tool_config point, the
+    hook must inject at most 3 message-level blocks so the tool_config cachePoint
+    appended by the Bedrock transform keeps the total at 4, not 5.
+    """
+    hook = AnthropicCacheControlHook()
+
+    messages: List[AllMessageValues] = [
+        {"role": "system", "content": f"System {i}"} for i in range(4)
+    ]
+    messages.append({"role": "user", "content": "hello"})
+
+    _, processed, non_default_params = hook.get_chat_completion_prompt(
+        model="bedrock/us.anthropic.claude-opus-4-6-v1:0",
+        messages=messages,
+        non_default_params={
+            "cache_control_injection_points": [
+                {
+                    "location": "message",
+                    "role": "system",
+                    "control": {"type": "ephemeral", "ttl": "1h"},
+                },
+                {"location": "tool_config"},
+            ]
+        },
+        prompt_id=None,
+        prompt_variables=None,
+        dynamic_callback_params={},
+    )
+
+    assert _count_cache_control(processed) == 3
+    # The tool_config point is passed through for the provider transform.
+    assert non_default_params["cache_control_injection_points"] == [
+        {"location": "tool_config"}
+    ]
+
+
+@pytest.mark.asyncio
+async def test_cache_control_hook_bedrock_payload_caps_with_tool_config_point():
+    """End-to-end: message + tool_config injection must not exceed 4 cachePoints."""
+    with patch.dict(
+        os.environ,
+        {
+            "AWS_ACCESS_KEY_ID": "fake_access_key_id",
+            "AWS_SECRET_ACCESS_KEY": "fake_secret_access_key",
+            "AWS_REGION_NAME": "us-east-1",
+        },
+    ):
+        litellm.callbacks = [AnthropicCacheControlHook()]
+
+        mock_response = MagicMock()
+        mock_response.json.return_value = {
+            "output": {"message": {"role": "assistant", "content": "ok"}},
+            "stopReason": "end_turn",
+            "usage": {"inputTokens": 100, "outputTokens": 4, "totalTokens": 104},
+        }
+        mock_response.status_code = 200
+
+        client = AsyncHTTPHandler()
+        with patch.object(client, "post", return_value=mock_response) as mock_post:
+            messages = [
+                {"role": "system", "content": f"System block {i}"} for i in range(4)
+            ]
+            messages.append({"role": "user", "content": "What is the weather?"})
+
+            await litellm.acompletion(
+                model="bedrock/us.anthropic.claude-opus-4-6-v1:0",
+                messages=messages,
+                max_tokens=32,
+                tools=[
+                    {
+                        "type": "function",
+                        "function": {
+                            "name": "get_weather",
+                            "description": "Get weather for a location",
+                            "parameters": {
+                                "type": "object",
+                                "properties": {"location": {"type": "string"}},
+                                "required": ["location"],
+                            },
+                        },
+                    }
+                ],
+                cache_control_injection_points=[
+                    {
+                        "location": "message",
+                        "role": "system",
+                        "control": {"type": "ephemeral", "ttl": "1h"},
+                    },
+                    {"location": "tool_config"},
+                ],
+                client=client,
+            )
+
+            request_body = json.loads(mock_post.call_args.kwargs["data"])
+
+            cache_points = sum(
+                1
+                for block in request_body.get("system", [])
+                if isinstance(block, dict) and "cachePoint" in block
+            )
+            for msg in request_body.get("messages", []):
+                content = msg.get("content", [])
+                if isinstance(content, list):
+                    cache_points += sum(
+                        1
+                        for block in content
+                        if isinstance(block, dict) and "cachePoint" in block
+                    )
+            for tool in request_body.get("toolConfig", {}).get("tools", []):
+                if isinstance(tool, dict) and "cachePoint" in tool:
+                    cache_points += 1
+
+            assert cache_points <= 4, (
+                f"Bedrock payload exceeded Anthropic's 4 cache_control block limit "
+                f"when mixing message and tool_config injection: found {cache_points}"
+            )

tests/test_litellm/llms/openai/test_use_chat_completions_api_no_leak.py

@@ -0,0 +1,74 @@
+"""
+Regression test for issue #28146.
+
+`use_chat_completions_api` is a LiteLLM-internal control flag (it forces the
+/responses -> /chat/completions bridge). When set as a model-level param in the
+proxy config, it must never be forwarded to the upstream provider's request
+body. OpenAI/Anthropic reject unknown body params with HTTP 400.
+"""
+
+import os
+import sys
+from unittest.mock import MagicMock
+
+sys.path.insert(0, os.path.abspath("../../../.."))
+
+import litellm
+from litellm.types.utils import all_litellm_params
+from litellm.utils import get_non_default_completion_params
+
+
+def test_use_chat_completions_api_is_a_known_litellm_param():
+    assert "use_chat_completions_api" in all_litellm_params
+
+
+def test_use_chat_completions_api_not_forwarded_as_provider_param():
+    forwarded = get_non_default_completion_params(
+        {"use_chat_completions_api": True, "temperature": 0.5}
+    )
+    assert "use_chat_completions_api" not in forwarded
+
+
+def test_completion_does_not_leak_flag_into_provider_request_body():
+    mock_response = MagicMock()
+    mock_response.model_dump.return_value = {
+        "id": "chatcmpl-1",
+        "object": "chat.completion",
+        "created": 1234567890,
+        "model": "gpt-4o-mini",
+        "choices": [
+            {
+                "index": 0,
+                "message": {"role": "assistant", "content": "hi"},
+                "finish_reason": "stop",
+            }
+        ],
+        "usage": {
+            "prompt_tokens": 1,
+            "completion_tokens": 1,
+            "total_tokens": 2,
+        },
+    }
+
+    mock_raw_response = MagicMock()
+    mock_raw_response.headers = {}
+    mock_raw_response.parse.return_value = mock_response
+
+    mock_client = MagicMock()
+    mock_client.chat.completions.with_raw_response.create.return_value = (
+        mock_raw_response
+    )
+
+    litellm.completion(
+        model="openai/gpt-4o-mini",
+        messages=[{"role": "user", "content": "hi"}],
+        use_chat_completions_api=True,
+        api_key="sk-test",
+        client=mock_client,
+    )
+
+    create_kwargs = (
+        mock_client.chat.completions.with_raw_response.create.call_args.kwargs
+    )
+    assert "use_chat_completions_api" not in create_kwargs
+    assert "use_chat_completions_api" not in (create_kwargs.get("extra_body") or {})

tests/test_litellm/proxy/test_common_request_processing.py

@@ -2244,6 +2244,41 @@ class TestHandleLLMApiExceptionDictDetail:
         assert proxy_exc.message == "Content blocked by guardrail"
         assert proxy_exc.provider_specific_fields is None
 
+    async def test_already_normalized_proxy_exception_is_honored(self):
+        """A ProxyException raised mid-request (e.g. a guardrail block) is already
+        the OpenAI wire format. The funnel must re-raise it untouched instead of
+        re-deriving the status from a (nonexistent) status_code attribute and
+        defaulting to 500. Regression for LIT-3751."""
+        from litellm.proxy._types import ProxyException
+
+        exc = ProxyException(
+            message='"Leroy Jenkins" detected as name',
+            type="invalid_request_error",
+            param=None,
+            code=400,
+            openai_code="content_policy_violation",
+        )
+        proxy_exc = await self._invoke(exc)
+        assert proxy_exc is exc
+        assert proxy_exc.code == "400"
+        assert proxy_exc.type == "invalid_request_error"
+        assert proxy_exc.param is None
+        assert proxy_exc.openai_code == "content_policy_violation"
+        assert proxy_exc.message == '"Leroy Jenkins" detected as name'
+
+        # The body the OpenAI-SDK client actually receives. The HTTP status line
+        # comes from int(exc.code) == 400; the wire ``code`` stays the status
+        # string. ``openai_code`` ("content_policy_violation") is intentionally
+        # NOT serialized here - to_dict() emits only ``code`` - so this asserts
+        # the real contract rather than the write-only attribute.
+        assert int(proxy_exc.code) == 400
+        assert proxy_exc.to_dict() == {
+            "message": '"Leroy Jenkins" detected as name',
+            "type": "invalid_request_error",
+            "param": None,
+            "code": "400",
+        }
+
 
 class TestAsyncStreamingDataGeneratorFastPath:
     """Fast/slow path branching in async_streaming_data_generator."""

tests/test_litellm/proxy/test_proxy_types.py

@@ -87,3 +87,76 @@ def test_user_api_key_auth_hashes_authorization_header_form_of_key():
         assert from_header.api_key == baseline.api_key
         assert from_header.token == baseline.token
         assert not from_header.api_key.lower().startswith("bearer")
+
+
+# === Regression tests for LIT-3094: ProxyException must populate Exception.args
+# so logging integrations using str(exc) record a non-empty error_message. ===
+
+
+def test_proxy_exception_str_returns_message():
+    """str(ProxyException) must return the stored message, not '' (LIT-3094)."""
+    from litellm.proxy._types import ProxyException
+
+    msg = "key not allowed to access model"
+    exc = ProxyException(message=msg, type="auth_error", param=None, code=401)
+    assert str(exc) == msg
+    assert exc.args == (msg,)
+    assert exc.message == msg
+
+
+def test_proxy_exception_populates_standard_logging_error_message():
+    """The full logging path used by proxy callbacks must capture the message
+    instead of recording an empty error_message (LIT-3094 report)."""
+    from litellm.litellm_core_utils.litellm_logging import (
+        StandardLoggingPayloadSetup,
+    )
+    from litellm.proxy._types import ProxyException
+
+    msg = "Authentication Error, Invalid proxy server token passed."
+    exc = ProxyException(message=msg, type="auth_error", param=None, code=401)
+    info = StandardLoggingPayloadSetup.get_error_information(original_exception=exc)
+    assert info["error_message"] == msg
+    assert info["error_class"] == "ProxyException"
+    assert info["error_code"] == "401"
+
+
+def test_proxy_exception_to_dict_unchanged():
+    """to_dict() shape must remain backwards-compatible after the fix."""
+    from litellm.proxy._types import ProxyException
+
+    exc = ProxyException(
+        message="boom", type="invalid_request_error", param="model", code=400
+    )
+    d = exc.to_dict()
+    assert d == {
+        "message": "boom",
+        "type": "invalid_request_error",
+        "param": "model",
+        "code": "400",
+    }
+
+
+def test_proxy_exception_routing_code_override_still_works():
+    """The 'No healthy deployment available' -> 429 remapping must survive
+    the super().__init__() addition."""
+    from litellm.proxy._types import ProxyException
+
+    exc = ProxyException(
+        message="No healthy deployment available for model=foo",
+        type="router_error",
+        param=None,
+        code=500,
+    )
+    assert exc.code == "429"
+    assert str(exc) == "No healthy deployment available for model=foo"
+
+
+def test_proxy_exception_non_string_message_coerced():
+    """Non-string `message` must still be coerced to str via self.message =
+    str(message), and Exception.args must reflect the coerced value."""
+    from litellm.proxy._types import ProxyException
+
+    exc = ProxyException(message=42, type="x", param=None, code=400)
+    assert exc.message == "42"
+    assert str(exc) == "42"
+    assert exc.args == ("42",)

tests/test_litellm/proxy/test_proxy_utils.py

@@ -15,7 +15,7 @@ sys.path.insert(
 )  # Adds the parent directory to the system path
 
 
-from unittest.mock import MagicMock
+from unittest.mock import MagicMock, patch
 
 from litellm.proxy.utils import get_custom_url, join_paths
 
@@ -368,3 +368,117 @@ class TestPostCallFailureHookLiftsFirstApiCallStartTime:
         await self._run(request_data)
         assert "first_api_call_start_time" not in request_data
         assert "litellm_logging_obj" not in request_data
+
+
+class TestPostCallFailureHookLLMExceptionAlerting:
+    """The llm_exceptions alert is for infra / LLM-API failures, not user
+    errors (https://github.com/BerriAI/litellm/issues/3395). Already-normalized
+    client errors must be excluded so a guardrail content-policy block never
+    pages on-call. ProxyException is such an error; before LIT-3751 only
+    HTTPException was excluded, so AIM blocks paged as if the LLM API failed."""
+
+    async def _alerted(self, exc) -> bool:
+        import asyncio
+        from unittest.mock import AsyncMock
+
+        from litellm.proxy._types import AlertType, UserAPIKeyAuth
+
+        proxy_logging_obj = ProxyLogging(user_api_key_cache=DualCache())
+        proxy_logging_obj.alert_types = [AlertType.llm_exceptions]
+        alerting_handler = AsyncMock()
+        with (
+            patch.object(proxy_logging_obj, "update_request_status", new=AsyncMock()),
+            patch.object(proxy_logging_obj, "alerting_handler", new=alerting_handler),
+        ):
+            await proxy_logging_obj.post_call_failure_hook(
+                request_data={},
+                original_exception=exc,
+                user_api_key_dict=UserAPIKeyAuth(),
+            )
+        await asyncio.sleep(0)  # let the fire-and-forget alert task run
+        return alerting_handler.called
+
+    @pytest.mark.asyncio
+    async def test_proxy_exception_does_not_alert(self):
+        from litellm.proxy._types import ProxyException
+
+        exc = ProxyException(
+            message="content blocked",
+            type="invalid_request_error",
+            param=None,
+            code=400,
+            openai_code="content_policy_violation",
+        )
+        assert await self._alerted(exc) is False
+
+    @pytest.mark.asyncio
+    async def test_http_exception_does_not_alert(self):
+        assert (
+            await self._alerted(HTTPException(status_code=400, detail="blocked"))
+            is False
+        )
+
+    @pytest.mark.asyncio
+    async def test_genuine_llm_api_error_still_alerts(self):
+        assert await self._alerted(Exception("upstream 503")) is True
+
+
+class TestPostCallFailureHookProxyExceptionLogging:
+    """A guardrail block raises a ProxyException; on an LLM route it must still
+    drive proxy-only failure logging (_handle_logging_proxy_only_error) so the
+    blocked request is recorded, exactly as the old HTTPException did. Before
+    LIT-3751 the classifier only matched HTTPException, so switching AIM to
+    ProxyException silently dropped the rejected prompt from failure logs."""
+
+    async def _logged(self, exc, *, request_route) -> bool:
+        from unittest.mock import AsyncMock
+
+        from litellm.proxy._types import UserAPIKeyAuth
+
+        proxy_logging_obj = ProxyLogging(user_api_key_cache=DualCache())
+        proxy_logging_obj.alert_types = []
+        handle_mock = AsyncMock()
+        with (
+            patch.object(proxy_logging_obj, "update_request_status", new=AsyncMock()),
+            patch.object(
+                proxy_logging_obj,
+                "_handle_logging_proxy_only_error",
+                new=handle_mock,
+            ),
+        ):
+            await proxy_logging_obj.post_call_failure_hook(
+                request_data={},
+                original_exception=exc,
+                user_api_key_dict=UserAPIKeyAuth(
+                    api_key="sk-test", request_route=request_route
+                ),
+            )
+        return handle_mock.await_count > 0
+
+    def _block(self):
+        from litellm.proxy._types import ProxyException
+
+        return ProxyException(
+            message="content blocked",
+            type="invalid_request_error",
+            param=None,
+            code=400,
+            openai_code="content_policy_violation",
+        )
+
+    @pytest.mark.asyncio
+    async def test_proxy_exception_on_llm_route_is_logged(self):
+        assert (
+            await self._logged(self._block(), request_route="/v1/chat/completions")
+            is True
+        )
+
+    @pytest.mark.asyncio
+    async def test_generic_exception_on_llm_route_is_not_logged(self):
+        # A raw provider/unknown exception is logged by the LLM call path, not here.
+        assert (
+            await self._logged(
+                Exception("upstream 503"), request_route="/v1/chat/completions"
+            )
+            is False
+        )

uv.lock

@@ -9,7 +9,7 @@ resolution-markers = [
 ]
 
 [options]
-exclude-newer = "2026-06-14T19:22:16.739045Z"
+exclude-newer = "2026-06-17T19:10:29.753403Z"
 exclude-newer-span = "P3D"
 
 [manifest]
@@ -3280,7 +3280,7 @@ wheels = [
 
 [[package]]
 name = "litellm"
-version = "1.88.3"
+version = "1.88.4"
 source = { editable = "." }
 dependencies = [
     { name = "aiohttp" },