diff --git a/app/protocols/anthropic.py b/app/protocols/anthropic.py
index aa18eced..fdb911d9 100644
--- a/app/protocols/anthropic.py
+++ b/app/protocols/anthropic.py
@@ -77,6 +77,7 @@ class AnthropicMessagesRequest(BaseModel):
     stream: bool = False
     metadata: dict | None = None
     thinking: dict | None = None
+    cache_control: dict | None = None
 
 
 # ── Request translation (Anthropic → OpenAI) ───────────────────────────
@@ -85,7 +86,16 @@ def _invalid(message: str) -> HTTPException:
     return HTTPException(status_code=400, detail=message)
 
 
-def _system_text(system: str | list[dict]) -> str:
+def _with_cache_control(translated: dict, original: dict) -> dict:
+    control = original.get("cache_control")
+    if control is not None:
+        if not isinstance(control, dict):
+            raise _invalid("cache_control must be an object")
+        translated["cache_control"] = dict(control)
+    return translated
+
+
+def _system_content(system: str | list[dict]) -> str | list[dict]:
     """The `system` param is text-only per the Anthropic API (string or
     text blocks). Anything else — an image block a client tucked into a
     role:"system" message inside messages[], which this surface tolerates
@@ -93,18 +103,20 @@ def _system_text(system: str | list[dict]) -> str:
     log rather than silently vanishing."""
     if isinstance(system, str):
         return system
-    parts: list[str] = []
+    parts: list[dict] = []
     dropped = 0
     for b in system:
         if isinstance(b, dict) and b.get("type") == "text":
             text = require_text(b.get("text"), field="system[].text")
             if text:
-                parts.append(text)
+                parts.append(_with_cache_control({"type": "text", "text": text}, b))
         else:
             dropped += 1
     if dropped:
         logger.warning("anthropic_non_text_system_block_dropped", count=dropped)
-    return "\n\n".join(parts)
+    if any("cache_control" in part for part in parts):
+        return parts
+    return "\n\n".join(part["text"] for part in parts)
 
 
 def _image_part(block: dict) -> dict:
@@ -112,24 +124,23 @@ def _image_part(block: dict) -> dict:
     stype = source.get("type")
     if stype == "base64":
         media = source.get("media_type", "application/octet-stream")
-        return {
+        return _with_cache_control({
             "type": "image_url",
             "image_url": {"url": f"data:{media};base64,{source.get('data', '')}"},
-        }
+        }, block)
     if stype == "url":
-        return {"type": "image_url", "image_url": {"url": source.get("url", "")}}
+        return _with_cache_control(
+            {"type": "image_url", "image_url": {"url": source.get("url", "")}}, block,
+        )
     raise _invalid(f"Unsupported image source type: {stype!r}")
 
 
-def _split_tool_result_content(content) -> tuple[str, list[dict]]:
+def _split_tool_result_content(content, *, preserve_blocks: bool = False) -> tuple[str | list[dict], list[dict]]:
     """Split tool_result.content into (text, image parts).
 
-    tool_result.content is a string or a list of text/image blocks. An
-    OpenAI `role: "tool"` message carries a plain string, so the text is
-    flattened into it and any images are handed back for the caller to
-    attach to the user message that follows the tool results — a tool
-    returning a screenshot is the common case, and dropping it would
-    silently compute the answer without context the caller supplied.
+    Unmarked results retain the existing text flattening/image handoff to
+    the following user message. Marked results use LiteLLM's multipart
+    tool-message representation to keep their cache breakpoint in place.
     """
     if content is None:
         return "", []
@@ -138,6 +149,7 @@ def _split_tool_result_content(content) -> tuple[str, list[dict]]:
     if not isinstance(content, list):
         return str(content), []
     texts: list[str] = []
+    parts: list[dict] = []
     images: list[dict] = []
     for b in content:
         if not isinstance(b, dict):
@@ -146,8 +158,16 @@ def _split_tool_result_content(content) -> tuple[str, list[dict]]:
             text = require_text(b.get("text"), field="tool_result.content[].text")
             if text:
                 texts.append(text)
