Add Anthropic prompt-cache clear-then-finalize

core/providers/anthropic.py

@@ -428,6 +428,59 @@ def _convert_tool_choice(
                 return {"type": "tool", "name": name}
         return {"type": "auto"}
 
+    @staticmethod
+    def _strip_cache_control(message: dict[str, Any]) -> dict[str, Any]:
+        """Return a copy of ``message`` with any cache_control marker removed.
+
+        Strips markers both from the message dict itself and from any content
+        block (string or block-list form) so a replayed request never carries
+        a stale breakpoint into the next finalize step.
+        """
+        if "cache_control" in message:
+            message = {k: v for k, v in message.items() if k != "cache_control"}
+        content = message.get("content")
+        if isinstance(content, list):
+            cleaned: list[Any] = []
+            changed = False
+            for block in content:
+                if isinstance(block, dict) and "cache_control" in block:
+                    cleaned.append(
+                        {k: v for k, v in block.items() if k != "cache_control"}
+                    )
+                    changed = True
+                else:
+                    cleaned.append(block)
+            if changed:
+                message = {**message, "content": cleaned}
+        return message
+
+    @staticmethod
+    def _place_cache_control(
+        message: dict[str, Any], marker: dict[str, Any]
+    ) -> dict[str, Any]:
+        """Return a copy of ``message`` with a fresh cache_control marker.
+
+        Handles both the plain-string content shape and the block-list shape
+        (used for tool_result / multimodal content). A message with no
+        cacheable content is returned unchanged — an image-only or empty
+        result must not host a breakpoint.
+        """
+        content = message.get("content")
+        if isinstance(content, str) and content:
+            return {
+                **message,
+                "content": [{"type": "text", "text": content, "cache_control": marker}],
+            }
+        if isinstance(content, list) and content:
+            nc = list(content)
+            nc[-1] = (
+                {**nc[-1], "cache_control": marker}
+                if isinstance(nc[-1], dict)
+                else {"type": "text", "text": str(nc[-1]), "cache_control": marker}
+            )
+            return {**message, "content": nc}
+        return message
+
     # ------------------------------------------------------------------
     # Prompt caching
     # ------------------------------------------------------------------
@@ -449,19 +502,23 @@ def _apply_cache_control(
             system = list(system)
             system[-1] = {**system[-1], "cache_control": marker}
 
-        new_msgs = list(messages)
-        if len(new_msgs) >= 3:
-            m = new_msgs[-2]
-            c = m.get("content")
-            if isinstance(c, str):
-                new_msgs[-2] = {
-                    **m,
-                    "content": [{"type": "text", "text": c, "cache_control": marker}],
-                }
-            elif isinstance(c, list) and c:
-                nc = list(c)
-                nc[-1] = {**nc[-1], "cache_control": marker}
-                new_msgs[-2] = {**m, "content": nc}
+        # --- Message cache breakpoint: clear-then-finalize ---
+        # The previous write point hardcoded messages[-2], which broke when the
+        # final message was not a cacheable content block (a tool_result
+        # carrying an image, a trailing system note, a truncated turn). Now
+        # (1) strip every stale marker from the whole message list, so a reused
+        # or replayed list can never carry a leftover breakpoint that would
+        # pin the cache at the wrong offset, then (2) place a single fresh
+        # marker on the *latest non-system message*. Anchoring at the newest
+        # cacheable message (instead of a fixed -2) keeps the write point
+        # exactly at the conversation frontier, maximizing the prefix that
+        # subsequent turns hit.
+        new_msgs = [cls._strip_cache_control(m) for m in messages]
+        for idx in range(len(new_msgs) - 1, -1, -1):
+            if new_msgs[idx].get("role") == "system":
+                continue
+            new_msgs[idx] = cls._place_cache_control(new_msgs[idx], marker)
+            break
 
         new_tools = tools
         if tools:

tests/test_anthropic_cache_control.py

@@ -0,0 +1,234 @@
+"""Anthropic prompt caching — clear-then-finalize at the conversation frontier.
+
+Every test drives ``AnthropicProvider._build_kwargs`` so the input messages go
+through the provider's own conversion exactly like a live request; assertions
+run on the converted Anthropic-format payload inside ``kwargs``.
+
+Contract under test:
+
+- the message breakpoint is written clear-then-finalize: every stale
+  ``cache_control`` is stripped from the converted list, then ONE fresh
+  marker lands on the LATEST message — the previous fixed ``messages[-2]``
+  write point missed the frontier whenever the tail shape changed;
+- a trailing system note never steals the frontier (it folds into the system
+  prompt, and the marker stays on the last real message);
+- the per-request marker budget is at most 4 (system 1 + tools <= 2 +
+  message frontier 1), with tool markers on the last builtin and the MCP tail;
+- a string system prompt is wrapped into exactly one marked block;
+- ``supports_caching=False`` emits no markers at all, and empty tool lists
+  never produce a ``tools`` kwarg.
+"""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+from typing import Any
+
+ROOT = Path(__file__).resolve().parents[1]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
+from core.providers.anthropic import AnthropicProvider
+
+
+def _provider() -> AnthropicProvider:
+    provider = AnthropicProvider(api_key="test-key", default_model="claude-sonnet-4-6")
+    provider.reasoning_supported = False
+    return provider
+
+
+def _build(
+    messages: list[dict[str, Any]],
+    tools: list[dict[str, Any]] | None = None,
+    *,
+    supports_caching: bool = True,
+) -> dict[str, Any]:
+    """Run the real kwargs path: OpenAI-format in, Anthropic-format payload out."""
+    return _provider()._build_kwargs(
+        messages,
+        tools,
+        "claude-sonnet-4-6",
+        1024,
+        0.0,
+        None,
+        None,
+        supports_caching=supports_caching,
+    )
+
+
+def _marked_indices(messages: list[dict[str, Any]]) -> list[int]:
+    """Indices of messages carrying a cache_control marker (key or block)."""
+    marked: list[int] = []
+    for i, msg in enumerate(messages):
+        if "cache_control" in msg:
+            marked.append(i)
+            continue
+        content = msg.get("content")
+        if isinstance(content, list) and any(
+            isinstance(b, dict) and "cache_control" in b for b in content
+        ):
+            marked.append(i)
+    return marked
+
+
+def _count_markers(*structures: Any) -> int:
+    """How many cache_control markers live across the given structures."""
+    total = 0
+    for items in structures:
+        if not items:
+            continue
+        for item in items:
+            if not isinstance(item, dict):
+                continue
+            if "cache_control" in item:
+                total += 1
+            content = item.get("content")
+            if isinstance(content, list):
+                total += sum(
+                    1 for b in content if isinstance(b, dict) and "cache_control" in b
+                )
+    return total
+
+
+def _conversation() -> list[dict[str, Any]]:
+    return [
+        {"role": "user", "content": "read the file"},
+        {
+            "role": "assistant",
+            "content": None,
+            "tool_calls": [
+                {
+                    "id": "call_1",
+                    "type": "function",
+                    "function": {"name": "read", "arguments": "{}"},
+                }
+            ],
+        },
+        {"role": "tool", "tool_call_id": "call_1", "content": "file body"},
+        {"role": "user", "content": "thanks"},
+    ]
+
+
+def _tools(*names: str) -> list[dict[str, Any]]:
+    return [
+        {"type": "function", "function": {"name": name, "parameters": {}}}
+        for name in names
+    ]
+
+
+# ---------------------------------------------------------------------------
+# Frontier placement (the change under review)
+# ---------------------------------------------------------------------------
+
+
+def test_frontier_marker_lands_on_latest_message() -> None:
+    kwargs = _build(_conversation())
+    out = kwargs["messages"]
+    assert len(out) >= 3
+    # ONE message-level breakpoint, on the LATEST message — not the fixed
+    # messages[-2] write point, which marked the tool result instead.
+    assert _marked_indices(out) == [len(out) - 1]
+
+
+def test_trailing_system_note_does_not_steal_the_frontier() -> None:
+    kwargs = _build(_conversation() + [{"role": "system", "content": "[note]"}])
+    out = kwargs["messages"]
+    # The note folds into the system prompt; the frontier marker stays on the
+    # last real conversation message.
+    assert "[note]" in str(kwargs.get("system"))
+    assert _marked_indices(out) == [len(out) - 1]
+
+
+def test_stale_markers_are_stripped_and_one_frontier_marker_remains() -> None:
+    messages = _conversation()
+    # A replayed list: a stale marker on an early user message, exactly the
+    # leftover state the old write point could never clean up.
+    messages[0] = {
+        "role": "user",
+        "content": [
+            {
+                "type": "text",
+                "text": "read the file",
+                "cache_control": {"type": "ephemeral"},
+            }
+        ],
+    }
+
+    kwargs = _build(messages)
+    out = kwargs["messages"]
+
+    # Exactly one message-level breakpoint, on the LATEST message.
+    assert _marked_indices(out) == [len(out) - 1]
+    # The stale marker on the early user message is gone after conversion.
+    first = out[0]
+    assert isinstance(first.get("content"), list)
+    assert all("cache_control" not in b for b in first["content"])
+
+
+# ---------------------------------------------------------------------------
+# Marker budget
+# ---------------------------------------------------------------------------
+
+
+def test_marker_budget_never_exceeds_four() -> None:
+    kwargs = _build(
+        _conversation(),
+        _tools(
+            "read",
+            "bash",
+            "edit",
+            "mcp__github__createIssue",
+            "mcp__github__listIssues",
+        ),
+    )
+    total = _count_markers(
+        kwargs.get("system") if isinstance(kwargs.get("system"), list) else None,
+        kwargs["messages"],
+        kwargs.get("tools"),
+    )
+    assert total <= 4
+    assert total >= 3  # system 1 + message 1 + at least one tool
+
+
+def test_tool_markers_cover_last_builtin_and_tail() -> None:
+    kwargs = _build(_conversation(), _tools("read", "bash", "mcp__ghidra__decompile"))
+    out_tools = kwargs.get("tools") or []
+    marked = [i for i, t in enumerate(out_tools) if "cache_control" in t]
+    # Last builtin (bash, index 1) and the MCP tail (index 2), never more.
+    assert marked == [1, 2]
+
+
+# ---------------------------------------------------------------------------
+# System + non-caching paths
+# ---------------------------------------------------------------------------
+
+
+def test_string_system_becomes_one_marked_block() -> None:
+    kwargs = _build(
+        [{"role": "system", "content": "base instructions"}] + _conversation()
+    )
+    assert kwargs["system"] == [
+        {
+            "type": "text",
+            "text": "base instructions",
+            "cache_control": {"type": "ephemeral"},
+        }
+    ]
+
+
+def test_no_markers_when_caching_disabled() -> None:
+    kwargs = _build(
+        [{"role": "system", "content": "base"}] + _conversation(),
+        _tools("read", "bash", "edit"),
+        supports_caching=False,
+    )
+    system = kwargs.get("system")
+    assert system == "base"  # plain string, never block-wrapped on this path
+    assert _count_markers(None, kwargs["messages"], kwargs.get("tools")) == 0
+    assert "cache_control" not in system
+
+
+def test_tools_absent_when_none_or_empty() -> None:
+    assert "tools" not in _build(_conversation(), None)
+    assert "tools" not in _build(_conversation(), [])