From 3c5ec0f0f84d4587ed248fa1cba341ec8fbeda05 Mon Sep 17 00:00:00 2001 From: CC#3 Kora Runtime Date: Sat, 23 May 2026 18:22:01 -0700 Subject: [PATCH] =?UTF-8?q?feat(kora):=20KR-CHEAP-PROMPT-CACHING=20?= =?UTF-8?q?=E2=80=94=20ephemeral=20cache=20on=20system=20prompt=20+=20tool?= =?UTF-8?q?=20descriptions?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Per Council R3 Lock R3-4, this is the build-list #1 baseline- setter: prompt caching ships FIRST so subsequent cheap-substrate telemetry measures the post-caching world. Expected ~50% input- token cost reduction on warm reasoning calls. # Two cache breakpoints (Anthropic API allows 4) System prompt + tool descriptions both marked with ``cache_control: {"type": "ephemeral"}``. The marker on a block caches everything UP TO AND INCLUDING that block, so two breakpoints cover both static input regions. # Call sites with cache_control 1. ``kora_cli/reasoning/anthropic_engine.py:_tool_use_loop`` — SDK kwargs["system"] is now a content-block list (was bare string) with cache_control on its single text block. Built ONCE outside the iteration loop so the same structure is sent on every roundtrip → cache key matches → reads hit. Wrapped via new module-level helper ``_wrap_system_as_ cacheable``. 2. Same loop — kwargs["tools"] now has cache_control on the LAST tool only (covers everything before it per API semantics). Wrapped via ``_wrap_tools_as_cacheable``; input list never mutated. Both wrappers tested as pure functions + via engine-level integration assertions. Tools-empty case preserved: registry fails → tools=[] → kwargs["tools"] omitted entirely (some SDK versions reject empty arrays). # Cost-ladder accounting patch The cost-ladder infrastructure ALREADY supported cache tokens via ``CanonicalUsage(cache_read_tokens, cache_write_tokens)`` + ``PricingEntry.cache_*_cost_per_million`` (Opus 4.7: $0.50 cache_read, $6.25 cache_write per million tokens). The gap was the engine→handler→holder data flow: 1. ``ResponseResult`` extended with ``cache_creation_input_tokens`` + ``cache_read_input_tokens`` fields (default 0, backwards-compatible). 2. ``_tool_use_loop`` accumulates both per iteration (mirror of input/output accumulation) reading from ``usage.cache_creation_input_tokens`` / ``usage.cache_read_input_tokens`` (getattr-default to 0 when older SDK / uncached call leaves them absent). 3. ``_project_final_response`` signature + body pass them through. Max-iter exit + the projection-failed branch also wired. 4. ``slack_dm_handler.py``: initial ``reasoning_meta`` dict + the engine-exception branch + the success-path meta build all gained the two cache fields. 5. ``_record_inference_to_cost_ladder`` reads the cache fields from meta + constructs ``CanonicalUsage(input_tokens, output_tokens, cache_write_tokens, cache_read_tokens)`` — PricingEntry's per-million rates do the rest. Early-return guard now checks ALL four token buckets (a pure-cache-read call still bills). 6. ``_append_outbound_log_entry`` extended with the same two optional kwargs so the cache totals persist into the slack DM outbound JSONL — reasoning panel can surface cache-hit rate per call without grepping logs. # K-DG verification Anthropic SDK 0.87.0 (one minor ahead of the spec's 0.86.0) verified to expose ``CacheControlEphemeralParam``, ``TextBlockParam``, ``ToolParam`` — cache_control syntax identical between minor versions per the docs at https://platform.claude.com/docs/en/agents-and-tools/prompt-caching. # Tests tests/kora_cli/reasoning/test_anthropic_engine_caching.py — 16 new tests: - Wrapper unit tests: system → content-block list with marker; tools → last marked, earlier unmarked; input not mutated; empty → empty; single-tool gets marker - Engine integration: system kwarg shape; tools[-1] has marker + tools[:-1] do NOT; tools kwarg omitted when registry returns empty; SAME system + tools sent on every iteration (cache key stability across tool-use loop) - ResponseResult surfaces cache_creation + cache_read totals; accumulates across iterations (write on iter 1, reads on iters 2+3); defaults to 0 when usage attrs missing - Handler cost-ladder: reasoning_meta cache fields flow into CanonicalUsage(cache_write_tokens, cache_read_tokens); None → 0 fallback; pure-cache-read call still bills; all- zero call still skips tests/kora_cli/reasoning/test_anthropic_engine.py — 1 pre- existing test updated: ``test_system_prompt_passed_to_sdk`` now asserts the content-block list shape instead of bare-string. # Regression 483/483 reasoning + handlers + listeners pass serially. Full repo xdist: 9228 passed, 43 failed identical to baseline (test_anthropic_adapter / test_backup / test_config / test_gateway_* / test_web_server* / test_kanban_db / test_list_picker_providers / test_model_switch_* / test_startup_plugin_gating / test_web_server_cron_profiles). Zero failures in reasoning or handlers. Co-Authored-By: Claude Opus 4.7 (1M context) --- kora_cli/handlers/slack_dm_handler.py | 55 +- kora_cli/reasoning/anthropic_engine.py | 120 +++- kora_cli/reasoning/engine.py | 9 + .../reasoning/test_anthropic_engine.py | 7 +- .../test_anthropic_engine_caching.py | 544 ++++++++++++++++++ 5 files changed, 729 insertions(+), 6 deletions(-) create mode 100644 tests/kora_cli/reasoning/test_anthropic_engine_caching.py diff --git a/kora_cli/handlers/slack_dm_handler.py b/kora_cli/handlers/slack_dm_handler.py index 7c1ebf860160..1d6663f80759 100644 --- a/kora_cli/handlers/slack_dm_handler.py +++ b/kora_cli/handlers/slack_dm_handler.py @@ -415,6 +415,13 @@ async def _send_echo_reply(self, payload: Dict[str, Any]) -> None: # the REASONING-PANEL can surface "this response used # N tools" without parsing structured logs. "tools_used": None, + # KR-CHEAP-PROMPT-CACHING — cache-token totals. Default + # None (engine didn't run / errored) → 0 when present. + # Handler bills cache_creation at ~1.25x base and + # cache_read at ~0.1x base via CanonicalUsage in + # _record_inference_to_cost_ladder. + "cache_creation_input_tokens": None, + "cache_read_input_tokens": None, } if engine is None: @@ -586,6 +593,8 @@ async def _call_reasoning_engine( "reasoning_duration_ms": None, "reasoning_error": f"engine_exception:{type(exc).__name__}", "tools_used": None, + "cache_creation_input_tokens": None, + "cache_read_input_tokens": None, }, ) @@ -612,6 +621,11 @@ async def _call_reasoning_engine( if result.error is None else None ), + # KR-CHEAP-PROMPT-CACHING — cache-token accumulators. + # Pass through whatever the engine accumulated; 0 when + # no caching engaged (uncached call OR pre-cache state). + "cache_creation_input_tokens": result.cache_creation_input_tokens, + "cache_read_input_tokens": result.cache_read_input_tokens, } if result.error is not None: @@ -669,7 +683,26 @@ def _record_inference_to_cost_ladder( model_name = reasoning_meta.get("model_used") input_tokens = reasoning_meta.get("input_tokens") or 0 output_tokens = reasoning_meta.get("output_tokens") or 0 - if not model_name or (input_tokens == 0 and output_tokens == 0): + # KR-CHEAP-PROMPT-CACHING — bill cache_creation + cache_read + # at their respective rates. CanonicalUsage(.cache_write_tokens + # → ~1.25x base) + (.cache_read_tokens → ~0.1x base) per the + # PricingEntry table in agent/usage_pricing.py. When the + # engine didn't cache (None or 0), CanonicalUsage's defaults + # keep this a no-op for those fields. + cache_creation_tokens = ( + reasoning_meta.get("cache_creation_input_tokens") or 0 + ) + cache_read_tokens = ( + reasoning_meta.get("cache_read_input_tokens") or 0 + ) + # Bail only when EVERY token bucket is 0 — a pure-cache-read + # call (input_tokens=0 but cache_read_tokens>0) still bills. + if not model_name or ( + input_tokens == 0 + and output_tokens == 0 + and cache_creation_tokens == 0 + and cache_read_tokens == 0 + ): return try: @@ -677,6 +710,8 @@ def _record_inference_to_cost_ladder( CanonicalUsage( input_tokens=int(input_tokens), output_tokens=int(output_tokens), + cache_write_tokens=int(cache_creation_tokens), + cache_read_tokens=int(cache_read_tokens), ), model_name=str(model_name), provider="anthropic", @@ -779,6 +814,14 @@ def _append_outbound_log_entry( # paths so the field's presence distinguishes "engine ran" # from "engine bypassed." tools_used: Optional[List[str]] = None, + # KR-CHEAP-PROMPT-CACHING — cache-token totals from the + # engine. None (omitted from JSONL) on non-reasoning paths; + # explicit 0 means the engine ran but no cache engaged + # (older SDK / uncached call). Persisted alongside + # input_tokens/output_tokens so the reasoning panel can + # show cache-hit rate per call without grepping logs. + cache_creation_input_tokens: Optional[int] = None, + cache_read_input_tokens: Optional[int] = None, ) -> None: """Outbound-side JSONL entry. Distinct schema from inbound entries (``sent_at`` instead of ``received_at``) so operator @@ -837,6 +880,16 @@ def _append_outbound_log_entry( # is a different signal than "engine didn't run." if tools_used is not None: entry["tools_used"] = list(tools_used) + # Cache-token persistence — same None-omit semantic as the + # other reasoning fields. + if cache_creation_input_tokens is not None: + entry["cache_creation_input_tokens"] = int( + cache_creation_input_tokens + ) + if cache_read_input_tokens is not None: + entry["cache_read_input_tokens"] = int( + cache_read_input_tokens + ) try: self._log_path.parent.mkdir(parents=True, exist_ok=True) diff --git a/kora_cli/reasoning/anthropic_engine.py b/kora_cli/reasoning/anthropic_engine.py index f3447c74f104..405104aae2c6 100644 --- a/kora_cli/reasoning/anthropic_engine.py +++ b/kora_cli/reasoning/anthropic_engine.py @@ -120,6 +120,73 @@ } +# --------------------------------------------------------------------------- +# KR-CHEAP-PROMPT-CACHING — cacheable input wrappers +# --------------------------------------------------------------------------- + + +# Per Anthropic API docs: +# https://platform.claude.com/docs/en/agents-and-tools/prompt-caching +# Cache breakpoints are marked with ``cache_control: {"type": +# "ephemeral"}``. The marker on a block caches everything UP TO AND +# INCLUDING that block. We use TWO breakpoints (API allows up to 4): +# 1. The system prompt — static across all calls. +# 2. The tool list — static unless the tool registry mutates. +# +# Cache TTL is ~5 minutes on Anthropic's side. Active reasoning +# sessions hit the warm cache repeatedly (~90% discount on the +# cached portion). Idle agent pays the cache-write premium (~25% +# above base rate) on the first call post-idle, then reads cheap +# until idle again. Net expected effect: ~50% input-cost +# reduction on warm sessions; spec PR body documents the cost +# shape change for operator visibility. + + +def _wrap_system_as_cacheable(system_prompt: str) -> list[dict]: + """Convert a bare-string system prompt to a content-block list + with ``cache_control: ephemeral`` on the last block (here: + the only block). + + The SDK accepts ``system: str`` OR + ``system: list[{type: "text", text: str, cache_control?: ...}]``. + The list form is required to attach the cache marker. + """ + return [ + { + "type": "text", + "text": system_prompt, + "cache_control": {"type": "ephemeral"}, + } + ] + + +def _wrap_tools_as_cacheable(tools: list[dict]) -> list[dict]: + """Return a NEW list of tool descriptors with + ``cache_control: ephemeral`` on the FINAL tool. + + The marker on the last tool covers the entire tool block + (system prompt's tool-system additions + every preceding + tool's schema). We never mutate the input list — caller + holds a reference to the registry's structures and we don't + want to surprise them with a side-effect. + + Empty input → empty output (caller skips ``tools=`` kwarg). + """ + if not tools: + return [] + wrapped: list[dict] = [] + for i, tool in enumerate(tools): + if i == len(tools) - 1: + # Last tool — attach the cache marker. Copy the dict so + # we don't mutate the registry's source structure. + new_tool = dict(tool) + new_tool["cache_control"] = {"type": "ephemeral"} + wrapped.append(new_tool) + else: + wrapped.append(tool) + return wrapped + + # --------------------------------------------------------------------------- # Errors # --------------------------------------------------------------------------- @@ -321,12 +388,32 @@ async def _tool_use_loop( # Accumulators across iterations. total_input_tokens = 0 total_output_tokens = 0 + # KR-CHEAP-PROMPT-CACHING — accumulate cache-write + + # cache-read tokens separately. The SDK surfaces them as + # ``usage.cache_creation_input_tokens`` (full-rate, billed + # 1x base + ~25% write premium per Anthropic pricing) and + # ``usage.cache_read_input_tokens`` (~90% discount vs base). + # Handler reads ResponseResult.cache_* and bills via + # CanonicalUsage(cache_write_tokens, cache_read_tokens) so + # the cost-ladder PricingEntry's cache_*_cost_per_million + # multipliers apply correctly. + total_cache_creation_tokens = 0 + total_cache_read_tokens = 0 # Track which tools Kora actually used — surfaced to the # handler via ResponseResult.tools_used (ST2 wires it into # the outbound JSONL; ST1 ships the field on the result # class so handler/test consumers don't churn between STs). tools_used: list[str] = [] + # KR-CHEAP-PROMPT-CACHING — build cacheable system block + + # cacheable tool list ONCE outside the loop. Each iteration + # sends the SAME structures so Anthropic recognizes the + # cache key + hits the warm cache after the first roundtrip. + # The ``cache_control: {type: ephemeral}`` marker covers + # everything UP TO AND INCLUDING the block it's attached to. + system_blocks = _wrap_system_as_cacheable(self._system_prompt) + cacheable_tools = _wrap_tools_as_cacheable(tools) + for iteration in range(1, MAX_TOOL_USE_ITERATIONS + 1): try: # tools= is optional per the SDK; omit when the @@ -334,13 +421,13 @@ async def _tool_use_loop( # array (some Anthropic SDK versions are strict). kwargs: Dict[str, Any] = { "model": model, - "system": self._system_prompt, + "system": system_blocks, "messages": messages, "max_tokens": self._max_output_tokens, "timeout": self._timeout, } - if tools: - kwargs["tools"] = tools + if cacheable_tools: + kwargs["tools"] = cacheable_tools response = await client.messages.create(**kwargs) except Exception as exc: return self._map_sdk_exception( @@ -353,6 +440,17 @@ async def _tool_use_loop( total_output_tokens += int( getattr(usage, "output_tokens", 0) or 0 ) + # Cache token accumulation. SDK exposes these as + # ``cache_creation_input_tokens`` + ``cache_read_input_tokens`` + # on the usage block. They're populated only when caching + # actually engages — uncached calls leave them at 0 (or + # the attr absent, which getattr-defaults to 0 here). + total_cache_creation_tokens += int( + getattr(usage, "cache_creation_input_tokens", 0) or 0 + ) + total_cache_read_tokens += int( + getattr(usage, "cache_read_input_tokens", 0) or 0 + ) # Detect tool-use vs end-of-turn. Anthropic SDK sets # ``response.stop_reason`` to one of: @@ -366,6 +464,8 @@ async def _tool_use_loop( started_at=started_at, total_input_tokens=total_input_tokens, total_output_tokens=total_output_tokens, + total_cache_creation_tokens=total_cache_creation_tokens, + total_cache_read_tokens=total_cache_read_tokens, tools_used=tools_used, ) @@ -386,6 +486,8 @@ async def _tool_use_loop( started_at=started_at, total_input_tokens=total_input_tokens, total_output_tokens=total_output_tokens, + total_cache_creation_tokens=total_cache_creation_tokens, + total_cache_read_tokens=total_cache_read_tokens, tools_used=tools_used, ) @@ -424,6 +526,8 @@ async def _tool_use_loop( reasoning_duration_ms=_elapsed_ms(started_at), error="tool_use_max_iterations_exceeded", tools_used=tools_used, + cache_creation_input_tokens=total_cache_creation_tokens, + cache_read_input_tokens=total_cache_read_tokens, ) def _extract_tool_use_blocks(self, response: Any) -> list: @@ -595,6 +699,8 @@ def _project_final_response( started_at: float, total_input_tokens: int, total_output_tokens: int, + total_cache_creation_tokens: int, + total_cache_read_tokens: int, tools_used: list[str], ) -> ResponseResult: """Multi-iteration variant of :meth:`_project_response`. @@ -602,7 +708,9 @@ def _project_final_response( Token totals are passed in (accumulated across all iterations) rather than read from the final response's usage block — important since intermediate roundtrips - billed tokens too. + billed tokens too. Cache totals are surfaced separately + so the handler can bill them at cache_read / cache_write + rates rather than the full input-token rate. """ text_parts: list[str] = [] try: @@ -623,6 +731,8 @@ def _project_final_response( reasoning_duration_ms=_elapsed_ms(started_at), error="response_projection_failed", tools_used=tools_used, + cache_creation_input_tokens=total_cache_creation_tokens, + cache_read_input_tokens=total_cache_read_tokens, ) text = "".join(text_parts).strip() @@ -634,6 +744,8 @@ def _project_final_response( reasoning_duration_ms=_elapsed_ms(started_at), error=None, tools_used=tools_used, + cache_creation_input_tokens=total_cache_creation_tokens, + cache_read_input_tokens=total_cache_read_tokens, ) async def close(self) -> None: diff --git a/kora_cli/reasoning/engine.py b/kora_cli/reasoning/engine.py index e63e90825478..6e38b66581fd 100644 --- a/kora_cli/reasoning/engine.py +++ b/kora_cli/reasoning/engine.py @@ -179,6 +179,15 @@ class ResponseResult: # backwards-compatible — existing ResponseResult construction # without this kwarg still works. tools_used: List[str] = field(default_factory=list) + # KR-CHEAP-PROMPT-CACHING — cache-token totals across all + # iterations of the tool-use loop. Both are 0 when no caching + # was used (uncached call, engine refused, or model didn't + # surface cache usage). Default 0 keeps the dataclass + # backwards-compatible. Handler reads these to bill against + # the cost-ladder's cache_read / cache_write rates rather than + # the full input-token rate. + cache_creation_input_tokens: int = 0 + cache_read_input_tokens: int = 0 # --------------------------------------------------------------------------- diff --git a/tests/kora_cli/reasoning/test_anthropic_engine.py b/tests/kora_cli/reasoning/test_anthropic_engine.py index 9bce997cfa1d..fca92082cfe1 100644 --- a/tests/kora_cli/reasoning/test_anthropic_engine.py +++ b/tests/kora_cli/reasoning/test_anthropic_engine.py @@ -426,8 +426,13 @@ async def test_system_prompt_passed_to_sdk( system_prompt_path=system_prompt_path, client=client ) await engine.respond(_msg(), _ctx()) + # KR-CHEAP-PROMPT-CACHING — system is now a content-block list + # (not a bare string) so cache_control: ephemeral can attach. + # See test_anthropic_engine_caching.py for the full cache-shape + # contract; this test just verifies the prompt text is reachable. system = client.messages.create.await_args.kwargs["system"] - assert "Kora" in system + assert isinstance(system, list) + assert "Kora" in system[0]["text"] # --------------------------------------------------------------------------- diff --git a/tests/kora_cli/reasoning/test_anthropic_engine_caching.py b/tests/kora_cli/reasoning/test_anthropic_engine_caching.py new file mode 100644 index 000000000000..ac141d5e8494 --- /dev/null +++ b/tests/kora_cli/reasoning/test_anthropic_engine_caching.py @@ -0,0 +1,544 @@ +"""Tests for KR-CHEAP-PROMPT-CACHING — ephemeral cache markers + +cost-ladder accuracy on cache tokens. + +Covers spec §2 + §3 acceptance: + + - kwargs["system"] is a content-block list with cache_control: + {"type": "ephemeral"} on the only block (not a bare string) + - kwargs["tools"]'s LAST tool carries cache_control: ephemeral; + earlier tools do NOT (the marker covers everything up to and + including the marked block, per Anthropic API semantics) + - Empty tool list → tools kwarg omitted (the original wrapper + behavior is preserved; no empty list ever sent) + - Tool list is NOT mutated in place (caller's registry structure + untouched between calls) + - SAME system + tools structures sent on every iteration so the + cache key matches across the tool-use loop + - ResponseResult surfaces cache_creation_input_tokens + + cache_read_input_tokens accumulated across iterations + - Handler's _record_inference_to_cost_ladder passes the cache + fields through to CanonicalUsage (cache_write_tokens + + cache_read_tokens) so the cost-ladder bills at the correct + rate (~1.25x for writes, ~0.1x for reads per the + PricingEntry's cache_*_cost_per_million) + - A pure-cache-read call (input_tokens=0 but cache_read_tokens>0) + still bills — the early-return guard checks ALL token buckets +""" + +from __future__ import annotations + +from datetime import datetime, timezone +from typing import Any, Dict, List +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from kora_cli.reasoning.anthropic_engine import ( + MODEL_OPUS, + OAUTH_TOKEN_ENV, + AnthropicReasoningEngine, + _wrap_system_as_cacheable, + _wrap_tools_as_cacheable, +) +from kora_cli.reasoning.engine import ( + ConversationContext, + IncomingMessage, + ResponseResult, +) + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _text_block(text: str): + b = MagicMock() + b.type = "text" + b.text = text + return b + + +def _tool_use_block(tool_id: str, name: str, tool_input: Dict[str, Any]): + b = MagicMock() + b.type = "tool_use" + b.id = tool_id + b.name = name + b.input = tool_input + return b + + +def _response( + content: List[Any], + stop_reason: str = "end_turn", + input_tokens: int = 100, + output_tokens: int = 50, + cache_creation_input_tokens: int = 0, + cache_read_input_tokens: int = 0, +): + usage = MagicMock() + usage.input_tokens = input_tokens + usage.output_tokens = output_tokens + usage.cache_creation_input_tokens = cache_creation_input_tokens + usage.cache_read_input_tokens = cache_read_input_tokens + r = MagicMock() + r.content = content + r.stop_reason = stop_reason + r.model = MODEL_OPUS + r.usage = usage + return r + + +def _make_client(responses): + client = MagicMock() + client.messages = MagicMock() + client.messages.create = AsyncMock(side_effect=list(responses)) + client.close = AsyncMock() + return client + + +def _msg(text: str = "hi") -> IncomingMessage: + return IncomingMessage( + text=text, + source="slack_dm", + received_at=datetime.now(timezone.utc), + metadata={}, + ) + + +def _ctx(rung: str = "normal", state: str = "ready") -> ConversationContext: + return ConversationContext( + recent_messages=[], + current_operational_state=state, + current_cost_ladder_rung=rung, + ) + + +@pytest.fixture +def system_prompt_path(tmp_path): + p = tmp_path / "kora_system_prompt.md" + p.write_text("You are Kora. Be useful.\n", encoding="utf-8") + return p + + +@pytest.fixture(autouse=True) +def _oauth(monkeypatch): + monkeypatch.setenv(OAUTH_TOKEN_ENV, "sk-ant-oat-test") + + +# --------------------------------------------------------------------------- +# Wrapper unit tests — pure functions +# --------------------------------------------------------------------------- + + +def test_wrap_system_as_cacheable_returns_content_block_list(): + out = _wrap_system_as_cacheable("hello world") + assert isinstance(out, list) + assert len(out) == 1 + assert out[0] == { + "type": "text", + "text": "hello world", + "cache_control": {"type": "ephemeral"}, + } + + +def test_wrap_tools_as_cacheable_marks_only_last_tool(): + """The cache marker on the last block caches everything UP TO + AND INCLUDING that block — so we only mark the last.""" + tools_in = [ + {"name": "a", "description": "A", "input_schema": {}}, + {"name": "b", "description": "B", "input_schema": {}}, + {"name": "c", "description": "C", "input_schema": {}}, + ] + out = _wrap_tools_as_cacheable(tools_in) + assert len(out) == 3 + assert "cache_control" not in out[0] + assert "cache_control" not in out[1] + assert out[2]["cache_control"] == {"type": "ephemeral"} + # Other fields preserved on the last tool. + assert out[2]["name"] == "c" + assert out[2]["description"] == "C" + + +def test_wrap_tools_as_cacheable_does_not_mutate_input(): + """Caller (the registry) holds a reference to the source list; + we must NOT add cache_control to its dicts.""" + tools_in = [{"name": "a", "description": "A", "input_schema": {}}] + _wrap_tools_as_cacheable(tools_in) + assert "cache_control" not in tools_in[0] + + +def test_wrap_tools_as_cacheable_empty_returns_empty(): + """Empty tool list → empty wrapper (caller skips the tools= + kwarg entirely; some SDK versions reject empty arrays).""" + assert _wrap_tools_as_cacheable([]) == [] + + +def test_wrap_tools_single_tool_gets_marker(): + """Edge case: only one tool — it IS the last.""" + out = _wrap_tools_as_cacheable( + [{"name": "solo", "description": "S", "input_schema": {}}] + ) + assert out[0]["cache_control"] == {"type": "ephemeral"} + + +# --------------------------------------------------------------------------- +# Engine integration — system block + tools block shape on SDK call +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_engine_sends_system_as_cacheable_content_block( + monkeypatch, system_prompt_path +): + """First SDK call's kwargs["system"] must be a list of content + blocks with cache_control on the last (here: only) block — NOT + a bare string.""" + from kora_cli.listeners import mcp_tools + + monkeypatch.setattr(mcp_tools, "_get_active_provider", lambda: None) + iter1 = _response([_text_block("hi")], stop_reason="end_turn") + client = _make_client([iter1]) + engine = AnthropicReasoningEngine( + system_prompt_path=system_prompt_path, client=client + ) + await engine.respond(_msg(), _ctx()) + + kwargs = client.messages.create.await_args_list[0].kwargs + assert isinstance(kwargs["system"], list), ( + "system must be a content-block list (not a bare string) " + "for cache_control to attach" + ) + assert len(kwargs["system"]) == 1 + block = kwargs["system"][0] + assert block["type"] == "text" + assert block["text"] == "You are Kora. Be useful.\n" + assert block["cache_control"] == {"type": "ephemeral"} + + +@pytest.mark.asyncio +async def test_engine_marks_last_tool_with_cache_control( + monkeypatch, system_prompt_path +): + """tools= kwarg's last entry carries cache_control: ephemeral. + Verifies the engine's wrap is engaged — not just the helper.""" + from kora_cli.listeners import mcp_tools + + monkeypatch.setattr(mcp_tools, "_get_active_provider", lambda: None) + iter1 = _response([_text_block("hi")], stop_reason="end_turn") + client = _make_client([iter1]) + engine = AnthropicReasoningEngine( + system_prompt_path=system_prompt_path, client=client + ) + await engine.respond(_msg(), _ctx()) + + kwargs = client.messages.create.await_args_list[0].kwargs + tools = kwargs["tools"] + assert len(tools) >= 1 + # Last tool marked. + assert tools[-1]["cache_control"] == {"type": "ephemeral"} + # Earlier tools NOT marked (if more than one). + for t in tools[:-1]: + assert "cache_control" not in t + + +@pytest.mark.asyncio +async def test_engine_omits_tools_kwarg_when_registry_empty( + monkeypatch, system_prompt_path +): + """Registry fails to load → tools=[] → SDK call must NOT include + tools= at all (some SDK versions reject empty arrays). The + cacheable wrapper preserves this contract.""" + monkeypatch.setattr( + "kora_cli.reasoning.tool_registry.get_reasoning_available_tools", + lambda: (_ for _ in ()).throw(RuntimeError("registry down")), + ) + iter1 = _response([_text_block("hi")], stop_reason="end_turn") + client = _make_client([iter1]) + engine = AnthropicReasoningEngine( + system_prompt_path=system_prompt_path, client=client + ) + await engine.respond(_msg(), _ctx()) + + kwargs = client.messages.create.await_args_list[0].kwargs + assert "tools" not in kwargs + + +@pytest.mark.asyncio +async def test_engine_sends_same_system_and_tools_each_iteration( + monkeypatch, system_prompt_path +): + """For the cache to HIT across iterations of the tool-use loop, + the engine must send byte-identical system + tools structures + on every iteration. The wrap is built ONCE outside the loop.""" + from kora_cli.listeners import mcp_tools + + monkeypatch.setattr(mcp_tools, "_get_active_provider", lambda: None) + iter1 = _response( + [ + _tool_use_block("toolu_a", "kora__get_operational_state", {}), + ], + stop_reason="tool_use", + ) + iter2 = _response([_text_block("done")], stop_reason="end_turn") + client = _make_client([iter1, iter2]) + engine = AnthropicReasoningEngine( + system_prompt_path=system_prompt_path, client=client + ) + await engine.respond(_msg(), _ctx()) + + call1_kwargs = client.messages.create.await_args_list[0].kwargs + call2_kwargs = client.messages.create.await_args_list[1].kwargs + # Same Python list object isn't required (the engine could + # build per-iter) but the SHAPE + CONTENT must match. + assert call1_kwargs["system"] == call2_kwargs["system"], ( + "system must be identical across iterations or the cache " + "key won't match + we pay the cache-write premium per iter" + ) + assert call1_kwargs["tools"] == call2_kwargs["tools"] + + +# --------------------------------------------------------------------------- +# Cache-token accounting — ResponseResult surfaces totals +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_response_result_carries_cache_token_totals( + monkeypatch, system_prompt_path +): + """A response with cache_creation + cache_read in usage → + ResponseResult.cache_creation_input_tokens + + cache_read_input_tokens are populated.""" + from kora_cli.listeners import mcp_tools + + monkeypatch.setattr(mcp_tools, "_get_active_provider", lambda: None) + iter1 = _response( + content=[_text_block("hi")], + stop_reason="end_turn", + input_tokens=200, + output_tokens=50, + cache_creation_input_tokens=1000, + cache_read_input_tokens=500, + ) + client = _make_client([iter1]) + engine = AnthropicReasoningEngine( + system_prompt_path=system_prompt_path, client=client + ) + result = await engine.respond(_msg(), _ctx()) + assert result.cache_creation_input_tokens == 1000 + assert result.cache_read_input_tokens == 500 + assert result.input_tokens == 200 + assert result.output_tokens == 50 + + +@pytest.mark.asyncio +async def test_cache_tokens_accumulate_across_iterations( + monkeypatch, system_prompt_path +): + """Tool-use loop with 3 iterations: cache_creation paid once + on iter 1; cache_read on iters 2 + 3. Totals sum correctly.""" + from kora_cli.listeners import mcp_tools + + monkeypatch.setattr(mcp_tools, "_get_active_provider", lambda: None) + iter1 = _response( + [_tool_use_block("toolu_a", "kora__get_operational_state", {})], + stop_reason="tool_use", + input_tokens=300, + output_tokens=20, + cache_creation_input_tokens=800, # write — first call + cache_read_input_tokens=0, + ) + iter2 = _response( + [_tool_use_block("toolu_b", "kora__get_health_rollup", {})], + stop_reason="tool_use", + input_tokens=350, + output_tokens=15, + cache_creation_input_tokens=0, + cache_read_input_tokens=800, # read — cache hit + ) + iter3 = _response( + [_text_block("done")], + stop_reason="end_turn", + input_tokens=400, + output_tokens=30, + cache_creation_input_tokens=0, + cache_read_input_tokens=800, # read — cache hit + ) + client = _make_client([iter1, iter2, iter3]) + engine = AnthropicReasoningEngine( + system_prompt_path=system_prompt_path, client=client + ) + result = await engine.respond(_msg(), _ctx()) + + assert result.cache_creation_input_tokens == 800 + assert result.cache_read_input_tokens == 1600 # 800 + 800 + assert result.input_tokens == 1050 # 300 + 350 + 400 + + +@pytest.mark.asyncio +async def test_cache_tokens_default_zero_when_usage_missing_attrs( + monkeypatch, system_prompt_path +): + """Older SDK versions OR uncached calls may leave the cache + attrs absent on usage. getattr-defaults to 0 → ResponseResult + fields default to 0 (no AttributeError).""" + from kora_cli.listeners import mcp_tools + + monkeypatch.setattr(mcp_tools, "_get_active_provider", lambda: None) + # Build a usage object WITHOUT the cache attrs. + usage = MagicMock(spec=["input_tokens", "output_tokens"]) + usage.input_tokens = 100 + usage.output_tokens = 25 + r = MagicMock() + r.content = [_text_block("hi")] + r.stop_reason = "end_turn" + r.model = MODEL_OPUS + r.usage = usage + + client = _make_client([r]) + engine = AnthropicReasoningEngine( + system_prompt_path=system_prompt_path, client=client + ) + result = await engine.respond(_msg(), _ctx()) + assert result.cache_creation_input_tokens == 0 + assert result.cache_read_input_tokens == 0 + + +# --------------------------------------------------------------------------- +# Handler cost-ladder integration — cache tokens flow through +# --------------------------------------------------------------------------- + + +def test_record_inference_passes_cache_tokens_to_canonical_usage(): + """When reasoning_meta carries cache_creation + + cache_read, the handler constructs CanonicalUsage with + BOTH cache_write_tokens AND cache_read_tokens populated — + not just input/output.""" + from unittest.mock import MagicMock, patch + + # The cost-ladder helper is a @staticmethod on SlackDMHandler; + # import via the module. + from kora_cli.handlers import slack_dm_handler as h_mod + + fake_holder = MagicMock() + captured_usage = [] + + def _capture(usage, *, model_name, provider): + captured_usage.append(usage) + + fake_holder.record_inference = _capture + + with patch.object( + h_mod, "get_cost_holder", return_value=fake_holder, create=True + ) as _mocked: + # The helper imports lazily; we monkeypatch get_cost_holder + # in the namespace it imports from. + with patch( + "agent.cost_state_holder.get_cost_holder", + return_value=fake_holder, + ): + h_mod.SlackDMHandler._record_inference_to_cost_ladder( + { + "model_used": "claude-opus-4-7", + "input_tokens": 200, + "output_tokens": 50, + "cache_creation_input_tokens": 1000, + "cache_read_input_tokens": 500, + } + ) + + assert len(captured_usage) == 1 + u = captured_usage[0] + assert u.input_tokens == 200 + assert u.output_tokens == 50 + assert u.cache_write_tokens == 1000 + assert u.cache_read_tokens == 500 + + +def test_record_inference_handles_none_cache_fields(): + """Engine errored (cache_* in meta is None) → record_inference + still bills the non-cache tokens; cache fields default to 0.""" + from unittest.mock import MagicMock, patch + + from kora_cli.handlers import slack_dm_handler as h_mod + + fake_holder = MagicMock() + captured = [] + fake_holder.record_inference = lambda usage, **kw: captured.append(usage) + + with patch( + "agent.cost_state_holder.get_cost_holder", + return_value=fake_holder, + ): + h_mod.SlackDMHandler._record_inference_to_cost_ladder( + { + "model_used": "claude-opus-4-7", + "input_tokens": 100, + "output_tokens": 25, + "cache_creation_input_tokens": None, + "cache_read_input_tokens": None, + } + ) + + assert len(captured) == 1 + assert captured[0].cache_write_tokens == 0 + assert captured[0].cache_read_tokens == 0 + assert captured[0].input_tokens == 100 + + +def test_record_inference_bills_pure_cache_read_calls(): + """A call that's ENTIRELY served from cache (input_tokens=0, + cache_read_tokens>0) must still bill — the early-return guard + bailed only when EVERY token bucket was 0.""" + from unittest.mock import MagicMock, patch + + from kora_cli.handlers import slack_dm_handler as h_mod + + fake_holder = MagicMock() + called = [] + fake_holder.record_inference = lambda usage, **kw: called.append(usage) + + with patch( + "agent.cost_state_holder.get_cost_holder", + return_value=fake_holder, + ): + h_mod.SlackDMHandler._record_inference_to_cost_ladder( + { + "model_used": "claude-opus-4-7", + "input_tokens": 0, + "output_tokens": 50, + "cache_creation_input_tokens": 0, + "cache_read_input_tokens": 1500, + } + ) + + assert len(called) == 1 + assert called[0].cache_read_tokens == 1500 + + +def test_record_inference_skips_when_every_bucket_zero(): + """All token buckets 0 → skip the call (no real inference).""" + from unittest.mock import MagicMock, patch + + from kora_cli.handlers import slack_dm_handler as h_mod + + fake_holder = MagicMock() + called = [] + fake_holder.record_inference = lambda usage, **kw: called.append(usage) + + with patch( + "agent.cost_state_holder.get_cost_holder", + return_value=fake_holder, + ): + h_mod.SlackDMHandler._record_inference_to_cost_ladder( + { + "model_used": "claude-opus-4-7", + "input_tokens": 0, + "output_tokens": 0, + "cache_creation_input_tokens": 0, + "cache_read_input_tokens": 0, + } + ) + + assert called == []