feat(anthropic_sdk): default-on automatic prompt caching

Adds 'cache_control: {type: ephemeral}' to messages.stream() calls. The
API auto-places a cache breakpoint at the longest cacheable prefix
(tools + system + accumulated messages) and moves it forward on each
turn -- multi-turn agent loops get cache reads on every turn after the
first.

Default-on because all current Claude models support cache_control and
CAPI accepts it (validated end-to-end against claude-mythos-5 via CAPI
on 2026-06-12). Callers pointed at proxies that strip / reject
cache_control can opt out with 'prompt_caching: false' in model_settings.

A string value (e.g. 'prompt_caching: 1h') sets a custom TTL.

Local validation against anticomputer/vulnerable-test-app on the same
audit pipeline, same model config, only changing prompt_caching:

  metric              | off       | on        | delta
  --------------------+-----------+-----------+------------
  requests            | 60        | 62        | +2 (noise)
  input tokens fresh  | 909,806   | 124       | -99.99%
  cache read tokens   | 0         | 728,079   | new
  cache write tokens  | 0         | 210,261   | new
  output tokens       | 42,300    | 44,933    | similar
  vulnerabilities     | 4         | 5         | +1
  est. mythos cost    | $11.21   | $5.60    | -50%

Same or better audit quality, half the token cost. Real audits with
larger system prompts + more tool definitions amortize the cache writes
over more reads, so production savings are typically larger than 50%.

Tests added:
- prompt_caching default-on emits cache_control
- prompt_caching=False suppresses cache_control (opt-out)
- prompt_caching='1h' includes the ttl field

23 tests pass total, hatch fmt clean.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: anticomputer

src/seclab_taskflow_agent/sdk/anthropic_sdk/backend.py

