Merge d0582d942b into 84617874b8

litellm/completion_extras/litellm_responses_transformation/transformation.py

@@ -36,7 +36,7 @@ from litellm.completion_extras.litellm_responses_transformation.history import (
 from litellm.integrations.anthropic_cache_control_hook import supports_openai_prompt_cache_breakpoint
 from litellm.litellm_core_utils.hidden_params import get_hidden_params, get_or_create_hidden_params
 from litellm.litellm_core_utils.prompt_templates.common_utils import (
-    prompt_cache_breakpoint_for_wire,
+    prompt_cache_breakpoint_with_readable_mode,
     responses_reasoning_items_from_thinking_blocks,
     with_prompt_cache_breakpoint,
 )
@@ -1666,7 +1666,9 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
                     if original_type == "text":
                         converted = with_prompt_cache_breakpoint(
                             self._convert_content_str_to_input_text(item.get("text", ""), role),
-                            prompt_cache_breakpoint_for_wire(item.get("prompt_cache_breakpoint"), drop_params),
+                            prompt_cache_breakpoint_with_readable_mode(
+                                item.get("prompt_cache_breakpoint"), drop_params
+                            ),
                         )
                         result.append(converted)
                         verbose_logger.debug("Chat provider:   text -> %s", converted)
@@ -1679,14 +1681,18 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
                                     cast(ChatCompletionImageObject, item), role
                                 ),
                             ),
-                            prompt_cache_breakpoint_for_wire(item.get("prompt_cache_breakpoint"), drop_params),
+                            prompt_cache_breakpoint_with_readable_mode(
+                                item.get("prompt_cache_breakpoint"), drop_params
+                            ),
                         )
                         result.append(converted)
                         verbose_logger.debug("Chat provider:   image_url -> %s", converted)
                     elif original_type == "input_audio":
                         converted = with_prompt_cache_breakpoint(
                             {"type": "input_audio", "input_audio": item.get("input_audio")},
-                            prompt_cache_breakpoint_for_wire(item.get("prompt_cache_breakpoint"), drop_params),
+                            prompt_cache_breakpoint_with_readable_mode(
+                                item.get("prompt_cache_breakpoint"), drop_params
+                            ),
                         )
                         result.append(converted)
                         verbose_logger.debug("Chat provider:   input_audio -> %s", converted)
@@ -1702,7 +1708,9 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
                                 _input_file_from_file_value(
                                     cast("ChatCompletionFileObject", item).get("file"),  # cast-ok: type tag checked
                                 ),
-                                prompt_cache_breakpoint_for_wire(item.get("prompt_cache_breakpoint"), drop_params),
+                                prompt_cache_breakpoint_with_readable_mode(
+                                    item.get("prompt_cache_breakpoint"), drop_params
+                                ),
                             )
                             result.append(converted)
                             verbose_logger.debug("Chat provider:   file -> %s", converted)
@@ -1726,7 +1734,9 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
                             # Default to input_text for unknown types
                             converted = with_prompt_cache_breakpoint(
                                 self._convert_content_str_to_input_text(str(item.get("text", item)), role),
-                                prompt_cache_breakpoint_for_wire(item.get("prompt_cache_breakpoint"), drop_params),
+                                prompt_cache_breakpoint_with_readable_mode(
+                                    item.get("prompt_cache_breakpoint"), drop_params
+                                ),
                             )
                             result.append(converted)
                             verbose_logger.debug("Chat provider:   unknown(%s) -> %s", original_type, converted)

litellm/integrations/anthropic_cache_control_hook.py

@@ -17,7 +17,7 @@ from collections.abc import Callable, Container, Iterable, Mapping, Sequence
 from typing import TYPE_CHECKING, Any, Final, cast
 from urllib.parse import urlparse
 
-from pydantic import TypeAdapter, ValidationError
+from pydantic import BaseModel, ConfigDict, TypeAdapter, ValidationError
 
 from litellm._logging import verbose_logger
 from litellm.integrations.custom_logger import CustomLogger
@@ -148,6 +148,30 @@ def openai_prompt_cache_breakpoint_support(model: str, custom_llm_provider: str
     return _named_model_breakpoint_support(model)
 
 
+class _DeclaredBaseModels(BaseModel):
+    model_config = ConfigDict(frozen=True, extra="ignore", from_attributes=True)
+
+    base_model: object = None
+    model_info: object = None
+
+    def names(self) -> tuple[str, ...]:
+        model_info: Final = _validated_object_mapping(self.model_info) or {}
+        candidates: Final = (self.base_model, model_info.get("base_model"))
+        return tuple(name for name in candidates if isinstance(name, str) and name)
+
+
+def deployment_prompt_cache_breakpoint_support(
+    model: str, custom_llm_provider: str, litellm_params: object
+) -> bool | None:
+    """``openai_prompt_cache_breakpoint_support`` for a deployment, read from its declared base model first."""
+    declared: Final = _DeclaredBaseModels.model_validate(litellm_params)
+    supports: Final = (
+        openai_prompt_cache_breakpoint_support(name, custom_llm_provider)
+        for name in dict.fromkeys((*declared.names(), model))
+    )
+    return next((support for support in supports if support is not None), None)
+
+
 def _named_model_breakpoint_support(model: str) -> bool | None:
     bare_model: Final = model.rsplit("/", 1)[-1].lower()
     if _O_SERIES_PATTERN.match(bare_model) is not None:

litellm/litellm_core_utils/prompt_templates/common_utils.py

@@ -29,6 +29,7 @@ from litellm.types.llms.openai import (
     ChatCompletionAudioObject,
     ChatCompletionFileObject,
     ChatCompletionImageObject,
+    ChatCompletionPromptCacheBreakpoint,
     ChatCompletionReasoningItem,
     ChatCompletionReasoningSummaryTextBlock,
     ChatCompletionRedactedThinkingBlock,
@@ -2179,16 +2180,62 @@ def with_prompt_cache_breakpoint(target: _MarkedT, marker: object) -> _MarkedT:
 
 
 _PROMPT_CACHE_BREAKPOINT: Final = TypeAdapter(PromptCacheBreakpoint)
+_CHAT_COMPLETION_PROMPT_CACHE_BREAKPOINT: Final = TypeAdapter(ChatCompletionPromptCacheBreakpoint)
+_AcceptedBreakpointT: Final = TypeVar("_AcceptedBreakpointT", bound=Mapping[str, object])
+
+
+def _accepted_prompt_cache_breakpoint(
+    accepted: TypeAdapter[_AcceptedBreakpointT], marker: object
+) -> _AcceptedBreakpointT | None:
+    try:
+        return accepted.validate_python(marker)
+    except ValidationError:
+        return None
+
+
+def prompt_cache_breakpoint_within(
+    accepted: TypeAdapter[_AcceptedBreakpointT], marker: object
+) -> _AcceptedBreakpointT | None:
+    """The client's marker cut down to the ``accepted`` shape: an unsupported ttl is dropped and the mode kept,
+    and a marker whose mode the target cannot read is dropped whole."""
+    whole: Final = _accepted_prompt_cache_breakpoint(accepted, marker)
+    if whole is not None:
+        return whole
+    fields: Final = _breakpoint_fields_or_none(marker)
+    without_ttl: Final = (
+        None
+        if fields is None
+        else _accepted_prompt_cache_breakpoint(accepted, {key: item for key, item in fields.items() if key != "ttl"})
+    )
+    verbose_logger.debug(
+        "Dropping malformed prompt_cache_breakpoint %r under drop_params"
+        if without_ttl is None
+        else "Dropping the unsupported ttl of prompt_cache_breakpoint %r under drop_params",
+        marker,
+    )
+    return without_ttl
 
 
 def prompt_cache_breakpoint_for_wire(marker: object, drop_params: bool) -> object:
     if marker is None or not drop_params:
         return marker
-    try:
-        return _PROMPT_CACHE_BREAKPOINT.validate_python(marker)
-    except ValidationError:
-        verbose_logger.debug("Dropping malformed prompt_cache_breakpoint %r under drop_params", marker)
-        return None
+    return prompt_cache_breakpoint_within(_PROMPT_CACHE_BREAKPOINT, marker)
+
+
+def prompt_cache_breakpoint_with_readable_mode(marker: object, drop_params: bool) -> object:
+    """The client's marker as written, or None under drop_params when no target could read its mode, so the
+    target's own config decides which ttl and keys it keeps."""
+    return None if prompt_cache_breakpoint_for_wire(marker, drop_params) is None else marker
+
+
+def whole_prompt_cache_breakpoint_or_none(marker: object) -> PromptCacheBreakpoint | None:
+    return _accepted_prompt_cache_breakpoint(_PROMPT_CACHE_BREAKPOINT, marker)
+
+
+def chat_completion_prompt_cache_breakpoint_for_wire(marker: object, drop_params: bool) -> object:
+    if marker is None or not drop_params:
+        return marker
+    return prompt_cache_breakpoint_within(_CHAT_COMPLETION_PROMPT_CACHE_BREAKPOINT, marker)
 
 
 def cache_control_as_prompt_cache_breakpoint(block: _MarkedT) -> _MarkedT:
@@ -2206,6 +2253,13 @@ _BREAKPOINT_FIELDS: Final = TypeAdapter(Mapping[str, object])
 _BREAKPOINT_BLOCKS: Final = TypeAdapter(list[object])
 
 
+def _breakpoint_fields_or_none(value: object) -> Mapping[str, object] | None:
+    try:
+        return _BREAKPOINT_FIELDS.validate_python(value)
+    except ValidationError:
+        return None
+
+
 def _without_prompt_cache_breakpoint(value: object) -> object:
     if not isinstance(value, Mapping):
         return value
@@ -2228,6 +2282,27 @@ def without_prompt_cache_breakpoints(block: _MarkedT) -> _MarkedT:
     return cast(_MarkedT, stripped)  # cast-ok: same block shape as the input minus the marker keys
 
 
+def _part_with_chat_completion_prompt_cache_breakpoint(part: object) -> object:
+    if not isinstance(part, Mapping):
+        return part
+    fields: Final = _BREAKPOINT_FIELDS.validate_python(part)
+    marker: Final = fields.get("prompt_cache_breakpoint")
+    if marker is None:
+        return fields
+    unmarked: Final = {key: item for key, item in fields.items() if key != "prompt_cache_breakpoint"}
+    return with_prompt_cache_breakpoint(unmarked, chat_completion_prompt_cache_breakpoint_for_wire(marker, True))
+
+
+def chat_completion_prompt_cache_breakpoints_for_wire(message: _MarkedT, drop_params: bool) -> _MarkedT:
+    content: Final = message.get("content")
+    if not drop_params or not isinstance(content, list):
+        return message
+    parts: Final = [
+        _part_with_chat_completion_prompt_cache_breakpoint(part) for part in _BREAKPOINT_BLOCKS.validate_python(content)
+    ]
+    return cast(_MarkedT, {**message, "content": parts})  # cast-ok: same message shape with its parts' markers trimmed
+
+
 LITELLM_RESPONSE_ONLY_MESSAGE_FIELDS: Final = frozenset(
     {"thinking_blocks", "reasoning_items", "provider_specific_fields"}
 )

litellm/llms/anthropic/messages/responses_adapters/handler.py

@@ -21,7 +21,7 @@ from litellm.litellm_core_utils.asyncify import (
 from litellm.litellm_core_utils.core_helpers import as_str_mapping, drops_unsupported_params
 from litellm.litellm_core_utils.prompt_templates.common_utils import (
     cache_control_as_prompt_cache_breakpoint,
-    prompt_cache_breakpoint_for_wire,
+    prompt_cache_breakpoint_with_readable_mode,
     with_prompt_cache_breakpoint,
 )
 from litellm.responses.utils import auto_reasoning_summary
@@ -207,7 +207,7 @@ def _without_prompt_cache_breakpoint(block: Mapping[str, object]) -> Mapping[str
 def _cache_control_as_wire_safe_prompt_cache_breakpoint(block: Mapping[str, object]) -> Mapping[str, object]:
     if "prompt_cache_breakpoint" not in block:
         return cache_control_as_prompt_cache_breakpoint(block)
-    wire_marker: Final = prompt_cache_breakpoint_for_wire(block["prompt_cache_breakpoint"], drop_params=True)
+    wire_marker: Final = prompt_cache_breakpoint_with_readable_mode(block["prompt_cache_breakpoint"], drop_params=True)
     return cache_control_as_prompt_cache_breakpoint(
         with_prompt_cache_breakpoint(_without_prompt_cache_breakpoint(block), wire_marker)
     )

litellm/llms/azure/chat/gpt_reasoning_transformation.py

@@ -1,6 +1,6 @@
 """Support for Azure OpenAI gpt-5 model family."""
 
-from collections.abc import Collection
+from collections.abc import Collection, Mapping
 from typing import ClassVar, Final
 
 import litellm
@@ -105,7 +105,7 @@ class AzureOpenAIGPT5Config(AzureOpenAIConfig, OpenAIGPT5Config):
         model: str,
         messages: list[AllMessageValues],
         optional_params: dict,
-        litellm_params: dict,
+        litellm_params: Mapping[str, object],
         headers: dict,
     ) -> dict:
         model = model.replace(self.GPT5_SERIES_ROUTE, "")

litellm/llms/azure/chat/gpt_transformation.py

@@ -5,12 +5,16 @@ from typing import TYPE_CHECKING, Any, Final
 from httpx._models import Headers, Response
 
 import litellm
+from litellm.integrations.anthropic_cache_control_hook import deployment_prompt_cache_breakpoint_support
+from litellm.litellm_core_utils.core_helpers import drops_unsupported_params
 from litellm.litellm_core_utils.prompt_templates.common_utils import (
+    chat_completion_prompt_cache_breakpoints_for_wire,
     drop_tool_reference_parts_from_tool_messages,
     flatten_combinators_and_drop_non_python_regex_patterns,
     hoist_images_from_tool_messages,
     system_messages_first,
     tool_with_sanitized_parameters,
+    without_prompt_cache_breakpoints,
 )
 from litellm.litellm_core_utils.prompt_templates.factory import (
     convert_to_azure_openai_messages,
@@ -290,11 +294,23 @@ class AzureOpenAIConfig(BaseConfig):
         model: str,
         messages: list[AllMessageValues],
         optional_params: dict,
-        litellm_params: dict,
+        litellm_params: Mapping[str, object],
         headers: dict,
     ) -> dict:
         ordered_messages: Final = system_messages_first(messages) if litellm.openai_system_messages_first else messages
-        stripped_messages: Final = drop_tool_reference_parts_from_tool_messages(ordered_messages)
+        known_support: Final = deployment_prompt_cache_breakpoint_support(model, "azure", litellm_params)
+        drop_params: Final = drops_unsupported_params(litellm_params.get("drop_params"))
+        wire_messages: Final = (
+            ordered_messages
+            if not drop_params
+            else [without_prompt_cache_breakpoints(message) for message in ordered_messages]
+            if known_support is False
+            else [
+                chat_completion_prompt_cache_breakpoints_for_wire(message, drop_params=True)
+                for message in ordered_messages
+            ]
+        )
+        stripped_messages: Final = drop_tool_reference_parts_from_tool_messages(wire_messages)
         azure_messages: Final = convert_to_azure_openai_messages(hoist_images_from_tool_messages(stripped_messages))
         request_params: Final = MappingProxyType(
             {

litellm/llms/azure/responses/transformation.py

@@ -1,6 +1,7 @@
 import re
-from collections.abc import Mapping
+from collections.abc import Callable, Mapping
 from copy import deepcopy
+from functools import partial
 from types import MappingProxyType
 from typing import TYPE_CHECKING, Any, Final, Literal
 from urllib.parse import urlsplit
@@ -11,6 +12,7 @@ from pydantic import TypeAdapter
 
 from litellm._logging import verbose_logger
 from litellm.constants import AZURE_DEFAULT_RESPONSES_API_VERSION
+from litellm.litellm_core_utils.prompt_templates.common_utils import prompt_cache_breakpoint_within
 from litellm.litellm_core_utils.url_utils import encode_url_path_segment
 from litellm.llms.azure.chat.gpt_reasoning_transformation import AzureOpenAIGPT5Config
 from litellm.llms.azure.common_utils import BaseAzureLLM
@@ -29,6 +31,7 @@ else:
     LiteLLMLoggingObj = Any
 
 _DECODED_JSON: Final = TypeAdapter(object)
+_AZURE_RESPONSES_PROMPT_CACHE_BREAKPOINT: Final = TypeAdapter(AzureResponsesPromptCacheBreakpoint)
 
 
 _DATED_GA_API_VERSION: Final = re.compile(r"\d{4}-\d{2}-\d{2}")
@@ -81,6 +84,13 @@ class AzureOpenAIResponsesAPIConfig(OpenAIResponsesAPIConfig):
     def _is_gpt_5_model(model: str) -> bool:
         return AzureOpenAIGPT5Config.is_model_gpt_5_model(model)
 
+    def _prompt_cache_breakpoint_to_wire(
+        self, litellm_params: GenericLiteLLMParams, reads_breakpoints: bool, known_support: bool | None
+    ) -> Callable[[object], object]:
+        if reads_breakpoints or known_support is True:
+            return partial(prompt_cache_breakpoint_within, _AZURE_RESPONSES_PROMPT_CACHE_BREAKPOINT)
+        return super()._prompt_cache_breakpoint_to_wire(litellm_params, reads_breakpoints, known_support)
+
     @staticmethod
     def _supports_reasoning_effort_none(model: str) -> bool:
         return AzureOpenAIGPT5Config.supports_reasoning_effort_level(model, "none")

litellm/llms/bedrock_mantle/responses/transformation.py

@@ -15,7 +15,8 @@ role / access key / profile / web identity), signed via the shared
 BaseAWSLLM._sign_request after the request body is finalized.
 """
 
-from collections.abc import Mapping, Sequence
+from collections.abc import Callable, Mapping, Sequence
+from functools import partial
 from typing import Final, cast  # noqa: TID251  # map_openai_params returns the filtered params as a bare dict
 
 import httpx
@@ -23,6 +24,7 @@ import httpx
 import litellm
 from litellm._logging import verbose_logger
 from litellm.litellm_core_utils.core_helpers import drops_unsupported_params
+from litellm.litellm_core_utils.prompt_templates.common_utils import chat_completion_prompt_cache_breakpoint_for_wire
 from litellm.litellm_core_utils.url_utils import join_api_base
 from litellm.llms.base_llm.chat.transformation import BaseLLMException
 from litellm.llms.base_llm.responses.codex_compat import drop_unsupported_tools, normalize_codex_input_items
@@ -113,6 +115,11 @@ class BedrockMantleResponsesAPIConfig(BedrockMantleAuthMixin, OpenAIResponsesAPI
             headers["OpenAI-Project"] = litellm_params.aws_bedrock_project_id
         return headers
 
+    def _prompt_cache_breakpoint_to_wire(
+        self, litellm_params: GenericLiteLLMParams, reads_breakpoints: bool, known_support: bool | None
+    ) -> Callable[[object], object]:
+        return partial(chat_completion_prompt_cache_breakpoint_for_wire, drop_params=True)
+
     def supports_native_file_search(self) -> bool:
         return False
 

litellm/llms/openai/chat/gpt_transformation.py

@@ -14,13 +14,14 @@ from pydantic import TypeAdapter, ValidationError
 
 import litellm
 from litellm.constants import OPENAI_SYSTEM_MESSAGES_FIRST_PROVIDERS
-from litellm.litellm_core_utils.core_helpers import map_finish_reason
+from litellm.litellm_core_utils.core_helpers import drops_unsupported_params, map_finish_reason
 from litellm.litellm_core_utils.llm_response_utils.convert_dict_to_response import (
     extract_reasoning_content,
     handle_invalid_parallel_tool_calls,
     should_convert_tool_call_to_json_mode,
 )
 from litellm.litellm_core_utils.prompt_templates.common_utils import (
+    chat_completion_prompt_cache_breakpoints_for_wire,
     drop_non_python_regex_patterns,
     drop_tool_reference_parts_from_tool_messages,
     flatten_combinators_and_drop_non_python_regex_patterns,
@@ -531,9 +532,10 @@ class OpenAIGPTConfig(BaseLLMModelInfo, BaseConfig):
         """
         A deployment that reads OpenAI's ``prompt_cache_breakpoint`` (gpt-5.6+ on
         api.openai.com, or one opted in through ``prompt_cache_options`` or the cost
-        map) gets each Anthropic ``cache_control`` mark rewritten in place. Any other
-        endpoint keeps or strips the marks as before, and under drop_params also loses
-        every ``prompt_cache_breakpoint`` it cannot read.
+        map) gets each Anthropic ``cache_control`` mark rewritten in place, and under
+        drop_params each content part's ``prompt_cache_breakpoint`` cut down to what chat
+        completions accepts. Any other endpoint keeps or strips the marks as before, and
+        under drop_params also loses every ``prompt_cache_breakpoint`` it cannot read.
         """
         from litellm.integrations.anthropic_cache_control_hook import AnthropicCacheControlHook
 
@@ -544,11 +546,15 @@ class OpenAIGPTConfig(BaseLLMModelInfo, BaseConfig):
         )
         if not reads_breakpoints and self._should_preserve_cache_control_for_endpoint(custom_llm_provider, api_base):
             return messages, tools
-        drops_breakpoints: Final = not reads_breakpoints and (
-            litellm.drop_params is True or litellm_params.get("drop_params") is True
-        )
+        drop_params: Final = drops_unsupported_params(litellm_params.get("drop_params"))
+        drops_breakpoints: Final = not reads_breakpoints and drop_params
         renamed: Final = (
-            [AnthropicCacheControlHook.cache_control_as_prompt_cache_breakpoints(message) for message in messages]
+            [
+                AnthropicCacheControlHook.cache_control_as_prompt_cache_breakpoints(
+                    chat_completion_prompt_cache_breakpoints_for_wire(message, drop_params)
+                )
+                for message in messages
+            ]
             if reads_breakpoints
             else [without_prompt_cache_breakpoints(message) for message in messages]
             if drops_breakpoints

litellm/llms/openai/responses/transformation.py

@@ -1,5 +1,5 @@
 from collections.abc import Callable, Collection, Iterable, Mapping, Sequence
-from functools import lru_cache
+from functools import lru_cache, partial
 from types import MappingProxyType
 from typing import TYPE_CHECKING, Any, Final, Protocol, cast, get_type_hints
 
@@ -12,7 +12,7 @@ import litellm
 from litellm._logging import verbose_logger
 from litellm.integrations.anthropic_cache_control_hook import (
     AnthropicCacheControlHook,
-    openai_prompt_cache_breakpoint_support,
+    deployment_prompt_cache_breakpoint_support,
     targets_openai_api,
 )
 from litellm.litellm_core_utils.core_helpers import normalize_drop_params, process_response_headers
@@ -25,6 +25,7 @@ from litellm.litellm_core_utils.prompt_templates.common_utils import (
     drop_non_python_regex_patterns,
     flatten_combinators_and_drop_non_python_regex_patterns,
     prompt_cache_breakpoint_for_wire,
+    whole_prompt_cache_breakpoint_or_none,
     with_prompt_cache_breakpoint,
 )
 from litellm.litellm_core_utils.safe_json_loads import safe_json_loads
@@ -103,13 +104,6 @@ _INPUT_ITEM: Final[TypeAdapter[Mapping[str, object]]] = TypeAdapter(Mapping[str,
 _CONTENT_PARTS: Final[TypeAdapter[Sequence[Mapping[str, object]]]] = TypeAdapter(Sequence[Mapping[str, object]])
 
 
-class _DeclaredModel(BaseModel):
-    model_config = ConfigDict(frozen=True, extra="ignore", from_attributes=True)
-
-    base_model: object = None
-    model_info: object = None
-
-
 class _DeploymentHost(BaseModel):
     """`responses()` builds the params from raw kwargs, where callers may spell the host `base_url`."""
 
@@ -241,12 +235,12 @@ def _json_array_or_empty(value: object) -> Sequence[object]:
     return value if isinstance(value, (list, tuple)) else ()
 
 
-def _item_with_wire_safe_prompt_cache_breakpoints(item: object, keeps_markers: bool) -> object:
+def _item_with_wire_safe_prompt_cache_breakpoints(item: object, to_wire: Callable[[object], object] | None) -> object:
     fields: Final = _json_object_or_none(item)
     if fields is None or not _carries_prompt_cache_breakpoint(fields):
         return item
     wire_safe_fields: Final = {
-        key: _parts_with_wire_safe_prompt_cache_breakpoints(key, value, keeps_markers)
+        key: _parts_with_wire_safe_prompt_cache_breakpoints(key, value, to_wire)
         for key, value in fields.items()
         if key != "prompt_cache_breakpoint"
     }
@@ -270,21 +264,23 @@ def _parts_carry_prompt_cache_breakpoint(parts: object) -> bool:
     return any(isinstance(part, Mapping) and "prompt_cache_breakpoint" in part for part in _json_array_or_empty(parts))
 
 
-def _parts_with_wire_safe_prompt_cache_breakpoints(key: str, value: object, keeps_markers: bool) -> object:
+def _parts_with_wire_safe_prompt_cache_breakpoints(
+    key: str, value: object, to_wire: Callable[[object], object] | None
+) -> object:
     if key not in _INPUT_PART_LIST_KEYS or not _parts_carry_prompt_cache_breakpoint(value):
         return value
     parts: Final = _json_array_or_empty(value)
-    wire_safe_parts: Final = tuple(_part_with_wire_safe_prompt_cache_breakpoint(part, keeps_markers) for part in parts)
+    wire_safe_parts: Final = tuple(_part_with_wire_safe_prompt_cache_breakpoint(part, to_wire) for part in parts)
     return value if _same_objects(wire_safe_parts, parts) else list(wire_safe_parts)
 
 
-def _part_with_wire_safe_prompt_cache_breakpoint(part: object, keeps_markers: bool) -> object:
+def _part_with_wire_safe_prompt_cache_breakpoint(part: object, to_wire: Callable[[object], object] | None) -> object:
     fields: Final = _json_object_or_none(part)
     if fields is None or "prompt_cache_breakpoint" not in fields:
         return part
     marker: Final = fields["prompt_cache_breakpoint"]
-    wire_marker: Final = prompt_cache_breakpoint_for_wire(marker, drop_params=True) if keeps_markers else None
-    if keeps_markers and wire_marker == marker:
+    wire_marker: Final = None if to_wire is None else to_wire(marker)
+    if to_wire is not None and wire_marker == marker:
         return part
     unmarked: Final = {key: value for key, value in fields.items() if key != "prompt_cache_breakpoint"}
     return with_prompt_cache_breakpoint(unmarked, wire_marker)
@@ -699,24 +695,30 @@ class OpenAIResponsesAPIConfig(BaseResponsesAPIConfig):
     ) -> str | ResponseInputParam:
         if not isinstance(input, list) or not drops_params(litellm_params):
             return input
-        keeps_markers: Final = reads_breakpoints or self._keeps_prompt_cache_breakpoints(
-            model=model, litellm_params=litellm_params
-        )
-        wire_safe_items: Final = tuple(
-            _item_with_wire_safe_prompt_cache_breakpoints(item, keeps_markers) for item in input
+        known_support: Final = self._prompt_cache_breakpoint_support(model=model, litellm_params=litellm_params)
+        to_wire: Final = (
+            self._prompt_cache_breakpoint_to_wire(litellm_params, reads_breakpoints, known_support)
+            if reads_breakpoints or known_support is not False
+            else None
         )
+        wire_safe_items: Final = tuple(_item_with_wire_safe_prompt_cache_breakpoints(item, to_wire) for item in input)
         if _same_objects(wire_safe_items, input):
             return input
         wire_safe_input: Final = list(wire_safe_items)
         return cast("ResponseInputParam", wire_safe_input)  # cast-ok: items keep their shape, minus a rejected marker
 
-    def _keeps_prompt_cache_breakpoints(self, model: str, litellm_params: GenericLiteLLMParams) -> bool:
-        provider: Final = _provider_name(self.custom_llm_provider)
-        base_model: Final = self._declared_base_model(model=model, litellm_params=litellm_params)
-        known_support: Final = (
-            openai_prompt_cache_breakpoint_support(name, provider) for name in dict.fromkeys((base_model, model))
+    def _prompt_cache_breakpoint_to_wire(
+        self, litellm_params: GenericLiteLLMParams, reads_breakpoints: bool, known_support: bool | None
+    ) -> Callable[[object], object]:
+        opted_in_openai_host: Final = reads_breakpoints and self.custom_llm_provider == LlmProviders.OPENAI
+        if opted_in_openai_host or self._targets_openai_api(litellm_params):
+            return partial(prompt_cache_breakpoint_for_wire, drop_params=True)
+        return whole_prompt_cache_breakpoint_or_none
+
+    def _prompt_cache_breakpoint_support(self, model: str, litellm_params: GenericLiteLLMParams) -> bool | None:
+        return deployment_prompt_cache_breakpoint_support(
+            model, _provider_name(self.custom_llm_provider), litellm_params
         )
-        return next((support for support in known_support if support is not None), None) is not False
 
     def _targets_openai_api(self, litellm_params: GenericLiteLLMParams) -> bool:
         host: Final = _DeploymentHost.model_validate(litellm_params)
@@ -885,13 +887,6 @@ class OpenAIResponsesAPIConfig(BaseResponsesAPIConfig):
         base_model: Final[object] = model_info.get("base_model") if isinstance(model_info, dict) else None
         return base_model if isinstance(base_model, str) and base_model else model
 
-    @staticmethod
-    def _declared_base_model(model: str, litellm_params: GenericLiteLLMParams) -> str:
-        declared: Final = _DeclaredModel.model_validate(litellm_params)
-        model_info: Final = _json_object_or_none(declared.model_info)
-        candidates: Final = (declared.base_model, None if model_info is None else model_info.get("base_model"))
-        return next((name for name in candidates if isinstance(name, str) and name), model)
-
     @staticmethod
     def _sanitized_tool_entry(
         entry: Mapping[str, object],

litellm/types/llms/openai.py

@@ -686,6 +686,15 @@ class PromptCacheBreakpoint(TypedDict):
     ttl: NotRequired[ReadOnly[Literal["30m", "12h"]]]
 
 
+class ChatCompletionPromptCacheBreakpoint(TypedDict):
+    mode: ReadOnly[Literal["explicit"]]
+
+
+class AzureResponsesPromptCacheBreakpoint(TypedDict):
+    mode: ReadOnly[Literal["explicit"]]
+    ttl: NotRequired[ReadOnly[Literal["30m"]]]
+
+
 class PromptCacheOptions(TypedDict, total=False):
     mode: ReadOnly[Literal["implicit", "explicit"]]
     ttl: ReadOnly[Literal["30m", "12h"]]

tests/integration/providers/test_openai_chat_prompt_cache_breakpoint_wire.py

@@ -41,6 +41,7 @@ class _Proxy:
     openai: Wire
     gpt_5_6_with_prompt_cache_options: str
     gpt_5_6_without_prompt_cache_options: str
+    gpt_5_6_dropping_params: str
 
     def openai_receives(
         self, model: str, messages: Sequence[JsonValue], *, stream: bool = False, tools: Sequence[JsonValue] = ()
@@ -79,6 +80,9 @@ def proxy() -> Iterator[_Proxy]:
                 model="openai/gpt-5.6", api_base=api_base, prompt_cache_options={"mode": "explicit"}
             ),
             gpt_5_6_without_prompt_cache_options=scenario.model(model="openai/gpt-5.6", api_base=api_base),
+            gpt_5_6_dropping_params=scenario.model(
+                model="openai/gpt-5.6", api_base=api_base, prompt_cache_options={"mode": "explicit"}, drop_params=True
+            ),
         )
 
 
@@ -202,3 +206,56 @@ def test_custom_api_base_without_prompt_cache_options_forwards_cache_control_unc
     assert received["messages"] == _SYSTEM_AND_USER_WITH_CACHE_CONTROL
     assert "prompt_cache_options" not in received
     assert "prompt_cache_breakpoint" not in str(received)
+
+
+_CLIENT_BREAKPOINTS_ON_PARTS: Final[list[JsonValue]] = [
+    {"type": "text", "text": "Stable rules", "prompt_cache_breakpoint": {"mode": "explicit", "ttl": "1h"}},
+    {"type": "text", "text": "More rules", "prompt_cache_breakpoint": {"mode": "implicit"}},
+    {"type": "text", "text": "Describe this", "prompt_cache_breakpoint": {"mode": "explicit"}},
+]
+_CLIENT_BREAKPOINTS_AS_CHAT_COMPLETIONS_TAKES_THEM: Final[list[JsonValue]] = [
+    {"type": "text", "text": "Stable rules", "prompt_cache_breakpoint": {"mode": "explicit"}},
+    {"type": "text", "text": "More rules"},
+    {"type": "text", "text": "Describe this", "prompt_cache_breakpoint": {"mode": "explicit"}},
+]
+_CHAT_COMPLETIONS_REJECTS_TTL_AND_IMPLICIT: Final = (
+    "live gpt-6.1-sol Chat Completions answers 400 on a part breakpoint with any ttl or mode implicit (2026-10-28)"
+)
+
+
+@pytest.mark.parametrize("stream", [False, True], ids=["unary", "stream"])
+def test_drop_params_cuts_client_breakpoints_down_to_what_chat_completions_takes(proxy: _Proxy, stream: bool) -> None:
+    received: Final = proxy.openai_receives(
+        proxy.gpt_5_6_dropping_params, [{"role": "user", "content": _CLIENT_BREAKPOINTS_ON_PARTS}], stream=stream
+    )
+
+    assert received["messages"] == [
+        {"role": "user", "content": _CLIENT_BREAKPOINTS_AS_CHAT_COMPLETIONS_TAKES_THEM}
+    ], _CHAT_COMPLETIONS_REJECTS_TTL_AND_IMPLICIT
+
+
+def test_without_drop_params_client_breakpoints_reach_openai_unchanged(proxy: _Proxy) -> None:
+    received: Final = proxy.openai_receives(
+        proxy.gpt_5_6_with_prompt_cache_options, [{"role": "user", "content": _CLIENT_BREAKPOINTS_ON_PARTS}]
+    )
+
+    assert received["messages"] == [{"role": "user", "content": _CLIENT_BREAKPOINTS_ON_PARTS}]
+
+
+def test_drop_params_cuts_client_breakpoints_on_v1_messages_routed_to_chat_completions(proxy: _Proxy) -> None:
+    _ = proxy.openai.drain()
+    response: Final = proxy.gateway.request(
+        "POST",
+        "/v1/messages",
+        {
+            "model": proxy.gpt_5_6_dropping_params,
+            "max_tokens": 16,
+            "stop_sequences": ["END"],
+            "messages": [{"role": "user", "content": _CLIENT_BREAKPOINTS_ON_PARTS}],
+        },
+    )
+    assert response.status_code == 200, response.text
+
+    assert proxy.received_by_openai()["messages"] == [
+        {"role": "user", "content": _CLIENT_BREAKPOINTS_AS_CHAT_COMPLETIONS_TAKES_THEM}
+    ], _CHAT_COMPLETIONS_REJECTS_TTL_AND_IMPLICIT

tests/integration/providers/test_responses_bridge_prompt_cache_breakpoint_wire.py

@@ -27,8 +27,9 @@ _MALFORMED: Final[tuple[tuple[str, JsonValue], ...]] = (
     ("5kb-string", "x" * 5000),
     ("empty-object", {}),
     ("bogus-mode", {"mode": "bogus"}),
-    ("bad-ttl", {"mode": "explicit", "ttl": "1h"}),
+    ("bogus-mode-with-ttl", {"mode": "bogus", "ttl": "30m"}),
 )
+_UNSUPPORTED_TTL: Final[Mapping[str, JsonValue]] = {"mode": "explicit", "ttl": "1h"}
 _MALFORMED_IDS: Final = tuple(name for name, _ in _MALFORMED)
 _MALFORMED_VALUES: Final = tuple(value for _, value in _MALFORMED)
 _CASES: Final[tuple[tuple[str, Mode, JsonValue, JsonValue], ...]] = (
@@ -192,6 +193,27 @@ def test_marker_with_an_unknown_key(bridge: _Bridge, mode: Mode, expected: JsonV
     bridge.spend.landed(bridge.model(mode), call_id, marker)
 
 
+@pytest.mark.parametrize(
+    ("mode", "expected"), (("on", None), ("off", _UNSUPPORTED_TTL)), ids=("dropped-on", "verbatim-off")
+)
+def test_marker_with_an_unsupported_ttl(bridge: _Bridge, mode: Mode, expected: JsonValue) -> None:
+    marker: Final = uuid.uuid4().hex
+    call_id: Final = _completion(_chat(bridge, bridge.model(mode), [_user(marker, _UNSUPPORTED_TTL)]), marker)
+    pcb.assert_marker(_user_block_on_wire(bridge, marker), expected)
+    bridge.spend.landed(bridge.model(mode), call_id, marker)
+
+
+def test_marker_with_an_unsupported_ttl_keeps_its_mode_on_a_host_opted_in_with_prompt_cache_options(
+    bridge: _Bridge,
+) -> None:
+    marker: Final = uuid.uuid4().hex
+    with bridge.gateway.scenario() as scenario:
+        model: Final = scenario.model(model=pcb.MODEL, api_base=bridge.api_base, drop_params=True, **_OPTED_IN)
+        call_id: Final = _completion(_chat(bridge, model, [_user(marker, _UNSUPPORTED_TTL)]), marker)
+        pcb.assert_marker(_user_block_on_wire(bridge, marker), pcb.EXPLICIT)
+        bridge.spend.landed(model, call_id, marker)
+
+
 def _second_block_on_wire(bridge: _Bridge, marker: str) -> dict[str, JsonValue]:
     request: Final = pcb.posted(bridge.wire, marker)
     _wire_body(request)

tests/unit/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_transformation.py

@@ -5884,9 +5884,9 @@ _HAND_WRITTEN_PROMPT_CACHE_BREAKPOINT_BLOCKS: Final = (
 )
 
 
-def test_convert_chat_completion_messages_to_responses_api_drops_malformed_prompt_cache_breakpoint_under_drop_params():
-    """OpenAI's Responses API answered "Supported values are: '30m'" for a 1h breakpoint ttl on 2026-10-07,
-    so an unsupported ttl drops the marker as a unit while an unknown key is dropped from a valid one."""
+def test_convert_chat_completion_messages_to_responses_api_drops_only_an_unreadable_prompt_cache_breakpoint_under_drop_params():
+    """The target's Responses config decides which ttl and keys its API takes, so the bridge keeps every marker
+    whose mode some target reads."""
     handler = LiteLLMResponsesTransformationHandler()
     messages = [{"role": "user", "content": list(_HAND_WRITTEN_PROMPT_CACHE_BREAKPOINT_BLOCKS)}]
 
@@ -5899,12 +5899,12 @@ def test_convert_chat_completion_messages_to_responses_api_drops_malformed_promp
         None,
         None,
         None,
-        None,
-        {"mode": "explicit"},
+        {"mode": "explicit", "ttl": "1h"},
+        {"mode": "explicit", "scope": "all"},
         {"mode": "explicit", "ttl": "30m"},
         {"mode": "explicit"},
     ]
-    assert all("prompt_cache_breakpoint" not in block for block in content[:4])
+    assert all("prompt_cache_breakpoint" not in block for block in content[:3])
 
 
 def test_convert_chat_completion_messages_to_responses_api_keeps_malformed_prompt_cache_breakpoint_by_default():

tests/unit/litellm_core_utils/prompt_templates/test_litellm_core_utils_prompt_templates_common_utils.py

@@ -2336,7 +2336,8 @@ _MALFORMED_PROMPT_CACHE_BREAKPOINTS: Final = (
     "",
     {},
     {"mode": "bogus"},
-    {"mode": "explicit", "ttl": "1h"},
+    {"mode": "bogus", "ttl": "30m"},
+    {1: "explicit"},
 )
 _MALFORMED_PROMPT_CACHE_BREAKPOINT_IDS: Final = (
     "string",
@@ -2345,7 +2346,8 @@ _MALFORMED_PROMPT_CACHE_BREAKPOINT_IDS: Final = (
     "empty-string",
     "empty-object",
     "bogus-mode",
-    "bad-ttl",
+    "bogus-mode-with-ttl",
+    "non-string-key",
 )
 _VALID_PROMPT_CACHE_BREAKPOINTS: Final = ({"mode": "explicit"}, {"mode": "explicit", "ttl": "30m"})
 
@@ -2375,6 +2377,18 @@ def test_prompt_cache_breakpoint_for_wire_strips_an_unknown_key_under_drop_param
     assert prompt_cache_breakpoint_for_wire({"mode": "explicit", "note": "kept"}, True) == {"mode": "explicit"}
 
 
+@pytest.mark.parametrize(
+    "marker",
+    ({"mode": "explicit", "ttl": "1h"}, {"mode": "explicit", "ttl": "1h", "note": "kept"}),
+    ids=("bad-ttl", "bad-ttl-and-unknown-key"),
+)
+def test_prompt_cache_breakpoint_for_wire_drops_an_unsupported_ttl_and_keeps_the_mode_under_drop_params(
+    marker: dict[str, str],
+) -> None:
+    assert prompt_cache_breakpoint_for_wire(marker, True) == {"mode": "explicit"}
+    assert prompt_cache_breakpoint_for_wire(marker, False) is marker
+
+
 def test_cache_control_is_replaced_by_an_explicit_prompt_cache_breakpoint_on_the_same_block() -> None:
     marked: Final = {"type": "text", "text": "rules", "cache_control": {"type": "ephemeral", "ttl": "1h"}}
 

tests/unit/llms/azure/chat/test_azure_chat_gpt_transformation.py

@@ -806,3 +806,89 @@ def test_azure_keeps_a_later_developer_message_in_place_as_system(config: AzureO
     )
     assert [message["role"] for message in messages] == ["system", "user", "assistant", "system", "user"]
     assert messages[3]["content"] == "Answer with exactly one word."
+
+
+_MARKED_AZURE_PARTS: Final = (
+    {"type": "text", "text": "Stable prefix", "prompt_cache_breakpoint": {"mode": "explicit", "ttl": "30m"}},
+    {"type": "text", "text": "Implicit", "prompt_cache_breakpoint": {"mode": "implicit"}},
+)
+_TRIMMED_AZURE_PARTS: Final = [
+    {"type": "text", "text": "Stable prefix", "prompt_cache_breakpoint": {"mode": "explicit"}},
+    {"type": "text", "text": "Implicit"},
+]
+
+
+@pytest.fixture
+def registered_azure_breakpoint_reader(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setitem(
+        litellm.model_cost,
+        "azure/unit-breakpoint-reader",
+        {"litellm_provider": "azure", "mode": "chat", "supports_prompt_cache_breakpoint": True},
+    )
+    monkeypatch.setitem(
+        litellm.model_cost,
+        "azure/unit-breakpoint-ignorer",
+        {"litellm_provider": "azure", "mode": "chat", "supports_prompt_cache_breakpoint": False},
+    )
+
+
+def _azure_request_parts(
+    config: AzureOpenAIConfig, model: str, litellm_params: dict[str, object]
+) -> object:
+    request: Final = config.transform_request(
+        model=model,
+        messages=[{"role": "user", "content": list(_MARKED_AZURE_PARTS)}],
+        optional_params={},
+        litellm_params=litellm_params,
+        headers={},
+    )
+    return request["messages"][0]["content"]
+
+
+@pytest.mark.parametrize("config", [AzureOpenAIConfig(), AzureOpenAIGPT5Config()])
+@pytest.mark.parametrize(
+    "model, litellm_params",
+    [
+        ("unit-breakpoint-reader", {"drop_params": True}),
+        ("my-azure-alias", {"drop_params": True, "base_model": "azure/unit-breakpoint-reader"}),
+        ("my-azure-alias", {"drop_params": True, "model_info": {"base_model": "azure/unit-breakpoint-reader"}}),
+        ("my-azure-alias", {"drop_params": True}),
+    ],
+    ids=("deployment-name", "litellm-params-base-model", "model-info-base-model", "unknown-deployment"),
+)
+def test_azure_drop_params_cuts_part_breakpoints_down_to_what_azure_chat_accepts(
+    registered_azure_breakpoint_reader: None,
+    config: AzureOpenAIConfig,
+    model: str,
+    litellm_params: dict[str, object],
+) -> None:
+    """Azure OpenAI chat answered 400 for a content part's breakpoint ttl and for implicit mode on 2026-10-29,
+    the same as OpenAI chat completions."""
+    assert _azure_request_parts(config, model, litellm_params) == _TRIMMED_AZURE_PARTS
+
+
+@pytest.mark.parametrize("litellm_params", [{}, {"drop_params": False}], ids=("no-drop-params", "drop-params-false"))
+def test_azure_sends_part_breakpoints_as_written_without_drop_params(
+    registered_azure_breakpoint_reader: None, litellm_params: dict[str, object]
+) -> None:
+    assert _azure_request_parts(AzureOpenAIConfig(), "unit-breakpoint-reader", litellm_params) == list(
+        _MARKED_AZURE_PARTS
+    )
+
+
+@pytest.mark.parametrize(
+    "model, litellm_params",
+    [
+        ("unit-breakpoint-ignorer", {"drop_params": True}),
+        ("unit-breakpoint-reader", {"drop_params": True, "base_model": "azure/unit-breakpoint-ignorer"}),
+        ("my-azure-alias", {"drop_params": True, "model_info": {"base_model": "azure/unit-breakpoint-ignorer"}}),
+    ],
+    ids=("known-non-reader", "base-model-known-non-reader", "model-info-base-model-known-non-reader"),
+)
+def test_azure_drop_params_strips_part_breakpoints_from_a_known_non_reader(
+    registered_azure_breakpoint_reader: None, model: str, litellm_params: dict[str, object]
+) -> None:
+    assert _azure_request_parts(AzureOpenAIConfig(), model, litellm_params) == [
+        {"type": "text", "text": "Stable prefix"},
+        {"type": "text", "text": "Implicit"},
+    ]

tests/unit/llms/azure_ai/chat/test_azure_ai_transformation.py

@@ -1271,6 +1271,12 @@ def test_azure_ai_falls_back_to_the_model_name_when_the_base_model_row_does_not_
             {"type": "text", "text": "stable", "prompt_cache_breakpoint": {"mode": "explicit"}},
         ),
         ({"type": "ephemeral"}, {"drop_params": True}, {"type": "text", "text": "stable"}),
+        ({"mode": "implicit"}, {"drop_params": True}, {"type": "text", "text": "stable"}),
+        (
+            {"mode": "explicit", "ttl": "1h"},
+            {"drop_params": True},
+            {"type": "text", "text": "stable", "prompt_cache_breakpoint": {"mode": "explicit"}},
+        ),
         (
             {"mode": "explicit", "ttl": "30m"},
             {},

tests/unit/llms/bedrock_mantle/test_bedrock_mantle_responses_transformation.py

@@ -1922,3 +1922,44 @@ class TestBedrockMantleResponsesPricing:
     def test_models_registered(self, local_cost_map):
         assert "bedrock_mantle/openai.gpt-5.5" in litellm.bedrock_mantle_models
         assert "bedrock_mantle/openai.gpt-5.4" in litellm.bedrock_mantle_models
+
+
+def _mantle_marked_part(litellm_params: GenericLiteLLMParams, marker: object) -> object:
+    body: Final = BedrockMantleResponsesAPIConfig().transform_responses_api_request(
+        model="openai.gpt-6.1-sol",
+        input=[
+            {
+                "type": "message",
+                "role": "user",
+                "content": [{"type": "input_text", "text": "Stable prefix", "prompt_cache_breakpoint": marker}],
+            }
+        ],
+        response_api_optional_request_params={},
+        litellm_params=litellm_params,
+        headers={},
+    )
+    return body["input"][0]["content"][0]
+
+
+@pytest.mark.parametrize(
+    "marker, wire_part",
+    [
+        (
+            {"mode": "explicit", "ttl": "30m"},
+            {"type": "input_text", "text": "Stable prefix", "prompt_cache_breakpoint": {"mode": "explicit"}},
+        ),
+        ({"mode": "implicit"}, {"type": "input_text", "text": "Stable prefix"}),
+    ],
+    ids=("ttl-dropped", "implicit-dropped"),
+)
+def test_mantle_drop_params_cuts_a_responses_breakpoint_down_to_its_mode(
+    monkeypatch: pytest.MonkeyPatch, marker: dict[str, str], wire_part: dict[str, object]
+) -> None:
+    """Bedrock Mantle Responses answered 400 for a 30m breakpoint ttl and 200 for a bare explicit mode on 2026-10-29."""
+    monkeypatch.setattr(litellm, "drop_params", False)
+    assert _mantle_marked_part(GenericLiteLLMParams(drop_params=True), marker) == wire_part
+    assert _mantle_marked_part(GenericLiteLLMParams(), marker) == {
+        "type": "input_text",
+        "text": "Stable prefix",
+        "prompt_cache_breakpoint": marker,
+    }

tests/unit/llms/openai/chat/test_openai_gpt_transformation.py

@@ -1659,6 +1659,125 @@ class TestOpenAIPromptCacheBreakpointChatPath:
         ]
 
 
+_CLIENT_BREAKPOINTS_AND_THEIR_CHAT_WIRE_FORM: Final = (
+    ({"mode": "explicit", "ttl": "1h"}, {"mode": "explicit"}),
+    ({"mode": "explicit", "ttl": "30m"}, {"mode": "explicit"}),
+    ({"mode": "explicit"}, {"mode": "explicit"}),
+    ({"mode": "implicit"}, None),
+    ({}, None),
+    ("explicit", None),
+)
+_CLIENT_BREAKPOINT_IDS: Final = ("ttl-1h", "ttl-30m", "explicit", "implicit", "empty", "string")
+
+
+class TestChatCompletionsClientBreakpointUnderDropParams:
+    """
+    Live OpenAI Chat Completions (gpt-6.1-sol, gpt-6.1-terra, gpt-6-sol, gpt-5.6, 2026-10-28) answers 400 on a
+    content part's prompt_cache_breakpoint carrying any ttl or a mode other than "explicit".
+    """
+
+    @staticmethod
+    def _request(marker: object, drop_params: bool) -> dict[str, object]:
+        return OpenAIGPTConfig().transform_request(
+            model="gpt-5.6",
+            messages=[
+                {"role": "system", "content": "Answer briefly"},
+                {
+                    "role": "user",
+                    "content": [
+                        {"type": "text", "text": "Describe this", "prompt_cache_breakpoint": marker},
+                        {"type": "image_url", "image_url": {"url": "https://example.com/a.png"}},
+                        {"type": "text", "text": "and nothing else"},
+                    ],
+                },
+            ],
+            optional_params={},
+            litellm_params={"custom_llm_provider": "openai", "drop_params": drop_params},
+            headers={},
+        )
+
+    @pytest.mark.parametrize(
+        ("marker", "wire_marker"), _CLIENT_BREAKPOINTS_AND_THEIR_CHAT_WIRE_FORM, ids=_CLIENT_BREAKPOINT_IDS
+    )
+    def test_drop_params_cuts_the_breakpoint_down_to_what_chat_completions_accepts(
+        self, marker: object, wire_marker: dict[str, str] | None
+    ):
+        text_part: Final = {"type": "text", "text": "Describe this"}
+
+        assert self._request(marker, drop_params=True)["messages"] == [
+            {"role": "system", "content": "Answer briefly"},
+            {
+                "role": "user",
+                "content": [
+                    text_part if wire_marker is None else {**text_part, "prompt_cache_breakpoint": wire_marker},
+                    {"type": "image_url", "image_url": {"url": "https://example.com/a.png"}},
+                    {"type": "text", "text": "and nothing else"},
+                ],
+            },
+        ]
+
+    @pytest.mark.parametrize(
+        "marker", [marker for marker, _ in _CLIENT_BREAKPOINTS_AND_THEIR_CHAT_WIRE_FORM], ids=_CLIENT_BREAKPOINT_IDS
+    )
+    def test_without_drop_params_the_client_breakpoint_reaches_openai_unchanged(self, marker: object):
+        assert self._request(marker, drop_params=False)["messages"][1]["content"][0] == {
+            "type": "text",
+            "text": "Describe this",
+            "prompt_cache_breakpoint": marker,
+        }
+
+    def test_cache_control_still_marks_a_part_whose_client_breakpoint_was_dropped(self):
+        request: Final = OpenAIGPTConfig().transform_request(
+            model="gpt-5.6",
+            messages=[
+                {
+                    "role": "user",
+                    "content": [
+                        {
+                            "type": "text",
+                            "text": "Describe this",
+                            "cache_control": {"type": "ephemeral"},
+                            "prompt_cache_breakpoint": {"mode": "implicit"},
+                        }
+                    ],
+                }
+            ],
+            optional_params={},
+            litellm_params={"custom_llm_provider": "openai", "drop_params": True},
+            headers={},
+        )
+
+        assert request["messages"] == [
+            {
+                "role": "user",
+                "content": [{"type": "text", "text": "Describe this", "prompt_cache_breakpoint": {"mode": "explicit"}}],
+            }
+        ]
+
+
+    @pytest.mark.parametrize("stray_part", ["Answer briefly", None], ids=["string", "null"])
+    def test_drop_params_leaves_a_part_that_is_not_an_object_for_openai_to_reject(self, stray_part: object):
+        assert OpenAIGPTConfig().transform_request(
+            model="gpt-5.6",
+            messages=[
+                {
+                    "role": "system",
+                    "content": [
+                        stray_part,
+                        {"type": "text", "text": "and nothing else", "prompt_cache_breakpoint": {"mode": "implicit"}},
+                    ],
+                },
+                {"role": "user", "content": "Describe this"},
+            ],
+            optional_params={},
+            litellm_params={"custom_llm_provider": "openai", "drop_params": True},
+            headers={},
+        )["messages"] == [
+            {"role": "system", "content": [stray_part, {"type": "text", "text": "and nothing else"}]},
+            {"role": "user", "content": "Describe this"},
+        ]
+
+
 _ARTIFACT_FIELD_PATTERN: Final = r'^(?!__.*__$)[^\p{Cc}\p{Cf}\p{Zl}\p{Zp}"\\./[\]]{1,200}$'
 
 

tests/unit/llms/openai/responses/test_openai_responses_transformation.py

@@ -4348,8 +4348,9 @@ async def test_extra_body_merges_with_request_data(extra_body_mock_response_data
         assert request_body["custom_field"] == "custom_value"
 
 
-_MALFORMED_MARKERS: Final = ("yes", 1, {"mode": "bogus"}, {"mode": "explicit", "ttl": "1h"})
-_MALFORMED_MARKER_IDS: Final = ("string", "int", "bogus-mode", "bad-ttl")
+_MALFORMED_MARKERS: Final = ("yes", 1, {"mode": "bogus"}, {"mode": "bogus", "ttl": "30m"})
+_MALFORMED_MARKER_IDS: Final = ("string", "int", "bogus-mode", "bogus-mode-with-ttl")
+_UNSUPPORTED_TTL_MARKER: Final = {"mode": "explicit", "ttl": "1h"}
 _VALID_MARKERS: Final = ({"mode": "explicit"}, {"mode": "explicit", "ttl": "30m"})
 _VALID_MARKER_IDS: Final = ("explicit", "explicit-30m")
 _PARTS: Final[Mapping[str, Mapping[str, object]]] = {
@@ -4435,6 +4436,76 @@ def test_valid_prompt_cache_breakpoint_reaches_the_wire_under_both_settings(
     assert _transformed_input(_CONFIGS[0], input, GenericLiteLLMParams(drop_params=drop_params)) == input
 
 
+@pytest.mark.parametrize("config", _CONFIGS, ids=_CONFIG_IDS)
+@pytest.mark.parametrize("part_type", tuple(_PARTS))
+def test_an_unsupported_ttl_is_dropped_and_the_explicit_mode_kept_under_drop_params(
+    monkeypatch: pytest.MonkeyPatch, config: BaseResponsesAPIConfig, part_type: str
+) -> None:
+    monkeypatch.setattr(litellm, "drop_params", False)
+    input: Final = _marked_message(part_type, _UNSUPPORTED_TTL_MARKER)
+    trimmed: Final = _transformed_input(config, input, GenericLiteLLMParams(drop_params=True))
+    assert _single_part(trimmed) == {**_PARTS[part_type], "prompt_cache_breakpoint": {"mode": "explicit"}}
+    assert _transformed_input(config, input, GenericLiteLLMParams(drop_params=False)) == input
+
+
+@pytest.mark.parametrize(
+    "config, marker, wire_marker",
+    [
+        (OpenAIResponsesAPIConfig(), {"mode": "explicit", "ttl": "12h"}, {"mode": "explicit", "ttl": "12h"}),
+        (AzureOpenAIResponsesAPIConfig(), {"mode": "explicit", "ttl": "30m"}, {"mode": "explicit", "ttl": "30m"}),
+        (AzureOpenAIResponsesAPIConfig(), {"mode": "explicit", "ttl": "12h"}, {"mode": "explicit"}),
+        (AzureAIResponsesAPIConfig(), {"mode": "explicit", "ttl": "12h"}, {"mode": "explicit"}),
+    ],
+    ids=("openai-keeps-12h", "azure-keeps-30m", "azure-drops-12h", "azure-ai-drops-12h"),
+)
+def test_drop_params_keeps_only_a_ttl_the_target_responses_api_accepts(
+    monkeypatch: pytest.MonkeyPatch,
+    config: BaseResponsesAPIConfig,
+    marker: Mapping[str, str],
+    wire_marker: Mapping[str, str],
+) -> None:
+    """Azure OpenAI Responses answered 400 for a 12h breakpoint ttl and 200 for 30m on 2026-10-29, and Azure AI
+    Foundry Responses did the same on 2026-10-30."""
+    monkeypatch.setattr(litellm, "drop_params", False)
+    input: Final = _marked_message("input_text", marker)
+    trimmed: Final = _transformed_input(config, input, GenericLiteLLMParams(drop_params=True))
+    assert _single_part(trimmed) == {**_PARTS["input_text"], "prompt_cache_breakpoint": wire_marker}
+    assert _transformed_input(config, input, GenericLiteLLMParams(drop_params=False)) == input
+
+
+@pytest.mark.parametrize(
+    "config, litellm_params",
+    [
+        (XAIResponsesAPIConfig(), GenericLiteLLMParams(drop_params=True)),
+        (HostedVLLMResponsesAPIConfig(), GenericLiteLLMParams(drop_params=True, api_base="http://localhost:8000/v1")),
+        (OpenAIResponsesAPIConfig(), GenericLiteLLMParams(drop_params=True, api_base="http://localhost:8000/v1")),
+    ],
+    ids=("xai", "hosted-vllm", "openai-compatible-host"),
+)
+def test_a_responses_api_not_known_to_take_breakpoints_gets_a_marker_whole_or_not_at_all(
+    monkeypatch: pytest.MonkeyPatch, config: BaseResponsesAPIConfig, litellm_params: GenericLiteLLMParams
+) -> None:
+    monkeypatch.setattr(litellm, "drop_params", False)
+    unsupported_ttl: Final = _marked_message("input_text", _UNSUPPORTED_TTL_MARKER)
+    assert _single_part(_transformed_input(config, unsupported_ttl, litellm_params)) == _PARTS["input_text"]
+    openai_ttl: Final = _marked_message("input_text", {"mode": "explicit", "ttl": "12h"})
+    assert _transformed_input(config, openai_ttl, litellm_params) == openai_ttl
+
+
+def test_a_custom_host_opted_in_with_prompt_cache_options_keeps_the_mode_of_a_marker_with_an_unsupported_ttl(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(litellm, "drop_params", False)
+    request: Final = OpenAIResponsesAPIConfig().transform_responses_api_request(
+        model="gpt-6.1-sol",
+        input=cast("ResponseInputParam", _marked_message("input_text", _UNSUPPORTED_TTL_MARKER)),
+        response_api_optional_request_params={"prompt_cache_options": {"mode": "explicit"}},
+        litellm_params=GenericLiteLLMParams(drop_params=True, api_base="https://gateway.example/v1"),
+        headers={},
+    )
+    assert _single_part(request["input"]) == {**_PARTS["input_text"], "prompt_cache_breakpoint": {"mode": "explicit"}}
+
+
 def test_prompt_cache_breakpoint_unknown_key_is_trimmed_under_drop_params(monkeypatch: pytest.MonkeyPatch) -> None:
     monkeypatch.setattr(litellm, "drop_params", False)
     input: Final = _marked_message("input_text", {"mode": "explicit", "note": "kept"})
@@ -4524,12 +4595,41 @@ def test_a_valid_marker_survives_drop_params_on_a_deployment_name_with_unknown_b
     monkeypatch.setattr(litellm, "drop_params", False)
     litellm_params: Final = GenericLiteLLMParams(drop_params=True)
     valid: Final = _marked_message("input_text", {"mode": "explicit"})
-    malformed: Final = _marked_message("input_text", {"mode": "explicit", "ttl": "1h"})
+    malformed: Final = _marked_message("input_text", {"mode": "bogus"})
     assert _transformed_input_for("litellm-qa-sol-alias", valid, litellm_params, config) == valid
     dropped: Final = _transformed_input_for("litellm-qa-sol-alias", malformed, litellm_params, config)
     assert _single_part(dropped) == _PARTS["input_text"]
 
 
+@pytest.mark.parametrize("ttl", ("30m", "12h"))
+@pytest.mark.parametrize(
+    "config", (AzureOpenAIResponsesAPIConfig(), AzureAIResponsesAPIConfig()), ids=("azure", "azure_ai")
+)
+def test_an_azure_deployment_with_unknown_breakpoint_support_gets_a_marker_whole_or_not_at_all(
+    monkeypatch: pytest.MonkeyPatch, config: BaseResponsesAPIConfig, ttl: str
+) -> None:
+    monkeypatch.setattr(litellm, "drop_params", False)
+    litellm_params: Final = GenericLiteLLMParams(drop_params=True)
+    supported_ttl: Final = _marked_message("input_text", {"mode": "explicit", "ttl": ttl})
+    unsupported_ttl: Final = _marked_message("input_text", _UNSUPPORTED_TTL_MARKER)
+    assert _transformed_input_for("litellm-qa-sol-alias", supported_ttl, litellm_params, config) == supported_ttl
+    dropped: Final = _transformed_input_for("litellm-qa-sol-alias", unsupported_ttl, litellm_params, config)
+    assert _single_part(dropped) == _PARTS["input_text"]
+
+
+@pytest.mark.parametrize(
+    "config", (AzureOpenAIResponsesAPIConfig(), AzureAIResponsesAPIConfig()), ids=("azure", "azure_ai")
+)
+def test_an_azure_deployment_declaring_a_breakpoint_reader_keeps_the_mode_of_a_marker_with_an_unsupported_ttl(
+    monkeypatch: pytest.MonkeyPatch, config: BaseResponsesAPIConfig
+) -> None:
+    monkeypatch.setattr(litellm, "drop_params", False)
+    litellm_params: Final = GenericLiteLLMParams(drop_params=True, base_model="gpt-6.1-sol")
+    unsupported_ttl: Final = _marked_message("input_text", _UNSUPPORTED_TTL_MARKER)
+    trimmed: Final = _transformed_input_for("litellm-qa-sol-alias", unsupported_ttl, litellm_params, config)
+    assert _single_part(trimmed) == {**_PARTS["input_text"], "prompt_cache_breakpoint": {"mode": "explicit"}}
+
+
 def _declaring_base_model(where: str, base_model: str) -> GenericLiteLLMParams:
     if where == "litellm_params":
         return GenericLiteLLMParams(drop_params=True, base_model=base_model)
@@ -4570,10 +4670,10 @@ def test_a_model_that_takes_markers_sends_kept_parts_byte_for_byte_and_drops_onl
         {"prompt_cache_breakpoint": {"mode": "explicit"}, "type": "input_text", "text": "b"},
     ]
     rejected: Final = {"type": "input_text", "text": "c"}
-    bad_ttl: Final = {"prompt_cache_breakpoint": {"mode": "explicit", "ttl": "1h"}}
+    bogus_mode: Final = {"prompt_cache_breakpoint": {"mode": "bogus"}}
     input: Final = [
         {"type": "message", "role": "user", "content": kept},
-        {"type": "message", "role": "user", "content": [{**bad_ttl, **rejected}]},
+        {"type": "message", "role": "user", "content": [{**bogus_mode, **rejected}]},
     ]
     transformed: Final = _transformed_input_for("gpt-6.1-sol", input, GenericLiteLLMParams(drop_params=True))
     assert json.dumps(transformed) == json.dumps(