fix(anthropic,bedrock): answer 400 on unsupported breakpoint ttl and malformed markers, count only sent marks, trim in linear time

A prompt_cache_breakpoint ttl Claude has no equivalent for (30m) now answers a 400 that names the fixes, and sends a 5m mark under drop_params (global, request, or deployment). A malformed marker answers a 400 without drop_params and sends no mark with it.

The four-breakpoint census counts only marks the provider body carries, so empty system text, part types the wire drops, unsent tools, and blank assistant strings on Converse no longer take a slot, while the Converse tool_config cachePoint does. A function-level tool marker on Converse now becomes a cachePoint.

Trimming the excess runs in linear time, and Converse also gives Claude breakpoints to a deployment whose base_model names Claude.

litellm/litellm_core_utils/prompt_templates/common_utils.py

@@ -14,8 +14,11 @@ from pathlib import Path
 from types import MappingProxyType
 from typing import TYPE_CHECKING, Any, Final, Literal, TypeVar, cast
 
+from pydantic import TypeAdapter, ValidationError
+
 import litellm
 from litellm import verbose_logger
+from litellm.exceptions import UnsupportedParamsError
 from litellm.router_utils.batch_utils import InMemoryFile
 from litellm.types.llms.openai import (
     AllMessageValues,
@@ -1787,6 +1790,8 @@ def check_is_function_call(logging_obj: "LoggingClass") -> bool:
 
 
 _MarkedT: Final = TypeVar("_MarkedT", bound=Mapping[str, object])
+_PROMPT_CACHE_BREAKPOINT: Final = TypeAdapter(PromptCacheBreakpoint)
+_DROP_PARAMS_HINT: Final = "set drop_params=True (proxy: litellm_settings: drop_params: true)"
 
 
 def with_prompt_cache_breakpoint(target: _MarkedT, marker: object) -> _MarkedT:
@@ -1807,18 +1812,52 @@ def cache_control_as_prompt_cache_breakpoint(block: _MarkedT) -> _MarkedT:
     )
 
 
