feat(kora): KR-CHEAP-PROMPT-CACHING — ephemeral cache on system prompt + tool descriptions

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,14 +683,35 @@ 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:
             holder.record_inference(
                 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)

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,26 +388,46 @@ 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
                 # registry returned empty so we don't send an empty
                 # 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,14 +699,18 @@ 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`.
 
         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:

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
 
 
 # ---------------------------------------------------------------------------

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"]
 
 
 # ---------------------------------------------------------------------------