fix(anthropic): count only the breakpoints Claude reads before trimming to four

A tool whose function carries its own cache_control now drops every breakpoint
on that tool, and completion() strips an unanswered tool_search call before the
mark census, so neither takes a slot from a mark Claude would read. The router
also pins a prefix whose breakpoint asks for a 1h ttl for an hour.

litellm/llms/anthropic/chat/prompt_cache_breakpoints.py

@@ -151,19 +151,25 @@ def _marked_message(index: int, message: object, mark: _Mark) -> object:
     }
 
 
-def _with_function_breakpoint_hoisted(tool: _Fields) -> _Fields:
-    """Converse reads only a tool's top-level mark, so a breakpoint written under ``function`` moves up."""
+def _with_one_tool_breakpoint(tool: _Fields) -> _Fields:
+    """Claude sends one mark per tool and Converse reads only the top-level one, so a breakpoint written under
+    ``function`` moves up, and a ``cache_control`` the client wrote at either level wins over every breakpoint."""
     function: Final = _fields(tool.get("function"))
-    marker: Final = None if function is None else function.get("prompt_cache_breakpoint")
-    if function is None or marker is None:
+    if function is None:
         return tool
-    keeps_client_mark: Final = (
-        function.get("cache_control") is not None or tool.get("prompt_cache_breakpoint") is not None
-    )
+    function_marker: Final = function.get("prompt_cache_breakpoint")
+    tool_marker: Final = tool.get("prompt_cache_breakpoint")
+    if function_marker is None and (tool_marker is None or function.get("cache_control") is None):
+        return tool
+    client_marked: Final = function.get("cache_control") is not None or tool.get("cache_control") is not None
     return {
-        **tool,
+        **{key: value for key, value in tool.items() if key != "prompt_cache_breakpoint"},
         "function": {key: value for key, value in function.items() if key != "prompt_cache_breakpoint"},
-        **({} if keeps_client_mark else {"prompt_cache_breakpoint": marker}),
+        **(
+            {}
+            if client_marked
+            else {"prompt_cache_breakpoint": function_marker if tool_marker is None else tool_marker}
+        ),
     }
 
 
@@ -171,7 +177,7 @@ def _marked_tool(index: int, tool: object, mark: _Mark) -> object:
     fields: Final = _fields(tool)
     if fields is None:
         return tool
-    marked: Final = mark(("tools", index), _with_function_breakpoint_hoisted(fields))
+    marked: Final = mark(("tools", index), _with_one_tool_breakpoint(fields))
     function: Final = _fields(marked.get("function"))
     return marked if function is None else {**marked, "function": mark(("tools", index, "function"), function)}
 
