fix(cache): count replayed server tool results and extra_body messages in the developer TTL check

A server tool result replayed from provider_specific_fields reaches Anthropic as is, and extra_body messages replace the request's messages while the developer prompt stays in system. So a 5m developer mark beside a 1h mark in either place made a request Anthropic rejects. The check now counts both

litellm/integrations/anthropic_cache_control_hook.py

@@ -146,6 +146,7 @@ def _validated_object_list(value: object) -> list[object] | None:
 _MessageT: Final = TypeVar("_MessageT")
 _CacheControlRewrite: TypeAlias = Callable[[object], ChatCompletionCachedContent | None]
 _MARK_CARRYING_MESSAGE_FIELDS: Final = ("content", "tool_calls")
+_REPLAYED_SERVER_TOOL_RESULT_FIELDS: Final = ("web_search_results", "tool_results")
 
 
 def _mark_carrying_blocks(fields: Mapping[str, object]) -> Sequence[object]:
@@ -243,12 +244,23 @@ def _each_cache_control(
     return tuple(itertools.chain.from_iterable(map(cache_controls_of, as_object_list(values) or ())))
 
 
+def _replayed_server_tool_results(fields: Mapping[str, object]) -> tuple[object, ...]:
+    provider_fields: Final = as_str_mapping(fields.get("provider_specific_fields")) or MappingProxyType({})
+    return tuple(
+        itertools.chain.from_iterable(
+            as_object_list(provider_fields.get(key)) or () for key in _REPLAYED_SERVER_TOOL_RESULT_FIELDS
+        )
+    )
+
+
 def _message_cache_controls(message: object) -> tuple[object, ...]:
     fields: Final = as_str_mapping(message)
     if fields is None:
         return ()
     return _own_cache_control(fields) + tuple(
-        itertools.chain.from_iterable(map(_own_cache_control, _mark_carrying_blocks(fields)))
+        itertools.chain.from_iterable(
+            map(_own_cache_control, (*_mark_carrying_blocks(fields), *_replayed_server_tool_results(fields)))
+        )
     )
 
 
@@ -260,8 +272,12 @@ def _tool_cache_controls(tool: object) -> tuple[object, ...]:
 def _request_cache_controls(fields: Mapping[str, object]) -> tuple[object, ...]:
     extra_body: Final = as_str_mapping(fields.get("extra_body")) or MappingProxyType({})
     top_level: Final = extra_body.get("cache_control", fields.get("cache_control"))
-    return (() if top_level is None else (top_level,)) + _each_cache_control(
-        _tool_cache_controls, extra_body["tools"] if "tools" in extra_body else fields.get("tools")
+    return (
+        (() if top_level is None else (top_level,))
+        + _each_cache_control(
+            _tool_cache_controls, extra_body["tools"] if "tools" in extra_body else fields.get("tools")
+        )
+        + _each_cache_control(_message_cache_controls, extra_body.get("messages"))
     )
 
 

tests/unit/integrations/test_anthropic_cache_control_hook.py

@@ -6276,6 +6276,65 @@ def test_one_hour_developer_mark_follows_the_ttls_of_the_marks_sent_outside_the_
     assert list(kept) == [marked_developer if keeps_the_mark else developer, question]
 
 
+def _server_search_replaying(results_field: str) -> dict[str, object]:
+    return {
+        "role": "assistant",
+        "content": None,
+        "tool_calls": [
+            {
+                "id": "srvtoolu_backup_rules",
+                "type": "function",
+                "function": {"name": "web_search", "arguments": '{"query": "backup rules"}'},
+            }
+        ],
+        "provider_specific_fields": {
+            results_field: [
+                {
+                    "type": "web_search_tool_result",
+                    "tool_use_id": "srvtoolu_backup_rules",
+                    "content": [],
+                    "cache_control": dict(_ONE_HOUR),
+                }
+            ]
+        },
+    }
+
+
+@pytest.mark.parametrize(
+    ("earlier_turns", "outside_fields"),
+    [
+        pytest.param([_server_search_replaying("web_search_results")], {}, id="replayed_web_search_result"),
+        pytest.param([_server_search_replaying("tool_results")], {}, id="replayed_server_tool_result"),
+        pytest.param(
+            [],
+            {
+                "extra_body": {
+                    "messages": [
+                        {
+                            "role": "user",
+                            "content": [{"type": "text", "text": "Which rule covers backups?", "cache_control": dict(_ONE_HOUR)}],
+                        }
+                    ]
+                }
+            },
+            id="envelope_messages",
+        ),
+    ],
+)
+def test_five_minute_developer_mark_is_dropped_before_a_one_hour_mark_the_provider_receives(
+    earlier_turns: list[dict[str, object]], outside_fields: dict[str, object]
+) -> None:
+    developer: Final = {"role": "developer", "content": _DEVELOPER_PREFIX}
+    turns: Final = [*earlier_turns, {"role": "user", "content": "Which rule covers backups?"}]
+
+    kept: Final = developer_messages_within_the_cache_limits(
+        cast(list[AllMessageValues], [{**developer, "cache_control": dict(_FIVE_MINUTES)}, *turns]),
+        MappingProxyType(outside_fields),
+    )
+
+    assert list(kept) == [developer, *turns]
+
+
 def test_system_injection_point_marks_a_developer_prompt_for_openai_prompt_cache_breakpoints(
     respx_mock: respx.MockRouter,
 ) -> None: