fix(responses): keep valid prompt cache markers in place and let cache_control fill a dropped one

Under drop_params the marker step now hands back the caller's own objects when every marker is
already wire-safe, so a request with valid markers is no longer copied and walked a second time.

A deployment that turns cache_control into markers keeps the caller's own markers too, whatever
its base_model says, so one request never gets two answers about breakpoint support.

On /v1/messages a malformed raw marker next to cache_control is dropped before the conversion,
so the block gets the cache hint cache_control alone would give, as on /v1/responses.

litellm/llms/anthropic/messages/responses_adapters/handler.py

@@ -15,7 +15,11 @@ from litellm.integrations.anthropic_cache_control_hook import (
     supports_openai_prompt_cache_breakpoint,
 )
 from litellm.litellm_core_utils.core_helpers import drops_unsupported_params
-from litellm.litellm_core_utils.prompt_templates.common_utils import cache_control_as_prompt_cache_breakpoint
+from litellm.litellm_core_utils.prompt_templates.common_utils import (
+    cache_control_as_prompt_cache_breakpoint,
+    prompt_cache_breakpoint_for_wire,
+    with_prompt_cache_breakpoint,
+)
 from litellm.responses.utils import auto_reasoning_summary
 from litellm.types.integrations.custom_logger import CONVERTED_STREAM_KEYS
 from litellm.types.llms.anthropic import (
@@ -90,14 +94,28 @@ def _without_prompt_cache_breakpoint(block: Mapping[str, object]) -> Mapping[str
     return {key: value for key, value in block.items() if key != "prompt_cache_breakpoint"}
 
 
+def _cache_control_as_wire_safe_prompt_cache_breakpoint(block: Mapping[str, object]) -> Mapping[str, object]:
+    if "prompt_cache_breakpoint" not in block:
+        return cache_control_as_prompt_cache_breakpoint(block)
+    wire_marker: Final = prompt_cache_breakpoint_for_wire(block["prompt_cache_breakpoint"], drop_params=True)
+    return cache_control_as_prompt_cache_breakpoint(
+        with_prompt_cache_breakpoint(_without_prompt_cache_breakpoint(block), wire_marker)
+    )
+
+
 def _prompt_cache_block_rewrite(model: str, forwarded_kwargs: Mapping[str, object]) -> BlockRewrite | None:
     """Client cache marks become breakpoints for a deployment that reads them, and a model that takes no breakpoints
     drops raw ones under drop_params."""
+    drops_params: Final = drops_unsupported_params(forwarded_kwargs.get("drop_params"))
     if _reads_prompt_cache_breakpoint(model, forwarded_kwargs):
-        return cache_control_as_prompt_cache_breakpoint
+        return (
+            _cache_control_as_wire_safe_prompt_cache_breakpoint
+            if drops_params
+            else cache_control_as_prompt_cache_breakpoint
+        )
     if _model_takes_prompt_cache_breakpoint(model, forwarded_kwargs):
         return None
-    if drops_unsupported_params(forwarded_kwargs.get("drop_params")):
+    if drops_params:
         return _without_prompt_cache_breakpoint
     return None
 

litellm/llms/openai/responses/transformation.py

@@ -218,16 +218,19 @@ def _item_with_wire_safe_prompt_cache_breakpoints(item: object, keeps_markers: b
     fields: Final = _json_object_or_none(item)
     if fields is None or not _carries_prompt_cache_breakpoint(fields):
         return item
-    return {
+    wire_safe_fields: Final = {
         key: _parts_with_wire_safe_prompt_cache_breakpoints(key, value, keeps_markers)
         for key, value in fields.items()
         if key != "prompt_cache_breakpoint"
     }
+    unchanged: Final = "prompt_cache_breakpoint" not in fields and _same_objects(
+        wire_safe_fields.values(), fields.values()
+    )
+    return item if unchanged else wire_safe_fields
 
 
-def _input_carries_marker(items: Iterable[object]) -> bool:
-    item_fields: Final = (_json_object_or_none(item) for item in items)
-    return any(_carries_prompt_cache_breakpoint(fields) for fields in item_fields if fields is not None)
+def _same_objects(rebuilt: Iterable[object], originals: Iterable[object]) -> bool:
+    return all(new is old for new, old in zip(rebuilt, originals, strict=True))
 
 
 def _carries_prompt_cache_breakpoint(fields: Mapping[str, object]) -> bool:
@@ -237,29 +240,27 @@ def _carries_prompt_cache_breakpoint(fields: Mapping[str, object]) -> bool:
 
 
 def _parts_carry_prompt_cache_breakpoint(parts: object) -> bool:
-    return any(_has_own_prompt_cache_breakpoint(part) for part in _json_array_or_empty(parts))
-
-
-def _has_own_prompt_cache_breakpoint(part: object) -> bool:
-    fields: Final = _json_object_or_none(part)
-    return fields is not None and "prompt_cache_breakpoint" in fields
+    return any(isinstance(part, Mapping) and "prompt_cache_breakpoint" in part for part in _json_array_or_empty(parts))
 
 
 def _parts_with_wire_safe_prompt_cache_breakpoints(key: str, value: object, keeps_markers: bool) -> object:
     if key not in _INPUT_PART_LIST_KEYS or not _parts_carry_prompt_cache_breakpoint(value):
         return value
-    return [_part_with_wire_safe_prompt_cache_breakpoint(part, keeps_markers) for part in _json_array_or_empty(value)]
+    parts: Final = _json_array_or_empty(value)
+    wire_safe_parts: Final = tuple(_part_with_wire_safe_prompt_cache_breakpoint(part, keeps_markers) for part in parts)
+    return value if _same_objects(wire_safe_parts, parts) else list(wire_safe_parts)
 
 
 def _part_with_wire_safe_prompt_cache_breakpoint(part: object, keeps_markers: bool) -> object:
     fields: Final = _json_object_or_none(part)
     if fields is None or "prompt_cache_breakpoint" not in fields:
         return part
+    marker: Final = fields["prompt_cache_breakpoint"]
+    wire_marker: Final = prompt_cache_breakpoint_for_wire(marker, drop_params=True) if keeps_markers else None
+    if keeps_markers and wire_marker == marker:
+        return part
     unmarked: Final = {key: value for key, value in fields.items() if key != "prompt_cache_breakpoint"}
-    if not keeps_markers:
-        return unmarked
-    marker: Final = prompt_cache_breakpoint_for_wire(fields["prompt_cache_breakpoint"], drop_params=True)
-    return with_prompt_cache_breakpoint(unmarked, marker)
+    return with_prompt_cache_breakpoint(unmarked, wire_marker)
 
 
 class _ReasoningSupportEntry(LiteLLMBaseModel):
@@ -588,12 +589,12 @@ class OpenAIResponsesAPIConfig(BaseResponsesAPIConfig):
         prompt_cache_options: object = None,
     ) -> tuple[str | ResponseInputParam, Sequence[ALL_RESPONSES_API_TOOL_PARAMS] | None]:
         validated_input: Final = self._validate_input_param(self._drop_bridge_minted_reasoning_items(input))
-        marker_safe_input: Final = self._input_with_wire_safe_prompt_cache_breakpoints(
-            model=model, input=validated_input, litellm_params=litellm_params
-        )
         reads_breakpoints: Final = self._reads_prompt_cache_breakpoints(
             model=model, litellm_params=litellm_params, prompt_cache_options=prompt_cache_options
         )
+        marker_safe_input: Final = self._input_with_wire_safe_prompt_cache_breakpoints(
+            model=model, input=validated_input, litellm_params=litellm_params, reads_breakpoints=reads_breakpoints
+        )
         maps_anthropic_tool_search: Final = self._targets_openai_api(litellm_params) and any(
             is_anthropic_tool_search_chat_tool(tool) for tool in tools or ()
         )
@@ -640,13 +641,24 @@ class OpenAIResponsesAPIConfig(BaseResponsesAPIConfig):
         )
 
     def _input_with_wire_safe_prompt_cache_breakpoints(
-        self, model: str, input: str | ResponseInputParam, litellm_params: GenericLiteLLMParams
+        self,
+        model: str,
+        input: str | ResponseInputParam,
+        litellm_params: GenericLiteLLMParams,
+        reads_breakpoints: bool,
     ) -> str | ResponseInputParam:
-        if not isinstance(input, list) or not _drops_params(litellm_params) or not _input_carries_marker(input):
+        if not isinstance(input, list) or not _drops_params(litellm_params):
             return input
-        keeps_markers: Final = self._keeps_prompt_cache_breakpoints(model=model, litellm_params=litellm_params)
-        wire_safe_items: Final = [_item_with_wire_safe_prompt_cache_breakpoints(item, keeps_markers) for item in input]
-        return cast("ResponseInputParam", wire_safe_items)  # cast-ok: items keep their shape, minus a rejected marker
+        keeps_markers: Final = reads_breakpoints or self._keeps_prompt_cache_breakpoints(
+            model=model, litellm_params=litellm_params
+        )
+        wire_safe_items: Final = tuple(
+            _item_with_wire_safe_prompt_cache_breakpoints(item, keeps_markers) for item in input
+        )
+        if _same_objects(wire_safe_items, input):
+            return input
+        wire_safe_input: Final = list(wire_safe_items)
+        return cast("ResponseInputParam", wire_safe_input)  # cast-ok: items keep their shape, minus a rejected marker
 
     def _keeps_prompt_cache_breakpoints(self, model: str, litellm_params: GenericLiteLLMParams) -> bool:
         provider: Final = _provider_name(self.custom_llm_provider)

tests/unit/llms/anthropic/messages/responses_adapters/test_responses_adapters_handler.py

@@ -556,6 +556,28 @@ def test_raw_prompt_cache_breakpoints_reach_a_deployment_unless_drop_params_drop
     ]
 
 
+def _gpt_5_6_input_for_a_cached_block(drop_params: bool, raw_mark: Mapping[str, object]) -> object:
+    responses_kwargs: Final = _build_responses_kwargs(
+        model="openai/gpt-5.6",
+        extra_kwargs={**OPENAI_API, "drop_params": drop_params},
+        max_tokens=1024,
+        messages=[{"role": "user", "content": [{"type": "text", "text": "look", "cache_control": EPHEMERAL, **raw_mark}]}],
+    )
+    return responses_kwargs["input"]
+
+
+def test_a_malformed_raw_marker_beside_cache_control_gives_way_to_it_only_under_drop_params(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(litellm, "drop_params", False)
+    malformed: Final = {"prompt_cache_breakpoint": "yes"}
+
+    assert _gpt_5_6_input_for_a_cached_block(True, malformed) == _gpt_5_6_input_for_a_cached_block(True, {})
+    assert _gpt_5_6_input_for_a_cached_block(False, malformed) == [
+        {"type": "message", "role": "user", "content": [{"type": "input_text", "text": "look", **malformed}]}
+    ]
+
+
 @pytest.mark.parametrize(
     ("model", "extra_kwargs"),
     [

tests/unit/llms/openai/responses/test_openai_responses_transformation.py

@@ -4356,6 +4356,19 @@ def test_a_declared_base_model_decides_whether_a_valid_marker_survives_drop_para
     assert _single_part(dropped) == _PARTS["input_text"]
 
 
+def test_a_deployment_that_marks_cache_control_keeps_the_callers_own_marker_whatever_its_base_model_says(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(litellm, "drop_params", False)
+    cached: Final = {"type": "input_text", "text": "a", "cache_control": {"type": "ephemeral"}}
+    marked: Final = {"type": "input_text", "text": "b", "prompt_cache_breakpoint": {"mode": "explicit"}}
+    input: Final = [{"type": "message", "role": "user", "content": [cached, marked]}]
+    transformed: Final = _transformed_input_for("gpt-6.1-sol", input, _declaring_base_model("model_info", "gpt-5.5"))
+    (item,) = cast(list[Mapping[str, object]], transformed)
+    parts: Final = cast(list[Mapping[str, object]], item["content"])
+    assert [part.get("prompt_cache_breakpoint") for part in parts] == [{"mode": "explicit"}, {"mode": "explicit"}]
+
+
 @pytest.mark.parametrize("content", ("Say ok", [{"type": "input_text", "text": "Say ok"}]), ids=("string", "parts"))
 @pytest.mark.parametrize("marker", (_VALID_MARKERS[0], "yes"), ids=("explicit", "string"))
 def test_a_prompt_cache_breakpoint_on_the_message_item_is_dropped_under_drop_params(