@@ -202,6 +202,22 @@ async def run_streamed(
                 create_kwargs["thinking"] = {"type": "adaptive"}
                 create_kwargs["output_config"] = {"effort": effort}
 
+        # Automatic prompt caching: place an ephemeral cache breakpoint at
+        # the longest cacheable prefix (tools + system + accumulated
+        # messages). The breakpoint moves forward on each turn, so
+        # multi-turn agent loops get cache reads on every turn after the
+        # first -- typically 50%+ cost reduction on token-heavy audits.
+        # All current Claude models (and the Anthropic-compatible CAPI
+        # proxy) support cache_control. Default on; explicit opt-out for
+        # callers pointed at proxies that don't support it.
+        prompt_caching = handle.model_settings.get("prompt_caching", True)
+        if prompt_caching:
+            ttl = prompt_caching if isinstance(prompt_caching, str) else "5m"
+            cache_block: dict[str, Any] = {"type": "ephemeral"}
+            if ttl != "5m":
+                cache_block["ttl"] = ttl
+            create_kwargs["cache_control"] = cache_block
+
         import anthropic
 
         for turn in range(max_turns):

tests/test_sdk_anthropic_adapter.py

@@ -227,3 +227,157 @@ async def _run():
 
     with pytest.raises(BackendBadRequestError, match="invalid reasoning effort"):
         asyncio.run(_run())
+
+
+# -- prompt caching --
+
+
+def test_prompt_caching_enabled_by_default():
+    """All Claude models support cache_control; default to on so callers
+    get the cost savings without explicit opt-in. Explicit opt-out via
+    prompt_caching=False remains available for proxies that don't support
+    cache_control."""
+    import asyncio
+
+    from seclab_taskflow_agent.sdk.anthropic_sdk.backend import _AnthropicHandle
+
+    captured = {}
+
+    class _FakeStreamCtx:
+        async def __aenter__(self): return self
+        async def __aexit__(self, *exc): return False
+        def __aiter__(self):
+            async def _gen():
+                return
+                yield
+            return _gen()
+        async def get_final_message(self):
+            return type("M", (), {"stop_reason": "end_turn", "content": []})()
+
+    class _FakeMessages:
+        def stream(self, **kwargs):
+            captured.update(kwargs)
+            return _FakeStreamCtx()
+
+    class _FakeClient:
+        def __init__(self):
+            self.messages = _FakeMessages()
+
+    handle = _AnthropicHandle(
+        client=_FakeClient(),
+        system_prompt="",
+        model="claude-mythos-5",
+        max_tokens=100,
+        tools=[],
+        mcp_server_map={},
+        model_settings={},
+    )
+    backend = AnthropicSDKBackend()
+
+    async def _run():
+        async for _ in backend.run_streamed(handle, "hi", max_turns=1):
+            pass
+
+    asyncio.run(_run())
+    assert captured.get("cache_control") == {"type": "ephemeral"}, (
+        f"expected default cache_control={{type: ephemeral}}, got {captured.get('cache_control')!r}"
+    )
+
+
+def test_prompt_caching_explicit_opt_out():
+    """prompt_caching=False must suppress cache_control entirely (for
+    callers pointed at proxies that don't support it)."""
+    import asyncio
+
+    from seclab_taskflow_agent.sdk.anthropic_sdk.backend import _AnthropicHandle
+
+    captured = {}
+
+    class _FakeStreamCtx:
+        async def __aenter__(self): return self
+        async def __aexit__(self, *exc): return False
+        def __aiter__(self):
+            async def _gen():
+                return
+                yield
+            return _gen()
+        async def get_final_message(self):
+            return type("M", (), {"stop_reason": "end_turn", "content": []})()
+
+    class _FakeMessages:
+        def stream(self, **kwargs):
+            captured.update(kwargs)
+            return _FakeStreamCtx()
+
+    class _FakeClient:
+        def __init__(self):
+            self.messages = _FakeMessages()
+
+    handle = _AnthropicHandle(
+        client=_FakeClient(),
+        system_prompt="",
+        model="claude-mythos-5",
+        max_tokens=100,
+        tools=[],
+        mcp_server_map={},
+        model_settings={"prompt_caching": False},
+    )
+    backend = AnthropicSDKBackend()
+
+    async def _run():
+        async for _ in backend.run_streamed(handle, "hi", max_turns=1):
+            pass
+
+    asyncio.run(_run())
+    assert "cache_control" not in captured, (
+        f"cache_control should be absent when explicitly opted out, got {captured}"
+    )
+
+
+def test_prompt_caching_1h_ttl_passes_ttl_field():
+    """When prompt_caching='1h', cache_control must include the 1h ttl."""
+    import asyncio
+
+    from seclab_taskflow_agent.sdk.anthropic_sdk.backend import _AnthropicHandle
+
+    captured = {}
+
+    class _FakeStreamCtx:
+        async def __aenter__(self): return self
+        async def __aexit__(self, *exc): return False
+        def __aiter__(self):
+            async def _gen():
+                return
+                yield
+            return _gen()
+        async def get_final_message(self):
+            return type("M", (), {"stop_reason": "end_turn", "content": []})()
+
+    class _FakeMessages:
+        def stream(self, **kwargs):
+            captured.update(kwargs)
+            return _FakeStreamCtx()
+
+    class _FakeClient:
+        def __init__(self):
+            self.messages = _FakeMessages()
+
+    handle = _AnthropicHandle(
+        client=_FakeClient(),
+        system_prompt="",
+        model="claude-mythos-5",
+        max_tokens=100,
+        tools=[],
+        mcp_server_map={},
+        model_settings={"prompt_caching": "1h"},
+    )
+    backend = AnthropicSDKBackend()
+
+    async def _run():
+        async for _ in backend.run_streamed(handle, "hi", max_turns=1):
+            pass
+
+    asyncio.run(_run())
+    assert captured.get("cache_control") == {"type": "ephemeral", "ttl": "1h"}, (
+        f"expected cache_control with 1h ttl, got {captured.get('cache_control')!r}"
+    )