scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-24 00:52:24 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 91

Merge pull request #41956 from BerriAI/litellm_explicit_cache_injection_points_survive_client_marks

Browse source 
fix: apply configured cache_control_injection_points beside client cache_control marks
This commit is contained in:
Mateo Wang 2026-09-21 12:28:02 -07:00 • committed by GitHub
commit d8267d507d
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
4 changed files with 543 additions and 225 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

320
litellm/integrations/anthropic_cache_control_hook.py
View file

@ -36,6 +36,7 @@ from litellm.types.integrations.anthropic_cache_control_hook import (
CacheControlMessageInjectionPoint,
)
from litellm.types.llms.anthropic import (
ANTHROPIC_TOOL_SEARCH_TOOL_TYPES,
AllAnthropicToolsValues,
AnthropicSystemMessageContent,
)
@ -124,6 +125,16 @@ def _carries_cache_breakpoint(block: object) -> bool:
return isinstance(block, dict) and any(block.get(key) is not None for key in CACHE_BREAKPOINT_KEYS)
def _tool_carries_cache_breakpoint(tool: object) -> bool:
return _carries_cache_breakpoint(tool) or (
isinstance(tool, dict) and _carries_cache_breakpoint(tool.get("function"))
)
def _chat_transform_drops_tool_cache_control(tool: object) -> bool:
return isinstance(tool, dict) and tool.get("type") in ANTHROPIC_TOOL_SEARCH_TOOL_TYPES
def _accepts_prompt_cache_breakpoint(block: object) -> bool:
return isinstance(block, dict) and block.get("type") in OPENAI_PROMPT_CACHE_BREAKPOINT_BLOCK_TYPES
@ -134,6 +145,8 @@ def _accepts_prompt_cache_breakpoint(block: object) -> bool:
# rather than spending them on a list that is still missing some of their targets.
CARRY_UNMATCHED_MESSAGE_POINTS: Final = "_litellm_carry_unmatched_cache_control_points"
EXTERNAL_BREAKPOINTS_STAMP: Final = "_litellm_external_breakpoints"
class AnthropicCacheControlHook(CustomPromptManagement):
@staticmethod
@ -199,19 +212,13 @@ class AnthropicCacheControlHook(CustomPromptManagement):
# Create a deep copy of messages to avoid modifying the original list
processed_messages = copy.deepcopy(messages)
# Separate message-level and non-message-level injection points
message_points: Final[list[CacheControlMessageInjectionPoint]] = []
remaining_points: Final[list[CacheControlInjectionPoint]] = []
for point in injection_points:
if point.get("location") == "message":
message_points.append(cast(CacheControlMessageInjectionPoint, point))
else:
remaining_points.append(point)
message_points: Final = tuple(
cast(CacheControlMessageInjectionPoint, point)
for point in injection_points
if point.get("location") == "message"
)
remaining_points: Final = tuple(point for point in injection_points if point.get("location") != "message")
# Non-message points (currently Bedrock tool_config) are handled in the
# provider transform, where each tool_config point appends at most one
# cachePoint to the tools. That block also counts toward Anthropic's
# limit, so reserve a slot for it here to leave room.
stamped_dialect: Final = injection_points[0].get("_litellm_openai_dialect")
openai_dialect: Final = (
stamped_dialect
@ -236,8 +243,10 @@ class AnthropicCacheControlHook(CustomPromptManagement):
if carry_unmatched
else tuple(message_points)
)
reserved_blocks: Final = (
1 if not openai_dialect and any(p.get("location") == "tool_config" for p in remaining_points) else 0
stamped_external: Final = injection_points[0].get(EXTERNAL_BREAKPOINTS_STAMP)
external_breakpoints: Final = stamped_external if isinstance(stamped_external, int) else 0
reserved_blocks: Final = AnthropicCacheControlHook._blocks_reserved_outside_messages(
remaining_points, external_breakpoints, openai_dialect
)
breakpoints_before: Final = AnthropicCacheControlHook.count_request_cache_breakpoints(processed_messages)
processed_messages = self._apply_message_injections(
@ -254,14 +263,19 @@ class AnthropicCacheControlHook(CustomPromptManagement):
# Points this pass did not place: non-message ones for the provider transform, and
# the deferred role-targeted ones. Deferring is what reaches the Responses API's
# `instructions`, which is only a system message once the bridge builds one. The
# judged stamp is what makes it safe: the next pass must not re-judge points
# against messages this pass already marked (see `_should_stand_down`).
carried_points: Final[Sequence[CacheControlInjectionPoint]] = (*remaining_points, *carried_message_points)
# `instructions`, which is only a system message once the bridge builds one. A later
# pass re-applies them safely: a target that already carries a mark is skipped and
# the census counts every mark on the wire, litellm's own included.
carried_points: Final[Sequence[CacheControlInjectionPoint]] = (
*AnthropicCacheControlHook._points_with_a_slot_left(
remaining_points,
AnthropicCacheControlHook.count_request_cache_breakpoints(processed_messages) + external_breakpoints,
openai_dialect,
),
*carried_message_points,
)
if carried_points:
non_default_params["cache_control_injection_points"] = AnthropicCacheControlHook._stamped_as_judged(
carried_points
)
non_default_params["cache_control_injection_points"] = list(carried_points)
return model, processed_messages, non_default_params
@ -296,6 +310,72 @@ class AnthropicCacheControlHook(CustomPromptManagement):
)
return system_blocks + sum(AnthropicCacheControlHook._count_cache_control_blocks(msg) for msg in messages)
@staticmethod
def count_external_cache_breakpoints(
tools: Iterable[object] | None, cache_control: object = None, request_kwargs: object = None
) -> int:
"""Client breakpoints outside messages and system that the provider cap still counts.
A tool carries its mark at the top level (Anthropic shape) or under ``function``
(OpenAI shape). A top-level ``cache_control`` is Anthropic's automatic caching,
which places one breakpoint of its own on top of the explicit ones. The
``extra_body`` envelope of ``request_kwargs`` is merged over the request on the
wire, so a ``tools`` or ``cache_control`` it carries replaces the direct value
and is counted in its place. Callers pass only the tools whose mark reaches the
provider on their path.
"""
extra_body: Final = (
_validated_object_mapping(AnthropicCacheControlHook._request_value(request_kwargs, "extra_body")) or {}
)
wire_cache_control: Final = extra_body.get("cache_control", cache_control)
wire_tools: Final = _validated_object_list(extra_body["tools"]) if "tools" in extra_body else tools
tool_blocks: Final = sum(1 for tool in wire_tools or () if _tool_carries_cache_breakpoint(tool))
envelope_blocks: Final = AnthropicCacheControlHook.count_request_cache_breakpoints(
_validated_object_list(extra_body.get("messages")) or (), extra_body.get("system")
)
return int(wire_cache_control is not None) + tool_blocks + envelope_blocks
@staticmethod
def count_external_cache_breakpoints_on_messages_route(
tools: Iterable[object] | None, cache_control: object, request_kwargs: object
) -> int:
"""The /v1/messages census before the route splits.
The native messages transforms drop the ``extra_body`` envelope while the
chat bridge merges it, so the cap reserves for whichever census is larger
rather than letting an envelope that unmarks a direct tool free a slot the
provider still counts.
"""
return max(
AnthropicCacheControlHook.count_external_cache_breakpoints(tools, cache_control),
AnthropicCacheControlHook.count_external_cache_breakpoints(tools, cache_control, request_kwargs),
)
@staticmethod
def _blocks_reserved_outside_messages(
remaining_points: Sequence[CacheControlInjectionPoint], external_breakpoints: int, openai_dialect: bool
) -> int:
"""Slots of the provider cap that the message census cannot see.
The client's breakpoints on tools and its automatic top-level ``cache_control``
are already on the wire, and a ``tool_config`` point becomes one more cachePoint
in the Bedrock converse transform. OpenAI's cap counts only its own block markers.
"""
if openai_dialect:
return 0
tool_config_blocks: Final = 1 if any(p.get("location") == "tool_config" for p in remaining_points) else 0
return external_breakpoints + tool_config_blocks
@staticmethod
def _points_with_a_slot_left(
remaining_points: Sequence[CacheControlInjectionPoint], breakpoints_on_wire: int, openai_dialect: bool
) -> tuple[CacheControlInjectionPoint, ...]:
"""A ``tool_config`` point becomes a cachePoint the Bedrock converse transform never
counts against the cap, so it is forwarded only while the wire still has a slot."""
if openai_dialect or breakpoints_on_wire < MAX_CACHE_CONTROL_BLOCKS:
return tuple(remaining_points)
return tuple(point for point in remaining_points if point.get("location") != "tool_config")
@staticmethod
def _apply_message_injections(
points: Sequence[CacheControlMessageInjectionPoint],
@ -476,11 +556,16 @@ class AnthropicCacheControlHook(CustomPromptManagement):
def apply_to_anthropic_messages_request(
messages: list[dict],
system: str | list | None,
injection_points: list[CacheControlInjectionPoint],
injection_points: Sequence[CacheControlInjectionPoint],
openai_dialect: bool = False,
external_breakpoints: int = 0,
) -> tuple[list[dict], str | list | None, list[CacheControlInjectionPoint]]:
"""Apply cache control injection for the Anthropic-native v1/messages endpoint.
``external_breakpoints`` is the client's breakpoint count outside ``messages`` and
``system`` (see ``count_external_cache_breakpoints``); it shrinks the budget so
the request never exceeds the provider cap.
Returns (messages, system, remaining_non_message_points).
"""
if not injection_points:
@ -489,22 +574,17 @@ class AnthropicCacheControlHook(CustomPromptManagement):
processed_messages: list[dict] = copy.deepcopy(messages)
processed_system = copy.deepcopy(system) if system is not None else None
message_points: Final[list[CacheControlMessageInjectionPoint]] = []
system_points: Final[list[CacheControlMessageInjectionPoint]] = []
remaining_points: Final[list[CacheControlInjectionPoint]] = []
role_points: Final = tuple(
cast(CacheControlMessageInjectionPoint, point)
for point in injection_points
if point.get("location") == "message"
)
system_points: Final = tuple(point for point in role_points if point.get("role") == "system")
message_points: Final = tuple(point for point in role_points if point.get("role") != "system")
remaining_points: Final = tuple(point for point in injection_points if point.get("location") != "message")
for point in injection_points:
if point.get("location") == "message":
msg_point = cast(CacheControlMessageInjectionPoint, point)
if msg_point.get("role") == "system":
system_points.append(msg_point)
else:
message_points.append(msg_point)
else:
remaining_points.append(point)
reserved_blocks: Final = (
1 if not openai_dialect and any(p.get("location") == "tool_config" for p in remaining_points) else 0
reserved_blocks: Final = AnthropicCacheControlHook._blocks_reserved_outside_messages(
remaining_points, external_breakpoints, openai_dialect
)
max_blocks: Final = MAX_CACHE_CONTROL_BLOCKS - reserved_blocks
@ -541,8 +621,14 @@ class AnthropicCacheControlHook(CustomPromptManagement):
max_blocks=max_blocks - system_blocks,
openai_dialect=openai_dialect,
)
forwarded_points: Final = AnthropicCacheControlHook._points_with_a_slot_left(
remaining_points,
AnthropicCacheControlHook.count_request_cache_breakpoints(processed_messages, processed_system)
+ external_breakpoints,
openai_dialect,
)
return processed_messages, processed_system, remaining_points
return processed_messages, processed_system, list(forwarded_points)
@staticmethod
def _default_control() -> ChatCompletionCachedContent:
@ -559,31 +645,26 @@ class AnthropicCacheControlHook(CustomPromptManagement):
return ChatCompletionCachedContent(type="ephemeral")
@staticmethod
def _stamped_as_judged(points: Sequence[CacheControlInjectionPoint]) -> Sequence[Mapping[str, object]]:
"""Mark written-back points as having passed the client cache_control judgment.
Builds copies because config-owned point dicts are shared across
requests; mutating them would leak the stamp into future requests.
"""
return AnthropicCacheControlHook._stamped(points, "_litellm_judged", True)
@staticmethod
def _judged_configured_points(
def _stamped_for_prompt_hook(
points: Sequence[CacheControlInjectionPoint],
messages: list[AllMessageValues],
tools: list[object] | None,
cache_control: object,
external_breakpoints: int,
model: str,
custom_llm_provider: str | None,
api_base: object,
prompt_cache_options: object,
request_kwargs: object,
) -> Sequence[Mapping[str, object]] | None:
if AnthropicCacheControlHook._should_stand_down(points, messages, None, tools, cache_control, request_kwargs):
return None
return AnthropicCacheControlHook._stamped_with_dialect(
) -> Sequence[Mapping[str, object]]:
"""Carry onto the points what the prompt-management hook never receives.
The hook sees neither the tools nor the request kwargs, so the target dialect
and the client's breakpoint count outside the message list ride on the points.
Builds copies because config-owned point dicts are shared across requests.
"""
with_dialect: Final = AnthropicCacheControlHook._stamped_with_dialect(
points, model, custom_llm_provider, api_base, prompt_cache_options
)
if external_breakpoints == 0:
return with_dialect
return AnthropicCacheControlHook._stamped(with_dialect, EXTERNAL_BREAKPOINTS_STAMP, external_breakpoints)
@staticmethod
def _stamped_with_dialect(
@ -604,35 +685,9 @@ class AnthropicCacheControlHook(CustomPromptManagement):
)
@staticmethod
def _stamped(
points: Sequence[CacheControlInjectionPoint], key: str, value: object
) -> Sequence[Mapping[str, object]]:
def _stamped(points: Sequence[Mapping[str, object]], key: str, value: object) -> Sequence[Mapping[str, object]]:
return [{**point, key: value} for point in points]
@staticmethod
def _should_stand_down(
points: Sequence[CacheControlInjectionPoint],
messages: list[AllMessageValues],
system: str | list | None,
tools: list | None,
cache_control: object = None,
request_kwargs: object = None,
) -> bool:
"""Whether configured injection points must yield to client-set cache_control.
Points that a prior pass over this request already judged and wrote
back carry the internal judged stamp; any re-entry (acompletion
re-entering completion, the async-to-sync /v1/messages dispatch,
interceptor sub-calls reusing the request kwargs) must not re-judge
them, because by then the messages carry litellm's own injected marks
and the judgment would misread those as client breakpoints.
"""
if all(point.get("_litellm_judged") for point in points):
return False
return AnthropicCacheControlHook._request_has_cache_control(
messages, system, tools, cache_control, request_kwargs
)
@staticmethod
def _request_has_cache_control(
messages: list[AllMessageValues],
@ -641,27 +696,18 @@ class AnthropicCacheControlHook(CustomPromptManagement):
cache_control: object = None,
request_kwargs: object = None,
) -> bool:
"""Client breakpoints own caching in both the request and its extra_body envelope."""
bodies: Final = (
{"messages": messages, "system": system, "tools": tools, "cache_control": cache_control},
_validated_object_mapping(AnthropicCacheControlHook._request_value(request_kwargs, "extra_body")) or {},
)
return any(
body.get("cache_control") is not None
or AnthropicCacheControlHook.count_request_cache_breakpoints(
_validated_object_list(body.get("messages")) or (), body.get("system")
)
> 0
or any(
AnthropicCacheControlHook._request_value(tool, "cache_control") is not None
or AnthropicCacheControlHook._request_value(
AnthropicCacheControlHook._request_value(tool, "function"), "cache_control"
)
is not None
for tool in (_validated_object_list(body.get("tools")) or ())
)
for body in bodies
)
"""Return True if the request already carries any client-supplied cache_control.
Only the automatic defaults stand down on it: a client that marks its own
breakpoints (Claude Code does) has a caching strategy the defaults would
clash with, whether the marks sit in the request or in its ``extra_body``
envelope. Configured injection points are an explicit instruction and are
applied alongside the client's marks, bounded by the provider cap.
"""
return (
AnthropicCacheControlHook.count_request_cache_breakpoints(messages, system)
+ AnthropicCacheControlHook.count_external_cache_breakpoints(tools, cache_control, request_kwargs)
) > 0
@staticmethod
def get_default_injection_points(
@ -769,34 +815,30 @@ class AnthropicCacheControlHook(CustomPromptManagement):
) -> None:
"""For /chat/completions: resolve the injection points the request should carry.
Configured injection points win over the automatic defaults, but stand
down entirely when the client already marked its own cache_control
breakpoints (messages or tools): injecting alongside them clashes with
the client's caching strategy and can exceed the provider's four-block
limit. The judgment happens once per request; points a prior pass
wrote back carry the judged stamp and are never re-judged (see
``_should_stand_down``). Seeding the param lets the existing
prompt-management gate and the AnthropicCacheControlHook run
unchanged.
Configured injection points win over the automatic defaults and are applied
even when the client marked its own cache_control elsewhere in the request;
the provider's four-block cap bounds them, counting the client's marks on
messages, tools and the top-level ``cache_control``. Only the defaults stand
down on client marks. Seeding the param lets the existing prompt-management
gate and the AnthropicCacheControlHook run unchanged.
"""
import litellm
if non_default_params.get("cache_control_injection_points"):
judged: Final = AnthropicCacheControlHook._judged_configured_points(
non_default_params["cache_control_injection_points"],
messages,
tools,
non_default_params.get("cache_control"),
configured: Final = non_default_params.get("cache_control_injection_points")
if configured:
tools_keeping_marks: Final = tuple(
tool for tool in tools or () if not _chat_transform_drops_tool_cache_control(tool)
)
non_default_params["cache_control_injection_points"] = AnthropicCacheControlHook._stamped_for_prompt_hook(
configured,
AnthropicCacheControlHook.count_external_cache_breakpoints(
tools_keeping_marks, non_default_params.get("cache_control"), non_default_params
),
model,
custom_llm_provider,
api_base,
non_default_params.get("prompt_cache_options"),
non_default_params,
)
if judged is None:
non_default_params.pop("cache_control_injection_points")
else:
non_default_params["cache_control_injection_points"] = judged
return
points: Final = AnthropicCacheControlHook.get_default_injection_points(
messages=messages,
@ -897,15 +939,14 @@ class AnthropicCacheControlHook(CustomPromptManagement):
) -> tuple[list[dict], str | list | None]:
"""Extract cache_control_injection_points from kwargs and apply if present.
Configured points stand down entirely when the client already marked
its own cache_control breakpoints anywhere in the request. The
judgment happens once per request; points a prior pass wrote back
carry the judged stamp and are never re-judged (see
``_should_stand_down``). When none are configured but
Configured points are applied even when the client marked its own
cache_control elsewhere in the request, bounded by the provider cap,
which counts the client's marks on messages, system, tools and the
top-level ``cache_control``. When none are configured but
``litellm.enable_anthropic_prompt_caching`` or the per-request
``enable_prompt_caching`` kwarg (stamped from key metadata) is on,
synthesize default breakpoints for the native /v1/messages path. Pops
both keys from kwargs;
synthesize default breakpoints for the native /v1/messages path; those
defaults alone stand down on client marks. Pops both keys from kwargs;
if remaining (non-message) points exist they are written back so
downstream transforms can handle them.
"""
@ -917,13 +958,8 @@ class AnthropicCacheControlHook(CustomPromptManagement):
configured: Final = cast( # cast-ok: kwargs is untyped; this key only holds the documented injection-point list
list[CacheControlInjectionPoint] | None, kwargs.pop("cache_control_injection_points", None)
)
if configured and AnthropicCacheControlHook._should_stand_down(
configured, typed_messages, system, tools, cache_control, kwargs
):
return messages, system
injection_points: list[CacheControlInjectionPoint] = configured or []
if not injection_points and model is not None:
injection_points = AnthropicCacheControlHook.get_default_injection_points(
injection_points: Final[Sequence[CacheControlInjectionPoint]] = configured or (
AnthropicCacheControlHook.get_default_injection_points(
messages=typed_messages,
system=system,
tools=tools,
@ -933,6 +969,9 @@ class AnthropicCacheControlHook(CustomPromptManagement):
cache_control=cache_control,
request_kwargs=kwargs,
)
if model is not None
else ()
)
if not injection_points:
return messages, system
@ -945,6 +984,9 @@ class AnthropicCacheControlHook(CustomPromptManagement):
system=system,
injection_points=injection_points,
openai_dialect=openai_dialect,
external_breakpoints=AnthropicCacheControlHook.count_external_cache_breakpoints_on_messages_route(
tools, cache_control, kwargs
),
)
breakpoints_added: Final = (
AnthropicCacheControlHook.count_request_cache_breakpoints(messages, system) - breakpoints_before
@ -953,7 +995,7 @@ class AnthropicCacheControlHook(CustomPromptManagement):
if openai_dialect and breakpoints_added > 0:
kwargs.setdefault("prompt_cache_options", PromptCacheOptions(mode="explicit"))
if remaining:
kwargs["cache_control_injection_points"] = AnthropicCacheControlHook._stamped_as_judged(remaining)
kwargs["cache_control_injection_points"] = remaining
return messages, system
@property

4
litellm/types/integrations/anthropic_cache_control_hook.py
View file

@ -17,8 +17,8 @@ class CacheControlMessageInjectionPoint(TypedDict):
role: Literal["user", "system", "assistant"] | None # Optional: target by role (user, system, assistant)
index: int | str | None # Optional: target by specific index
control: ChatCompletionCachedContent | None
_litellm_judged: NotRequired[bool] # Internal: written back by litellm once the client cache_control judgment ran
_litellm_openai_dialect: NotRequired[ReadOnly[bool]]
_litellm_external_breakpoints: NotRequired[ReadOnly[int]]
class CacheControlToolConfigInjectionPoint(TypedDict):
@ -26,8 +26,8 @@ class CacheControlToolConfigInjectionPoint(TypedDict):
location: Literal["tool_config"]
control: ChatCompletionCachedContent | None
_litellm_judged: NotRequired[bool] # Internal: written back by litellm once the client cache_control judgment ran
_litellm_openai_dialect: NotRequired[ReadOnly[bool]]
_litellm_external_breakpoints: NotRequired[ReadOnly[int]]
CacheControlInjectionPoint = CacheControlMessageInjectionPoint | CacheControlToolConfigInjectionPoint

4
litellm/types/llms/anthropic.py
View file

@ -756,6 +756,10 @@ class ANTHROPIC_BETA_HEADER_VALUES(str, Enum):
# Tool search beta header constant (for Anthropic direct API and Microsoft Foundry)
ANTHROPIC_TOOL_SEARCH_BETA_HEADER: Final = "advanced-tool-use-2025-11-20"
ANTHROPIC_TOOL_SEARCH_TOOL_TYPES: Final = frozenset(
{"tool_search_tool_regex_20251119", "tool_search_tool_bm25_20251119"}
)
# Effort beta header constant
ANTHROPIC_EFFORT_BETA_HEADER: Final = "effort-2025-11-24"

440
tests/test_litellm/integrations/test_anthropic_cache_control_hook.py
View file

@ -4,10 +4,11 @@ import os
import subprocess
import sys
import textwrap
from typing import List, Optional, Tuple
from typing import Final, List, Optional, Tuple
from unittest.mock import MagicMock, patch
import pytest
from pydantic import BaseModel, ConfigDict
import litellm
from litellm.integrations.anthropic_cache_control_hook import (
@ -1276,11 +1277,7 @@ def test_cache_control_hook_reserves_slot_for_tool_config_point():
)
assert _count_cache_control(processed) == 3
# The tool_config point is passed through for the provider transform,
# stamped so re-entries never re-judge it against litellm's own marks.
assert non_default_params["cache_control_injection_points"] == [
{"location": "tool_config", "_litellm_judged": True}
]
assert non_default_params["cache_control_injection_points"] == [{"location": "tool_config"}]
@pytest.mark.asyncio
@ -1338,18 +1335,8 @@ async def test_cache_control_hook_bedrock_payload_caps_with_tool_config_point(mo
client=client,
)
request_body = json.loads(mock_post.call_args.kwargs["data"])
cache_points = sum(
1 for block in request_body.get("system", []) if isinstance(block, dict) and "cachePoint" in block
)
for msg in request_body.get("messages", []):
content = msg.get("content", [])
if isinstance(content, list):
cache_points += sum(1 for block in content if isinstance(block, dict) and "cachePoint" in block)
for tool in request_body.get("toolConfig", {}).get("tools", []):
if isinstance(tool, dict) and "cachePoint" in tool:
cache_points += 1
request_body = _ConverseBody.model_validate_json(mock_post.call_args.kwargs["data"])
cache_points = _count_converse_cache_points(request_body)
assert cache_points <= 4, (
f"Bedrock payload exceeded Anthropic's 4 cache_control block limit "
@ -1357,6 +1344,97 @@ async def test_cache_control_hook_bedrock_payload_caps_with_tool_config_point(mo
)
class _ConverseMessage(BaseModel):
model_config = ConfigDict(frozen=True)
content: tuple[dict[str, object], ...] = ()
class _ConverseToolConfig(BaseModel):
model_config = ConfigDict(frozen=True)
tools: tuple[dict[str, object], ...] = ()
class _ConverseBody(BaseModel):
model_config = ConfigDict(frozen=True)
system: tuple[dict[str, object], ...] = ()
messages: tuple[_ConverseMessage, ...] = ()
toolConfig: _ConverseToolConfig = _ConverseToolConfig()
def _count_converse_cache_points(request_body: _ConverseBody) -> int:
blocks: Final = (
*request_body.system,
*(block for message in request_body.messages for block in message.content),
*request_body.toolConfig.tools,
)
return sum(1 for block in blocks if "cachePoint" in block)
@pytest.mark.asyncio
async def test_cache_control_hook_bedrock_tool_config_point_stands_down_when_client_marks_fill_the_cap(
monkeypatch: pytest.MonkeyPatch,
):
with patch.dict(
os.environ,
{
"AWS_ACCESS_KEY_ID": "fake_access_key_id",
"AWS_SECRET_ACCESS_KEY": "fake_secret_access_key",
"AWS_REGION_NAME": "us-east-1",
},
):
monkeypatch.setattr(litellm, "callbacks", [AnthropicCacheControlHook()])
mock_response = MagicMock()
mock_response.json.return_value = {
"output": {"message": {"role": "assistant", "content": "ok"}},
"stopReason": "end_turn",
"usage": {"inputTokens": 100, "outputTokens": 4, "totalTokens": 104},
}
mock_response.status_code = 200
client = AsyncHTTPHandler()
with patch.object(client, "post", return_value=mock_response) as mock_post:
marked = {"type": "ephemeral"}
messages = [
{"role": "system", "content": [{"type": "text", "text": "sys", "cache_control": marked}]},
*(
{"role": "user", "content": [{"type": "text", "text": f"turn {i}", "cache_control": marked}]}
for i in range(3)
),
{"role": "user", "content": "What is the weather?"},
]
await litellm.acompletion(
model="bedrock/us.anthropic.claude-opus-4-6-v1:0",
messages=messages,
max_tokens=32,
tools=[
{
"type": "function",
"function": {
"name": "get_weather",
"description": "Get weather for a location",
"parameters": {
"type": "object",
"properties": {"location": {"type": "string"}},
"required": ["location"],
},
},
}
],
cache_control_injection_points=[{"location": "tool_config"}],
client=client,
)
request_body = _ConverseBody.model_validate_json(mock_post.call_args.kwargs["data"])
assert _count_converse_cache_points(request_body) == 4
assert not any("cachePoint" in tool for tool in request_body.toolConfig.tools)
class TestApplyToAnthropicMessagesRequest:
"""Tests for apply_to_anthropic_messages_request (v1/messages cache control)."""
@ -1683,13 +1761,17 @@ class TestEnableAnthropicPromptCaching:
result_messages, result_system = AnthropicCacheControlHook.maybe_inject_cache_control(
messages, system, kwargs, model, provider, tools=tools,
)
if client_control != "none":
if client_control != "none" and not configured:
assert (result_messages, result_system, tools) == original
assert kwargs["metadata"] == {}
else:
assert kwargs["metadata"]["litellm_gateway_injected_cache"] == "selected-deployment"
assert sum(AnthropicCacheControlHook._count_cache_control_blocks(m) for m in result_messages) == 1
assert result_system[0]["cache_control"] == control
assert result_messages[-1]["content"][-1]["cache_control"] == control
assert tools == original[2]
assert (result_messages == original[0]) == (envelope == "request" and client_control == "message")
assert (result_system == original[1]) == (envelope == "request" and client_control == "system")
if provider == "vertex_ai":
wire = VertexAIAnthropicConfig().transform_request(
model=model, messages=[{"role": "system", "content": result_system}, *result_messages],
@ -1706,7 +1788,7 @@ class TestEnableAnthropicPromptCaching:
AnthropicCacheControlHook.maybe_seed_default_injection_points(
seeded, [{"role": "system", "content": original[1]}, *original[0]], model, provider, tools=tools,
)
assert bool(seeded.get("cache_control_injection_points")) == (client_control == "none")
assert bool(seeded.get("cache_control_injection_points")) == (client_control == "none" or configured)
@pytest.mark.asyncio
@pytest.mark.parametrize("asynchronous", [False, True])
@ -2257,13 +2339,11 @@ class TestPerKeyEnablePromptCaching:
assert result_msgs == messages
class TestConfiguredInjectionPointsStandDown:
"""Configured cache_control_injection_points must stand down entirely when the
client already set its own cache_control anywhere in the request (LIT-4582);
injecting alongside client breakpoints clashes with the client's caching
strategy and can push the request past Anthropic's four-block limit."""
class TestConfiguredInjectionPointsSurviveClientMarks:
CONFIGURED = [{"location": "message", "role": "system"}]
TAIL_POINT = [{"location": "message", "index": -1}]
TOOL_CONFIG_POINT = [{"location": "tool_config"}]
EPHEMERAL = {"type": "ephemeral"}
CLEAN_MESSAGES: List[AllMessageValues] = [
{"role": "system", "content": "sys"},
@ -2277,6 +2357,37 @@ class TestConfiguredInjectionPointsStandDown:
V1_MESSAGES = [{"role": "user", "content": [{"type": "text", "text": "hi"}]}]
MARKED_TOOL_TOP_LEVEL = {
"type": "function",
"function": {"name": "t", "parameters": {}},
"cache_control": {"type": "ephemeral"},
}
MARKED_TOOL_NESTED = {
"type": "function",
"function": {"name": "t", "parameters": {}, "cache_control": {"type": "ephemeral"}},
}
UNMARKED_TOOL = {"type": "function", "function": {"name": "t", "parameters": {}}}
MARKED_V1_TOOL = {"name": "t", "input_schema": {}, "cache_control": {"type": "ephemeral"}}
UNMARKED_V1_TOOL = {"name": "t", "input_schema": {}}
MARKED_SYSTEM = [{"type": "text", "text": "sys", "cache_control": EPHEMERAL}]
MARKED_TOOL_SEARCH_REGEX = {
"type": "tool_search_tool_regex_20251119",
"name": "tool_search",
"cache_control": {"type": "ephemeral"},
}
MARKED_TOOL_SEARCH_BM25 = {
"type": "tool_search_tool_bm25_20251119",
"name": "tool_search",
"cache_control": {"type": "ephemeral"},
}
@staticmethod
def _marked_user_turns(count: int) -> List[AllMessageValues]:
return [
{"role": "user", "content": [{"type": "text", "text": f"turn {i}", "cache_control": {"type": "ephemeral"}}]}
for i in range(count)
]
def _seed(self, params, messages, tools=None):
AnthropicCacheControlHook.maybe_seed_default_injection_points(
non_default_params=params,
@ -2286,6 +2397,17 @@ class TestConfiguredInjectionPointsStandDown:
tools=tools,
)
def _chat(self, params: dict[str, object], messages: List[AllMessageValues]) -> List[AllMessageValues]:
_, processed, _ = AnthropicCacheControlHook().get_chat_completion_prompt(
model="claude-sonnet-4-5",
messages=messages,
non_default_params=params,
prompt_id=None,
prompt_variables=None,
dynamic_callback_params={},
)
return processed
def _inject(self, messages, kwargs, system="sys", tools=None):
return AnthropicCacheControlHook.maybe_inject_cache_control(
messages,
@ -2296,23 +2418,79 @@ class TestConfiguredInjectionPointsStandDown:
tools=tools,
)
def test_configured_points_dropped_when_messages_carry_cache_control(self):
def test_chat_tail_point_applies_when_client_marked_the_system_block(self):
messages: List[AllMessageValues] = [
{"role": "system", "content": [{"type": "text", "text": "sys", "cache_control": {"type": "ephemeral"}}]},
{"role": "user", "content": "history"},
{"role": "assistant", "content": "reply"},
{"role": "user", "content": "question"},
]
params = {"cache_control_injection_points": copy.deepcopy(self.TAIL_POINT)}
self._seed(params, messages)
processed = self._chat(params, messages)
assert processed[0] == messages[0]
assert processed[-1] == {"role": "user", "content": "question", "cache_control": self.EPHEMERAL}
assert _count_cache_control(processed) == 2
def test_chat_configured_points_apply_when_messages_carry_cache_control(self):
params = {"cache_control_injection_points": copy.deepcopy(self.CONFIGURED)}
self._seed(params, copy.deepcopy(self.MARKED_MESSAGES))
assert "cache_control_injection_points" not in params
processed = self._chat(params, copy.deepcopy(self.MARKED_MESSAGES))
assert processed[0] == {"role": "system", "content": "sys", "cache_control": self.EPHEMERAL}
assert processed[1] == self.MARKED_MESSAGES[1]
@pytest.mark.parametrize(
"tool",
[
{"type": "function", "function": {"name": "t", "parameters": {}}, "cache_control": {"type": "ephemeral"}},
{"type": "function", "function": {"name": "t", "parameters": {}, "cache_control": {"type": "ephemeral"}}},
],
ids=["top_level", "nested_in_function"],
"tool", [MARKED_TOOL_TOP_LEVEL, MARKED_TOOL_NESTED], ids=["top_level", "nested_in_function"]
)
def test_configured_points_dropped_when_tools_carry_cache_control(self, tool):
def test_chat_configured_points_apply_when_tools_carry_cache_control(self, tool):
params = {"cache_control_injection_points": copy.deepcopy(self.CONFIGURED)}
self._seed(params, copy.deepcopy(self.CLEAN_MESSAGES), tools=[tool])
assert "cache_control_injection_points" not in params
processed = self._chat(params, copy.deepcopy(self.CLEAN_MESSAGES))
assert processed[0] == {"role": "system", "content": "sys", "cache_control": self.EPHEMERAL}
@pytest.mark.parametrize(
"tool,injected",
[(MARKED_TOOL_TOP_LEVEL, 0), (MARKED_TOOL_NESTED, 0), (UNMARKED_TOOL, 1)],
ids=["marked_top_level", "marked_nested_in_function", "unmarked"],
)
def test_chat_cap_counts_client_marked_tools(self, tool, injected):
messages = [{"role": "system", "content": "sys"}, *self._marked_user_turns(3)]
params = {"cache_control_injection_points": copy.deepcopy(self.CONFIGURED)}
self._seed(params, copy.deepcopy(messages), tools=[tool])
processed = self._chat(params, copy.deepcopy(messages))
assert _count_cache_control(processed) == 3 + injected
@pytest.mark.parametrize("tool", [MARKED_TOOL_SEARCH_REGEX, MARKED_TOOL_SEARCH_BM25], ids=["regex", "bm25"])
def test_chat_cap_ignores_marked_tool_search_tools(self, tool):
messages = [{"role": "system", "content": "sys"}, *self._marked_user_turns(3)]
params = {"cache_control_injection_points": copy.deepcopy(self.CONFIGURED)}
self._seed(params, copy.deepcopy(messages), tools=[tool])
processed = self._chat(params, copy.deepcopy(messages))
assert _count_cache_control(processed) == 4
@pytest.mark.parametrize("marked_turns,forwarded", [(3, ["tool_config"]), (4, [])], ids=["slot_left", "cap_full"])
def test_chat_forwards_tool_config_point_only_while_a_slot_is_left(self, marked_turns, forwarded):
messages = [{"role": "system", "content": "sys"}, *self._marked_user_turns(marked_turns)]
params = {"cache_control_injection_points": copy.deepcopy(self.TOOL_CONFIG_POINT)}
self._seed(params, copy.deepcopy(messages), tools=[self.UNMARKED_TOOL])
self._chat(params, copy.deepcopy(messages))
assert [p["location"] for p in params.get("cache_control_injection_points", [])] == forwarded
@pytest.mark.parametrize("marked_turns,forwarded", [(3, ["tool_config"]), (4, [])], ids=["slot_left", "cap_full"])
def test_v1_messages_forwards_tool_config_point_only_while_a_slot_is_left(self, marked_turns, forwarded):
kwargs = {"cache_control_injection_points": copy.deepcopy(self.TOOL_CONFIG_POINT)}
self._inject(self._marked_user_turns(marked_turns), kwargs, tools=[self.UNMARKED_V1_TOOL])
assert [p["location"] for p in kwargs.get("cache_control_injection_points", [])] == forwarded
@pytest.mark.parametrize("marked_turns,injected", [(2, 1), (3, 0)])
def test_chat_root_cache_control_reserves_a_slot(self, marked_turns, injected):
messages = [{"role": "system", "content": "sys"}, *self._marked_user_turns(marked_turns)]
root_cache_control = {"type": "ephemeral"}
params = {"cache_control_injection_points": copy.deepcopy(self.CONFIGURED), "cache_control": root_cache_control}
self._seed(params, copy.deepcopy(messages))
processed = self._chat(params, copy.deepcopy(messages))
assert _count_cache_control(processed) == marked_turns + injected
assert params["cache_control"] is root_cache_control
def test_configured_points_kept_when_request_is_unmarked(self):
configured = copy.deepcopy(self.CONFIGURED)
@ -2320,43 +2498,59 @@ class TestConfiguredInjectionPointsStandDown:
self._seed(params, copy.deepcopy(self.CLEAN_MESSAGES))
assert params["cache_control_injection_points"] is configured
def test_judged_remainder_survives_reentry_despite_injected_marks(self):
"""acompletion() re-enters completion() after injection ran, with only the
stamped non-message points written back; the re-entry must not misread
litellm's own marks as client ones and drop that remainder."""
remainder = [{"location": "tool_config", "_litellm_judged": True}]
params = {"cache_control_injection_points": remainder}
self._seed(params, copy.deepcopy(self.MARKED_MESSAGES))
assert params["cache_control_injection_points"] is remainder
def test_chat_reentry_over_injected_messages_adds_no_duplicate_marks(self):
points = [{"location": "message", "role": "system"}, {"location": "tool_config"}]
first_params = {"cache_control_injection_points": copy.deepcopy(points)}
self._seed(first_params, copy.deepcopy(self.MARKED_MESSAGES))
first = self._chat(first_params, copy.deepcopy(self.MARKED_MESSAGES))
assert _count_cache_control(first) == 2
assert first_params["cache_control_injection_points"] == [{"location": "tool_config"}]
def test_v1_messages_stand_down_when_content_block_marked(self):
second_params = {"cache_control_injection_points": copy.deepcopy(points)}
self._seed(second_params, copy.deepcopy(first))
second = self._chat(second_params, copy.deepcopy(first))
assert second == first
assert second_params["cache_control_injection_points"] == [{"location": "tool_config"}]
def test_v1_messages_configured_point_applies_when_content_block_marked(self):
messages = [
{"role": "user", "content": [{"type": "text", "text": "hi", "cache_control": {"type": "ephemeral"}}]}
]
kwargs = {"cache_control_injection_points": copy.deepcopy(self.CONFIGURED)}
result_msgs, result_sys = self._inject(copy.deepcopy(messages), kwargs)
assert result_msgs == messages
assert result_sys == "sys"
assert result_sys == [{"type": "text", "text": "sys", "cache_control": self.EPHEMERAL}]
assert "cache_control_injection_points" not in kwargs
def test_v1_messages_stand_down_when_system_block_marked(self):
"""A configured point targeting a message must not fire when the client
marked the system prompt; the old behavior injected into the message
because only the exact targeted position was guarded."""
def test_v1_messages_tail_point_applies_when_system_block_marked(self):
system = [{"type": "text", "text": "s", "cache_control": {"type": "ephemeral"}}]
kwargs = {"cache_control_injection_points": [{"location": "message", "role": "user"}]}
kwargs = {"cache_control_injection_points": copy.deepcopy(self.TAIL_POINT)}
result_msgs, result_sys = self._inject(copy.deepcopy(self.V1_MESSAGES), kwargs, system=system)
assert result_msgs == self.V1_MESSAGES
assert result_msgs == [
{"role": "user", "content": [{"type": "text", "text": "hi", "cache_control": self.EPHEMERAL}]}
]
assert result_sys == system
assert "cache_control_injection_points" not in kwargs
def test_v1_messages_stand_down_when_tools_marked(self):
tools = [{"name": "t", "input_schema": {}, "cache_control": {"type": "ephemeral"}}]
def test_v1_messages_configured_point_applies_when_tools_marked(self):
kwargs = {"cache_control_injection_points": copy.deepcopy(self.CONFIGURED)}
result_msgs, result_sys = self._inject(copy.deepcopy(self.V1_MESSAGES), kwargs, tools=tools)
result_msgs, result_sys = self._inject(copy.deepcopy(self.V1_MESSAGES), kwargs, tools=[self.MARKED_V1_TOOL])
assert result_msgs == self.V1_MESSAGES
assert result_sys == "sys"
assert "cache_control_injection_points" not in kwargs
assert result_sys == [{"type": "text", "text": "sys", "cache_control": self.EPHEMERAL}]
@pytest.mark.parametrize(
"tool,expected_system",
[
(MARKED_V1_TOOL, "sys"),
(MARKED_TOOL_SEARCH_REGEX, "sys"),
(MARKED_TOOL_SEARCH_BM25, "sys"),
(UNMARKED_V1_TOOL, [{"type": "text", "text": "sys", "cache_control": {"type": "ephemeral"}}]),
],
ids=["marked", "marked_tool_search_regex", "marked_tool_search_bm25", "unmarked"],
)
def test_v1_messages_cap_counts_client_marked_tools(self, tool, expected_system):
kwargs = {"cache_control_injection_points": copy.deepcopy(self.CONFIGURED)}
_, result_sys = self._inject(self._marked_user_turns(3), kwargs, tools=[tool])
assert result_sys == expected_system
def test_v1_messages_configured_points_apply_when_unmarked(self):
kwargs = {"cache_control_injection_points": copy.deepcopy(self.CONFIGURED)}
@ -2364,16 +2558,73 @@ class TestConfiguredInjectionPointsStandDown:
assert result_sys == [{"type": "text", "text": "sys", "cache_control": {"type": "ephemeral"}}]
@pytest.mark.parametrize(
"configured",
[None, CONFIGURED],
ids=["automatic_defaults", "configured_points"],
"extra_body,injected",
[
({"tools": [MARKED_TOOL_TOP_LEVEL]}, 0),
({"cache_control": {"type": "ephemeral"}}, 0),
({"tools": [UNMARKED_TOOL]}, 1),
],
ids=["marked_tool", "root_cache_control", "unmarked_tool"],
)
def test_v1_messages_stands_down_for_root_cache_control(self, monkeypatch, configured):
def test_chat_cap_counts_client_marks_sent_through_extra_body(self, extra_body, injected):
messages = [{"role": "system", "content": "sys"}, *self._marked_user_turns(3)]
params = {"cache_control_injection_points": copy.deepcopy(self.CONFIGURED), "extra_body": extra_body}
self._seed(params, copy.deepcopy(messages))
processed = self._chat(params, copy.deepcopy(messages))
assert _count_cache_control(processed) == 3 + injected
@pytest.mark.parametrize(
"extra_body,expected_system",
[
({"cache_control": {"type": "ephemeral"}}, "sys"),
({"tools": [MARKED_V1_TOOL]}, "sys"),
({"tools": [UNMARKED_V1_TOOL]}, [{"type": "text", "text": "sys", "cache_control": {"type": "ephemeral"}}]),
],
ids=["root_cache_control", "marked_tool", "unmarked_tool"],
)
def test_v1_messages_cap_counts_client_marks_sent_through_extra_body(self, extra_body, expected_system):
kwargs = {"cache_control_injection_points": copy.deepcopy(self.CONFIGURED), "extra_body": extra_body}
_, result_sys = self._inject(self._marked_user_turns(3), kwargs)
assert result_sys == expected_system
@pytest.mark.parametrize(
"params,tools,marked_turns,injected",
[
({"extra_body": {"tools": [MARKED_TOOL_TOP_LEVEL]}}, [MARKED_TOOL_TOP_LEVEL], 2, 1),
({"extra_body": {"tools": [UNMARKED_TOOL]}}, [MARKED_TOOL_TOP_LEVEL], 3, 1),
({"extra_body": {"tools": [MARKED_TOOL_TOP_LEVEL]}}, [UNMARKED_TOOL], 3, 0),
({"extra_body": {"cache_control": EPHEMERAL}, "cache_control": EPHEMERAL}, None, 2, 1),
],
ids=["same_marked_tool_both_ways", "extra_body_unmarks", "extra_body_marks", "root_cache_control_both_ways"],
)
def test_chat_cap_counts_extra_body_fields_in_place_of_the_direct_ones(self, params, tools, marked_turns, injected):
messages = [{"role": "system", "content": "sys"}, *self._marked_user_turns(marked_turns)]
params = {"cache_control_injection_points": copy.deepcopy(self.CONFIGURED), **copy.deepcopy(params)}
self._seed(params, copy.deepcopy(messages), tools=tools)
processed = self._chat(params, copy.deepcopy(messages))
assert _count_cache_control(processed) == marked_turns + injected
@pytest.mark.parametrize(
"kwargs,tools,marked_turns,expected_system",
[
({"extra_body": {"tools": [MARKED_V1_TOOL]}}, [MARKED_V1_TOOL], 2, MARKED_SYSTEM),
({"extra_body": {"tools": [UNMARKED_V1_TOOL]}}, [MARKED_V1_TOOL], 3, "sys"),
({"extra_body": {"tools": [MARKED_V1_TOOL]}}, [UNMARKED_V1_TOOL], 3, "sys"),
({"extra_body": {"cache_control": EPHEMERAL}, "cache_control": EPHEMERAL}, None, 2, MARKED_SYSTEM),
],
ids=["same_marked_tool_both_ways", "extra_body_unmarks", "extra_body_marks", "root_cache_control_both_ways"],
)
def test_v1_messages_cap_reserves_for_the_larger_of_direct_and_extra_body_marks(
self, kwargs, tools, marked_turns, expected_system
):
kwargs = {"cache_control_injection_points": copy.deepcopy(self.CONFIGURED), **copy.deepcopy(kwargs)}
_, result_sys = self._inject(self._marked_user_turns(marked_turns), kwargs, tools=tools)
assert result_sys == expected_system
def test_v1_messages_automatic_defaults_stand_down_for_root_cache_control(self, monkeypatch):
monkeypatch.setattr(litellm, "enable_anthropic_prompt_caching", True)
root_cache_control = {"type": "ephemeral"}
kwargs = {"cache_control": root_cache_control, "litellm_metadata": {}}
if configured is not None:
kwargs["cache_control_injection_points"] = copy.deepcopy(configured)
result_messages, result_system = self._inject(copy.deepcopy(self.V1_MESSAGES), kwargs)
@ -2382,17 +2633,28 @@ class TestConfiguredInjectionPointsStandDown:
assert kwargs["cache_control"] is root_cache_control
assert "litellm_gateway_injected_cache" not in kwargs["litellm_metadata"]
@pytest.mark.parametrize(
"marked_turns,expected_system",
[(2, [{"type": "text", "text": "sys", "cache_control": {"type": "ephemeral"}}]), (3, "sys")],
)
def test_v1_messages_configured_points_apply_with_root_cache_control_reserving_a_slot(
self, marked_turns, expected_system
):
root_cache_control = {"type": "ephemeral"}
kwargs = {
"cache_control": root_cache_control,
"cache_control_injection_points": copy.deepcopy(self.CONFIGURED),
}
_, result_system = self._inject(self._marked_user_turns(marked_turns), kwargs)
assert result_system == expected_system
assert kwargs["cache_control"] is root_cache_control
def test_v1_messages_reentry_flow_preserves_tool_config_remainder(self):
"""The advisor interceptor re-enters anthropic_messages() with the outer
request's kwargs and post-injection messages. The first pass applies the
message point and writes back a stamped tool_config remainder; the
re-entry must keep that remainder even though the messages and system
now carry litellm's own marks."""
points = [{"location": "message", "role": "system"}, {"location": "tool_config"}]
kwargs = {"cache_control_injection_points": copy.deepcopy(points)}
msgs1, sys1 = self._inject(copy.deepcopy(self.V1_MESSAGES), kwargs)
assert sys1[0]["cache_control"] == {"type": "ephemeral"}
expected_remainder = [{"location": "tool_config", "_litellm_judged": True}]
expected_remainder = [{"location": "tool_config"}]
assert kwargs["cache_control_injection_points"] == expected_remainder
msgs2, sys2 = self._inject(msgs1, kwargs, system=sys1)
@ -2631,22 +2893,26 @@ class TestOpenAIPromptCacheBreakpoint:
assert system == [{"type": "text", "text": "sys", "cache_control": {"type": "ephemeral"}}]
assert kwargs == {}
def test_v1_messages_client_content_breakpoint_makes_configured_points_stand_down(self):
messages = [{"role": "user", "content": [{"type": "text", "text": "hi", "prompt_cache_breakpoint": self.EXPLICIT}]}]
def test_v1_messages_configured_points_apply_beside_client_content_breakpoint(self):
messages = [
{"role": "user", "content": [{"type": "text", "text": "hi", "prompt_cache_breakpoint": self.EXPLICIT}]}
]
kwargs = {"cache_control_injection_points": copy.deepcopy(self.SYSTEM_POINT)}
result, system = self._inject(messages, "sys", kwargs)
assert result == messages
assert system == "sys"
assert kwargs == {}
assert system == [{"type": "text", "text": "sys", "prompt_cache_breakpoint": self.EXPLICIT}]
assert kwargs == {"prompt_cache_options": self.EXPLICIT}
def test_v1_messages_client_system_breakpoint_makes_configured_points_stand_down(self):
def test_v1_messages_tail_point_applies_beside_client_system_breakpoint(self):
system = [{"type": "text", "text": "sys", "prompt_cache_breakpoint": self.EXPLICIT}]
messages = [{"role": "user", "content": [{"type": "text", "text": "hi"}]}]
kwargs = {"cache_control_injection_points": [{"location": "message", "index": -1}]}
result, result_system = self._inject(messages, system, kwargs)
assert result == messages
assert result == [
{"role": "user", "content": [{"type": "text", "text": "hi", "prompt_cache_breakpoint": self.EXPLICIT}]}
]
assert result_system == system
assert kwargs == {}
assert kwargs == {"prompt_cache_options": self.EXPLICIT}
def test_chat_system_string_wrapped_with_block_breakpoint(self):
params = {"cache_control_injection_points": copy.deepcopy(self.SYSTEM_POINT)}
@ -2710,18 +2976,25 @@ class TestOpenAIPromptCacheBreakpoint:
assert processed[0] == {"role": "system", "content": "sys", "cache_control": {"type": "ephemeral"}}
assert params == {}
def test_chat_client_breakpoint_makes_seeded_points_stand_down(self):
def test_chat_seeded_points_apply_beside_client_breakpoint(self):
params = {"cache_control_injection_points": copy.deepcopy(self.SYSTEM_POINT)}
messages = [
{"role": "system", "content": "sys"},
{"role": "user", "content": [{"type": "text", "text": "hi", "prompt_cache_breakpoint": self.EXPLICIT}]},
]
AnthropicCacheControlHook.maybe_seed_default_injection_points(
non_default_params=params,
messages=[
{"role": "system", "content": "sys"},
{"role": "user", "content": [{"type": "text", "text": "hi", "prompt_cache_breakpoint": self.EXPLICIT}]},
],
messages=messages,
model="openai/gpt-5.6",
custom_llm_provider="openai",
)
assert params == {}
assert params["cache_control_injection_points"] == [
{"location": "message", "role": "system", "_litellm_openai_dialect": True}
]
_, processed, _ = self._chat(messages, params)
assert processed[0]["content"] == [{"type": "text", "text": "sys", "prompt_cache_breakpoint": self.EXPLICIT}]
assert processed[1] == messages[1]
assert params["prompt_cache_options"] == self.EXPLICIT
def test_cap_counts_client_breakpoints_of_both_kinds(self):
messages = [
@ -3315,7 +3588,6 @@ class TestRecordGatewayInjection:
assert kwargs["litellm_metadata"][self.KEY] == self.DEPLOYMENT
def test_configured_points_skipping_a_marked_target_record_nothing(self):
"""Configured injection stands down on client breakpoints, so no marker lands."""
kwargs: dict = {
"litellm_metadata": {},
"cache_control_injection_points": [{"location": "message", "role": "system", "index": None}],