-def prompt_cache_breakpoint_as_cache_control(block: _MarkedT) -> _MarkedT:
+def _claude_cache_control_for_prompt_cache_breakpoint(
+    marker: object, drop_params: bool
+) -> ChatCompletionCachedContent | None:
+    """The 5m ``cache_control`` Claude takes for an OpenAI ``prompt_cache_breakpoint``. A malformed marker answers
+    a 400, or gives no mark under ``drop_params``. A ``ttl`` Claude has no equivalent for answers a 400 naming the
+    fix, or gives a plain 5m mark under ``drop_params``."""
+    try:
+        valid_marker: Final = _PROMPT_CACHE_BREAKPOINT.validate_python(marker)
+    except ValidationError:
+        if not drop_params:
+            raise UnsupportedParamsError(
+                message=(
+                    f"prompt_cache_breakpoint must be {{'mode': 'explicit'}}, optionally with ttl '30m', got "
+                    f"{marker!r}. Fix the marker, or {_DROP_PARAMS_HINT} to send the block without it."
+                ),
+                status_code=400,
+            ) from None
+        verbose_logger.debug("Dropping malformed prompt_cache_breakpoint=%r under drop_params", marker)
+        return None
+    ttl: Final = valid_marker.get("ttl")
+    if ttl is not None and not drop_params:
+        raise UnsupportedParamsError(
+            message=(
+                f"Claude caches a prompt for 5m or 1h, so prompt_cache_breakpoint ttl={ttl!r} has no equivalent. "
+                f"Remove the ttl, {_DROP_PARAMS_HINT} to send a 5m breakpoint instead, or send "
+                "cache_control {'type': 'ephemeral', 'ttl': '1h'} on the block for a 1h cache."
+            ),
+            status_code=400,
+        )
+    return ChatCompletionCachedContent(type="ephemeral")
+
+
+def prompt_cache_breakpoint_as_cache_control(block: _MarkedT, drop_params: bool) -> _MarkedT:
     """OpenAI's ``prompt_cache_breakpoint`` rewritten as Anthropic's ``cache_control`` on the same block.
 
-    A ``cache_control`` the client already wrote wins. OpenAI's 30m TTL has no Anthropic equivalent and is dropped."""
+    A ``cache_control`` the client already wrote wins, and the marker is then dropped unread."""
-    if block.get("prompt_cache_breakpoint") is None:
+    marker: Final = block.get("prompt_cache_breakpoint")
+    if marker is None:
         return block
     unmarked: Final = {key: value for key, value in block.items() if key != "prompt_cache_breakpoint"}
-    marked: Final = (
+    control: Final = (
-        unmarked
+        None
         if unmarked.get("cache_control") is not None
-        else {**unmarked, "cache_control": ChatCompletionCachedContent(type="ephemeral")}
+        else _claude_cache_control_for_prompt_cache_breakpoint(marker, drop_params)
     )
+    marked: Final = unmarked if control is None else {**unmarked, "cache_control": control}
     return cast(_MarkedT, marked)  # cast-ok: same block shape as the input with the marker key renamed
 
 
@@ -1839,14 +1878,16 @@ def _without_key(block: Mapping[str, object], key: str) -> Mapping[str, object]:
     return {name: value for name, value in block.items() if name != key}
 
 
-def _block_with_prompt_cache_breakpoint_as_cache_control(block: object) -> object:
+def _block_with_prompt_cache_breakpoint_as_cache_control(block: object, drop_params: bool) -> object:
     fields: Final = _as_mapping(block)
-    return prompt_cache_breakpoint_as_cache_control(fields) if fields is not None else block
+    return prompt_cache_breakpoint_as_cache_control(fields, drop_params) if fields is not None else block
 
 
-def message_with_prompt_cache_breakpoints_as_cache_control(message: AllMessageValues) -> AllMessageValues:
+def message_with_prompt_cache_breakpoints_as_cache_control(
+    message: AllMessageValues, drop_params: bool
+) -> AllMessageValues:
     """The message-level mark, every content part's mark and every tool call's mark renamed to ``cache_control``."""
-    renamed: Final = prompt_cache_breakpoint_as_cache_control(message)
+    renamed: Final = prompt_cache_breakpoint_as_cache_control(message, drop_params)
     fields: Final = message_fields(renamed)
     content: Final = _as_list(fields.get("content"))
     tool_calls: Final = _as_list(fields.get("tool_calls"))
@@ -1857,12 +1898,20 @@ def message_with_prompt_cache_breakpoints_as_cache_control(message: AllMessageVa
         {
             **fields,
             **(
-                {"content": [_block_with_prompt_cache_breakpoint_as_cache_control(block) for block in content]}
+                {
+                    "content": [
+                        _block_with_prompt_cache_breakpoint_as_cache_control(block, drop_params) for block in content
+                    ]
+                }
                 if content is not None
                 else {}
             ),
             **(
-                {"tool_calls": [_block_with_prompt_cache_breakpoint_as_cache_control(call) for call in tool_calls]}
+                {
+                    "tool_calls": [
+                        _block_with_prompt_cache_breakpoint_as_cache_control(call, drop_params) for call in tool_calls
+                    ]
+                }
                 if tool_calls is not None
                 else {}
             ),
@@ -1871,17 +1920,24 @@ def message_with_prompt_cache_breakpoints_as_cache_control(message: AllMessageVa
 
 
 def messages_with_prompt_cache_breakpoints_as_cache_control(
-    messages: Sequence[AllMessageValues],
+    messages: Sequence[AllMessageValues], drop_params: bool
 ) -> tuple[AllMessageValues, ...]:
-    return tuple(message_with_prompt_cache_breakpoints_as_cache_control(message) for message in messages)
+    return tuple(message_with_prompt_cache_breakpoints_as_cache_control(message, drop_params) for message in messages)
 
 
-def tool_with_prompt_cache_breakpoint_as_cache_control(tool: Mapping[str, object]) -> Mapping[str, object]:
+def tool_with_prompt_cache_breakpoint_as_cache_control(
+    tool: Mapping[str, object], drop_params: bool
+) -> Mapping[str, object]:
     """A tool definition's mark, written at the top level (Anthropic shape) or under ``function`` (OpenAI shape),
     renamed to a top-level ``cache_control``, which every Claude transform reads."""
     function: Final = _as_mapping(tool.get("function"))
-    function_marked: Final = function is not None and function.get("prompt_cache_breakpoint") is not None
+    top_level_marker: Final = tool.get("prompt_cache_breakpoint")
-    if tool.get("prompt_cache_breakpoint") is None and not function_marked:
+    marker: Final = (
+        top_level_marker
+        if top_level_marker is not None
+        else (function.get("prompt_cache_breakpoint") if function is not None else None)
+    )
+    if marker is None:
         return tool
     unmarked_function: Final = _without_key(function, "prompt_cache_breakpoint") if function is not None else None
     unmarked: Final = {
@@ -1891,7 +1947,8 @@ def tool_with_prompt_cache_breakpoint_as_cache_control(tool: Mapping[str, object
     already_marked: Final = unmarked.get("cache_control") is not None or (
         unmarked_function is not None and unmarked_function.get("cache_control") is not None
     )
-    return unmarked if already_marked else {**unmarked, "cache_control": ChatCompletionCachedContent(type="ephemeral")}
+    control: Final = None if already_marked else _claude_cache_control_for_prompt_cache_breakpoint(marker, drop_params)
+    return unmarked if control is None else {**unmarked, "cache_control": control}
 
 
 def _block_without_cache_control(block: object) -> object:
@@ -1924,14 +1981,16 @@ def message_without_cache_control_marks(
     )
 
 
-def _tool_block_with_prompt_cache_breakpoint_as_cache_control(tool: object) -> object:
+def _tool_block_with_prompt_cache_breakpoint_as_cache_control(tool: object, drop_params: bool) -> object:
     fields: Final = _as_mapping(tool)
-    return tool_with_prompt_cache_breakpoint_as_cache_control(fields) if fields is not None else tool
+    return tool_with_prompt_cache_breakpoint_as_cache_control(fields, drop_params) if fields is not None else tool
 
 
-def tools_with_prompt_cache_breakpoints_as_cache_control(tools: object) -> tuple[object, ...]:
+def tools_with_prompt_cache_breakpoints_as_cache_control(tools: object, drop_params: bool) -> tuple[object, ...]:
     """Every tool definition in ``tools`` (anything but a list or tuple is treated as none) with its mark renamed."""
-    return tuple(_tool_block_with_prompt_cache_breakpoint_as_cache_control(tool) for tool in _as_list(tools) or ())
+    return tuple(
+        _tool_block_with_prompt_cache_breakpoint_as_cache_control(tool, drop_params) for tool in _as_list(tools) or ()
+    )
 
 
 LITELLM_INTERNAL_MESSAGE_FIELDS: Final = frozenset({"thinking_blocks", "reasoning_content", "provider_specific_fields"})

litellm/llms/anthropic/chat/prompt_cache_breakpoints.py

@@ -6,8 +6,12 @@ has no Claude equivalent and is rejected or dropped like any other unsupported p
 
 from collections.abc import Iterator, Mapping, Sequence
 from dataclasses import dataclass
-from itertools import chain
+from itertools import chain, groupby
-from typing import Final, Literal, TypeAlias, assert_never
+from operator import itemgetter
+from types import MappingProxyType
+from typing import Final, Literal, TypeAlias
+
+from typing_extensions import assert_never
 
 from litellm._logging import verbose_logger
 from litellm.constants import MAX_CACHE_CONTROL_BLOCKS
@@ -19,11 +23,21 @@ from litellm.litellm_core_utils.prompt_templates.common_utils import (
     tools_with_prompt_cache_breakpoints_as_cache_control,
 )
 from litellm.llms.anthropic.common_utils import tool_call_is_rebuilt_as_server_tool_use
+from litellm.types.llms.anthropic import ANTHROPIC_TOOL_SEARCH_TOOL_TYPES
 from litellm.types.llms.openai import AllMessageValues
 
 PROMPT_CACHE_BREAKPOINT_ON_TOOLS: Final = "_litellm_prompt_cache_breakpoint_on_tools"
 _Site: TypeAlias = tuple[Literal["tool", "message", "part", "tool_call", "tool_message"], int, int]
 ClaudeWire: TypeAlias = Literal["anthropic", "bedrock_converse"]
+_ANTHROPIC_PART_TYPES_BY_ROLE: Final = MappingProxyType(
+    {
+        "user": frozenset({"text", "image_url", "document", "file"}),
+        "tool": frozenset({"text", "image_url", "file"}),
+        "function": frozenset({"text", "image_url", "file"}),
+        "assistant": frozenset({"text", "thinking", "server_tool_use"}),
+    }
+)
+_CONVERSE_TOOL_BLOCK_KEYS: Final = frozenset({"systemTool", "toolSpec", "cachePoint"})
 
 
 @dataclass(frozen=True, slots=True)
@@ -59,13 +73,37 @@ def tools_carry_prompt_cache_breakpoint(tools: object) -> bool:
     return any(_tool_carries(tool, "prompt_cache_breakpoint") for tool in _as_blocks(tools) or ())
 
 
+def _text(block: Mapping[str, object]) -> str | None:
+    text: Final = block.get("text")
+    return text if isinstance(text, str) else None
+
+
 def _renders_message_mark(message: Mapping[str, object], wire: ClaudeWire) -> bool:
-    """Claude's message builder keeps a message-level mark on string content and on tool results only. Anthropic's
+    """Claude's message builders keep a message-level mark on string content and on tool results only, and skip it
-    builder skips an empty assistant string, mark included, where Converse still emits its ``cachePoint``."""
+    where the string itself is skipped: an empty system prompt on both wires, and on Converse an empty user string
+    or a blank assistant string."""
     content: Final = message.get("content")
-    if isinstance(content, str):
+    role: Final = message.get("role")
-        return bool(content) or message.get("role") != "assistant" or wire == "bedrock_converse"
+    if not isinstance(content, str):
-    return message.get("role") in ("tool", "function")
+        return role in ("tool", "function")
+    if wire == "bedrock_converse" and role == "assistant":
+        return bool(content.strip())
+    return bool(content) or (wire == "anthropic" and role != "system")
+
+
+def _renders_part_mark(role: object, part: Mapping[str, object], wire: ClaudeWire) -> bool:
+    """Whether the builder for ``wire`` keeps ``part``, and so its mark, in the body it sends."""
+    part_type: Final = part.get("type")
+    if role == "system" and wire == "bedrock_converse":
+        return part_type == "text" and bool(_text(part))
+    if role == "system":
+        return part_type != "text" or bool(_text(part))
+    if wire == "bedrock_converse":
+        return part_type != "text" or bool((_text(part) or "").strip())
+    if role == "assistant" and isinstance(part_type, str) and part_type.endswith("_tool_result"):
+        return True
+    kept_types: Final = _ANTHROPIC_PART_TYPES_BY_ROLE.get(role) if isinstance(role, str) else None
+    return kept_types is None or part_type in kept_types
 
 
 def _renders_tool_call_mark(message: Mapping[str, object], call: object, wire: ClaudeWire) -> bool:
@@ -75,15 +113,24 @@ def _renders_tool_call_mark(message: Mapping[str, object], call: object, wire: C
     )
 
 
+def _part_sites(index: int, message: Mapping[str, object], key: str, wire: ClaudeWire) -> Iterator[_Site]:
+    role: Final = message.get("role")
+    for position, part in enumerate(_as_blocks(message.get("content")) or ()):
+        if (
+            (fields := _as_mapping(part)) is not None
+            and _carries(fields, key)
+            and _renders_part_mark(role, fields, wire)
+        ):
+            yield ("part", index, position)
+
+
 def _message_sites(index: int, message: object, key: str, wire: ClaudeWire) -> Iterator[_Site]:
     fields: Final = _as_mapping(message)
     if fields is None:
         return
     if fields.get(key) is not None and _renders_message_mark(fields, wire):
         yield ("message", index, -1)
-    for position, part in enumerate(_as_blocks(fields.get("content")) or ()):
+    yield from _part_sites(index, fields, key, wire)
-        if _carries(part, key):
-            yield ("part", index, position)
     for position, call in enumerate(_as_blocks(fields.get("tool_calls")) or ()):
         if _carries(call, key) and _renders_tool_call_mark(fields, call, wire):
             yield ("tool_call", index, position)
@@ -99,22 +146,55 @@ def _converse_message_sites(index: int, message: object, key: str) -> Iterator[_
         yield ("tool_message", index, -1)
 
 
-def _sites(messages: Sequence[object], tools: Sequence[object], key: str, wire: ClaudeWire) -> tuple[_Site, ...]:
+def _messages_sites(messages: Sequence[object], key: str, wire: ClaudeWire) -> tuple[_Site, ...]:
-    """Every mark of ``key`` the ``wire`` renders, in wire order: tool definitions first, then the messages as
-    listed. Converse reads only a tool definition's top-level mark."""
     if wire == "bedrock_converse":