@@ -318,7 +324,7 @@ def claude_prompt_cache(
     listed_tools: Final = tools or ()
     if not any(map(_message_carries_breakpoint, messages)) and not any(map(_tool_carries_breakpoint, listed_tools)):
         return ClaudePromptCache(tuple(messages), None if tools is None else tuple(tools))
-    blocks: Final = tuple(block for _, block in _markable(messages, listed_tools))
+    blocks: Final = tuple(block for _, block in _markable(*_marked_request(messages, listed_tools, _unchanged)))
     problem: Final = next(filter(None, map(_breakpoint_problem, blocks)), None)
     if problem is not None and not drop_params:
         return problem

litellm/main.py

@@ -5579,6 +5579,13 @@ def completion(
                 base_model=base_model,
             )
 
+        request_messages: Final = cast(list[object], messages)  # cast-ok: completion types messages as a bare List
+        history_for_route: Final = (
+            None
+            if provider_config is None
+            else _history_without_unanswered_tool_search_calls(request_messages, custom_llm_provider, provider_config)
+        )
+        routed_messages: Final = request_messages if history_for_route is None else list(history_for_route)
         claude_drop_params: Final = litellm.drop_params is True or (
             "drop_params" in kwargs
             and normalize_drop_params(cast(object, kwargs["drop_params"])) is True  # cast-ok: untyped request kwargs
@@ -5586,7 +5593,7 @@ def completion(
         claude_cache: Final = (
             resolved_claude_prompt_cache(
                 claude_prompt_cache(
-                    cast(list[object], messages),  # cast-ok: completion types messages as a bare List
+                    routed_messages,
                     cast(list[object] | None, tools),  # cast-ok: completion types tools as a bare List
                     non_default_params,
                     drop_params=claude_drop_params,
@@ -5602,20 +5609,12 @@ def completion(
             )
             else None
         )
-        claude_messages: Final = (
-            None
-            if claude_cache is None
-            else cast(list[AllMessageValues], list(claude_cache.messages))  # cast-ok: marks renamed in place
+        provider_messages: Final = cast(  # cast-ok: completion() only renames marks and drops tool_search calls
+            list[AllMessageValues], routed_messages if claude_cache is None else list(claude_cache.messages)
         )
         if provider_config is not None:
-            translated_messages: Final = provider_config.translate_developer_role_to_system_role(
-                messages=messages if claude_messages is None else claude_messages
-            )
-            history_for_route: Final = _history_without_unanswered_tool_search_calls(
-                translated_messages, custom_llm_provider, provider_config
-            )
             messages = (  # rebind-ok: completion() rewrites messages step by step for the provider
-                translated_messages if history_for_route is None else list(history_for_route)
+                provider_config.translate_developer_role_to_system_role(messages=provider_messages)
             )
 
         if (

litellm/router_utils/prompt_caching_cache.py

@@ -61,12 +61,20 @@ def _canonical_bytes(value: object) -> bytes:
     return json.dumps(value, sort_keys=True, separators=(",", ":")).encode()
 
 
-def _carries_cache_mark(block: Mapping[str, object]) -> bool:
+def _mark_pin_ttl_seconds(block: Mapping[str, object]) -> int | None:
     cache_control: Final = block.get("cache_control")
     marker: Final = block.get("prompt_cache_breakpoint")
-    return (isinstance(cache_control, dict) and cache_control.get("type") == "ephemeral") or (
-        isinstance(marker, dict) and marker.get("mode") == "explicit"
-    )
+    if isinstance(cache_control, dict) and cache_control.get("type") == "ephemeral":
+        return (
+            PROMPT_CACHE_ONE_HOUR_PIN_TTL_SECONDS if cache_control.get("ttl") == "1h" else PROMPT_CACHE_PIN_TTL_SECONDS
+        )
+    if isinstance(marker, dict) and marker.get("mode") == "explicit":
+        return PROMPT_CACHE_ONE_HOUR_PIN_TTL_SECONDS if marker.get("ttl") == "1h" else PROMPT_CACHE_PIN_TTL_SECONDS
+    return None
+
+
+def _carries_cache_mark(block: Mapping[str, object]) -> bool:
+    return _mark_pin_ttl_seconds(block) is not None
 
 
 def _block_unit(
@@ -129,12 +137,7 @@ def _lookback_keys(positions: tuple[PrefixPosition, ...]) -> tuple[str, ...]:
 
 
 def _breakpoint_pin_ttl_seconds(carrier: JsonValue | Mapping[str, JsonValue]) -> int | None:
-    if not isinstance(carrier, Mapping):
-        return None
-    cache_control: Final = carrier.get("cache_control")
-    if not isinstance(cache_control, dict) or cache_control.get("type") != "ephemeral":
-        return None
-    return PROMPT_CACHE_ONE_HOUR_PIN_TTL_SECONDS if cache_control.get("ttl") == "1h" else PROMPT_CACHE_PIN_TTL_SECONDS
+    return _mark_pin_ttl_seconds(carrier) if isinstance(carrier, Mapping) else None
 
 
 def _breakpoint_carriers(message: Mapping[str, JsonValue]) -> tuple[JsonValue | Mapping[str, JsonValue], ...]:

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

@@ -90,6 +90,20 @@ def test_a_client_cache_control_on_the_same_block_wins():
     assert list(_resolved(messages).messages) == [{"role": "user", "content": [_part("hi", cache_control=hour)]}]
 
 
+def test_a_breakpoint_under_a_function_the_tool_cache_control_overrides_is_never_reported():
+    tools: Final = [
+        {
+            "type": "function",
+            "cache_control": _EPHEMERAL,
+            "function": {"name": "t", "prompt_cache_breakpoint": {"mode": "explicit", "ttl": "30m"}},
+        }
+    ]
+
+    result: Final = _resolved([{"role": "user", "content": "hi"}], tools)
+
+    assert list(result.tools or ()) == [{"type": "function", "cache_control": _EPHEMERAL, "function": {"name": "t"}}]
+
+
 @pytest.mark.parametrize(
     "marker, kept",
     [({"mode": "explicit", "ttl": "30m"}, _EPHEMERAL), ({"mode": "sometimes"}, None), ("yes", None)],
@@ -318,6 +332,28 @@ def test_a_tool_marked_at_both_levels_takes_one_slot_under_drop_params():
     ]
 
 
+def test_a_tool_breakpoint_beside_a_function_cache_control_frees_a_slot_claude_still_reads_under_drop_params():
+    tools: Final = [
+        {
+            "type": "function",
+            "prompt_cache_breakpoint": _EXPLICIT,
+            "function": {"name": "t", "cache_control": _EPHEMERAL},
+        }
+    ]
+    messages: Final = [
+        {"role": "system", "content": [_part("s", prompt_cache_breakpoint=_EXPLICIT)]},
+        {"role": "user", "content": [_part(str(index), prompt_cache_breakpoint=_EXPLICIT) for index in range(3)]},
+    ]
+
+    result: Final = _resolved(messages, tools, drop_params=True)
+
+    assert list(result.tools or ()) == [{"type": "function", "function": {"name": "t", "cache_control": _EPHEMERAL}}]
+    assert list(result.messages) == [
+        {"role": "system", "content": [_part("s")]},
+        {"role": "user", "content": [_part(str(index), cache_control=_EPHEMERAL) for index in range(3)]},
+    ]
+
+
 def test_a_message_level_breakpoint_on_a_tool_call_turn_without_content_takes_no_slot_under_drop_params():
     messages: Final = [
         {"role": "system", "content": [_part("s", prompt_cache_breakpoint=_EXPLICIT)]},

tests/unit/router_utils/test_prompt_caching_cache.py

@@ -102,6 +102,19 @@ def _tool(cache_control: dict[str, str] | None) -> ChatCompletionToolParam:
             PROMPT_CACHE_ONE_HOUR_PIN_TTL_SECONDS,
             id="1h-system-then-5m-turn",
         ),
+        pytest.param(
+            [
+                {
+                    "role": "system",
+                    "content": "Here is the full text of a complex legal agreement" * 200,
+                    "prompt_cache_breakpoint": {"mode": "explicit", "ttl": "1h"},
+                },
+                {"role": "user", "content": "Reply with the single word ok."},
+            ],
+            None,
+            PROMPT_CACHE_ONE_HOUR_PIN_TTL_SECONDS,
+            id="1h-breakpoint",
+        ),
         pytest.param(
             _system_block_messages({"type": "ephemeral"}),
             [_tool({"type": "ephemeral", "ttl": "1h"})],

tests/unit/test_main.py

@@ -621,6 +621,57 @@ def test_completion_strips_an_unanswered_claude_tool_search_call_only_for_routes
     assert called_names(json.loads(route.calls.last.request.content)) == sent_names
 
 
+_CONVERSE_ROUTE: Final = r"https://bedrock-runtime\.us-east-1\.amazonaws\.com/model/.*/converse"
+_CONVERSE_REPLY: Final = {
+    "output": {"message": {"role": "assistant", "content": [{"text": "18C"}]}},
+    "stopReason": "end_turn",
+    "usage": {"inputTokens": 1, "outputTokens": 1, "totalTokens": 2},
+}
+_FAKE_AWS: Final = {"aws_access_key_id": "fake", "aws_secret_access_key": "fake", "aws_region_name": "us-east-1"}
+_BREAKPOINT: Final = {"mode": "explicit"}
+
+
+def _converse_cache_points(body: Mapping[str, object]) -> tuple[str, ...]:
+    system: Final = body.get("system")
+    messages: Final = body["messages"]
+    assert isinstance(system, list) and isinstance(messages, list)
+    in_messages: Final = tuple(block for message in messages for block in message["content"])
+    return (
+        *("system" for block in system if "cachePoint" in block),
+        *("message" for block in in_messages if "cachePoint" in block),
+    )
+
+
+def test_a_breakpoint_on_an_unanswered_tool_search_call_takes_no_slot_from_the_four_under_drop_params(
+    respx_mock: respx.MockRouter,
+):
+    route: Final = respx_mock.post(url__regex=_CONVERSE_ROUTE).respond(json=_CONVERSE_REPLY)
+
+    litellm.completion(
+        model="bedrock/converse/us.anthropic.claude-opus-5-5",
+        messages=[
+            {"role": "system", "content": [{"type": "text", "text": "sys", "prompt_cache_breakpoint": _BREAKPOINT}]},
+            {
+                "role": "user",
+                "content": [
+                    {"type": "text", "text": str(index), "prompt_cache_breakpoint": _BREAKPOINT} for index in range(3)
+                ],
+            },
+            {
+                "role": "assistant",
+                "content": None,
+                "tool_calls": [{**_CLAUDE_TOOL_SEARCH_CALL, "prompt_cache_breakpoint": _BREAKPOINT}],
+            },
+            {"role": "user", "content": "And tomorrow?"},
+        ],
+        tools=[{"type": "function", "function": {"name": "get_weather", "parameters": {"type": "object"}}}],
+        drop_params=True,
+        **_FAKE_AWS,
+    )
+
+    assert _converse_cache_points(json.loads(route.calls.last.request.content)) == ("system", *("message",) * 3)
+
+
 def test_completion_keeps_a_function_call_whose_id_only_looks_like_a_claude_server_tool(
     respx_mock: respx.MockRouter, openai_api_response
 ):