feat(llm): enable Bedrock/Anthropic prompt caching for Claude models (#772)

Co-authored-by: Sean Turner <sean.turner@zerohash.com>
Co-authored-by: Ahmed Allam <ahmed39652003@gmail.com>

tests/test_inputs.py

@@ -5,6 +5,7 @@ from __future__ import annotations
 from itertools import pairwise
 from typing import Any
 
+import litellm
 import pytest
 
 from strix.core.inputs import build_root_task, child_initial_input, make_model_settings
@@ -55,6 +56,103 @@ def test_child_initial_input_no_consecutive_same_role(parent_history: list[Any])
     assert all(prev != nxt for prev, nxt in pairwise(roles))
 
 
+def _cache_points(model_name: str) -> Any:
+    extra = make_model_settings(None, model_name=model_name).extra_args or {}
+    return extra.get("cache_control_injection_points")
+
+
+def test_make_model_settings_enables_prompt_cache_for_bedrock_claude() -> None:
+    assert _cache_points("bedrock/global.anthropic.claude-opus-4-8") == [
+        {"location": "message", "role": "system"},
+        {"location": "tool_config"},
+        {"location": "message", "index": -1},
+    ]
+
+
+@pytest.mark.parametrize(
+    "model_name",
+    [
+        "anthropic/claude-sonnet-4-5",
+        "openrouter/anthropic/claude-3.5-sonnet",
+        "vertex_ai/claude-sonnet-4-5",
+    ],
+)
+def test_make_model_settings_enables_prompt_cache_for_non_bedrock_claude(model_name: str) -> None:
+    assert _cache_points(model_name) == [
+        {"location": "message", "role": "system"},
+        {"location": "message", "index": -1},
+    ]
+
+
+def test_tool_config_point_not_leaked_to_non_bedrock_claude() -> None:
+    # LiteLLM only consumes tool_config on Bedrock; elsewhere it leaks onto the
+    # wire and native Anthropic 400s.
+    for model in ("anthropic/claude-sonnet-4-5", "openrouter/anthropic/claude-3.5-sonnet"):
+        points = _cache_points(model) or []
+        assert all(p.get("location") != "tool_config" for p in points)
+
+
+def test_prompt_cache_can_be_disabled() -> None:
+    assert (
+        make_model_settings(
+            None, model_name="anthropic/claude-sonnet-4-5", prompt_cache=False
+        ).extra_args
+        is None
+    )
+
+
+@pytest.mark.parametrize("model_name", ["gpt-5", "vertex_ai/gemini-2.5-pro", "openai/o3"])
+def test_make_model_settings_no_prompt_cache_for_non_claude(model_name: str) -> None:
+    assert make_model_settings(None, model_name=model_name).extra_args is None
+
+
+def test_no_prompt_cache_for_unmapped_bedrock_claude_model(monkeypatch: Any) -> None:
+    # A Bedrock Claude model LiteLLM hasn't mapped must run uncached, not crash.
+    unmapped = "bedrock/global.anthropic.claude-brand-new-9"
+    monkeypatch.setattr(litellm, "model_cost", {}, raising=False)
+    if getattr(getattr(litellm, "utils", None), "supports_prompt_caching", None):
+        monkeypatch.setattr(litellm.utils, "supports_prompt_caching", lambda *_a, **_k: False)
+
+    assert make_model_settings(None, model_name=unmapped).extra_args is None
+
+
+def test_prompt_cache_kept_for_non_bedrock_claude_even_if_unmapped(monkeypatch: Any) -> None:
+    # Only Bedrock hard-rejects unknown cache fields, so only Bedrock is guarded.
+    monkeypatch.setattr(litellm, "model_cost", {}, raising=False)
+    if getattr(getattr(litellm, "utils", None), "supports_prompt_caching", None):
+        monkeypatch.setattr(litellm.utils, "supports_prompt_caching", lambda *_a, **_k: False)
+
+    for model in ("anthropic/claude-brand-new-9", "openrouter/anthropic/claude-brand-new"):
+        assert _cache_points(model) == [
+            {"location": "message", "role": "system"},
+            {"location": "message", "index": -1},
+        ]
+
+
+def test_conversation_tail_breakpoint_moves_with_appended_transcript() -> None:
+    # LiteLLM must place the index=-1 cache_control on the last message however
+    # long the transcript grows.
+    hook_mod = pytest.importorskip("litellm.integrations.anthropic_cache_control_hook")
+    apply = hook_mod.AnthropicCacheControlHook._apply_message_injections
+    points = _cache_points("bedrock/global.anthropic.claude-opus-4-8")
+    msg_points = [p for p in points if p.get("location") == "message"]
+
+    def last_msg_cache_control(n_turns: int) -> Any:
+        messages: list[dict[str, Any]] = [{"role": "system", "content": "stable prompt"}]
+        for i in range(n_turns):
+            messages.append({"role": "assistant", "content": f"turn {i} action"})
+            messages.append({"role": "user", "content": f"turn {i} tool result"})
+        processed = apply(msg_points, messages, 4)
+        last = processed[-1]
+        content = last.get("content")
+        if isinstance(content, list):
+            return content[-1].get("cache_control")
+        return last.get("cache_control")
+
+    assert last_msg_cache_control(2) == {"type": "ephemeral"}
+    assert last_msg_cache_control(20) == {"type": "ephemeral"}
+
+
 def test_build_root_task_empty_config() -> None:
     assert build_root_task({}) == ""