+                parts.append(_with_cache_control({"type": "text", "text": text}, b))
         elif b.get("type") == "image":
-            images.append(_image_part(b))
+            image = _image_part(b)
+            images.append(image)
+            parts.append(image)
+    if preserve_blocks:
+        # LiteLLM supports multipart tool messages. Keep a marked result's
+        # boundaries and images inside that result, so its breakpoint still
+        # ends at the content the caller selected.
+        return parts, []
     return "\n\n".join(texts), images
 
 
@@ -155,31 +175,33 @@ def _split_tool_result_content(content) -> tuple[str, list[dict]]:
 
 
 def _translate_assistant_message(blocks: list[dict]) -> dict:
-    texts: list[str] = []
+    parts: list[dict] = []
     tool_calls: list[dict] = []
     for block in blocks:
         btype = block.get("type")
         if btype == "text":
-            texts.append(require_text(block.get("text"), field="messages[].content[].text"))
+            text = require_text(block.get("text"), field="messages[].content[].text")
+            parts.append(_with_cache_control({"type": "text", "text": text}, block))
         elif btype == "tool_use":
             name = block.get("name")
             if not isinstance(name, str) or not name:
                 # The upstream rejects an empty function name with a 400
                 # the engine would report as a retryable 503 — fail here.
                 raise _invalid("tool_use block requires a non-empty string 'name'")
-            tool_calls.append({
+            tool_calls.append(_with_cache_control({
                 "id": block.get("id") or f"toolu_{uuid.uuid4().hex[:24]}",
                 "type": "function",
                 "function": {
                     "name": name,
                     "arguments": json.dumps(block.get("input") or {}),
                 },
-            })
+            }, block))
         elif btype in _DROPPED_BLOCK_TYPES:
             continue
         else:
             raise _invalid(f"Unsupported content block type in assistant message: {btype!r}")
-    out: dict = {"role": "assistant", "content": "".join(texts) or None}
+    content = parts if any("cache_control" in part for part in parts) else "".join(p["text"] for p in parts)
+    out: dict = {"role": "assistant", "content": content or None}
     if tool_calls:
         out["tool_calls"] = tool_calls
     if out["content"] is None and not tool_calls:
@@ -195,9 +217,9 @@ def _translate_assistant_message(blocks: list[dict]) -> dict:
 def _translate_user_message(blocks: list[dict]) -> list[dict]:
     """One Anthropic user message may carry tool_result blocks plus regular
     content. OpenAI needs the tool results as their own `role: "tool"`
-    messages (first), then the remaining content as a user message —
-    which is also where images returned by a tool end up, since an OpenAI
-    tool message can only hold text."""
+    messages (first), then the remaining content as a user message.
+    Unmarked tool images retain the existing handoff to that user message;
+    marked results stay multipart until LiteLLM's provider conversion."""
     tool_messages: list[dict] = []
     parts: list[dict] = []
     for block in blocks:
@@ -209,12 +231,18 @@ def _translate_user_message(blocks: list[dict]) -> list[dict]:
                 # the upstream as a malformed tool message and comes back
                 # as a retryable 503 for a request that can never succeed.
                 raise _invalid("tool_result block requires a non-empty string 'tool_use_id'")
-            text, images = _split_tool_result_content(block.get("content"))
-            tool_messages.append({
+            content = block.get("content")
+            marked = block.get("cache_control") is not None or (
+                isinstance(content, list) and any(
+                    isinstance(part, dict) and part.get("cache_control") is not None for part in content
+                )
+            )
+            text, images = _split_tool_result_content(content, preserve_blocks=marked)
+            tool_messages.append(_with_cache_control({
                 "role": "tool",
                 "tool_call_id": tool_use_id,
                 "content": text,
-            })
+            }, block))
             parts.extend(images)
         elif btype == "text":
             text = require_text(block.get("text"), field="messages[].content[].text")
@@ -224,7 +252,7 @@ def _translate_user_message(blocks: list[dict]) -> list[dict]:
                 # as a BadRequest — reported back as a retryable 5xx for a
                 # request that can never succeed. A turn left with no
                 # content at all is rejected below, as `content: []` is.
-                parts.append({"type": "text", "text": text})
+                parts.append(_with_cache_control({"type": "text", "text": text}, block))
         elif btype == "image":
             parts.append(_image_part(block))
         elif btype in _DROPPED_BLOCK_TYPES:
@@ -236,7 +264,7 @@ def _translate_user_message(blocks: list[dict]) -> list[dict]:
 
     out = tool_messages
     if parts:
-        if len(parts) == 1 and parts[0]["type"] == "text":
+        if len(parts) == 1 and parts[0]["type"] == "text" and "cache_control" not in parts[0]:
             out.append({"role": "user", "content": parts[0]["text"]})
         else:
             out.append({"role": "user", "content": parts})
@@ -257,7 +285,7 @@ def _translate_message(m: AnthropicMessage) -> list[dict]:
         # Code 2.x internal requests) also put role:"system" entries inside
         # messages[]. OpenAI's format allows system messages there, so
         # translate instead of rejecting — a 400 here breaks Claude Code.
-        text = m.content if isinstance(m.content, str) else _system_text(m.content)
+        text = _system_content(m.content)
         return [{"role": "system", "content": text}]
     if m.role not in ("user", "assistant"):
         raise _invalid(f"Unsupported message role: {m.role!r}")
@@ -304,7 +332,7 @@ def _translate_tool(tool: dict) -> dict:
     }
     if description:
         fn["description"] = description
-    return {"type": "function", "function": fn}
+    return _with_cache_control({"type": "function", "function": fn}, tool)
 
 
 def _translate_tool_choice(tc: dict) -> str | dict:
@@ -323,12 +351,58 @@ def _translate_tool_choice(tc: dict) -> str | dict:
     raise _invalid(f"Unsupported tool_choice type: {tctype!r}")
 
 
+def _apply_automatic_cache_control(out: dict, policy: dict | None) -> None:
+    """Lower automatic policy to its documented last-cacheable-block marker.
+
+    https://platform.claude.com/docs/en/build-with-claude/prompt-caching#automatic-caching
+    Block markers are portable through LiteLLM; a raw Anthropic-only top-level
+    extra would also reach unrelated providers selected by the shared router.
+    """
+    if policy is None:
+        return
+    candidates = list(out.get("tools") or [])
+    for message in out["messages"]:
+        if message["role"] == "tool":
+            candidates.append(message)
+            continue
+        content = message.get("content")
+        if isinstance(content, str) and content:
+            content = [{"type": "text", "text": content}]
+            message["content"] = content
+        if isinstance(content, list):
+            candidates.extend(
+                b for b in content if b.get("type") in {"text", "image_url"}
+                and (b.get("type") != "text" or b.get("text"))
+            )
+        # LiteLLM renders assistant text before its tool_calls, matching the
+        # established translation. Select the last rendered block, not the
+        # last native block before that grouping.
+        candidates.extend(message.get("tool_calls") or [])
+    if not candidates:
+        return
+    last = candidates[-1]
+    existing = last.get("cache_control")
+    if existing is not None:
+        if not isinstance(existing, dict):
+            raise _invalid("cache_control must be an object")
+        if existing.get("ttl", "5m") != policy.get("ttl", "5m"):
+            raise _invalid("automatic cache_control conflicts with the last block's TTL")
+        return
+    cache_blocks = list(candidates)
+    for block in candidates:
+        if block.get("role") == "tool" and isinstance(block.get("content"), list):
+            cache_blocks.extend(block["content"])
+    if sum(b.get("cache_control") is not None for b in cache_blocks) >= 4:
+        raise _invalid("automatic cache_control requires a free cache breakpoint slot")
+    last["cache_control"] = dict(policy)
+
+
 def to_openai_request(req: AnthropicMessagesRequest) -> dict:
     """Translate a validated Anthropic Messages request into the internal
     OpenAI chat.completions dict consumed by the chat engine."""
     messages: list[dict] = []
     if req.system is not None:
-        text = _system_text(req.system)
+        text = _system_content(req.system)
         if text:
             messages.append({"role": "system", "content": text})
     for m in req.messages:
@@ -361,6 +435,7 @@ def to_openai_request(req: AnthropicMessagesRequest) -> dict:
         logger.warning("anthropic_thinking_param_dropped")
     if req.top_k is not None:
         logger.info("anthropic_top_k_dropped")
+    _apply_automatic_cache_control(out, req.cache_control)
     return out
 
 
diff --git a/app/schemas.py b/app/schemas.py
index 0be15a10..2c9a3d02 100644
--- a/app/schemas.py
+++ b/app/schemas.py
@@ -15,6 +15,8 @@ class ChatMessage(BaseModel):
     name: str | None = None
     tool_calls: list[dict] | None = None
     tool_call_id: str | None = None
+    # LiteLLM uses a tool-message marker on the native tool_result block.
+    cache_control: dict | None = None
 
 
 class ChatCompletionRequest(BaseModel):
diff --git a/integrations/claude-code.md b/integrations/claude-code.md
index d6dadd85..1dcc6940 100644
--- a/integrations/claude-code.md
+++ b/integrations/claude-code.md
@@ -36,6 +36,13 @@ print(message.content[0].text)
 
 Streaming (`client.messages.stream(...)`) and tool use work as usual.
 
+`cache_control` markers and their TTLs are preserved on supported system,
+text, image, custom-tool, tool-use and tool-result blocks. A top-level
+automatic policy becomes the equivalent explicit marker on the last
+cacheable block, preserving existing breakpoints. Provider caching still
+depends on the selected provider/model and LiteLLM version; these markers
+do not enable Anthropic caching on other providers.
+
 ## Known limitations (v1)
 
 - `thinking` (extended thinking) is accepted but **dropped** — the request
@@ -46,10 +53,10 @@ Streaming (`client.messages.stream(...)`) and tool use work as usual.
   `max_tokens` must be >= 1, likewise.
 - `stop_sequences` beyond the OpenAI wire cap of 4 are truncated to the
   first 4 (a `structlog` warning is emitted).
-- Images inside a `tool_result` are forwarded as a user message following
-  the tool result, since the internal format's tool messages are
-  text-only. Non-text blocks in `system` (which the API itself does not
-  allow) are dropped with a warning.
+- Images inside an unmarked `tool_result` are forwarded as a user message
+  following the tool result. Marked multipart results stay together until
+  LiteLLM converts them for the selected provider. Non-text blocks in
+  `system` (which the API itself does not allow) are dropped with a warning.
 - Streaming `message_start.usage.input_tokens` is the same estimate
   `/v1/messages/count_tokens` returns — the upstream only reports the
   exact count at end-of-stream, where it is sent in `message_delta`.