-        return (
+        return tuple(
-            *(("tool", index, -1) for index, tool in enumerate(tools) if _carries(tool, key)),
+            chain.from_iterable(_converse_message_sites(index, message, key) for index, message in enumerate(messages))
-            *chain.from_iterable(
-                _converse_message_sites(index, message, key) for index, message in enumerate(messages)
-            ),
         )
-    return (
+    return tuple(
-        *(("tool", index, -1) for index, tool in enumerate(tools) if _tool_carries(tool, key)),
+        chain.from_iterable(_message_sites(index, message, key, wire) for index, message in enumerate(messages))
-        *chain.from_iterable(_message_sites(index, message, key, wire) for index, message in enumerate(messages)),
     )
 
 
+def _converse_sends_tool(tool: object) -> bool:
+    """Whether Converse keeps ``tool`` in ``toolConfig``. Tool search and built-in tools with no function schema
+    are skipped there."""
+    fields: Final = _as_mapping(tool)
+    if fields is None:
+        return False
+    if "systemTool" in fields:
+        return True
+    return fields.get("type") not in ANTHROPIC_TOOL_SEARCH_TOOL_TYPES and any(
+        key in fields for key in ("function", "input_schema", *_CONVERSE_TOOL_BLOCK_KEYS)
+    )
+
+
+def _converse_sends_tool_mark(tool: object) -> bool:
+    """Converse adds a ``cachePoint`` only after a tool it builds a ``toolSpec`` for, and reads only the top-level
+    mark."""
+    fields: Final = _as_mapping(tool)
+    return (
+        fields is not None
+        and fields.get("cache_control") is not None
+        and _converse_sends_tool(fields)
+        and not any(key in fields for key in _CONVERSE_TOOL_BLOCK_KEYS)
+    )
+
+
+def converse_appends_tool_config_cache_point(tools: object, injection_points: object) -> bool:
+    """Whether Converse appends a ``cachePoint`` to ``toolConfig``: a ``tool_config`` injection point does once the
+    request keeps at least one tool."""
+    return any(_converse_sends_tool(tool) for tool in _as_blocks(tools) or ()) and any(
+        (_as_mapping(point) or {}).get("location") == "tool_config" for point in _as_blocks(injection_points) or ()
+    )
+
+
+def _tool_sites(tools: Sequence[object], wire: ClaudeWire) -> tuple[_Site, ...]:
+    if wire == "bedrock_converse":
+        return tuple(("tool", index, -1) for index, tool in enumerate(tools) if _converse_sends_tool_mark(tool))
+    return tuple(("tool", index, -1) for index, tool in enumerate(tools) if _tool_carries(tool, "cache_control"))
+
+
 def _without_cache_control(block: Mapping[str, object]) -> Mapping[str, object]:
     return {key: value for key, value in block.items() if key != "cache_control"}
 
@@ -130,23 +210,17 @@ def _tool_without_cache_control(tool: object) -> object:
     }
 
 
-def _dropped_positions(dropped: frozenset[_Site], kind: str, index: int) -> frozenset[int]:
+def _message_without(message: AllMessageValues, dropped: Sequence[_Site]) -> AllMessageValues:
-    return frozenset(
+    if any(kind == "tool_message" for kind, _, _ in dropped):
-        position for site_kind, site_index, position in dropped if (site_kind, site_index) == (kind, index)
-    )
-
-
-def _message_without(index: int, message: AllMessageValues, dropped: frozenset[_Site]) -> AllMessageValues:
-    if ("tool_message", index, -1) in dropped:
         content: Final = _as_blocks(message_fields(message).get("content")) or ()
         return message_without_cache_control_marks(
             message, message_mark=True, parts=frozenset(range(len(content))), tool_calls=frozenset()
         )
     return message_without_cache_control_marks(
         message,
-        message_mark=("message", index, -1) in dropped,
+        message_mark=any(kind == "message" for kind, _, _ in dropped),
-        parts=_dropped_positions(dropped, "part", index),
+        parts=frozenset(position for kind, _, position in dropped if kind == "part"),
-        tool_calls=_dropped_positions(dropped, "tool_call", index),
+        tool_calls=frozenset(position for kind, _, position in dropped if kind == "tool_call"),
     )
 
 
@@ -157,38 +231,53 @@ def claude_cache_marks(
     automatic_breakpoint: bool,
     drop_params: bool,
     tools_carried_breakpoint: bool = False,
+    tool_config_breakpoint: bool = False,
     wire: ClaudeWire = "anthropic",
 ) -> ClaudeCacheMarks | CacheBreakpointCapExceeded:
     """The request's ``prompt_cache_breakpoint`` marks renamed to ``cache_control`` and, when it carried any, held to
     Claude's cap: ``drop_params`` keeps the latest four in wire order, otherwise the excess is reported. A request
     that carried none is forwarded as written so the provider's own answer reaches the client.
 
-    ``automatic_breakpoint`` is the request-level ``cache_control`` (Anthropic's automatic caching), which takes one
+    ``automatic_breakpoint`` is the request-level ``cache_control`` (Anthropic's automatic caching) and
-    slot of its own and is never dropped. ``tools_carried_breakpoint`` covers tools already mapped to Anthropic's
+    ``tool_config_breakpoint`` the ``cachePoint`` Converse appends to ``toolConfig``. Each takes one slot of its own
-    shape, where the mark was renamed before this census ran. ``wire`` is the body the marks are rendered into, so
+    and is never dropped. ``tools_carried_breakpoint`` covers tools already mapped to Anthropic's shape, where the
-    the count matches the breakpoints the provider receives."""
+    mark was renamed before this census ran. ``wire`` is the body the marks are rendered into, so the count covers
+    only the breakpoints the provider receives."""
     listed_tools: Final = _as_blocks(tools) or ()
-    if not tools_carried_breakpoint and not _sites(messages, listed_tools, "prompt_cache_breakpoint", wire):
+    carried_breakpoint: Final = (
+        tools_carried_breakpoint
+        or tools_carry_prompt_cache_breakpoint(listed_tools)
+        or bool(_messages_sites(messages, "prompt_cache_breakpoint", wire))
+    )
+    if not carried_breakpoint:
         return ClaudeCacheMarks(tuple(messages), tuple(listed_tools))
-    renamed_messages: Final = messages_with_prompt_cache_breakpoints_as_cache_control(messages)
+    renamed_messages: Final = messages_with_prompt_cache_breakpoints_as_cache_control(messages, drop_params)
-    renamed_tools: Final = tools_with_prompt_cache_breakpoints_as_cache_control(listed_tools)
+    renamed_tools: Final = tools_with_prompt_cache_breakpoints_as_cache_control(listed_tools, drop_params)
-    sites: Final = _sites(renamed_messages, renamed_tools, "cache_control", wire)
+    tool_sites: Final = _tool_sites(renamed_tools, wire)
-    found: Final = len(sites) + int(automatic_breakpoint)
+    message_sites: Final = _messages_sites(renamed_messages, "cache_control", wire)
+    found: Final = len(tool_sites) + len(message_sites) + int(automatic_breakpoint) + int(tool_config_breakpoint)
     if found <= MAX_CACHE_CONTROL_BLOCKS:
         return ClaudeCacheMarks(renamed_messages, renamed_tools)
     if not drop_params:
         return CacheBreakpointCapExceeded(found)
-    dropped: Final = frozenset(sites[: found - MAX_CACHE_CONTROL_BLOCKS])
+    excess: Final = found - MAX_CACHE_CONTROL_BLOCKS
+    dropped_tools: Final = frozenset(index for _, index, _ in tool_sites[:excess])
+    dropped_by_message: Final = MappingProxyType(
+        {index: tuple(sites) for index, sites in groupby(message_sites[: excess - len(dropped_tools)], itemgetter(1))}
+    )
     verbose_logger.debug(
         "Dropping the earliest %d of %d cache breakpoints: Claude accepts at most %d.",
-        len(dropped),
+        excess,
         found,
         MAX_CACHE_CONTROL_BLOCKS,
     )
     return ClaudeCacheMarks(
-        messages=tuple(_message_without(index, message, dropped) for index, message in enumerate(renamed_messages)),
+        messages=tuple(
+            _message_without(message, dropped_by_message[index]) if index in dropped_by_message else message
+            for index, message in enumerate(renamed_messages)
+        ),
         tools=tuple(
-            _tool_without_cache_control(tool) if ("tool", index, -1) in dropped else tool
+            _tool_without_cache_control(tool) if index in dropped_tools else tool
             for index, tool in enumerate(renamed_tools)
         ),
     )

litellm/llms/anthropic/chat/transformation.py

@@ -1578,7 +1578,11 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
             elif param == "tools":
                 requested_tools = cast(object, value)  # cast-ok: the loop yields untyped values
                 anthropic_tools, mcp_servers = self._map_tools(
-                    list(tools_with_prompt_cache_breakpoints_as_cache_control(requested_tools))
+                    list(
+                        tools_with_prompt_cache_breakpoints_as_cache_control(
+                            requested_tools, drop_params or litellm.drop_params is True
+                        )
+                    )
                 )
                 optional_params = self._add_tools_to_optional_params(
                     optional_params=optional_params, tools=anthropic_tools

litellm/llms/bedrock/chat/converse_transformation.py

@@ -46,6 +46,7 @@ from litellm.litellm_core_utils.prompt_templates.mid_conversation_system import
 )
 from litellm.llms.anthropic.chat.prompt_cache_breakpoints import (
     claude_cache_marks,
+    converse_appends_tool_config_cache_point,
     reject_prompt_cache_options,
     resolved_claude_cache_marks,
 )
@@ -57,7 +58,6 @@ from litellm.llms.anthropic.chat.transformation import (
 )
 from litellm.llms.anthropic.common_utils import AnthropicModelInfo
 from litellm.llms.base_llm.chat.transformation import BaseConfig, BaseLLMException
-from litellm.llms.bedrock.base_aws_llm import BaseAWSLLM
 from litellm.llms.bedrock.chat.tool_result_images import place_tool_result_images
 from litellm.llms.bedrock.common_utils import bedrock_model_supports_regex_lookaround
 from litellm.llms.bedrock.request_metadata import (
@@ -110,6 +110,7 @@ from ..common_utils import (
     BedrockError,
     BedrockModelInfo,
     bedrock_converse_supports_parallel_tool_use_config,
+    bedrock_deployment_serves_claude,
     bedrock_model_accepts_cache_points,
     bedrock_reasoning_effort_disabled,
     bedrock_rejects_stop_sequences,
@@ -2028,7 +2029,7 @@ class AmazonConverseConfig(BaseConfig):
         """On Claude, OpenAI ``prompt_cache_breakpoint`` marks as ``cache_control`` so the factory emits
         ``cachePoint`` blocks, held to Claude's cap, with ``prompt_cache_options`` rejected or dropped. A Claude model
         that never emits ``cachePoint`` blocks keeps its marks as written. Other models are left as they are."""
-        if BaseAWSLLM.get_bedrock_invoke_provider(model) != "anthropic":
+        if not bedrock_deployment_serves_claude(model, litellm_params):
             return tuple(messages)
         emits_cache_points: Final = bedrock_model_accepts_cache_points(model)
         drop_params: Final = litellm.drop_params or litellm_params.get("drop_params") is True
@@ -2037,12 +2038,16 @@ class AmazonConverseConfig(BaseConfig):
         )
         if not emits_cache_points:
             return tuple(messages)
+        tools: Final = optional_params.get("tools")
         cache_marks: Final = resolved_claude_cache_marks(
             claude_cache_marks(
                 messages=messages,
-                tools=optional_params.get("tools"),
+                tools=tools,
                 automatic_breakpoint=optional_params.get("cache_control") is not None,
                 drop_params=drop_params,
+                tool_config_breakpoint=converse_appends_tool_config_cache_point(
+                    tools, optional_params.get("cache_control_injection_points")
+                ),
                 wire="bedrock_converse",
             ),
             model=model,

litellm/llms/bedrock/common_utils.py

@@ -1217,6 +1217,14 @@ def bedrock_model_accepts_cache_points(model: str | None) -> bool:
     )
 
 
+def bedrock_deployment_serves_claude(model: str, litellm_params: Mapping[str, object]) -> bool:
+    """Whether ``model`` names Claude, or the deployment's ``base_model`` does, as for an application inference
+    profile ARN that hides the model behind it."""
+    base_model: Final = litellm_params.get("base_model")
+    candidates: Final = (*((base_model,) if isinstance(base_model, str) else ()), model)
+    return any(get_bedrock_base_model(candidate).startswith("anthropic") for candidate in candidates)
+
+
 def bedrock_supports_tool_search(model: str) -> bool:
     """
     Whether Bedrock InvokeModel admits the ``tool_search_tool_*`` tool types on ``model``.

tests/integration/providers/test_claude_chat_prompt_cache_breakpoint_wire.py

@@ -1,7 +1,7 @@
 from collections.abc import Iterator, Mapping
 from pathlib import Path
 from types import MappingProxyType
-from typing import Final, Literal, assert_never
+from typing import Final, Literal
 
 import pytest
 from integration._support.client import Gateway, Scenario, gateway_from_environment
@@ -38,6 +38,7 @@ from integration.providers._cache_control_marks_support import (
     tool_use_label,
 )
 from pydantic import BaseModel, ConfigDict, JsonValue
+from typing_extensions import assert_never
 
 Provider = Literal["anthropic", "bedrock_converse", "bedrock_invoke", "vertex_ai", "azure_ai"]
 Site = Literal["tool", "system", "user_part", "tool_call", "tool_result_part", "string_message"]
@@ -231,12 +232,36 @@ def test_breakpoint_becomes_a_cache_mark_on_the_same_block(gateway: Gateway, pro
     assert _labels(provider, request) == [_expected(provider, site, marker)]
 
 
-def test_breakpoint_ttl_is_not_forwarded(gateway: Gateway) -> None:
+@pytest.mark.parametrize(
+    ("breakpoint", "hint"),
+    ((_EXPLICIT_30M, "Remove the ttl"), ({"mode": "implicit"}, "prompt_cache_breakpoint must be")),
+    ids=("30m-ttl", "malformed"),
+)
+@pytest.mark.parametrize("provider", _PROVIDERS)
+def test_a_breakpoint_claude_cannot_take_is_a_400_before_any_provider_call_without_drop_params(
+    gateway: Gateway, provider: Provider, breakpoint: dict[str, JsonValue], hint: str
+) -> None:
+    marker: Final = new_marker()
+    marks: Final = _marked(("user_part",), {"prompt_cache_breakpoint": breakpoint})
+    with wire_server(lambda request: _respond(provider, request)) as wire, gateway.scenario() as scenario:
+        model: Final = _deployment(gateway, scenario, wire, provider)
+        status, _, text = post_chat(gateway, _body(model, marker, marks))
+        received: Final = wire.drain()
+    assert status == 400, text
+    assert hint in text, text
+    assert "drop_params" in text, text
+    assert received == (), [request.target for request in received]
+
+
+@pytest.mark.parametrize("source", ("deployment", "request"))
+def test_a_30m_breakpoint_is_sent_as_a_5m_mark_under_drop_params(gateway: Gateway, source: str) -> None:
     marker: Final = new_marker()
     marks: Final = _marked(("user_part",), {"prompt_cache_breakpoint": _EXPLICIT_30M})
+    deployment: Final[dict[str, JsonValue]] = {"drop_params": True} if source == "deployment" else {}
+    request_fields: Final[dict[str, JsonValue]] = {"drop_params": True} if source == "request" else {}
     with wire_server(anthropic_peer) as wire, gateway.scenario() as scenario:
-        model: Final = _deployment(gateway, scenario, wire, "anthropic")
+        model: Final = _deployment(gateway, scenario, wire, "anthropic", **deployment)
-        status, _, text = post_chat(gateway, _body(model, marker, marks))
+        status, _, text = post_chat(gateway, _body(model, marker, marks, **request_fields))
         assert status == 200, text
         request: Final = _only_request(wire)
     assert anthropic_marks(request) == (Mark(ASK_LABEL, None),)

tests/unit/litellm_core_utils/prompt_templates/test_litellm_core_utils_prompt_templates_common_utils.py

@@ -7,6 +7,7 @@ from typing import Final
 
 import pytest
 
+from litellm.exceptions import UnsupportedParamsError
 from litellm.litellm_core_utils.prompt_templates.common_utils import (
     ENCRYPTED_REASONING_SIGNATURE_PREFIX,
     TOOL_RESULT_IMAGE_BOUNDARY,
@@ -2323,16 +2324,45 @@ def _lookup_call(identity: str, **fields: object) -> dict[str, object]:
     return {"id": identity, "type": "function", "function": {"name": "lookup", "arguments": "{}"}, **fields}
 
 
-def test_prompt_cache_breakpoint_becomes_an_ephemeral_cache_control_without_the_ttl() -> None:
+@pytest.mark.parametrize("marker", [_EXPLICIT, _EXPLICIT_30M], ids=["no-ttl", "30m-ttl"])
-    marked: Final = {"type": "text", "text": "rules", "prompt_cache_breakpoint": _EXPLICIT_30M}
+def test_prompt_cache_breakpoint_becomes_a_5m_cache_control_under_drop_params(marker: dict[str, str]) -> None:
+    marked: Final = {"type": "text", "text": "rules", "prompt_cache_breakpoint": marker}
 
-    assert prompt_cache_breakpoint_as_cache_control(marked) == {
+    assert prompt_cache_breakpoint_as_cache_control(marked, drop_params=True) == {
         "type": "text",
         "text": "rules",
         "cache_control": _EPHEMERAL,
     }
 
 
+def test_a_30m_breakpoint_is_a_400_naming_every_fix_without_drop_params() -> None:
+    marked: Final = {"type": "text", "text": "rules", "prompt_cache_breakpoint": _EXPLICIT_30M}
+
+    with pytest.raises(UnsupportedParamsError) as caught:
+        prompt_cache_breakpoint_as_cache_control(marked, drop_params=False)
+
+    assert caught.value.status_code == 400
+    assert "Remove the ttl" in caught.value.message
+    assert "drop_params=True" in caught.value.message
+    assert "'ttl': '1h'" in caught.value.message
+
+
+@pytest.mark.parametrize(
+    "marker",
+    [{"mode": "implicit"}, {"mode": "explicit", "ttl": "1h"}, "explicit", {}],
+    ids=["implicit-mode", "1h-ttl", "string", "empty"],
+)
+def test_a_malformed_breakpoint_is_a_400_without_drop_params_and_no_mark_with_it(marker: object) -> None:
+    marked: Final = {"type": "text", "text": "rules", "prompt_cache_breakpoint": marker}
+
+    with pytest.raises(UnsupportedParamsError) as caught:
+        prompt_cache_breakpoint_as_cache_control(marked, drop_params=False)
+
+    assert caught.value.status_code == 400
+    assert "drop_params=True" in caught.value.message
+    assert prompt_cache_breakpoint_as_cache_control(marked, drop_params=True) == {"type": "text", "text": "rules"}
+
+
 def test_a_cache_control_the_client_wrote_wins_over_the_breakpoint_on_the_same_block() -> None:
     both: Final = {
         "type": "text",
@@ -2341,7 +2371,7 @@ def test_a_cache_control_the_client_wrote_wins_over_the_breakpoint_on_the_same_b
         "prompt_cache_breakpoint": _EXPLICIT_30M,
     }
 
-    assert prompt_cache_breakpoint_as_cache_control(both) == {
+    assert prompt_cache_breakpoint_as_cache_control(both, drop_params=False) == {
         "type": "text",
         "text": "rules",
         "cache_control": {"type": "ephemeral", "ttl": "1h"},
@@ -2351,18 +2381,18 @@ def test_a_cache_control_the_client_wrote_wins_over_the_breakpoint_on_the_same_b
 def test_a_block_without_a_breakpoint_is_returned_untouched() -> None:
     unmarked: Final = {"type": "text", "text": "rules", "cache_control": _EPHEMERAL}
 
-    assert prompt_cache_breakpoint_as_cache_control(unmarked) is unmarked
+    assert prompt_cache_breakpoint_as_cache_control(unmarked, drop_params=False) is unmarked
 
 
 def test_breakpoints_are_renamed_at_the_message_level_on_parts_and_on_tool_calls() -> None:
     message: Final = {
         "role": "assistant",
         "content": [{"type": "text", "text": "a", "prompt_cache_breakpoint": _EXPLICIT}, {"type": "text", "text": "b"}],
-        "tool_calls": [_lookup_call("call_1", prompt_cache_breakpoint=_EXPLICIT_30M), _lookup_call("call_2")],
+        "tool_calls": [_lookup_call("call_1", prompt_cache_breakpoint=_EXPLICIT), _lookup_call("call_2")],
         "prompt_cache_breakpoint": _EXPLICIT,
     }
 
-    assert message_with_prompt_cache_breakpoints_as_cache_control(message) == {
+    assert message_with_prompt_cache_breakpoints_as_cache_control(message, drop_params=False) == {
         "role": "assistant",
         "content": [{"type": "text", "text": "a", "cache_control": _EPHEMERAL}, {"type": "text", "text": "b"}],
         "tool_calls": [_lookup_call("call_1", cache_control=_EPHEMERAL), _lookup_call("call_2")],
@@ -2374,7 +2404,7 @@ def test_a_message_level_breakpoint_on_string_content_is_renamed_and_other_messa
     unmarked: Final = {"role": "user", "content": "unmarked"}
 
     renamed: Final = messages_with_prompt_cache_breakpoints_as_cache_control(
-        [{"role": "user", "content": "hi", "prompt_cache_breakpoint": _EXPLICIT}, unmarked]
+        [{"role": "user", "content": "hi", "prompt_cache_breakpoint": _EXPLICIT}, unmarked], drop_params=False
     )
 
     assert renamed == ({"role": "user", "content": "hi", "cache_control": _EPHEMERAL}, unmarked)
@@ -2390,13 +2420,34 @@ def test_a_message_level_breakpoint_on_string_content_is_renamed_and_other_messa
     ids=["top-level", "under-function"],
 )
 def test_a_tool_breakpoint_becomes_a_top_level_cache_control(tool: dict[str, object]) -> None:
-    assert tool_with_prompt_cache_breakpoint_as_cache_control(tool) == {
+    assert tool_with_prompt_cache_breakpoint_as_cache_control(tool, drop_params=False) == {
         "type": "function",
         "function": {"name": "lookup", "parameters": {}},
         "cache_control": _EPHEMERAL,
     }
 
 
+@pytest.mark.parametrize(
+    "tool",
+    [
+        {"type": "function", "function": {"name": "lookup"}, "prompt_cache_breakpoint": _EXPLICIT_30M},
+        {"type": "function", "function": {"name": "lookup", "prompt_cache_breakpoint": _EXPLICIT_30M}},
+    ],
+    ids=["top-level", "under-function"],
+)
+def test_a_30m_tool_breakpoint_is_a_400_without_drop_params_and_a_5m_cache_control_with_it(
+    tool: dict[str, object],
+) -> None:
+    with pytest.raises(UnsupportedParamsError, match="Remove the ttl"):
+        tool_with_prompt_cache_breakpoint_as_cache_control(tool, drop_params=False)
+
+    assert tool_with_prompt_cache_breakpoint_as_cache_control(tool, drop_params=True) == {
+        "type": "function",
+        "function": {"name": "lookup"},
+        "cache_control": _EPHEMERAL,
+    }
+
+
 def test_a_tool_cache_control_the_client_wrote_wins_over_the_breakpoint() -> None:
     tool: Final = {
         "type": "function",
@@ -2407,7 +2458,7 @@ def test_a_tool_cache_control_the_client_wrote_wins_over_the_breakpoint() -> Non
         },
     }
 
-    assert tool_with_prompt_cache_breakpoint_as_cache_control(tool) == {
+    assert tool_with_prompt_cache_breakpoint_as_cache_control(tool, drop_params=False) == {
         "type": "function",
         "function": {"name": "lookup", "cache_control": {"type": "ephemeral", "ttl": "1h"}},
     }
@@ -2416,8 +2467,11 @@ def test_a_tool_cache_control_the_client_wrote_wins_over_the_breakpoint() -> Non
 def test_tools_that_are_not_a_list_carry_no_marks_and_unmarked_tools_are_kept() -> None:
     unmarked: Final = {"type": "function", "function": {"name": "lookup"}}
 
-    assert tools_with_prompt_cache_breakpoints_as_cache_control(None) == ()
+    assert tools_with_prompt_cache_breakpoints_as_cache_control(None, drop_params=False) == ()
-    assert tools_with_prompt_cache_breakpoints_as_cache_control([unmarked, "opaque"]) == (unmarked, "opaque")
+    assert tools_with_prompt_cache_breakpoints_as_cache_control([unmarked, "opaque"], drop_params=False) == (
+        unmarked,
+        "opaque",
+    )
 
 
 def test_message_without_cache_control_marks_strips_only_the_listed_positions() -> None:

tests/unit/llms/anthropic/chat/test_anthropic_chat_transformation.py

@@ -8781,7 +8781,7 @@ async def test_prompt_cache_breakpoints_on_system_user_and_tool_definition_becom
         monkeypatch,
         messages=[
             {"role": "system", "content": [{"type": "text", "text": "rules", "prompt_cache_breakpoint": EXPLICIT}]},
-            {"role": "user", "content": [{"type": "text", "text": "hi", "prompt_cache_breakpoint": EXPLICIT_30M}]},
+            {"role": "user", "content": [{"type": "text", "text": "hi", "prompt_cache_breakpoint": EXPLICIT}]},
         ],
         tools=[{"type": "function", "function": {**_WEATHER_FUNCTION, "prompt_cache_breakpoint": EXPLICIT}}],
     )
@@ -8821,7 +8821,7 @@ async def test_prompt_cache_breakpoints_on_tool_calls_tool_results_and_string_me
                 "content": [{"type": "text", "text": "sunny", "prompt_cache_breakpoint": EXPLICIT}],
             },
             {"role": "assistant", "content": "It is sunny."},
-            {"role": "user", "content": "thanks", "prompt_cache_breakpoint": EXPLICIT_30M},
+            {"role": "user", "content": "thanks", "prompt_cache_breakpoint": EXPLICIT},
         ],
         tools=[{"type": "function", "function": _WEATHER_FUNCTION}],
     )
@@ -9085,6 +9085,120 @@ async def test_a_message_level_breakpoint_on_list_content_does_not_count_toward_
     ]
 
 
+def _text_part(text: str, marker: dict[str, str]) -> dict[str, object]:
+    return {"type": "text", "text": text, "prompt_cache_breakpoint": marker}
+
+
+_MARKED_30M_PART: Final = _text_part("hi", EXPLICIT_30M)
+_WEATHER_TOOL: Final = {"type": "function", "function": _WEATHER_FUNCTION}
+
+
+@pytest.mark.parametrize(
+    ("messages", "tools"),
+    [
+        ([{"role": "user", "content": [_MARKED_30M_PART]}], [_WEATHER_TOOL]),
+        ([{"role": "user", "content": "hi"}], [{**_WEATHER_TOOL, "prompt_cache_breakpoint": EXPLICIT_30M}]),
+    ],
+    ids=["message-part", "tool-definition"],
+)
+async def test_a_30m_breakpoint_is_a_bad_request_naming_every_fix_without_drop_params(
+    respx_mock: respx.MockRouter,
+    monkeypatch: pytest.MonkeyPatch,
+    messages: list[dict[str, object]],
+    tools: list[dict[str, object]],
+) -> None:
+    monkeypatch.setattr(litellm, "drop_params", False)
+    monkeypatch.setenv("DISABLE_AIOHTTP_TRANSPORT", "True")
+    route: Final = respx_mock.post("https://api.anthropic.com/v1/messages").mock(
+        return_value=httpx.Response(200, json=_ANTHROPIC_REPLY)
+    )
+
+    with pytest.raises(litellm.BadRequestError) as caught:
+        await litellm.acompletion(
+            api_key="mock_api_key",
+            model=f"anthropic/{PROMPT_CACHING_MODEL}",
+            messages=messages,
+            tools=tools,
+            num_retries=0,
+        )
+
+    assert caught.value.status_code == 400
+    assert "Remove the ttl" in str(caught.value)
+    assert "drop_params=True" in str(caught.value)
+    assert "'ttl': '1h'" in str(caught.value)
+    assert route.call_count == 0
+
+
+@pytest.mark.parametrize("drop_params_source", ["kwarg", "global"])
+async def test_a_30m_breakpoint_sends_a_5m_cache_control_under_drop_params(
+    respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch, drop_params_source: str
+) -> None:
+    monkeypatch.setattr(litellm, "drop_params", drop_params_source == "global")
+    request: Final = await _send_prompt_caching_request(
+        respx_mock,
+        monkeypatch,
+        messages=[{"role": "user", "content": [_MARKED_30M_PART]}],
+        tools=[{**_WEATHER_TOOL, "prompt_cache_breakpoint": EXPLICIT_30M}],
+        **({"drop_params": True} if drop_params_source == "kwarg" else {}),
+    )
+
+    body: Final = json.loads(request.content)
+    assert body["messages"] == [{"role": "user", "content": [{"type": "text", "text": "hi", "cache_control": EPHEMERAL}]}]
+    assert body["tools"] == [{**_ANTHROPIC_WEATHER_TOOL, "cache_control": EPHEMERAL}]
+
+
+@pytest.mark.parametrize(
+    "marker", [{"mode": "implicit"}, {"mode": "explicit", "ttl": "1h"}], ids=["implicit-mode", "1h-ttl"]
+)
+async def test_a_malformed_breakpoint_is_a_bad_request_without_drop_params_and_sends_no_mark_with_it(
+    respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch, marker: dict[str, str]
+) -> None:
+    monkeypatch.setattr(litellm, "drop_params", False)
+    messages: Final = [{"role": "user", "content": [{"type": "text", "text": "hi", "prompt_cache_breakpoint": marker}]}]
+
+    with pytest.raises(litellm.BadRequestError, match="prompt_cache_breakpoint must be") as caught:
+        await litellm.acompletion(
+            api_key="mock_api_key", model=f"anthropic/{PROMPT_CACHING_MODEL}", messages=messages, num_retries=0
+        )
+    request: Final = await _send_prompt_caching_request(respx_mock, monkeypatch, messages=messages, drop_params=True)
+
+    assert caught.value.status_code == 400
+    assert json.loads(request.content)["messages"] == [{"role": "user", "content": [{"type": "text", "text": "hi"}]}]
+
+
+@pytest.mark.parametrize(
+    "unsent",
+    [
+        {"role": "system", "content": [{"type": "text", "text": "rules"}, _text_part("", EXPLICIT)]},
+        {
+            "role": "user",
+            "content": [
+                {"type": "input_audio", "input_audio": {"data": "AAAA", "format": "wav"}, "prompt_cache_breakpoint": EXPLICIT}
+            ],
+        },
+        {
+            "role": "assistant",
+            "content": [
+                {"type": "image_url", "image_url": {"url": "https://example.com/a.png"}, "prompt_cache_breakpoint": EXPLICIT}
+            ],
+        },
+    ],
+    ids=["empty-system-text", "user-input-audio", "assistant-image"],
+)
+async def test_a_mark_on_a_block_anthropic_does_not_send_takes_none_of_the_four_slots(
+    respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch, unsent: dict[str, object]
+) -> None:
+    monkeypatch.setattr(litellm, "drop_params", False)
+    marked: Final = [_text_part(f"part {index}", EXPLICIT) for index in range(4)]
+
+    request: Final = await _send_prompt_caching_request(
+        respx_mock, monkeypatch, messages=[unsent, {"role": "user", "content": marked}]
+    )
+
+    assert request.content.decode().count('"cache_control"') == 4
+    assert "prompt_cache_breakpoint" not in request.content.decode()
+
+
 def test_map_openai_params_maps_a_tuple_of_tools() -> None:
     optional_params: Final = AnthropicConfig().map_openai_params(
         non_default_params={"tools": ({"type": "function", "function": _WEATHER_FUNCTION},)},

tests/unit/llms/anthropic/chat/test_prompt_cache_breakpoints.py

@@ -1,3 +1,5 @@
+import json
+import time
 from typing import Final
 
 import pytest
@@ -6,6 +8,7 @@ import litellm
 from litellm.llms.anthropic.chat.prompt_cache_breakpoints import (
     CacheBreakpointCapExceeded,
     ClaudeCacheMarks,
+    ClaudeWire,
     claude_cache_marks,
     reject_prompt_cache_options,
     resolved_claude_cache_marks,
@@ -14,7 +17,6 @@ from litellm.llms.anthropic.chat.prompt_cache_breakpoints import (
 
 EPHEMERAL: Final = {"type": "ephemeral"}
 EXPLICIT: Final = {"mode": "explicit"}
-EXPLICIT_30M: Final = {"mode": "explicit", "ttl": "30m"}
 MODEL: Final = "claude-opus-5-5"
 
 
@@ -35,7 +37,7 @@ def _six_breakpoints() -> tuple[list[dict[str, object]], list[dict[str, object]]
     one marked, in the order Claude reads them"""
     messages: Final = [
         {"role": "system", "content": [_text("rules", prompt_cache_breakpoint=EXPLICIT)]},
-        {"role": "user", "content": [_text("hi", prompt_cache_breakpoint=EXPLICIT_30M)]},
+        {"role": "user", "content": [_text("hi", prompt_cache_breakpoint=EXPLICIT)]},
         {"role": "assistant", "content": "", "tool_calls": [_call("call_1", prompt_cache_breakpoint=EXPLICIT)]},
         {"role": "tool", "tool_call_id": "call_1", "content": [_text("sunny", prompt_cache_breakpoint=EXPLICIT)]},
         {"role": "user", "content": "done", "prompt_cache_breakpoint": EXPLICIT},
@@ -172,16 +174,42 @@ def _four_marked_parts_then(message: dict[str, object]) -> list[dict[str, object
     ]
 
 
-def test_a_mark_on_an_empty_assistant_string_takes_no_anthropic_slot_but_a_converse_slot() -> None:
+@pytest.mark.parametrize("content", ["", " "], ids=["empty", "blank"])
-    messages: Final = _four_marked_parts_then({"role": "assistant", "content": "", "prompt_cache_breakpoint": EXPLICIT})
+def test_a_mark_on_a_blank_assistant_string_takes_an_anthropic_slot_but_no_converse_slot(content: str) -> None:
+    messages: Final = _four_marked_parts_then(
+        {"role": "assistant", "content": content, "prompt_cache_breakpoint": EXPLICIT}
+    )
 
     anthropic: Final = claude_cache_marks(messages, None, automatic_breakpoint=False, drop_params=False)
     converse: Final = claude_cache_marks(
         messages, None, automatic_breakpoint=False, drop_params=False, wire="bedrock_converse"
     )
 
-    assert isinstance(anthropic, ClaudeCacheMarks), anthropic
+    assert anthropic == CacheBreakpointCapExceeded(found=5)
-    assert converse == CacheBreakpointCapExceeded(found=5)
+    assert isinstance(converse, ClaudeCacheMarks), converse
+
+
+@pytest.mark.parametrize("wire", ["anthropic", "bedrock_converse"])
+def test_trimming_thousands_of_marks_to_four_takes_linear_time(wire: ClaudeWire) -> None:
+    messages: Final = [
+        {"role": "user", "content": [_text(f"turn {index}", prompt_cache_breakpoint=EXPLICIT)]}
+        for index in range(10000)
+    ]
+
+    started: Final = time.perf_counter()
+    marks: Final = claude_cache_marks(messages, None, automatic_breakpoint=False, drop_params=True, wire=wire)
+    elapsed: Final = time.perf_counter() - started
+
+    assert isinstance(marks, ClaudeCacheMarks), marks
+    assert json.dumps(marks.messages).count("cache_control") == 4
+    assert marks.messages[-5:] == (
+        {"role": "user", "content": [_text("turn 9995")]},
+        *(
+            {"role": "user", "content": [_text(f"turn {index}", cache_control=EPHEMERAL)]}
+            for index in range(9996, 10000)
+        ),
+    )
+    assert elapsed < 5, f"trimming 10000 marks took {elapsed:.1f}s"
 
 
 def test_a_mark_on_a_tool_call_rebuilt_as_server_tool_use_takes_no_anthropic_slot() -> None:

tests/unit/llms/azure_ai/claude/test_azure_anthropic_transformation.py

@@ -494,7 +494,7 @@ def test_prompt_cache_breakpoint_becomes_cache_control_on_the_same_block():
         messages=[
             {
                 "role": "user",
-                "content": [{"type": "text", "text": "hi", "prompt_cache_breakpoint": {"mode": "explicit", "ttl": "30m"}}],
+                "content": [{"type": "text", "text": "hi", "prompt_cache_breakpoint": {"mode": "explicit"}}],
             }
         ],
         optional_params={},

tests/unit/llms/bedrock/chat/invoke_transformations/test_bedrock_chat_invoke_transformations_anthropic_claude3_transformation.py

@@ -1369,7 +1369,7 @@ class TestBedrockAnthropicPromptCachingRegression:
                     {
                         "type": "text",
                         "text": LARGE_DOCUMENT_FOR_CACHING,
-                        "prompt_cache_breakpoint": {"mode": "explicit", "ttl": "30m"},
+                        "prompt_cache_breakpoint": {"mode": "explicit"},
                     },
                     {"type": "text", "text": "What are the payment terms?"},
                 ],

tests/unit/llms/bedrock/chat/test_converse_transformation.py

@@ -10171,7 +10171,7 @@ _LOOKUP_TOOL_SPEC: Final = {
 def _breakpoint_conversation() -> list[dict[str, object]]:
     return [
         {"role": "system", "content": [{"type": "text", "text": "rules", "prompt_cache_breakpoint": _EXPLICIT}]},
-        {"role": "user", "content": [{"type": "text", "text": "hi", "prompt_cache_breakpoint": _EXPLICIT_30M}]},
+        {"role": "user", "content": [{"type": "text", "text": "hi", "prompt_cache_breakpoint": _EXPLICIT}]},
         {
             "role": "assistant",
             "content": "",
@@ -10429,6 +10429,174 @@ async def test_a_tool_result_with_two_marked_parts_is_one_cache_point_within_the
     ]
 
 
+def _four_marked_user_parts() -> list[dict[str, object]]:
+    return [
+        {
+            "role": "user",
+            "content": [
+                {"type": "text", "text": f"part {index}", "prompt_cache_breakpoint": _EXPLICIT} for index in range(4)
+            ],
+        }
+    ]
+
+
+@pytest.mark.parametrize(
+    ("marker", "error", "sent_under_drop_params"),
+    [
+        (_EXPLICIT_30M, "Remove the ttl", [{"text": "hi"}, _CACHE_POINT]),
+        ({"mode": "implicit"}, "prompt_cache_breakpoint must be", [{"text": "hi"}]),
+    ],
+    ids=["30m-ttl", "malformed"],
+)
+@pytest.mark.parametrize("sync_mode", [True, False], ids=["sync", "async"])
+@pytest.mark.asyncio
+async def test_a_breakpoint_claude_cannot_take_is_a_bad_request_unless_drop_params(
+    sync_mode: bool,
+    marker: dict[str, str],
+    error: str,
+    sent_under_drop_params: list[dict[str, object]],
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(litellm, "drop_params", False)
+
+    def marked() -> list[dict[str, object]]:
+        return [{"role": "user", "content": [{"type": "text", "text": "hi", "prompt_cache_breakpoint": marker}]}]
+
+    with pytest.raises(litellm.BadRequestError, match=error) as caught:
+        await _breakpoint_converse_request(sync_mode, _BREAKPOINT_MODEL, marked())
+    dropped: Final = await _breakpoint_converse_request(
+        sync_mode, _BREAKPOINT_MODEL, marked(), litellm_params={"drop_params": True}
+    )
+
+    assert caught.value.status_code == 400
+    assert dropped["messages"] == [{"role": "user", "content": sent_under_drop_params}]
+
+
+@pytest.mark.parametrize(
+    ("messages", "tools"),
+    [
+        (
+            [
+                {
+                    "role": "system",
+                    "content": [
+                        {"type": "text", "text": "rules"},
+                        {"type": "text", "text": "", "prompt_cache_breakpoint": _EXPLICIT},
+                    ],
+                },
+                *_four_marked_user_parts(),
+            ],
+            [_LOOKUP_TOOL],
+        ),
+        (
+            [
+                *_four_marked_user_parts(),
+                {"role": "assistant", "content": " ", "prompt_cache_breakpoint": _EXPLICIT},
+                {"role": "user", "content": "next"},
+            ],
+            [_LOOKUP_TOOL],
+        ),
+        (
+            _four_marked_user_parts(),
+            [
+                _LOOKUP_TOOL,
+                {"type": "tool_search_tool_regex_20251119", "name": "find", "prompt_cache_breakpoint": _EXPLICIT},
+            ],
+        ),
+        (_four_marked_user_parts(), [_LOOKUP_TOOL, {"type": "web_search", "cache_control": {"type": "ephemeral"}}]),
+    ],
+    ids=["empty-system-text", "blank-assistant-string", "tool-search-tool", "built-in-web-search-tool"],
+)
+@pytest.mark.parametrize("sync_mode", [True, False], ids=["sync", "async"])
+@pytest.mark.asyncio
+async def test_a_mark_converse_does_not_send_takes_none_of_the_four_cache_points(
+    sync_mode: bool, messages: list[dict[str, object]], tools: list[dict[str, object]], monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setattr(litellm, "drop_params", False)
+
+    result: Final = await _breakpoint_converse_request(
+        sync_mode, _BREAKPOINT_MODEL, copy.deepcopy(messages), optional_params={"tools": copy.deepcopy(tools)}
+    )
+
+    assert json.dumps(result).count("cachePoint") == 4
+    assert result["toolConfig"]["tools"] == [_LOOKUP_TOOL_SPEC]
+
+
+@pytest.mark.parametrize("sync_mode", [True, False], ids=["sync", "async"])
+@pytest.mark.asyncio
+async def test_a_tool_config_injection_point_takes_one_of_the_four_cache_points(
+    sync_mode: bool, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setattr(litellm, "drop_params", False)
+
+    def optional_params() -> dict[str, object]:
+        return {
+            "tools": [_LOOKUP_TOOL],
+            "cache_control_injection_points": [{"location": "tool_config", "control": {"type": "ephemeral"}}],
+        }
+
+    with pytest.raises(litellm.BadRequestError, match=r"Claude accepts at most 4 cache breakpoints, found 5"):
+        await _breakpoint_converse_request(
+            sync_mode, _BREAKPOINT_MODEL, _four_marked_user_parts(), optional_params=optional_params()
+        )
+    trimmed: Final = await _breakpoint_converse_request(
+        sync_mode,
+        _BREAKPOINT_MODEL,
+        _four_marked_user_parts(),
+        optional_params=optional_params(),
+        litellm_params={"drop_params": True},
+    )
+
+    assert json.dumps(trimmed).count("cachePoint") == 4
+    assert trimmed["toolConfig"]["tools"] == [_LOOKUP_TOOL_SPEC, _CACHE_POINT]
+
+
+@pytest.mark.parametrize("sync_mode", [True, False], ids=["sync", "async"])
+@pytest.mark.asyncio
+async def test_a_breakpoint_under_a_tool_function_becomes_a_cache_point_after_the_tool(sync_mode: bool) -> None:
+    tool: Final = {
+        "type": "function",
+        "function": {
+            "name": "lookup",
+            "description": "lookup",
+            "parameters": {"type": "object", "properties": {}, "required": []},
+            "prompt_cache_breakpoint": _EXPLICIT,
+        },
+    }
+
+    result: Final = await _breakpoint_converse_request(
+        sync_mode, _BREAKPOINT_MODEL, [{"role": "user", "content": "hi"}], optional_params={"tools": [tool]}
+    )
+
+    assert result["toolConfig"]["tools"] == [_LOOKUP_TOOL_SPEC, _CACHE_POINT]
+
+
+@pytest.mark.parametrize(
+    ("base_model", "system"),
+    [
+        ("anthropic.claude-opus-5-5", [{"text": "rules"}, _CACHE_POINT]),
+        ("amazon.nova-2-lite-v1:0", [{"text": "rules"}]),
+    ],
+    ids=["claude", "nova"],
+)
+@pytest.mark.parametrize("sync_mode", [True, False], ids=["sync", "async"])
+@pytest.mark.asyncio
+async def test_a_profile_arn_gets_claude_breakpoints_when_its_base_model_is_claude(
+    sync_mode: bool, base_model: str, system: list[dict[str, object]]
+) -> None:
+    result: Final = await _breakpoint_converse_request(
+        sync_mode,
+        "arn:aws:bedrock:us-east-1:123456789012:application-inference-profile/abc123xyz",
+        [
+            {"role": "system", "content": [{"type": "text", "text": "rules", "prompt_cache_breakpoint": _EXPLICIT}]},
+            {"role": "user", "content": "hi"},
+        ],
+        litellm_params={"base_model": base_model},
+    )
+
+    assert result["system"] == system
+
+
 @pytest.mark.parametrize("sync_mode", [True, False], ids=["sync", "async"])
 @pytest.mark.asyncio
 async def test_prompt_cache_options_on_a_non_claude_converse_model_are_forwarded_as_before(

tests/unit/llms/databricks/chat/test_databricks_chat_transformation.py

@@ -2282,7 +2282,7 @@ def test_prompt_cache_breakpoints_reach_databricks_claude_as_cache_control(
         model="databricks/databricks-claude-haiku-4-5",
         messages=[
             {"role": "system", "content": [{"type": "text", "text": "rules", "prompt_cache_breakpoint": _EXPLICIT}]},
-            {"role": "user", "content": "hi", "prompt_cache_breakpoint": {"mode": "explicit", "ttl": "30m"}},
+            {"role": "user", "content": "hi", "prompt_cache_breakpoint": _EXPLICIT},
         ],
     )
 

tests/unit/llms/vertex_ai/vertex_ai_partner_models/anthropic/test_vertex_ai_partner_models_anthropic_transformation.py

@@ -1130,7 +1130,7 @@ async def test_vertex_ai_anthropic_prompt_cache_breakpoint_becomes_cache_control
             {
                 "role": "user",
                 "content": [
-                    {"type": "text", "text": "Cache this request", "prompt_cache_breakpoint": {"mode": "explicit", "ttl": "30m"}}
+                    {"type": "text", "text": "Cache this request", "prompt_cache_breakpoint": {"mode": "explicit"}}
                 ],
             }
         ],