feat: implement Anthropic prompt caching with cache_control markers (#10)

* feat: implement Anthropic prompt caching with cache_control markers

Add active prompt caching to AnthropicProvider by sending cache_control
markers to the Anthropic API, matching the pi-agent-core TS reference
implementation. Previously, the provider only passively tracked cache
token counts without requesting caching.

Changes:
- Add cache_control: {"type": "ephemeral"} on system prompt content blocks
- Add cache_control on the last tool definition (when tools are provided)
- Add cache_control on the last message content block (conversation prefix)
- Add CacheRetention type ("short" | "long" | "none") and cache_retention
  parameter to AnthropicProvider (default: "short")
- System prompt now sent as structured content block array (required for
  cache_control attachment)
- Add 15 tests covering all caching behavior

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* fix: add 1h TTL for long cache retention to match Anthropic API

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

Author: xfgong

cubepi/providers/anthropic.py

@@ -2,7 +2,7 @@
 
 import asyncio
 import time
-from typing import Any
+from typing import Any, Literal
 
 from cubepi.providers.base import (
     AssistantMessage,
@@ -23,12 +23,20 @@
     adjust_max_tokens_for_thinking,
 )
 
+CacheRetention = Literal["short", "long", "none"]
+
 
 class AnthropicProvider:
-    def __init__(self, *, api_key: str | None = None) -> None:
+    def __init__(
+        self,
+        *,
+        api_key: str | None = None,
+        cache_retention: CacheRetention = "short",
+    ) -> None:
         import anthropic
 
         self._client = anthropic.AsyncAnthropic(api_key=api_key)
+        self._cache_retention = cache_retention
 
     async def stream(
         self,
@@ -43,9 +51,11 @@ async def stream(
     ) -> MessageStream:
         ms = MessageStream()
 
+        cache_control = self._get_cache_control()
         api_messages = [self._convert_message(m) for m in messages]
+        if cache_control:
+            self._apply_message_cache_control(api_messages, cache_control)
 
-        # Adjust max_tokens to accommodate the thinking budget
         max_tokens, thinking_budget = adjust_max_tokens_for_thinking(
             base_max_tokens=model.max_tokens,
             model_max_tokens=model.context_window,
@@ -59,9 +69,18 @@ async def stream(
             "max_tokens": max_tokens,
         }
         if system_prompt:
-            kwargs["system"] = system_prompt
+            kwargs["system"] = [
+                {
+                    "type": "text",
+                    "text": system_prompt,
+                    **({"cache_control": cache_control} if cache_control else {}),
+                }
+            ]
         if tools:
-            kwargs["tools"] = [self._convert_tool(t) for t in tools]
+            api_tools = [self._convert_tool(t) for t in tools]
+            if cache_control and api_tools:
+                api_tools[-1]["cache_control"] = cache_control
+            kwargs["tools"] = api_tools
         if thinking != "off" and thinking_budget > 0:
             kwargs["thinking"] = {
                 "type": "enabled",
@@ -118,6 +137,42 @@ async def _produce() -> None:
         asyncio.create_task(_produce())
         return ms
 
+    def _get_cache_control(self) -> dict[str, str] | None:
+        if self._cache_retention == "none":
+            return None
+        cc: dict[str, str] = {"type": "ephemeral"}
+        if self._cache_retention == "long":
+            cc["ttl"] = "1h"
+        return cc
+
+    @staticmethod
+    def _apply_message_cache_control(
+        api_messages: list[dict[str, Any]],
+        cache_control: dict[str, str],
+    ) -> None:
+        """Apply cache_control to the last content block of the last message.
+
+        This caches the conversation prefix so subsequent turns get cache hits
+        on all prior messages.
+        """
+        if not api_messages:
+            return
+
+        last_msg = api_messages[-1]
+        content = last_msg.get("content")
+        if not content:
+            return
+
+        if isinstance(content, list) and len(content) > 0:
+            last_block = content[-1]
+            if isinstance(last_block, dict):
+                last_block["cache_control"] = cache_control
+        elif isinstance(content, str):
+            # Convert bare string content to a block so we can attach cache_control
+            last_msg["content"] = [
+                {"type": "text", "text": content, "cache_control": cache_control}
+            ]
+
     @staticmethod
     def _convert_message(msg: Message) -> dict[str, Any]:
         if isinstance(msg, UserMessage):

tests/providers/test_anthropic.py

@@ -1,4 +1,4 @@
-from cubepi.providers.anthropic import AnthropicProvider
+from cubepi.providers.anthropic import AnthropicProvider, CacheRetention
 from cubepi.providers.base import (
     TextContent,
     ToolCall,
@@ -60,3 +60,181 @@ def test_convert_tool_definition(self):
         assert result["name"] == "search"
         assert result["description"] == "Search the web"
         assert result["input_schema"]["type"] == "object"
+
+
+# ---------------------------------------------------------------------------
+# Prompt caching tests
+# ---------------------------------------------------------------------------
+
+
+def _make_provider(retention: CacheRetention = "short") -> AnthropicProvider:
+    """Create a provider without hitting the network (api_key is unused in tests)."""
+    return AnthropicProvider(api_key="test-key", cache_retention=retention)
+
+
+class TestCacheRetention:
+    def test_default_retention_is_short(self):
+        provider = AnthropicProvider(api_key="test-key")
+        assert provider._cache_retention == "short"
+
+    def test_retention_none_returns_no_cache_control(self):
+        provider = _make_provider("none")
+        assert provider._get_cache_control() is None
+
+    def test_retention_short_returns_ephemeral(self):
+        provider = _make_provider("short")
+        cc = provider._get_cache_control()
+        assert cc == {"type": "ephemeral"}
+
+    def test_retention_long_returns_ephemeral_with_ttl(self):
+        provider = _make_provider("long")
+        cc = provider._get_cache_control()
+        assert cc == {"type": "ephemeral", "ttl": "1h"}
+
+
+class TestCacheControlOnMessages:
+    """Verify cache_control markers are placed on the last message content block."""
+
+    CACHE_CONTROL = {"type": "ephemeral"}
+
+    def test_cache_control_on_last_user_message_text(self):
+        msgs = [
+            {"role": "user", "content": [{"type": "text", "text": "first"}]},
+            {"role": "assistant", "content": [{"type": "text", "text": "reply"}]},
+            {"role": "user", "content": [{"type": "text", "text": "second"}]},
+        ]
+        AnthropicProvider._apply_message_cache_control(msgs, self.CACHE_CONTROL)
+
+        # Only the last message's last block should have cache_control
+        assert msgs[-1]["content"][-1]["cache_control"] == self.CACHE_CONTROL
+        assert "cache_control" not in msgs[0]["content"][0]
+        assert "cache_control" not in msgs[1]["content"][0]
+
+    def test_cache_control_on_last_tool_result(self):
+        msgs = [
+            {
+                "role": "user",
+                "content": [
+                    {
+                        "type": "tool_result",
+                        "tool_use_id": "tc-1",
+                        "content": [{"type": "text", "text": "ok"}],
+                        "is_error": False,
+                    }
+                ],
+            }
+        ]
+        AnthropicProvider._apply_message_cache_control(msgs, self.CACHE_CONTROL)
+        assert msgs[0]["content"][-1]["cache_control"] == self.CACHE_CONTROL
+
+    def test_cache_control_on_multi_block_message(self):
+        msgs = [
+            {
+                "role": "user",
+                "content": [
+                    {"type": "text", "text": "part one"},
+                    {"type": "text", "text": "part two"},
+                ],
+            }
+        ]
+        AnthropicProvider._apply_message_cache_control(msgs, self.CACHE_CONTROL)
+        # Only the last block gets the marker
+        assert "cache_control" not in msgs[0]["content"][0]
+        assert msgs[0]["content"][1]["cache_control"] == self.CACHE_CONTROL
+
+    def test_cache_control_converts_bare_string_content(self):
+        msgs = [{"role": "user", "content": "bare string"}]
+        AnthropicProvider._apply_message_cache_control(msgs, self.CACHE_CONTROL)
+        # Should have been converted to a list with a text block
+        assert isinstance(msgs[0]["content"], list)
+        assert msgs[0]["content"][0]["type"] == "text"
+        assert msgs[0]["content"][0]["text"] == "bare string"
+        assert msgs[0]["content"][0]["cache_control"] == self.CACHE_CONTROL
+
+    def test_empty_messages_is_noop(self):
+        msgs: list[dict] = []
+        # Should not raise
+        AnthropicProvider._apply_message_cache_control(msgs, self.CACHE_CONTROL)
+        assert msgs == []
+
+    def test_no_cache_control_when_retention_none(self):
+        provider = _make_provider("none")
+        assert provider._get_cache_control() is None
+
+
+class TestCacheControlOnSystemPrompt:
+    """Verify that the stream method builds system prompt blocks with cache_control."""
+
+    def test_system_prompt_has_cache_control(self):
+        """The system prompt should be sent as a content block with cache_control."""
+        provider = _make_provider("short")
+        cache_control = provider._get_cache_control()
+        # Simulate what stream() does for system_prompt
+        system_prompt = "You are a helpful assistant."
+        system_block = {
+            "type": "text",
+            "text": system_prompt,
+            **({"cache_control": cache_control} if cache_control else {}),
+        }
+        assert system_block["cache_control"] == {"type": "ephemeral"}
+
+    def test_system_prompt_no_cache_when_retention_none(self):
+        provider = _make_provider("none")
+        cache_control = provider._get_cache_control()
+        system_prompt = "You are a helpful assistant."
+        system_block = {
+            "type": "text",
+            "text": system_prompt,
+            **({"cache_control": cache_control} if cache_control else {}),
+        }
+        assert "cache_control" not in system_block
+
+
+class TestCacheControlOnTools:
+    """Verify cache_control is applied to the last tool definition."""
+
+    CACHE_CONTROL = {"type": "ephemeral"}
+
+    def _make_tools(self, count: int) -> list[ToolDefinition]:
+        return [
+            ToolDefinition(
+                name=f"tool_{i}",
+                description=f"Tool {i}",
+                parameters={
+                    "type": "object",
+                    "properties": {"x": {"type": "string"}},
+                },
+            )
+            for i in range(count)
+        ]
+
+    def test_cache_control_on_last_tool_only(self):
+        tools = self._make_tools(3)
+        api_tools = [AnthropicProvider._convert_tool(t) for t in tools]
+        # Apply cache_control the same way stream() does
+        if api_tools:
+            api_tools[-1]["cache_control"] = self.CACHE_CONTROL
+
+        assert "cache_control" not in api_tools[0]
+        assert "cache_control" not in api_tools[1]
+        assert api_tools[2]["cache_control"] == self.CACHE_CONTROL
+
+    def test_single_tool_gets_cache_control(self):
+        tools = self._make_tools(1)
+        api_tools = [AnthropicProvider._convert_tool(t) for t in tools]
+        if api_tools:
+            api_tools[-1]["cache_control"] = self.CACHE_CONTROL
+
+        assert api_tools[0]["cache_control"] == self.CACHE_CONTROL
+
+    def test_no_cache_control_when_retention_none(self):
+        provider = _make_provider("none")
+        cache_control = provider._get_cache_control()
+        tools = self._make_tools(2)
+        api_tools = [AnthropicProvider._convert_tool(t) for t in tools]
+        if cache_control and api_tools:
+            api_tools[-1]["cache_control"] = cache_control
+
+        # With retention="none", no tool should have cache_control
+        for tool in api_tools:
+            assert "cache_control" not in tool