diff --git a/tests/integration/test_anthropic_cache_controls.py b/tests/integration/test_anthropic_cache_controls.py
new file mode 100644
index 00000000..fbf88cab
--- /dev/null
+++ b/tests/integration/test_anthropic_cache_controls.py
@@ -0,0 +1,262 @@
+# ruff: noqa: F811  (fixture imported from a sibling module)
+"""Native cache policies must survive the real LiteLLM Anthropic payload.
+
+Provider dispatch uses the real LiteLLM request transformations and a
+deterministic response. No optional MCP server or network is needed.
+"""
+
+from __future__ import annotations
+
+import json
+from copy import deepcopy
+from functools import partial
+
+import httpx
+import pytest
+from openai import AsyncOpenAI
+
+from tests.integration.test_anthropic_messages import native_client  # noqa: F401
+
+
+@pytest.fixture
+async def anthropic_provider(native_client, monkeypatch):
+    import litellm
+    from litellm.llms.anthropic.chat.transformation import AnthropicConfig
+
+    from app import router_cache
+    from packages.litellm_adapter.client import OrcaLiteLLMClient
+    from packages.litellm_adapter.types import ProviderDeployment
+
+    client, fake, key = native_client
+    received = []
+    reply = fake.acompletion.side_effect
+
+    async def transform_and_reply(**kwargs):
+        config = AnthropicConfig()
+        params = {k: v for k, v in kwargs.items() if k not in {"model", "messages"}}
+        optional = config.map_openai_params(
+            non_default_params=params, optional_params={}, model=kwargs["model"], drop_params=False,
+        )
+        received.append(config.transform_request(
+            model=kwargs["model"], messages=deepcopy(kwargs["messages"]),
+            optional_params=optional, litellm_params={}, headers={},
+        ))
+        return await reply(**kwargs)
+
+    fake.acompletion.side_effect = transform_and_reply
+    monkeypatch.setattr(litellm, "Router", lambda **_: fake)
+    adapter = OrcaLiteLLMClient(deployments=[ProviderDeployment(
+        model_name="claude-3-5-sonnet-latest",
+        litellm_model="anthropic/claude-3-5-sonnet-latest",
+        api_key="sk-test", provider="anthropic",
+    )], num_retries=0)
+    async def get_router(_session):
+        return adapter
+
+    monkeypatch.setattr(router_cache, "get_router", get_router)
+    yield client, key, received
+
+
+@pytest.fixture
+async def openai_provider(native_client, monkeypatch):
+    from app import router_cache
+    from packages.litellm_adapter.client import OrcaLiteLLMClient
+    from packages.litellm_adapter.types import ProviderDeployment
+
+    client, _, key = native_client
+    received = []
+
+    def respond(request):
+        body = json.loads(request.content)
+        received.append(body)
+        common = {"id": "chatcmpl-policy", "model": body["model"], "created": 1}
+        if body.get("stream"):
+            chunk = {**common, "object": "chat.completion.chunk", "choices": [
+                {"index": 0, "delta": {"content": "ok"}, "finish_reason": "stop"},
+            ]}
+            return httpx.Response(
+                200, headers={"content-type": "text/event-stream"},
+                text=f"data: {json.dumps(chunk)}\n\ndata: [DONE]\n\n",
+            )
+        return httpx.Response(200, json={
+            **common, "object": "chat.completion", "choices": [
+                {"index": 0, "message": {"role": "assistant", "content": "ok"}, "finish_reason": "stop"},
+            ], "usage": {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2},
+        })
+
+    async with httpx.AsyncClient(transport=httpx.MockTransport(respond)) as upstream:
+        async with AsyncOpenAI(api_key="sk-test", http_client=upstream) as sdk:
+            adapter = OrcaLiteLLMClient(deployments=[ProviderDeployment(
+                model_name="gpt-4o-mini", litellm_model="openai/gpt-4o-mini",
+                api_key="sk-test", provider="openai",
+            )], num_retries=0)
+            monkeypatch.setattr(
+                adapter._router, "acompletion", partial(adapter._router.acompletion, client=sdk),
+            )
+
+            async def get_router(_session):
+                return adapter
+
+            monkeypatch.setattr(router_cache, "get_router", get_router)
+            yield client, key, received
+
+
+async def _send(provider, *, stream=False, **overrides):
+    client, key, received = provider
+    response = await client.post("/v1/messages", headers={"x-api-key": key}, json={
+        "model": "claude-3-5-sonnet-latest", "max_tokens": 32,
+        "messages": [{"role": "user", "content": "hi"}], "stream": stream,
+        **overrides,
+    })
+    assert response.status_code == 200, response.text
+    assert len(received) == 1
+    return received[0]
+
+
+@pytest.mark.parametrize("stream", [False, True])
+async def test_system_content_and_tool_breakpoints_reach_provider(anthropic_provider, stream):
+    wire = await _send(anthropic_provider, stream=stream,
+        system=[
+            {"type": "text", "text": "prefix", "cache_control": {"type": "ephemeral", "ttl": "1h"}},
+            {"type": "text", "text": "suffix"},
+        ],
+        tools=[{"name": "lookup", "input_schema": {"type": "object"},
+                "cache_control": {"type": "ephemeral", "ttl": "1h"}}],
+        messages=[{"role": "user", "content": [
+            {"type": "text", "text": "hi", "cache_control": {"type": "ephemeral", "ttl": "5m"}},
+        ]}],
+    )
+    assert wire["system"] == [
+        {"type": "text", "text": "prefix", "cache_control": {"type": "ephemeral", "ttl": "1h"}},
+        {"type": "text", "text": "suffix"},
+    ]
+    assert wire["tools"][0]["cache_control"] == {"type": "ephemeral", "ttl": "1h"}
+    assert wire["messages"][0]["content"] == [
+        {"type": "text", "text": "hi", "cache_control": {"type": "ephemeral", "ttl": "5m"}},
+    ]
+
+
+@pytest.mark.parametrize("stream", [False, True])
+@pytest.mark.parametrize("policy", [
+    {"type": "ephemeral"}, {"type": "ephemeral", "ttl": "5m"},
+    {"type": "ephemeral", "ttl": "1h"},
+])
+async def test_automatic_cache_policy_reaches_provider(anthropic_provider, stream, policy):
+    wire = await _send(anthropic_provider, stream=stream, cache_control=policy)
+    # The documented automatic-policy equivalent is an explicit breakpoint
+    # on the last cacheable block, avoiding Anthropic-only top-level extras
+    # when the same native ingress resolves to another provider.
+    assert wire["messages"][0]["content"] == [
+        {"type": "text", "text": "hi", "cache_control": policy},
+    ]
+    assert "cache_control" not in wire
+
+
+async def test_automatic_policy_uses_last_rendered_block_after_existing_grouping(anthropic_provider):
+    wire = await _send(anthropic_provider,
+        cache_control={"type": "ephemeral", "ttl": "1h"},
+        tools=[{"name": "lookup", "input_schema": {"type": "object"}}],
+        messages=[
+            {"role": "user", "content": "lookup"},
+            {"role": "assistant", "content": [
+                {"type": "tool_use", "id": "toolu_1", "name": "lookup", "input": {}},
+                {"type": "text", "text": "checking"},
+            ]},
+        ],
+    )
+    # The established translator groups assistant text before calls. The
+    # automatic policy must cover the final rendered call, as a top-level
+    # policy on this same Anthropic payload would, rather than stop early.
+    assert wire["messages"][-1]["content"] == [
+        {"type": "text", "text": "checking"},
+        {"type": "tool_use", "id": "toolu_1", "name": "lookup", "input": {},
+         "cache_control": {"type": "ephemeral", "ttl": "1h"}},
+    ]
+
+
+@pytest.mark.parametrize("stream", [False, True])
+async def test_tool_use_and_result_breakpoints_keep_content(anthropic_provider, stream):
+    wire = await _send(anthropic_provider, stream=stream,
+        tools=[{"name": "lookup", "input_schema": {"type": "object"}}],
+        messages=[
+            {"role": "user", "content": "lookup"},
+            {"role": "assistant", "content": [
+                {"type": "text", "text": "checking", "cache_control": {"type": "ephemeral", "ttl": "1h"}},
+                {"type": "tool_use", "id": "toolu_1", "name": "lookup", "input": {"q": "x"},
+                 "cache_control": {"type": "ephemeral", "ttl": "1h"}},
+            ]},
+            {"role": "user", "content": [
+                {"type": "tool_result", "tool_use_id": "toolu_1", "content": "found",
+                 "cache_control": {"type": "ephemeral", "ttl": "5m"}},
+            ]},
+        ],
+    )
+    assert wire["messages"][1]["content"] == [
+        {"type": "text", "text": "checking", "cache_control": {"type": "ephemeral", "ttl": "1h"}},
+        {"type": "tool_use", "id": "toolu_1", "name": "lookup", "input": {"q": "x"},
+         "cache_control": {"type": "ephemeral", "ttl": "1h"}},
+    ]
+    assert wire["messages"][2]["content"] == [
+        {"type": "tool_result", "tool_use_id": "toolu_1", "content": "found",
+         "cache_control": {"type": "ephemeral", "ttl": "5m"}},
+    ]
+
+
+@pytest.mark.parametrize("stream", [False, True])
+async def test_native_cache_policy_does_not_leak_to_openai_wire(openai_provider, stream):
+    wire = await _send(openai_provider, stream=stream, model="gpt-4o-mini",
+        cache_control={"type": "ephemeral", "ttl": "1h"},
+        system=[{"type": "text", "text": "prefix", "cache_control": {"type": "ephemeral", "ttl": "1h"}}],
+        messages=[
+            {"role": "user", "content": "look"},
+            {"role": "assistant", "content": [
+                {"type": "tool_use", "id": "toolu_1", "name": "lookup", "input": {},
+                 "cache_control": {"type": "ephemeral", "ttl": "1h"}},
+            ]},
+            {"role": "user", "content": [
+                {"type": "tool_result", "tool_use_id": "toolu_1", "content": [
+                    {"type": "text", "text": "found"},
+                    {"type": "image", "source": {
+                        "type": "base64", "media_type": "image/png", "data": "aGk=",
+                    }},
+                ]},
+            ]},
+        ],
+    )
+    assert "cache_control" not in json.dumps(wire)
+    tool_message = next(m for m in wire["messages"] if m["role"] == "tool")
+    assert tool_message["tool_call_id"] == "toolu_1"
+    assert tool_message["content"] == "found"
+    assert wire["messages"][-1]["role"] == "user"
+    assert wire["messages"][-1]["content"][-1] == {
+        "type": "image_url", "image_url": {"url": "data:image/png;base64,aGk="},
+    }
+
+
+@pytest.mark.parametrize("stream", [False, True])
+async def test_marked_image_and_nested_tool_content_reach_provider(anthropic_provider, stream):
+    image = {"type": "image", "source": {
+        "type": "base64", "media_type": "image/png", "data": "aGk=",
+    }, "cache_control": {"type": "ephemeral", "ttl": "5m"}}
+    wire = await _send(anthropic_provider, stream=stream,
+        tools=[{"name": "lookup", "input_schema": {"type": "object"}}],
+        messages=[
+            {"role": "user", "content": [image]},
+            {"role": "assistant", "content": [
+                {"type": "tool_use", "id": "toolu_1", "name": "lookup", "input": {}},
+            ]},
+            {"role": "user", "content": [
+                {"type": "tool_result", "tool_use_id": "toolu_1", "content": [
+                    {"type": "text", "text": "found", "cache_control": {"type": "ephemeral", "ttl": "5m"}},
+                    image,
+                ]},
+            ]},
+        ],
+    )
+    assert wire["messages"][0]["content"] == [image]
+    assert wire["messages"][2]["content"] == [
+        {"type": "tool_result", "tool_use_id": "toolu_1", "content": [
+            {"type": "text", "text": "found", "cache_control": {"type": "ephemeral", "ttl": "5m"}},
+            image,
+        ]},
+    ]
diff --git a/tests/unit/test_anthropic_translation.py b/tests/unit/test_anthropic_translation.py
index 596a2ded..1eb957c2 100644
--- a/tests/unit/test_anthropic_translation.py
+++ b/tests/unit/test_anthropic_translation.py
@@ -65,14 +65,97 @@ def test_system_string_becomes_leading_system_message():
     assert out["messages"][1]["role"] == "user"
 
 
