fix(anthropic): keep Claude's ttl order and the client's own marks when translating prompt_cache_breakpoint

Under drop_params only translated breakpoints are dropped past the cap, earliest first in Claude's read order. A breakpoint with no ttl ahead of a 1h mark goes out as 1h, and under drop_params a client ttl that breaks the order is dropped. A Bedrock model id that does not name Claude is translated only when base_model names Claude.

litellm/llms/anthropic/chat/prompt_cache_breakpoints.py

@@ -3,9 +3,7 @@
 from collections.abc import Callable, Iterator, Mapping, Sequence
 from dataclasses import dataclass
 from types import MappingProxyType
-from typing import Final, NoReturn, TypeAlias
-
-from pydantic import TypeAdapter
+from typing import Final, Literal, NoReturn, TypeAlias, cast  # noqa: TID251  # isinstance-checked hot-path JSON
 
 from litellm._logging import verbose_logger
 from litellm.constants import MAX_CACHE_CONTROL_BLOCKS
@@ -15,9 +13,9 @@ from litellm.types.llms.openai import ChatCompletionCachedContent
 _Path: TypeAlias = tuple[str | int, ...]
 _Fields: TypeAlias = Mapping[str, object]
 _Mark: TypeAlias = Callable[[_Path, _Fields], _Fields]
+_Kind: TypeAlias = Literal["written", "explicit", "free"]
 _MESSAGE_BLOCK_KEYS: Final = ("content", "tool_calls")