-def test_system_blocks_join_text_and_drop_cache_control():
+def test_system_blocks_preserve_cache_control_and_boundaries():
     out = to_openai_request(_req(system=[
         {"type": "text", "text": "part one", "cache_control": {"type": "ephemeral"}},
         {"type": "text", "text": "part two"},
     ]))
+    assert out["messages"][0] == {"role": "system", "content": [
+        {"type": "text", "text": "part one", "cache_control": {"type": "ephemeral"}},
+        {"type": "text", "text": "part two"},
+    ]}
+
+
+def test_unmarked_system_blocks_still_join_text():
+    out = to_openai_request(_req(system=[
+        {"type": "text", "text": "part one"}, {"type": "text", "text": "part two"},
+    ]))
     assert out["messages"][0] == {"role": "system", "content": "part one\n\npart two"}
 
 
+@pytest.mark.parametrize("last_type", ["text", "tool_use", "tool_result"])
+def test_automatic_policy_marks_last_cacheable_block_without_mutating_input(last_type):
+    block = {
+        "text": {"type": "text", "text": "answer"},
+        "tool_use": {"type": "tool_use", "id": "toolu_1", "name": "f", "input": {}},
+        "tool_result": {"type": "tool_result", "tool_use_id": "toolu_1", "content": "result"},
+    }[last_type]
+    role = "user" if last_type == "tool_result" else "assistant"
+    req = _req(cache_control={"type": "ephemeral", "ttl": "1h"}, messages=[
+        {"role": role, "content": [block, {"type": "thinking", "thinking": "hidden"}]},
+    ])
+    before = req.model_dump()
+    out = to_openai_request(req)
+    assert req.model_dump() == before
+    last = out["messages"][-1]
+    if last_type == "tool_use":
+        marked = last["tool_calls"][0]
+    elif last_type == "tool_result":
+        marked = last
+    else:
+        marked = last["content"][0]
+    assert marked["cache_control"] == {"type": "ephemeral", "ttl": "1h"}
+
+
+def test_automatic_policy_keeps_matching_explicit_marker():
+    out = to_openai_request(_req(cache_control={"type": "ephemeral", "ttl": "5m"}, messages=[
+        {"role": "user", "content": [
+            {"type": "text", "text": "hi", "cache_control": {"type": "ephemeral"}},
+        ]},
+    ]))
+    assert out["messages"][0]["content"][0]["cache_control"] == {"type": "ephemeral"}
+
+
+def test_automatic_policy_rejects_conflicting_last_marker_ttl():
+    with pytest.raises(HTTPException) as exc:
+        to_openai_request(_req(cache_control={"type": "ephemeral", "ttl": "1h"}, messages=[
+            {"role": "user", "content": [
+                {"type": "text", "text": "hi", "cache_control": {"type": "ephemeral"}},
+            ]},
+        ]))
+    assert exc.value.status_code == 400
+
+
+@pytest.mark.parametrize("existing,allowed", [(3, True), (4, False)])
+def test_automatic_policy_respects_breakpoint_slots(existing, allowed):
+    req = _req(cache_control={"type": "ephemeral"}, system=[
+        {"type": "text", "text": str(i), "cache_control": {"type": "ephemeral"}}
+        for i in range(existing)
+    ])
+    if allowed:
+        out = to_openai_request(req)
+        assert out["messages"][-1]["content"][0]["cache_control"] == {"type": "ephemeral"}
+    else:
+        with pytest.raises(HTTPException) as exc:
+            to_openai_request(req)
+        assert exc.value.status_code == 400
+
+
+def test_automatic_policy_skips_when_no_cacheable_content():
+    out = to_openai_request(_req(cache_control={"type": "ephemeral"}, messages=[
+        {"role": "assistant", "content": [{"type": "thinking", "thinking": "hidden"}]},
+    ]))
+    assert out["messages"] == [{"role": "assistant", "content": ""}]
+
+
+def test_cache_marker_does_not_enable_unsupported_document_block():
+    with pytest.raises(HTTPException) as exc:
+        to_openai_request(_req(messages=[{"role": "user", "content": [
+            {"type": "document", "cache_control": {"type": "ephemeral"}},
+        ]}]))
+    assert exc.value.status_code == 400
+
+
 def test_image_base64_block_becomes_data_uri_part():
     out = to_openai_request(_req(messages=[{
         "role": "user",