-_FIELDS: Final = TypeAdapter(Mapping[str, object])
-_BLOCKS: Final = TypeAdapter(Sequence[object])
+_REQUEST: Final[_Path] = ("cache_control",)
 _CLAUDE_TTL_MARKS: Final = MappingProxyType(
     {
         "5m": ChatCompletionCachedContent(type="ephemeral", ttl="5m"),
@@ -40,12 +38,19 @@ class UnsupportedPromptCacheParam:
     fallback: ChatCompletionCachedContent | None
 
 
+@dataclass(frozen=True, slots=True)
+class _Placed:
+    path: _Path
+    ttl: Literal["5m", "1h"]
+    kind: _Kind
+
+
 def _fields(value: object) -> _Fields | None:
-    return _FIELDS.validate_python(value) if isinstance(value, Mapping) else None
+    return cast(_Fields, value) if isinstance(value, Mapping) else None  # cast-ok: request JSON has string keys
 
 
 def _blocks(value: object) -> Sequence[object] | None:
-    return _BLOCKS.validate_python(value) if isinstance(value, list) else None
+    return cast(Sequence[object], value) if isinstance(value, list) else None  # cast-ok: elements checked per use
 
 
 def _ttl_mark(ttl: object, param: str, value: object) -> ChatCompletionCachedContent | UnsupportedPromptCacheParam:
@@ -92,6 +97,28 @@ def _markable_in_tool(index: int, tool: _Fields) -> Iterator[tuple[_Path, _Field
         yield ("tools", index, "function"), function
 
 
+def _carries_breakpoint(value: object) -> bool:
+    fields: Final = _fields(value)
+    return fields is not None and fields.get("prompt_cache_breakpoint") is not None
+
+
+def _any_block_carries_breakpoint(blocks: object) -> bool:
+    return any(map(_carries_breakpoint, _blocks(blocks) or ()))
+
+
+def _message_carries_breakpoint(message: object) -> bool:
+    fields: Final = _fields(message) or {}
+    return (
+        fields.get("prompt_cache_breakpoint") is not None
+        or _any_block_carries_breakpoint(fields.get("content"))
+        or _any_block_carries_breakpoint(fields.get("tool_calls"))
+    )
+
+
+def _tool_carries_breakpoint(tool: object) -> bool:
+    return _carries_breakpoint(tool) or _carries_breakpoint((_fields(tool) or {}).get("function"))
+
+
 def _markable(messages: Sequence[object], tools: Sequence[object]) -> Iterator[tuple[_Path, _Fields]]:
     """Every block that can carry a cache mark, in the order Claude renders them: tool definitions, then messages."""
     for index, tool in _indexed_fields(tools):
@@ -177,13 +204,82 @@ def _marked_request(
     )
 
 
-def _without_cache_control(dropped: frozenset[_Path]) -> _Mark:
+def _settled_marks(dropped: frozenset[_Path], replacements: Mapping[_Path, ChatCompletionCachedContent]) -> _Mark:
     def mark(path: _Path, block: _Fields) -> _Fields:
-        return {key: value for key, value in block.items() if key != "cache_control"} if path in dropped else block
+        if path in dropped:
+            return {key: value for key, value in block.items() if key != "cache_control"}
+        replacement: Final = replacements.get(path)
+        return block if replacement is None else {**block, "cache_control": replacement}
 
     return mark
 
 
+def _unchanged(_path: _Path, block: _Fields) -> _Fields:
+    return block
+
+
+def _translated(messages: Sequence[object], tools: Sequence[object]) -> frozenset[_Path]:
+    """Blocks whose mark comes from a breakpoint rather than a ``cache_control`` the client wrote."""
+    hoisted_messages, hoisted_tools = _marked_request(messages, tools, _unchanged)
+    return frozenset(
+        path
+        for path, block in _markable(hoisted_messages, hoisted_tools)
+        if block.get("prompt_cache_breakpoint") is not None and block.get("cache_control") is None
+    )
+
+
+def _wire_position(path: _Path, messages: Sequence[object]) -> tuple[int, ...]:
+    """Where Claude reads a mark: tool definitions, then system messages wherever they sit, then the other messages,
+    with a message-level mark on the message's last block."""
+    if path[0] == "tools":
+        return (0, *(index for index in path if isinstance(index, int)))
+    index: Final = int(path[1])
+    role: Final = (_fields(messages[index]) or {}).get("role")
+    group: Final = 1 if role in ("system", "developer") else 2
+    block: Final = (
+        (_MESSAGE_BLOCK_KEYS.index(str(path[2])), int(path[3])) if len(path) == 4 else (len(_MESSAGE_BLOCK_KEYS),)
+    )
+    return (group, index, *block)
+
+
+def _ttl(cache_control: object) -> Literal["5m", "1h"]:
+    return "1h" if (_fields(cache_control) or {}).get("ttl") == "1h" else "5m"
+
+
+def _kind(path: _Path, cache_control: object, translated: frozenset[_Path]) -> _Kind:
+    if path not in translated:
+        return "written"
+    return "free" if (_fields(cache_control) or {}).get("ttl") is None else "explicit"
+
+
+def _out_of_order(index: int, marks: Sequence[_Placed]) -> bool:
+    """Claude rejects a 1h mark read after a 5m one."""
+    if marks[index].ttl == "1h":
+        return any(other.kind != "free" and other.ttl == "5m" for other in marks[:index])
+    return any(other.kind != "free" and other.ttl == "1h" for other in marks[index + 1 :])
+
+
+def _loosened(marks: Sequence[_Placed], drop_params: bool) -> tuple[_Placed, ...]:
+    return tuple(
+        _Placed(mark.path, "5m", "free")
+        if drop_params and mark.kind == "explicit" and _out_of_order(index, marks)
+        else mark
+        for index, mark in enumerate(marks)
+    )
+
+
+def _settled(marks: Sequence[_Placed], drop_params: bool) -> tuple[_Placed, ...]:
+    """A breakpoint without a ttl takes 1h when a 1h mark is read after it, so the request stays valid for Claude;
+    under ``drop_params`` a ttl the client set on a breakpoint that breaks that order is dropped first."""
+    loosened: Final = _loosened(marks, drop_params)
+    return tuple(
+        _Placed(mark.path, "1h", "free")
+        if mark.kind == "free" and any(later.kind != "free" and later.ttl == "1h" for later in loosened[index + 1 :])
+        else mark
+        for index, mark in enumerate(loosened)
+    )
+
+
 def claude_prompt_cache(
     messages: Sequence[object],
     tools: Sequence[object] | None,
@@ -193,30 +289,56 @@ def claude_prompt_cache(
 ) -> ClaudePromptCache | UnsupportedPromptCacheParam:
     """``prompt_cache_breakpoint`` marks renamed to ``cache_control``. A mark Claude cannot honor is reported unless
     ``drop_params``, which keeps what Claude can use and, when the request ends up with more than Claude's four
-    breakpoints, the latest four. A request-level ``cache_control`` in ``params`` takes one of the four. A request
-    carrying no breakpoint is returned as written."""
+    breakpoints, drops the earliest translated ones. A request-level ``cache_control`` in ``params`` takes one of the
+    four. A request carrying no breakpoint is returned as written."""
     listed_tools: Final = tools or ()
-    blocks: Final = tuple(block for _, block in _markable(messages, listed_tools))
-    if all(block.get("prompt_cache_breakpoint") is None for block in blocks):
+    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))
     problem: Final = next(filter(None, map(_breakpoint_problem, blocks)), None)
     if problem is not None and not drop_params:
         return problem
     if problem is not None:
         verbose_logger.debug("Claude: adjusting %s=%r because %s", problem.param, problem.value, problem.reason)
     renamed_messages, renamed_tools = _marked_request(messages, listed_tools, _renamed)
-    sites: Final = tuple(
-        path for path, block in _markable(renamed_messages, renamed_tools) if block.get("cache_control") is not None
+    translated: Final = _translated(messages, listed_tools)
+    block_marks: Final = sorted(
+        (
+            _Placed(path, _ttl(block.get("cache_control")), _kind(path, block.get("cache_control"), translated))
+            for path, block in _markable(renamed_messages, renamed_tools)
+            if block.get("cache_control") is not None
+        ),
+        key=lambda mark: _wire_position(mark.path, messages),
     )
-    excess: Final = len(sites) + int(params.get("cache_control") is not None) - MAX_CACHE_CONTROL_BLOCKS
-    if excess > 0 and drop_params:
-        verbose_logger.debug("Claude: dropping the earliest %d cache breakpoints", excess)
-    trimmed_messages, trimmed_tools = (
-        _marked_request(renamed_messages, renamed_tools, _without_cache_control(frozenset(sites[:excess])))
+    request_mark: Final = params.get("cache_control")
+    marks: Final = (
+        *block_marks,
+        *(() if request_mark is None else (_Placed(_REQUEST, _ttl(request_mark), "written"),)),
+    )
+    excess: Final = len(marks) - MAX_CACHE_CONTROL_BLOCKS
+    dropped: Final[frozenset[_Path]] = (
+        frozenset(tuple(mark.path for mark in marks if mark.kind != "written")[:excess])
         if excess > 0 and drop_params
-        else (renamed_messages, renamed_tools)
+        else frozenset()
     )
-    return ClaudePromptCache(trimmed_messages, None if tools is None else trimmed_tools)
+    if dropped:
+        verbose_logger.debug("Claude: dropping the earliest %d prompt_cache_breakpoint marks", len(dropped))
+    kept: Final = tuple(mark for mark in marks if mark.path not in dropped)
+    replacements: Final = MappingProxyType(
+        {
+            settled.path: (
+                ChatCompletionCachedContent(type="ephemeral", ttl="1h")
+                if settled.ttl == "1h"
+                else ChatCompletionCachedContent(type="ephemeral")
+            )
+            for mark, settled in zip(kept, _settled(kept, drop_params))
+            if settled.ttl != mark.ttl
+        }
+    )
+    final_messages, final_tools = _marked_request(
+        renamed_messages, renamed_tools, _settled_marks(dropped, replacements)
+    )
+    return ClaudePromptCache(final_messages, None if tools is None else final_tools)
 
 
 def resolved_claude_prompt_cache(

litellm/main.py

@@ -1090,6 +1090,12 @@ def _is_claude_tool_target(custom_llm_provider: str | None, model: str) -> bool:
     return False
 
 
+def _is_claude_prompt_cache_target(custom_llm_provider: str | None, model: str, base_model: str | None) -> bool:
+    if custom_llm_provider == "bedrock" and "claude" not in model.lower():
+        return base_model is not None and "claude" in base_model.lower()
+    return _is_claude_tool_target(custom_llm_provider=custom_llm_provider, model=model)
+
+
 def _without_anthropic_only_tool_keys(tool: dict[str, object]) -> dict[str, object]:
     kept: Final = {key: value for key, value in tool.items() if key not in _ANTHROPIC_ONLY_TOOL_KEYS}
     function: Final = tool.get("function")
@@ -5572,7 +5578,11 @@ def completion(
                 llm_provider=custom_llm_provider,
             )
             if provider_config is not None
-            and _is_claude_tool_target(custom_llm_provider=custom_llm_provider, model=model)
+            and _is_claude_prompt_cache_target(
+                custom_llm_provider=custom_llm_provider,
+                model=model,
+                base_model=base_model if isinstance(base_model, str) else None,
+            )
             else None
         )
         claude_messages: Final = (

tests/integration/providers/test_claude_chat_prompt_cache_breakpoint_wire.py

@@ -53,11 +53,13 @@ _PROVIDERS: Final[tuple[Provider, ...]] = (
     "vertex_ai",
     "azure_ai",
 )
+_ANTHROPIC_BODY_PROVIDERS: Final = tuple(provider for provider in _PROVIDERS if provider != "bedrock_converse")
 _SITES: Final[tuple[Site, ...]] = ("tool", "system", "user_part", "tool_call", "tool_result", "string_message")
 _FIVE_SITES: Final[tuple[Site, ...]] = ("tool", "system", "user_part", "tool_call", "string_message")
 _EXPLICIT: Final[dict[str, JsonValue]] = {"mode": "explicit"}
 _BREAKPOINT: Final[Mapping[str, JsonValue]] = MappingProxyType({"prompt_cache_breakpoint": _EXPLICIT})
 _NATIVE: Final[Mapping[str, JsonValue]] = MappingProxyType({"cache_control": EPHEMERAL})
+_ONE_HOUR: Final[dict[str, JsonValue]] = {"type": "ephemeral", "ttl": "1h"}
 _CITY: Final = CITIES[0]
 _BEDROCK_TOKEN: Final = "synthetic-bedrock-bearer"
 _VERTEX_PROJECT: Final = "scripted-project"
@@ -290,6 +292,48 @@ def test_five_marks_without_drop_params_reach_the_provider_and_its_error_reaches
     assert [len(_labels(provider, request)) for request in received] == [5], [request.target for request in received]
 
 
+@pytest.mark.parametrize("provider", _PROVIDERS)
+def test_drop_params_drops_the_breakpoint_past_the_cap_and_keeps_every_mark_the_client_wrote(
+    gateway: Gateway, provider: Provider
+) -> None:
+    marker: Final = new_marker()
+    written: Final = _FIVE_SITES[:-1]
+    marks: Final = MappingProxyType({**_marked(written, _NATIVE), _FIVE_SITES[-1]: _BREAKPOINT})
+    with wire_server(lambda request: _respond(provider, request)) as wire, gateway.scenario() as scenario:
+        model: Final = _deployment(gateway, scenario, wire, provider, drop_params=True)
+        status, _, text = post_chat(gateway, _body(model, marker, marks))
+        assert status == 200, text
+        request: Final = _only_request(wire)
+    assert _labels(provider, request) == [_expected(provider, site, marker) for site in written]
+
+
+@pytest.mark.parametrize("provider", _ANTHROPIC_BODY_PROVIDERS)
+def test_a_breakpoint_ahead_of_a_one_hour_mark_reaches_claude_as_one_hour(gateway: Gateway, provider: Provider) -> None:
+    marker: Final = new_marker()
+    marks: Final = MappingProxyType(
+        {"system": _BREAKPOINT, "user_part": MappingProxyType({"cache_control": _ONE_HOUR})}
+    )
+    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))
+        assert status == 200, text
+        request: Final = _only_request(wire)
+    assert anthropic_marks(request) == (Mark(SYSTEM_LABEL, "1h"), Mark(ASK_LABEL, "1h"))
+
+
+def test_a_one_hour_breakpoint_after_a_five_minute_mark_loses_its_ttl_under_drop_params(gateway: Gateway) -> None:
+    marker: Final = new_marker()
+    marks: Final = MappingProxyType(
+        {"system": _NATIVE, "user_part": MappingProxyType({"prompt_cache_breakpoint": {**_EXPLICIT, "ttl": "1h"}})}
+    )
+    with wire_server(anthropic_peer) as wire, gateway.scenario() as scenario:
+        model: Final = _deployment(gateway, scenario, wire, "anthropic", drop_params=True)
+        status, _, text = post_chat(gateway, _body(model, marker, marks))
+        assert status == 200, text
+        request: Final = _only_request(wire)
+    assert anthropic_marks(request) == (Mark(SYSTEM_LABEL, None), Mark(ASK_LABEL, None))
+
+
 def test_injection_points_stand_down_behind_the_latest_four_breakpoints(gateway: Gateway) -> None:
     marker: Final = new_marker()
     with wire_server(anthropic_peer) as wire, gateway.scenario() as scenario:
@@ -302,6 +346,26 @@ def test_injection_points_stand_down_behind_the_latest_four_breakpoints(gateway:
     assert gateway_injected(response_id) is False
 
 
+@pytest.mark.parametrize(
+    ("base_model", "marked"), ((None, False), ("bedrock/mistral.mistral-large-2407-v1:0", False), (BEDROCK_MODEL, True))
+)
+def test_an_application_inference_profile_is_marked_only_when_its_base_model_is_claude(
+    gateway: Gateway, base_model: str | None, marked: bool
+) -> None:
+    marker: Final = new_marker()
+    profile: Final = "arn:aws:bedrock:us-east-1:000000000000:application-inference-profile/scripted"
+    with wire_server(bedrock_peer) as wire, gateway.scenario() as scenario:
+        model: Final = scenario.model(
+            model_info=None if base_model is None else {"base_model": base_model},
+            **{**_litellm_params(gateway, wire, "bedrock_converse"), "model": f"bedrock/converse/{profile}"},
+        )
+        status, _, text = post_chat(gateway, _body(model, marker, _marked(("system",))))
+        assert status == 200, text
+        request: Final = _only_request(wire)
+    assert b"prompt_cache_breakpoint" not in request.body, request.body
+    assert bedrock_labels(request) == ([_expected("bedrock_converse", "system", marker)] if marked else [])
+
+
 def _drop_params_config(directory: Path) -> Path:
     return owned_config(directory, [], litellm_settings={"drop_params": True})
 

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

@@ -12,6 +12,7 @@ from litellm.llms.anthropic.chat.prompt_cache_breakpoints import (
 
 _EXPLICIT: Final = {"mode": "explicit"}
 _EPHEMERAL: Final = {"type": "ephemeral"}
+_ONE_HOUR: Final = {"type": "ephemeral", "ttl": "1h"}
 
 
 def _cache(
@@ -58,8 +59,8 @@ def test_breakpoints_become_cache_control_on_every_block_that_carries_one():
     result: Final = _resolved(messages, tools)
 
     assert list(result.messages) == [
-        {"role": "system", "content": [_part("sys", cache_control=_EPHEMERAL)]},
-        {"role": "user", "content": "hi", "cache_control": {"type": "ephemeral", "ttl": "1h"}},
+        {"role": "system", "content": [_part("sys", cache_control=_ONE_HOUR)]},
+        {"role": "user", "content": "hi", "cache_control": _ONE_HOUR},
         {
             "role": "assistant",
             "content": None,
@@ -67,8 +68,8 @@ def test_breakpoints_become_cache_control_on_every_block_that_carries_one():
         },
     ]
     assert list(result.tools or ()) == [
-        {"type": "function", "function": {"name": "a"}, "cache_control": _EPHEMERAL},
-        {"type": "function", "function": {"name": "b"}, "cache_control": _EPHEMERAL},
+        {"type": "function", "function": {"name": "a"}, "cache_control": _ONE_HOUR},
+        {"type": "function", "function": {"name": "b"}, "cache_control": _ONE_HOUR},
     ]
 
 
@@ -116,19 +117,116 @@ def test_an_unsupported_breakpoint_raises_a_400_naming_drop_params():
     assert "30m" in raised.value.message and "drop_params" in raised.value.message
 
 
-def test_drop_params_keeps_the_latest_four_marks_in_the_order_claude_renders_them():
+def test_drop_params_drops_the_earliest_breakpoints_and_keeps_every_mark_the_client_wrote():
     tools: Final = [{"type": "function", "function": {"name": "t"}, "prompt_cache_breakpoint": _EXPLICIT}]
     messages: Final = [
-        {"role": "system", "content": [_part("s", cache_control=_EPHEMERAL)]},
         {"role": "user", "content": [_part(str(index), prompt_cache_breakpoint=_EXPLICIT) for index in range(3)]},
+        {"role": "system", "content": [_part("s", cache_control=_EPHEMERAL)]},
     ]
 
     result: Final = _resolved(messages, tools, {"cache_control": _EPHEMERAL}, drop_params=True)
 
     assert list(result.tools or ()) == [{"type": "function", "function": {"name": "t"}}]
+    assert list(result.messages) == [
+        {
+            "role": "user",
+            "content": [_part("0"), _part("1", cache_control=_EPHEMERAL), _part("2", cache_control=_EPHEMERAL)],
+        },
+        {"role": "system", "content": [_part("s", cache_control=_EPHEMERAL)]},
+    ]
+
+
+def test_marks_the_client_wrote_past_the_cap_reach_claude_as_written_under_drop_params():
+    messages: Final = [
+        {"role": "system", "content": [_part("s", prompt_cache_breakpoint=_EXPLICIT)]},
+        {"role": "user", "content": [_part(str(index), cache_control=_EPHEMERAL) for index in range(5)]},
+    ]
+
+    result: Final = _resolved(messages, drop_params=True)
+
     assert list(result.messages) == [
         {"role": "system", "content": [_part("s")]},
-        {"role": "user", "content": [_part(str(index), cache_control=_EPHEMERAL) for index in range(3)]},
+        {"role": "user", "content": [_part(str(index), cache_control=_EPHEMERAL) for index in range(5)]},
+    ]
+
+
+@pytest.mark.parametrize(
+    ("later", "params"),
+    [
+        ({"cache_control": _ONE_HOUR}, {}),
+        ({"prompt_cache_breakpoint": {"mode": "explicit", "ttl": "1h"}}, {}),
+        ({}, {"cache_control": _ONE_HOUR}),
+    ],
+    ids=["client-mark", "breakpoint", "top-level-cache-control"],
+)
+def test_a_breakpoint_read_before_a_one_hour_mark_is_sent_as_one_hour(
+    later: dict[str, object], params: dict[str, object]
+):
+    messages: Final = [
+        {"role": "user", "content": [_part("doc", prompt_cache_breakpoint=_EXPLICIT), _part("ask", **later)]},
+        {"role": "system", "content": [_part("s", prompt_cache_breakpoint=_EXPLICIT)]},
+    ]
+
+    result: Final = _resolved(messages, params=params)
+
+    ask: Final = _part("ask", cache_control=_ONE_HOUR) if later else _part("ask")
+    assert list(result.messages) == [
+        {"role": "user", "content": [_part("doc", cache_control=_ONE_HOUR), ask]},
+        {"role": "system", "content": [_part("s", cache_control=_ONE_HOUR)]},
+    ]
+
+
+def test_a_breakpoint_read_after_the_last_one_hour_mark_keeps_the_default_ttl():
+    messages: Final = [
+        {"role": "system", "content": [_part("s", cache_control=_ONE_HOUR)]},
+        {"role": "user", "content": [_part("ask", prompt_cache_breakpoint=_EXPLICIT)]},
+    ]
+
+    result: Final = _resolved(messages)
+
+    assert list(result.messages) == [
+        {"role": "system", "content": [_part("s", cache_control=_ONE_HOUR)]},
+        {"role": "user", "content": [_part("ask", cache_control=_EPHEMERAL)]},
+    ]
+
+
+@pytest.mark.parametrize(
+    ("system", "ask", "as_written", "under_drop_params"),
+    [
+        (
+            {"cache_control": _EPHEMERAL},
+            {"prompt_cache_breakpoint": {"mode": "explicit", "ttl": "1h"}},
+            ({"cache_control": _EPHEMERAL}, {"cache_control": _ONE_HOUR}),
+            ({"cache_control": _EPHEMERAL}, {"cache_control": _EPHEMERAL}),
+        ),
+        (
+            {"prompt_cache_breakpoint": {"mode": "explicit", "ttl": "5m"}},
+            {"cache_control": _ONE_HOUR},
+            ({"cache_control": {"type": "ephemeral", "ttl": "5m"}}, {"cache_control": _ONE_HOUR}),
+            ({"cache_control": _ONE_HOUR}, {"cache_control": _ONE_HOUR}),
+        ),
+    ],
+    ids=["one-hour-breakpoint-after-a-five-minute-mark", "five-minute-breakpoint-before-a-one-hour-mark"],
+)
+@pytest.mark.parametrize("drop_params", [False, True])
+def test_a_ttl_the_client_set_out_of_claude_order_is_sent_as_written_or_dropped_under_drop_params(
+    system: dict[str, object],
+    ask: dict[str, object],
+    as_written: tuple[dict[str, object], dict[str, object]],
+    under_drop_params: tuple[dict[str, object], dict[str, object]],
+    drop_params: bool,
+):
+    messages: Final = [
+        {"role": "system", "content": [_part("s", **system)]},
+        {"role": "user", "content": [_part("ask", **ask)]},
+    ]
+
+    result: Final = _resolved(messages, drop_params=drop_params)
+
+    expected_system, expected_ask = under_drop_params if drop_params else as_written
+    assert list(result.messages) == [
+        {"role": "system", "content": [_part("s", **expected_system)]},
+        {"role": "user", "content": [_part("ask", **expected_ask)]},
     ]