From f923d6468b6ee7a7bd2e4bec727347f5f24ab873 Mon Sep 17 00:00:00 2001 From: pufit Date: Mon, 6 Jul 2026 12:16:02 -0400 Subject: [PATCH] Add cadence-aware prompt-cache TTL policy (1h caching for sparse sessions) --- docs/config.md | 2 + docs/cron.md | 1 + nerve/agent/cache_policy.py | 274 +++++++++++++++++++++++ nerve/agent/engine.py | 125 ++++++++++- nerve/agent/prompts.py | 22 +- nerve/config.py | 14 ++ nerve/cron/jobs.py | 5 + nerve/cron/service.py | 2 + nerve/db/usage.py | 30 +++ nerve/gateway/routes/diagnostics.py | 20 ++ scripts/backtest_cache_ttl.py | 278 ++++++++++++++++++++++++ tests/test_cache_policy.py | 325 ++++++++++++++++++++++++++++ 12 files changed, 1082 insertions(+), 16 deletions(-) create mode 100644 nerve/agent/cache_policy.py create mode 100644 scripts/backtest_cache_ttl.py create mode 100644 tests/test_cache_policy.py diff --git a/docs/config.md b/docs/config.md index f24c5b3..90f9160 100644 --- a/docs/config.md +++ b/docs/config.md @@ -40,6 +40,8 @@ from any working directory: | `agent.cron_model` | string | `claude-sonnet-4-6` | Model for cron jobs (cheaper) | | `agent.max_turns` | int | `50` | Max agentic turns per request | | `agent.max_concurrent` | int | `4` | Max concurrent agent sessions | +| `agent.cache_ttl` | string | `"5m"` | Prompt-cache write TTL policy: `5m` (status quo), `1h` (always request the 1-hour TTL), or `auto` (per session at client-build time: sparse-cadence sessions — persistent crons, wakeup loops, spaced chats — get `1h`; dense sessions stay on `5m`). Per-cron-job override via `cache_ttl` in jobs.yaml. See `nerve/agent/cache_policy.py` | +| `agent.cache_ttl_excluded_models` | list | `[]` | Model-name substrings that never request the 1h TTL | | `agent.prompt_rewrite.enabled` | bool | `true` | Offer the first-prompt rewrite feature in the web UI (per-user toggle lives in the composer) | | `agent.prompt_rewrite.model` | string | `""` | Model for prompt rewriting (empty = `agent.model`, the chat model) | | `agent.prompt_rewrite.max_tokens` | int | `1024` | Max tokens for the rewritten prompt | diff --git a/docs/cron.md b/docs/cron.md index 0ed3d99..4054897 100644 --- a/docs/cron.md +++ b/docs/cron.md @@ -82,6 +82,7 @@ prompt definition. | `prompt_file` | string | yes* | Path to a file containing the prompt (relative to the YAML's directory). Read fresh each run; shareable between jobs. *One of `prompt`/`prompt_file` is required | | `description` | string | no | Human-readable description | | `model` | string | no | Override model (default: `agent.cron_model`) | +| `cache_ttl` | string | no | Prompt-cache TTL override for this job's sessions: `5m`, `1h`, or `auto` (default: `agent.cache_ttl`). Sparse-schedule persistent jobs benefit from `1h` — see `nerve/agent/cache_policy.py` | | `target` | string | no | Delivery channel (default: `telegram`) | | `session_mode` | string | no | `isolated` (new session per run), `persistent` (reuse context), or `main` | | `context_rotate_hours` | int | no | Hours before a persistent job rotates to a fresh chat (default: 24, 0 = never). The old chat is preserved | diff --git a/nerve/agent/cache_policy.py b/nerve/agent/cache_policy.py new file mode 100644 index 0000000..8553e91 --- /dev/null +++ b/nerve/agent/cache_policy.py @@ -0,0 +1,274 @@ +"""Cadence-aware prompt-cache TTL policy. + +Anthropic prompt caching has two write TTLs: 5 minutes (1.25x base input) +and 1 hour (2.0x base input); reads are 0.1x. A cache write is therefore +~12.5x the price of a read of the same tokens, so sessions whose turn +cadence exceeds 5 minutes by design (persistent crons, ScheduleWakeup +monitoring loops, spaced web conversations) re-buy their entire context +on every turn under the default TTL. + +This module decides, per SDK-client build, which TTL a session should +request. The Claude Code CLI has native support via env vars: + +- ``ENABLE_PROMPT_CACHING_1H=1`` — request the 1h TTL (API-key auth; + the CLI adds the ``extended-cache-ttl-2025-04-11`` beta itself). +- ``ENABLE_PROMPT_CACHING_1H_BEDROCK=1`` — same, for Bedrock. +- ``FORCE_PROMPT_CACHING_5M=1`` — upstream kill switch (not set here). + +Policy modes (``agent.cache_ttl`` in config, or a per-session override +in session metadata): + +- ``"5m"`` — status quo, never request the beta. +- ``"1h"`` — always request it (minus excluded models). +- ``"auto"`` — per session at client-build time: + 1. observed cadence wins: median of the session's recent turn gaps + in (5min, 1h] → 1h; any other observed cadence → 5m (gaps beyond + the TTL expire either way, so the 2x write premium buys nothing); + 2. no history: wakeup-driven turns and persistent-mode cron sessions + → 1h (the canonical sparse-cadence cases), everything else → 5m. + +The 1h TTL only pays off if the prompt bytes are identical across turns +— which in Nerve holds *within* an SDK-client lifetime, and across +client rebuilds only if the system prompt is byte-stable (see the +``Current date`` + frozen-recall changes in prompts.py / engine.py). +""" + +from __future__ import annotations + +import logging +import statistics +from typing import Any, Iterable + +from nerve.db.usage import _get_pricing + +logger = logging.getLogger(__name__) + +FIVE_MIN_S = 300.0 +ONE_HOUR_S = 3600.0 + +# How many recent turn gaps to consider for the cadence heuristic. +CADENCE_WINDOW = 12 + +VALID_TTL_MODES = ("5m", "1h", "auto") + + +def cache_ttl_env(ttl: str, is_bedrock: bool = False) -> dict[str, str]: + """Env vars for the CLI subprocess implementing the resolved TTL. + + ``"5m"`` returns an empty dict — the CLI default is already 5m and we + deliberately do NOT set ``FORCE_PROMPT_CACHING_5M`` (that would also + override a claude.ai-subscriber allowlist upstream). + """ + if ttl != "1h": + return {} + env = {"ENABLE_PROMPT_CACHING_1H": "1"} + if is_bedrock: + env["ENABLE_PROMPT_CACHING_1H_BEDROCK"] = "1" + return env + + +async def resolve_cache_ttl( + agent_config: Any, + db: Any, + session_id: str, + source: str, + model: str | None, + session_meta: dict | None = None, + is_claude_model: bool = True, +) -> str: + """Resolve the cache TTL ("5m" | "1h") for a session's next client. + + ``session_meta`` is the parsed session metadata dict; recognised keys: + + - ``cache_ttl_override`` — per-session mode override (e.g. from a + cron job's ``cache_ttl`` in jobs.yaml). Same values as the config. + - ``cron_session_mode`` — "persistent" | "isolated", written by the + cron runners; used as the no-history prior for cron sessions. + """ + if not is_claude_model: + return "5m" # Ollama/OpenAI-translated models have no Anthropic cache + + meta = session_meta or {} + mode = meta.get("cache_ttl_override") or getattr( + agent_config, "cache_ttl", "5m", + ) + if mode not in VALID_TTL_MODES: + logger.warning( + "Unknown cache_ttl mode %r for session %s — falling back to 5m", + mode, session_id, + ) + return "5m" + if mode == "5m": + return "5m" + + # Model exclusion applies to both "1h" and "auto". + resolved = (model or getattr(agent_config, "model", "") or "").lower() + excluded = getattr(agent_config, "cache_ttl_excluded_models", []) or [] + if any(tok and tok.lower() in resolved for tok in excluded): + return "5m" + + if mode == "1h": + return "1h" + + # --- auto: observed cadence first, source priors on no data + gaps: list[float] = [] + try: + gaps = await get_recent_turn_gaps(db, session_id, CADENCE_WINDOW) + except Exception as e: # never fail a client build over the heuristic + logger.warning( + "cache_ttl cadence query failed for %s: %s", session_id, e, + ) + + if gaps: + med = statistics.median(gaps) + return "1h" if FIVE_MIN_S < med <= ONE_HOUR_S else "5m" + + if source == "wakeup": + return "1h" + if source == "cron" and meta.get("cron_session_mode") == "persistent": + return "1h" + return "5m" + + +async def get_recent_turn_gaps( + db: Any, session_id: str, window: int = CADENCE_WINDOW, +) -> list[float]: + """Seconds between the session's most recent turns (indexed query).""" + async with db.db.execute( + """ + SELECT CAST(strftime('%s', created_at) AS REAL) + FROM session_usage WHERE session_id = ? + ORDER BY id DESC LIMIT ? + """, + (session_id, window + 1), + ) as cursor: + ts = [row[0] async for row in cursor if row[0] is not None] + ts.reverse() # chronological + return [b - a for a, b in zip(ts, ts[1:])] + + +# --------------------------------------------------------------------------- +# Live counterfactual: what would the observed traffic have cost on 5m? +# --------------------------------------------------------------------------- + +def estimate_live_ttl_delta( + turns: Iterable[tuple], + ttl_threshold: float = ONE_HOUR_S, +) -> dict: + """Estimate savings of observed (possibly 1h-cached) traffic vs a + pure-5m baseline. + + ``turns`` are chronological rows for ONE session: + ``(ts_epoch, model, input_tokens, cache_read, write_5m, write_1h)``. + + Model: a turn whose gap from the previous turn is in (5min, 1h] and + whose predecessor wrote 1h-TTL cache benefited from the extended TTL + — under 5m its first-iteration prefix read would have been a re-write + at 1.25x. The warm-prefix size is tracked from creation tokens + (reads multi-count across agentic-loop iterations, creations don't). + Turns are also charged the 1h-vs-5m write premium they actually paid. + + Returns ``{"actual": $, "baseline_5m": $, "savings": $}`` where + positive savings mean the 1h TTL is paying off. + """ + actual = 0.0 + baseline = 0.0 + warm_prefix = 0 + prev_ts: float | None = None + prev_model: str | None = None + prev_wrote_1h = False + + for ts, model, inp, reads, w5m, w1h in turns: + p_in, _o, p_read, p_c5m, p_c1h, _w = _get_pricing(model) + creation = (w5m or 0) + (w1h or 0) + gap = None if prev_ts is None else ts - prev_ts + cold_boundary = ( + gap is None or model != prev_model or gap > ttl_threshold + ) + + actual += ( + inp * p_in + reads * p_read + w5m * p_c5m + w1h * p_c1h + ) / 1_000_000 + + # Baseline: same turn under a pure-5m policy. + converted = 0 + if ( + not cold_boundary + and gap is not None + and gap > FIVE_MIN_S + and prev_wrote_1h + ): + # This read survived only thanks to the 1h TTL; under 5m the + # prefix would have been re-written once at the 5m rate. + converted = min(reads, warm_prefix) + baseline += ( + inp * p_in + + (reads - converted) * p_read + + converted * p_c5m + + (w5m + w1h) * p_c5m + ) / 1_000_000 + + # Warm-prefix bookkeeping (observed world). + if cold_boundary or (gap is not None and gap > FIVE_MIN_S and not prev_wrote_1h): + warm_prefix = creation + else: + warm_prefix += creation + prev_ts = ts + prev_model = model + prev_wrote_1h = w1h > 0 + + return { + "actual": round(actual, 4), + "baseline_5m": round(baseline, 4), + "savings": round(baseline - actual, 4), + } + + +def build_ttl_report(rows: list[tuple]) -> dict: + """Aggregate the live 1h-vs-5m estimate per source for diagnostics. + + ``rows`` come from ``UsageStore.get_cache_ttl_turn_rows``: + ``(session_id, source, ts, model, input_tokens, reads, w5m, w1h)``, + ordered by session then time. + + Guardrail: a source whose 1h traffic is estimated to cost *more* than + the 5m baseline (the auto policy misclassified its cadence) lands in + ``regressions`` and is logged at WARNING — manual revert is one + config line (``agent.cache_ttl``) or a per-job ``cache_ttl: "5m"``. + """ + by_session: dict[str, tuple[str, list[tuple]]] = {} + for sid, source, ts, model, inp, reads, w5m, w1h in rows: + if ts is None: + continue + by_session.setdefault(sid, (source, []))[1].append( + (ts, model, inp or 0, reads or 0, w5m or 0, w1h or 0), + ) + + per_source: dict[str, dict] = {} + total = {"actual": 0.0, "baseline_5m": 0.0, "savings": 0.0} + for _sid, (source, turns) in by_session.items(): + est = estimate_live_ttl_delta(turns) + agg = per_source.setdefault( + source, {"actual": 0.0, "baseline_5m": 0.0, "savings": 0.0}, + ) + for k in agg: + agg[k] = round(agg[k] + est[k], 4) + total[k] = round(total[k] + est[k], 4) + + regressions = [ + src for src, agg in per_source.items() if agg["savings"] < -0.5 + ] + for src in regressions: + logger.warning( + "cache_ttl guardrail: 1h caching for source %r cost " + "$%.2f MORE than the 5m baseline over the window — the auto " + "policy may be misclassifying its cadence (revert via " + "agent.cache_ttl or a per-job cache_ttl override)", + src, -per_source[src]["savings"], + ) + + return { + "by_source": per_source, + "total": total, + "regressions": regressions, + } diff --git a/nerve/agent/engine.py b/nerve/agent/engine.py index cb520d1..6dbe9d2 100644 --- a/nerve/agent/engine.py +++ b/nerve/agent/engine.py @@ -34,13 +34,18 @@ from claude_agent_sdk._errors import CLIConnectionError from claude_agent_sdk.types import HookMatcher, HookJSONOutput, HookContext +from nerve.agent.cache_policy import cache_ttl_env, resolve_cache_ttl from nerve.agent.interactive import ( InteractiveToolHandler, register_handler, unregister_handler, get_handler, ) -from nerve.agent.prompts import build_system_prompt, set_skill_manager +from nerve.agent.prompts import ( + build_system_prompt, + current_time_str, + set_skill_manager, +) from nerve.agent.sessions import SessionManager, SessionStatus from nerve.agent.streaming import broadcaster from nerve.agent.tools import ( @@ -959,6 +964,7 @@ def _build_options( resume: str | None = None, fork_session: bool = False, can_use_tool=None, + cache_ttl: str = "5m", ) -> ClaudeAgentOptions: """Build SDK client options for a session.""" # Get skill summaries for system prompt injection @@ -972,8 +978,12 @@ def _build_options( if loop.is_running(): # We're in an async context — schedule and await later # For now, use cached data from the manager + # Sorted for deterministic system-prompt bytes — scan + # order varies across restarts, and a reordered skill + # list would silently invalidate every session's prompt + # cache after a restart. skill_summaries = [] - for sid, meta in self._skill_manager._cache.items(): + for sid, meta in sorted(self._skill_manager._cache.items()): if meta.enabled and meta.model_invocable: skill_summaries.append({ "id": meta.id, @@ -1106,7 +1116,7 @@ def _cli_stderr(line: str) -> None: # CLI's own autonomous firing is suppressed via the # CLAUDE_CODE_DISABLE_CRON env var set in ``_build_env``. disallowed_tools=["CronCreate", "CronList", "CronDelete"], - env=self._build_env(), + env=self._build_env(cache_ttl=cache_ttl), cwd=str(self.config.workspace), mcp_servers=self._build_mcp_servers(session_id), # Claude Code plugins — loaded via --plugin-dir so the CLI @@ -1156,9 +1166,14 @@ def _write_system_prompt_file(self, session_id: str, content: str) -> str: path.write_text(content, encoding="utf-8") return str(path) - def _build_env(self) -> dict[str, str]: + def _build_env(self, cache_ttl: str = "5m") -> dict[str, str]: """Build environment variables for the SDK subprocess.""" env: dict[str, str] = {} + # Prompt-cache TTL: the CLI natively supports the 1-hour TTL via + # this env var (it adds the extended-cache-ttl beta header and the + # cache_control ttl itself). Resolved per client build by + # nerve.agent.cache_policy — see that module for the policy. + env.update(cache_ttl_env(cache_ttl, self.config.provider.is_bedrock)) # Disable the CLI's built-in cron/wakeup scheduler. It fires # autonomously inside the subprocess, but Nerve only reads the SDK # stream during an active run() — so a fired turn lands in an unread @@ -1622,17 +1637,19 @@ async def _get_or_create_client( # Check for stored SDK session ID for resume session = await self.db.get_session(session_id) sdk_resume_id = session.get("sdk_session_id") if session else None + try: + session_meta = json.loads( + (session.get("metadata") if session else None) or "{}", + ) + except (TypeError, ValueError): + session_meta = {} # Seed the serving-model baseline from the last persisted # observation so downgrade detection survives restarts without # re-firing an event on every resumed session. if session and session_id not in self._observed_models: - try: - _meta = json.loads(session.get("metadata") or "{}") - except (TypeError, ValueError): - _meta = {} - if _meta.get("observed_model"): - self._observed_models[session_id] = _meta["observed_model"] + if session_meta.get("observed_model"): + self._observed_models[session_id] = session_meta["observed_model"] # For forks, use the source session's SDK ID if fork_from and not sdk_resume_id: @@ -1671,18 +1688,55 @@ async def _get_or_create_client( session_id, sdk_resume_id[:12], ) - # Pre-recall memories for new session context + # Pre-recall memories for new session context. The first + # successful recall is frozen in session metadata and reused on + # every rebuild of the same session: byte-identical system + # prompts across rebuilds are what let a resumed conversation + # hit the prompt cache (see nerve/agent/cache_policy.py), and a + # session keeping its original priors is more consistent anyway + # — live recall stays available via the memory_recall tool. recalled_memories: list[str] = [] - if self._memory_bridge and self._memory_bridge.available: + meta_updates: dict[str, Any] = {} + frozen_recall = session_meta.get("recalled_memories") + if isinstance(frozen_recall, list): + recalled_memories = [str(m) for m in frozen_recall] + elif self._memory_bridge and self._memory_bridge.available: try: raw = await self._memory_bridge.recall( f"context for {source} session", limit=8, ) recalled_memories = [m["summary"] for m in raw] + meta_updates["recalled_memories"] = recalled_memories except Exception as e: logger.warning("Pre-recall failed: %s", e) + # Resolve the prompt-cache TTL for this client build (cadence- + # aware; see nerve/agent/cache_policy.py). Recomputed on every + # client (re)build — client recycling at the idle sweep makes + # the policy naturally track cadence changes. + requested = model or self.config.agent.model + is_claude = not ( + self.config.ollama.enabled + and "claude" not in requested.lower() + ) + cache_ttl = await resolve_cache_ttl( + self.config.agent, self.db, session_id, source, requested, + session_meta=session_meta, is_claude_model=is_claude, + ) + logger.info( + "Session %s: prompt-cache TTL %s (source=%s, mode=%s)", + session_id, cache_ttl, source, + session_meta.get("cache_ttl_override") + or self.config.agent.cache_ttl, + ) + if session_meta.get("cache_ttl") != cache_ttl: + meta_updates["cache_ttl"] = cache_ttl + + if meta_updates and session: + session_meta.update(meta_updates) + await self.db.update_session_metadata(session_id, session_meta) + # Determine if this is a fork is_fork = fork_from is not None @@ -1704,6 +1758,7 @@ async def _get_or_create_client( resume=sdk_resume_id, fork_session=is_fork, can_use_tool=handler.can_use_tool, + cache_ttl=cache_ttl, ) client = ClaudeSDKClient(options=options) await client.connect() @@ -2842,6 +2897,24 @@ async def _run_inner( if query_text and query_text.startswith("/"): query_text = "\u200b" + query_text + # Trailing wall-clock reminder. Precise time deliberately does + # NOT live in the system prompt (only the date does): a + # per-build timestamp there changes the prompt bytes on every + # client rebuild, invalidating the prompt cache for the entire + # conversation replay (see nerve/agent/cache_policy.py). As a + # message-tail reminder it is fresher \u2014 per turn instead of per + # client build \u2014 and costs nothing cache-wise. Not persisted to + # the DB message (db_text above), so the UI stays clean. + if query_text or images: + _time_note = ( + "Current time: " + f"{current_time_str(self.config.timezone)}" + "" + ) + query_text = ( + f"{query_text}\n\n{_time_note}" if query_text else _time_note + ) + # Build multi-modal content blocks once (reused on retry) if images: content_blocks: list[dict[str, Any]] = [] @@ -3531,12 +3604,37 @@ async def _teardown_oneshot_client( # lock and must not gate the run lifecycle. await self._discard_client(session_id, background_memorize=True) + async def _stamp_cron_session_meta( + self, session_id: str, mode: str, cache_ttl: str = "", + ) -> None: + """Stamp cron-session hints consumed by the cache-TTL policy. + + ``mode`` ("persistent" | "isolated") is the no-history prior for + auto TTL resolution; ``cache_ttl`` is the per-job override from + jobs.yaml (empty = no override). + """ + session = await self.db.get_session(session_id) + if not isinstance(session, dict): + return + try: + meta = json.loads(session.get("metadata") or "{}") + except (TypeError, ValueError): + meta = {} + updates: dict[str, Any] = {"cron_session_mode": mode} + if cache_ttl: + updates["cache_ttl_override"] = cache_ttl + if all(meta.get(k) == v for k, v in updates.items()): + return + meta.update(updates) + await self.db.update_session_metadata(session_id, meta) + async def run_cron( self, job_id: str, prompt: str, model: str | None = None, run_id: str | None = None, + cache_ttl: str = "", ) -> str: """Run an agent turn for a cron job in an isolated session. @@ -3550,6 +3648,7 @@ async def run_cron( run_id = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S") session = await self.sessions.create_cron_session(job_id, run_id=run_id) session_id = session["id"] + await self._stamp_cron_session_meta(session_id, "isolated", cache_ttl) try: return await self.run( session_id=session_id, @@ -3566,6 +3665,7 @@ async def run_persistent_cron( prompt: str, model: str | None = None, session_id: str | None = None, + cache_ttl: str = "", ) -> str: """Run a persistent cron job that maintains context across runs. @@ -3586,6 +3686,7 @@ async def run_persistent_cron( await self.sessions.get_or_create( session_id, title=f"Cron: {job_id}", source="cron", ) + await self._stamp_cron_session_meta(session_id, "persistent", cache_ttl) try: return await self.run( session_id=session_id, diff --git a/nerve/agent/prompts.py b/nerve/agent/prompts.py index 2466919..c28f9ef 100644 --- a/nerve/agent/prompts.py +++ b/nerve/agent/prompts.py @@ -59,6 +59,15 @@ def _format_tool_list() -> str: PROMPT_FILES = ["SOUL.md", "TASK.md", "IDENTITY.md", "USER.md", "AGENTS.md", "TOOLS.md"] +def current_time_str(timezone_name: str = "America/New_York") -> str: + """Minute-resolution local time, for the per-turn message reminder.""" + try: + tz = ZoneInfo(timezone_name) + return datetime.now(tz).strftime("%Y-%m-%d %H:%M %Z") + except Exception: + return datetime.now().strftime("%Y-%m-%d %H:%M") + + def _read_if_exists(path: Path) -> str | None: """Read file content if it exists, otherwise return None.""" try: @@ -117,17 +126,22 @@ def build_system_prompt( else: parts.append(memory_content) - # Session context + # Session context. Deliberately date-resolution only: a minute-level + # timestamp here changes the system-prompt bytes on every client + # rebuild, invalidating the prompt cache for the entire conversation + # replay (prefix match — see nerve/agent/cache_policy.py). Precise + # wall-clock time is injected per turn as a trailing message reminder + # by the engine instead, which is fresher and costs nothing cache-wise. try: tz = ZoneInfo(timezone_name) - now = datetime.now(tz).strftime("%Y-%m-%d %H:%M %Z") + today = datetime.now(tz).strftime("%Y-%m-%d %Z") except Exception: - now = datetime.now().strftime("%Y-%m-%d %H:%M") + today = datetime.now().strftime("%Y-%m-%d") context = f"""# Session Context - **Session ID:** {session_id} - **Source:** {source} -- **Current time:** {now} +- **Current date:** {today} - **Workspace:** {workspace} You have access to the following custom tools: diff --git a/nerve/config.py b/nerve/config.py index 22e6464..3198690 100644 --- a/nerve/config.py +++ b/nerve/config.py @@ -151,6 +151,16 @@ class AgentConfig: # this subscription"). Match is case-insensitive substring on the resolved # model name. Empty list = send beta for all models when context_1m=True. context_1m_excluded_models: list[str] = field(default_factory=list) + # Prompt-cache write TTL policy: "5m" (status quo — every write uses the + # default 5-minute TTL), "1h" (always request the 1-hour TTL: writes cost + # 2x base input instead of 1.25x but survive sparse turn cadences), or + # "auto" (per session at client-build time — sparse-cadence sessions such + # as persistent crons, wakeup loops and spaced conversations get 1h; + # dense sessions stay on 5m). See nerve/agent/cache_policy.py. + cache_ttl: str = "5m" + # Substrings of model names that must never request the 1h cache TTL + # (same matching semantics as context_1m_excluded_models). + cache_ttl_excluded_models: list[str] = field(default_factory=list) # Hung-CLI detection: max idle time between SDK messages on a single # turn before the engine treats the subprocess as dead and falls into # the existing CLI-crash retry path. Set to 0 to disable (legacy @@ -185,6 +195,10 @@ def from_dict(cls, d: dict) -> AgentConfig: context_1m_excluded_models=list( d.get("context_1m_excluded_models", []) or [] ), + cache_ttl=str(d.get("cache_ttl", "5m")), + cache_ttl_excluded_models=list( + d.get("cache_ttl_excluded_models", []) or [] + ), cli_idle_timeout_seconds=int(d.get("cli_idle_timeout_seconds", 900)), background_agent_permissions=bool( d.get("background_agent_permissions", True) diff --git a/nerve/cron/jobs.py b/nerve/cron/jobs.py index f071382..60567e6 100644 --- a/nerve/cron/jobs.py +++ b/nerve/cron/jobs.py @@ -33,6 +33,9 @@ class CronJob: prompt_file: str = "" description: str = "" model: str = "" # Override model; empty = use config default + # Per-job prompt-cache TTL override: "5m", "1h" or "auto"; empty = use + # agent.cache_ttl from config. See nerve/agent/cache_policy.py. + cache_ttl: str = "" session_mode: str = "isolated" # "isolated" (new session per run) or "persistent" (reuse context) context_rotate_hours: int = 24 # Hours before persistent context is rotated (0 = never) context_rotate_at: str = "" # Time of day to rotate (e.g. "04:00"); overrides hours-based rotation @@ -118,6 +121,7 @@ def from_dict(cls, d: dict, base_dir: Path | None = None) -> CronJob: prompt_file=d.get("prompt_file", ""), description=d.get("description", ""), model=d.get("model", ""), + cache_ttl=d.get("cache_ttl", ""), session_mode=d.get("session_mode", "isolated"), context_rotate_hours=int(d.get("context_rotate_hours", 24)), context_rotate_at=d.get("context_rotate_at", ""), @@ -179,6 +183,7 @@ def save_jobs(jobs: list[CronJob], jobs_file: Path) -> None: "prompt_file": job.prompt_file, "description": job.description, "model": job.model, + "cache_ttl": job.cache_ttl, "session_mode": job.session_mode, "context_rotate_hours": job.context_rotate_hours, "context_rotate_at": job.context_rotate_at, diff --git a/nerve/cron/service.py b/nerve/cron/service.py index 67621a2..c3b0bae 100644 --- a/nerve/cron/service.py +++ b/nerve/cron/service.py @@ -664,6 +664,7 @@ async def _run_job_inner(self, job: CronJob) -> None: prompt=prompt, model=model, session_id=session_id, + cache_ttl=job.cache_ttl, ) else: response = await self.engine.run_cron( @@ -671,6 +672,7 @@ async def _run_job_inner(self, job: CronJob) -> None: prompt=base_prompt, model=model, run_id=run_id, + cache_ttl=job.cache_ttl, ) # Keep the tail of the response — for multi-message runs the diff --git a/nerve/db/usage.py b/nerve/db/usage.py index 10fcf0e..18d56cf 100644 --- a/nerve/db/usage.py +++ b/nerve/db/usage.py @@ -281,6 +281,36 @@ async def get_usage_summary(self, days: int = 7) -> dict: d["total_cost_usd"] = round(d["total_cost_usd"], 4) return d + async def get_cache_ttl_turn_rows(self, days: int = 7) -> list[tuple]: + """Per-turn rows for the cache-TTL savings estimate. + + Returns ``(session_id, source, ts_epoch, model, input_tokens, + cache_read, write_5m, write_1h)`` ordered by session and time. + Legacy rows that predate the TTL split report their aggregate + ``cache_creation_input_tokens`` as 5m writes (the historical + default). + """ + async with self.db.execute( + """ + SELECT u.session_id, COALESCE(s.source, 'unknown'), + CAST(strftime('%s', u.created_at) AS REAL), + u.model, u.input_tokens, u.cache_read_input_tokens, + CASE + WHEN u.cache_creation_5m_input_tokens + + u.cache_creation_1h_input_tokens > 0 + THEN u.cache_creation_5m_input_tokens + ELSE u.cache_creation_input_tokens + END, + u.cache_creation_1h_input_tokens + FROM session_usage u + LEFT JOIN sessions s ON s.id = u.session_id + WHERE u.created_at >= DATETIME('now', ?) + ORDER BY u.session_id, u.created_at, u.id + """, + (f"-{days} days",), + ) as cursor: + return [tuple(row) async for row in cursor] + async def delete_session_usage(self, session_id: str) -> None: """Delete all usage records for a session (cascade on delete).""" await self._write( diff --git a/nerve/gateway/routes/diagnostics.py b/nerve/gateway/routes/diagnostics.py index 9b36c28..bf0350b 100644 --- a/nerve/gateway/routes/diagnostics.py +++ b/nerve/gateway/routes/diagnostics.py @@ -66,6 +66,7 @@ async def diagnostics(user: dict = Depends(require_auth)): daily_usage_res, source_usage_res, model_usage_res, + cache_ttl_rows_res, ) = await asyncio.gather( deps.db.get_cron_logs(limit=10), deps.db.get_known_source_names(), @@ -77,6 +78,7 @@ async def diagnostics(user: dict = Depends(require_auth)): deps.db.get_usage_by_period(days=7), deps.db.get_usage_by_source(days=7), deps.db.get_usage_by_model(days=7), + deps.db.get_cache_ttl_turn_rows(days=7), return_exceptions=True, ) @@ -105,6 +107,24 @@ def _ok(v, default): "by_source": source_usage_res, "by_model": model_usage_res, } + # Cache-TTL policy block: 1h write share + estimated savings vs a + # pure-5m baseline (guardrail — see cache_policy.build_ttl_report). + try: + if not isinstance(cache_ttl_rows_res, BaseException): + from nerve.agent.cache_policy import build_ttl_report + + report = build_ttl_report(cache_ttl_rows_res) + creation = usage_summary_res.get("total_cache_creation") or 0 + creation_1h = ( + usage_summary_res.get("total_cache_creation_1h") or 0 + ) + report["cache_creation_1h_share"] = round( + creation_1h / creation, 4, + ) if creation else 0.0 + report["policy"] = config.agent.cache_ttl + usage_data["cache_ttl"] = report + except Exception: + logger.exception("Failed to build cache TTL diagnostics") # Augment known sources with registered runners (includes sources that # haven't logged a run yet). diff --git a/scripts/backtest_cache_ttl.py b/scripts/backtest_cache_ttl.py new file mode 100644 index 0000000..2887aaa --- /dev/null +++ b/scripts/backtest_cache_ttl.py @@ -0,0 +1,278 @@ +#!/usr/bin/env python3 +"""Backtest prompt-cache TTL policies against historical session_usage data. + +Simulates three cache-TTL policies over real per-turn token traffic and +reports the weekly USD delta of each vs the status quo: + +- ``5m`` — status quo: every cache write uses the default 5-minute TTL. +- ``1h`` — every session requests the 1-hour TTL (writes 2.0x base + input instead of 1.25x, but turns arriving within an hour of the + previous turn read the prefix at 0.1x instead of re-writing it). +- ``auto`` — per-session: 1h iff the session's median turn gap falls in + (5min, 1h] (sparse cadence that actually benefits), else 5m. + +Simulation model (per session, turns ordered by created_at): + +The observed data was produced under the 5m policy, so the observed +``cache_creation`` of a turn whose gap from the previous turn exceeds +5 minutes is (mostly) a *full re-write* of the accumulated prefix. +Under a 1h TTL that prefix would still be warm, so the re-written +portion converts into cache reads. The warm-prefix size after turn i +is estimated with a running counter ``C``: + + C = creation[i] if turn i looked cold (gap > 5m / first + turn / model switch) + C = C + creation[i] if turn i looked warm (gap <= 5m) + +(``cache_read`` sums the prefix once per agentic-loop iteration inside a +run, so it multi-counts and is NOT a usable prefix estimate; creation +tokens are written once and are.) + +For a turn with gap in (5min, 1h] the 1h simulation converts +``min(creation[i], C_prev)`` write-tokens into read-tokens. Gaps > 1h +are cold under both policies (and in Nerve the SDK client is recycled at +the 60-min idle sweep, changing the system-prompt bytes anyway, so >1h +continuity would not materialize even if the TTL allowed it). + +Output-token and web-search costs are identical across policies and are +excluded everywhere. + +Usage: + python3 scripts/backtest_cache_ttl.py [--db ~/.nerve/nerve.db] \ + [--days 28] [--ttl-threshold 3600] [--top 20] +""" + +from __future__ import annotations + +import argparse +import sqlite3 +import statistics +import sys +from dataclasses import dataclass +from pathlib import Path + +# Reuse the real pricing table — no invented prices. +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) +from nerve.db.usage import _get_pricing # noqa: E402 + +FIVE_MIN = 300.0 +ONE_HOUR = 3600.0 + + +@dataclass +class Turn: + """One session_usage row (input side only).""" + + ts: float # unix epoch seconds + model: str | None + input_tokens: int + cache_creation: int + cache_read: int + + +@dataclass +class SimResult: + cost_5m: float + cost_1h: float + converted_tokens: int # write-tokens that became reads under 1h + + +def _input_cost( + pricing: tuple, fresh: int, reads: int, w5m: int, w1h: int, +) -> float: + p_in, _p_out, p_read, p_c5m, p_c1h, _p_ws = pricing + return ( + fresh * p_in + reads * p_read + w5m * p_c5m + w1h * p_c1h + ) / 1_000_000 + + +def simulate_session( + turns: list[Turn], ttl_threshold: float = ONE_HOUR, +) -> SimResult: + """Simulate 5m (status quo) and 1h policies over one session's turns. + + Returns input-side costs only (output/web-search excluded — identical + across policies). + """ + cost_5m = 0.0 + cost_1h = 0.0 + converted_total = 0 + warm_prefix = 0 # C: estimated cached-prefix tokens after prev turn + prev_ts: float | None = None + prev_model: str | None = None + + for t in turns: + pricing = _get_pricing(t.model) + gap = None if prev_ts is None else t.ts - prev_ts + model_switch = prev_model is not None and t.model != prev_model + + # --- status quo: everything observed was a 5m write + cost_5m += _input_cost( + pricing, t.input_tokens, t.cache_read, t.cache_creation, 0, + ) + + # --- 1h policy counterfactual + if gap is None or model_switch or gap > ttl_threshold: + # Cold under 1h too: same token flows, writes at the 1h rate. + converted = 0 + elif gap <= FIVE_MIN: + # Warm under both policies: same token flows. + converted = 0 + else: + # Cold observed (5m expired) but warm under 1h: the re-written + # prefix converts into reads. + converted = min(t.cache_creation, warm_prefix) + cost_1h += _input_cost( + pricing, + t.input_tokens, + t.cache_read + converted, + 0, + t.cache_creation - converted, + ) + converted_total += converted + + # --- update the observed-world warm-prefix estimate + if gap is None or model_switch or gap > FIVE_MIN: + warm_prefix = t.cache_creation # full re-write observed + else: + warm_prefix += t.cache_creation # incremental suffix write + + prev_ts = t.ts + prev_model = t.model + + return SimResult(cost_5m, cost_1h, converted_total) + + +def auto_policy_for_session( + turns: list[Turn], ttl_threshold: float = ONE_HOUR, +) -> str: + """Cadence heuristic: 1h iff the median turn gap is in (5min, ttl]. + + Sessions with <2 turns (no gaps) stay on 5m — a lone write at 2x can + only lose money. Median gaps > ttl also stay on 5m: the prefix dies + before the next turn either way, so the 2x write premium buys nothing. + """ + if len(turns) < 2: + return "5m" + gaps = [b.ts - a.ts for a, b in zip(turns, turns[1:])] + med = statistics.median(gaps) + return "1h" if FIVE_MIN < med <= ttl_threshold else "5m" + + +def load_sessions( + db_path: str, days: int, +) -> dict[str, tuple[str, list[Turn]]]: + """Load per-turn usage rows grouped by session: {sid: (source, turns)}.""" + conn = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True) + try: + rows = conn.execute( + """ + SELECT u.session_id, COALESCE(s.source, 'unknown'), + CAST(strftime('%s', u.created_at) AS REAL), + u.model, u.input_tokens, + u.cache_creation_input_tokens, u.cache_read_input_tokens + FROM session_usage u + LEFT JOIN sessions s ON s.id = u.session_id + WHERE u.created_at >= DATETIME('now', ?) + ORDER BY u.session_id, u.created_at, u.id + """, + (f"-{days} days",), + ).fetchall() + finally: + conn.close() + + sessions: dict[str, tuple[str, list[Turn]]] = {} + for sid, source, ts, model, inp, cc, cr in rows: + if ts is None: + continue + entry = sessions.setdefault(sid, (source, [])) + entry[1].append(Turn(ts, model, inp or 0, cc or 0, cr or 0)) + return sessions + + +def main() -> None: + ap = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + ap.add_argument("--db", default=str(Path.home() / ".nerve" / "nerve.db")) + ap.add_argument("--days", type=int, default=28) + ap.add_argument("--ttl-threshold", type=float, default=ONE_HOUR, + help="Effective 1h-hit window in seconds (default 3600)") + ap.add_argument("--top", type=int, default=20) + args = ap.parse_args() + + sessions = load_sessions(args.db, args.days) + if not sessions: + print("No usage rows in window.") + return + + weeks = args.days / 7.0 + per_source: dict[str, dict[str, float]] = {} + per_session: list[tuple[float, str, str, int, str]] = [] + tot = {"5m": 0.0, "1h": 0.0, "auto": 0.0, "converted": 0} + losers_1h = 0.0 + + for sid, (source, turns) in sessions.items(): + sim = simulate_session(turns, args.ttl_threshold) + policy = auto_policy_for_session(turns, args.ttl_threshold) + cost_auto = sim.cost_1h if policy == "1h" else sim.cost_5m + + agg = per_source.setdefault( + source, {"5m": 0.0, "1h": 0.0, "auto": 0.0, "sessions": 0, + "auto_1h_sessions": 0}, + ) + agg["5m"] += sim.cost_5m + agg["1h"] += sim.cost_1h + agg["auto"] += cost_auto + agg["sessions"] += 1 + agg["auto_1h_sessions"] += policy == "1h" + + tot["5m"] += sim.cost_5m + tot["1h"] += sim.cost_1h + tot["auto"] += cost_auto + tot["converted"] += sim.converted_tokens + if sim.cost_1h > sim.cost_5m: + losers_1h += sim.cost_1h - sim.cost_5m + + per_session.append( + (sim.cost_5m - cost_auto, sid, source, len(turns), policy), + ) + + print(f"Backtest window: {args.days} days " + f"({len(sessions)} sessions, " + f"{sum(len(t) for _, t in sessions.values())} turns)\n") + + hdr = (f"{'source':<10} {'sess':>5} {'auto=1h':>7} " + f"{'5m $/wk':>9} {'1h $/wk':>9} {'auto $/wk':>9} " + f"{'Δ1h/wk':>8} {'Δauto/wk':>9}") + print(hdr) + print("-" * len(hdr)) + for source, a in sorted(per_source.items(), key=lambda kv: -kv[1]["5m"]): + print(f"{source:<10} {a['sessions']:>5} {a['auto_1h_sessions']:>7} " + f"{a['5m'] / weeks:>9.2f} {a['1h'] / weeks:>9.2f} " + f"{a['auto'] / weeks:>9.2f} " + f"{(a['5m'] - a['1h']) / weeks:>+8.2f} " + f"{(a['5m'] - a['auto']) / weeks:>+9.2f}") + print("-" * len(hdr)) + print(f"{'TOTAL':<10} {len(sessions):>5} {'':>7} " + f"{tot['5m'] / weeks:>9.2f} {tot['1h'] / weeks:>9.2f} " + f"{tot['auto'] / weeks:>9.2f} " + f"{(tot['5m'] - tot['1h']) / weeks:>+8.2f} " + f"{(tot['5m'] - tot['auto']) / weeks:>+9.2f}") + + print(f"\nWrite-tokens converted to reads under 1h: " + f"{tot['converted'] / 1e6:.1f}M over {args.days}d") + print(f"Blanket-1h losses on dense/one-shot sessions: " + f"${losers_1h / weeks:.2f}/wk (avoided by auto)") + + print(f"\nTop {args.top} sessions by auto-policy savings (window total):") + per_session.sort(reverse=True) + for delta, sid, source, n, policy in per_session[: args.top]: + print(f" {delta:>+8.2f} USD {sid:<44} {source:<9} " + f"turns={n:<4} auto={policy}") + + gate = (tot["5m"] - tot["auto"]) / weeks + print(f"\nGate: auto saves ${gate:.2f}/week " + f"({'PASS ≥ $25' if gate >= 25 else 'FAIL < $25'})") + + +if __name__ == "__main__": + main() diff --git a/tests/test_cache_policy.py b/tests/test_cache_policy.py new file mode 100644 index 0000000..9eeb7e2 --- /dev/null +++ b/tests/test_cache_policy.py @@ -0,0 +1,325 @@ +"""Tests for nerve.agent.cache_policy — cadence-aware prompt-cache TTL.""" + +import importlib.util +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import AsyncMock, patch + +import pytest + +from nerve.agent.cache_policy import ( + build_ttl_report, + cache_ttl_env, + estimate_live_ttl_delta, + resolve_cache_ttl, +) +from nerve.agent.engine import AgentEngine +from nerve.config import AgentConfig + +# --------------------------------------------------------------------------- +# The backtest simulator (scripts/) shares the gap→conversion model; import +# it from the script file so the hand-computed cases below pin both. +# --------------------------------------------------------------------------- +_spec = importlib.util.spec_from_file_location( + "backtest_cache_ttl", + Path(__file__).resolve().parent.parent / "scripts" / "backtest_cache_ttl.py", +) +backtest = importlib.util.module_from_spec(_spec) +import sys # noqa: E402 + +sys.modules["backtest_cache_ttl"] = backtest # dataclasses need the module registered +_spec.loader.exec_module(backtest) + + +def _cfg(**kw) -> AgentConfig: + return AgentConfig(**kw) + + +def _resolve(agent_cfg, *, source="web", model="claude-fable-5", + meta=None, gaps=None, is_claude=True, gaps_raise=False): + """Run resolve_cache_ttl with the cadence query stubbed out.""" + import asyncio + + mock = AsyncMock(return_value=gaps or []) + if gaps_raise: + mock.side_effect = RuntimeError("db down") + with patch("nerve.agent.cache_policy.get_recent_turn_gaps", mock): + return asyncio.run( + resolve_cache_ttl( + agent_cfg, db=object(), session_id="s1", source=source, + model=model, session_meta=meta, is_claude_model=is_claude, + ) + ) + + +# --------------------------------------------------------------------------- +# cache_ttl_env +# --------------------------------------------------------------------------- + +def test_env_5m_is_empty(): + assert cache_ttl_env("5m") == {} + assert cache_ttl_env("5m", is_bedrock=True) == {} + + +def test_env_1h_sets_cli_flag(): + assert cache_ttl_env("1h") == {"ENABLE_PROMPT_CACHING_1H": "1"} + + +def test_env_1h_bedrock_sets_both_flags(): + env = cache_ttl_env("1h", is_bedrock=True) + assert env["ENABLE_PROMPT_CACHING_1H"] == "1" + assert env["ENABLE_PROMPT_CACHING_1H_BEDROCK"] == "1" + + +# --------------------------------------------------------------------------- +# resolve_cache_ttl — mode / exclusion / prior matrix +# --------------------------------------------------------------------------- + +class TestResolveModes: + def test_default_config_is_5m(self): + assert _cfg().cache_ttl == "5m" + + def test_mode_5m_never_requests(self): + cfg = _cfg(cache_ttl="5m") + for source in ("web", "cron", "wakeup", "telegram"): + assert _resolve(cfg, source=source, gaps=[1800, 1800]) == "5m" + + def test_mode_1h_always_requests(self): + cfg = _cfg(cache_ttl="1h") + assert _resolve(cfg, source="web") == "1h" + assert _resolve(cfg, source="cron") == "1h" + + def test_mode_1h_respects_model_exclusion(self): + cfg = _cfg(cache_ttl="1h", + cache_ttl_excluded_models=["sonnet-4-6"]) + assert _resolve(cfg, model="claude-sonnet-4-6") == "5m" + assert _resolve(cfg, model="claude-fable-5") == "1h" + + def test_auto_respects_model_exclusion(self): + cfg = _cfg(cache_ttl="auto", cache_ttl_excluded_models=["haiku"]) + assert _resolve(cfg, model="claude-haiku-4-5", + gaps=[1800, 1800]) == "5m" + + def test_non_claude_model_never_requests(self): + cfg = _cfg(cache_ttl="1h") + assert _resolve(cfg, model="qwen3:32b", is_claude=False) == "5m" + + def test_invalid_mode_falls_back_to_5m(self): + cfg = _cfg(cache_ttl="2h") + assert _resolve(cfg, gaps=[1800, 1800]) == "5m" + + def test_metadata_override_beats_config(self): + cfg = _cfg(cache_ttl="5m") + assert _resolve(cfg, meta={"cache_ttl_override": "1h"}) == "1h" + cfg = _cfg(cache_ttl="1h") + assert _resolve(cfg, meta={"cache_ttl_override": "5m"}) == "5m" + + +class TestResolveAutoCadence: + def test_sparse_median_gets_1h(self): + cfg = _cfg(cache_ttl="auto") + # 30-minute cadence — the canonical sparse session. + assert _resolve(cfg, source="web", gaps=[1800, 1750, 1900]) == "1h" + + def test_dense_median_stays_5m_even_for_cron(self): + cfg = _cfg(cache_ttl="auto") + assert _resolve( + cfg, source="cron", gaps=[30, 60, 45], + meta={"cron_session_mode": "persistent"}, + ) == "5m" + + def test_median_beyond_ttl_stays_5m(self): + # Gaps > 1h expire either way — the 2x write premium buys nothing. + cfg = _cfg(cache_ttl="auto") + assert _resolve(cfg, source="web", gaps=[7200, 8000]) == "5m" + + def test_no_history_web_defaults_5m(self): + cfg = _cfg(cache_ttl="auto") + assert _resolve(cfg, source="web") == "5m" + assert _resolve(cfg, source="telegram") == "5m" + + def test_no_history_wakeup_defaults_1h(self): + cfg = _cfg(cache_ttl="auto") + assert _resolve(cfg, source="wakeup") == "1h" + + def test_no_history_persistent_cron_defaults_1h(self): + cfg = _cfg(cache_ttl="auto") + assert _resolve( + cfg, source="cron", meta={"cron_session_mode": "persistent"}, + ) == "1h" + + def test_no_history_isolated_cron_defaults_5m(self): + cfg = _cfg(cache_ttl="auto") + assert _resolve( + cfg, source="cron", meta={"cron_session_mode": "isolated"}, + ) == "5m" + assert _resolve(cfg, source="cron") == "5m" + + def test_cadence_query_failure_falls_back_to_priors(self): + cfg = _cfg(cache_ttl="auto") + assert _resolve(cfg, source="wakeup", gaps_raise=True) == "1h" + assert _resolve(cfg, source="web", gaps_raise=True) == "5m" + + +# --------------------------------------------------------------------------- +# Engine env wiring — the switch reaches the CLI subprocess iff resolved 1h +# --------------------------------------------------------------------------- + +def _make_env_engine(is_bedrock: bool = False) -> AgentEngine: + engine = AgentEngine.__new__(AgentEngine) + engine.config = SimpleNamespace( + provider=SimpleNamespace( + is_bedrock=is_bedrock, aws_region="", aws_profile="", + aws_access_key_id="", aws_secret_access_key="", + ), + proxy=SimpleNamespace(enabled=False, host="", port=0), + effective_api_key="", + ) + return engine + + +def test_build_env_5m_has_no_cache_flag(): + env = _make_env_engine()._build_env(cache_ttl="5m") + assert "ENABLE_PROMPT_CACHING_1H" not in env + assert "FORCE_PROMPT_CACHING_5M" not in env # never force upstream off + + +def test_build_env_1h_sets_cache_flag(): + env = _make_env_engine()._build_env(cache_ttl="1h") + assert env["ENABLE_PROMPT_CACHING_1H"] == "1" + + +def test_build_env_1h_bedrock_sets_bedrock_flag(): + env = _make_env_engine(is_bedrock=True)._build_env(cache_ttl="1h") + assert env["ENABLE_PROMPT_CACHING_1H"] == "1" + assert env["ENABLE_PROMPT_CACHING_1H_BEDROCK"] == "1" + + +def test_build_env_default_is_5m(): + env = _make_env_engine()._build_env() + assert "ENABLE_PROMPT_CACHING_1H" not in env + + +# --------------------------------------------------------------------------- +# Backtest simulator — hand-computed costs (fable pricing: input 10, +# read 1.0, write5m 12.5, write1h 20 per 1M tokens) +# --------------------------------------------------------------------------- + +FABLE = "claude-fable-5" + + +def _turn(ts, cc, cr=0, inp=0): + return backtest.Turn(ts=ts, model=FABLE, input_tokens=inp, + cache_creation=cc, cache_read=cr) + + +class TestBacktestSimulator: + def test_sparse_session_converts_prefix(self): + # t=0: cold write 100k. t=30min: observed cold re-write of 110k + # (prefix 100k + 10k new); under 1h the 100k prefix reads instead. + sim = backtest.simulate_session([ + _turn(0, 100_000), _turn(1800, 110_000), + ]) + assert sim.cost_5m == pytest.approx(210_000 * 12.5 / 1e6) # 2.625 + assert sim.cost_1h == pytest.approx( + 100_000 * 20 / 1e6 # turn 1 cold write @ 1h + + 100_000 * 1.0 / 1e6 # converted prefix read + + 10_000 * 20 / 1e6 # genuinely-new suffix write + ) # = 2.3 + assert sim.converted_tokens == 100_000 + + def test_dense_session_pays_write_premium(self): + # 60s gap: warm under both policies; 1h only raises write price. + sim = backtest.simulate_session([ + _turn(0, 100_000), _turn(60, 5_000, cr=100_000), + ]) + assert sim.converted_tokens == 0 + assert sim.cost_5m == pytest.approx( + 105_000 * 12.5 / 1e6 + 100_000 * 1.0 / 1e6, + ) + assert sim.cost_1h == pytest.approx( + 105_000 * 20 / 1e6 + 100_000 * 1.0 / 1e6, + ) + assert sim.cost_1h > sim.cost_5m + + def test_gap_beyond_ttl_is_cold_for_both(self): + sim = backtest.simulate_session([ + _turn(0, 100_000), _turn(7200, 120_000), + ]) + assert sim.converted_tokens == 0 + assert sim.cost_5m == pytest.approx(220_000 * 12.5 / 1e6) + assert sim.cost_1h == pytest.approx(220_000 * 20 / 1e6) + + def test_model_switch_breaks_conversion(self): + turns = [_turn(0, 100_000), _turn(1800, 110_000)] + turns[1].model = "claude-haiku-4-5" + sim = backtest.simulate_session(turns) + assert sim.converted_tokens == 0 + + def test_conversion_capped_by_warm_prefix(self): + # Second write smaller than tracked prefix (compaction): convert + # only what was actually re-written. + sim = backtest.simulate_session([ + _turn(0, 100_000), _turn(1800, 40_000), + ]) + assert sim.converted_tokens == 40_000 + + def test_auto_policy_thresholds(self): + pol = backtest.auto_policy_for_session + assert pol([_turn(0, 1)]) == "5m" # no gaps + assert pol([_turn(0, 1), _turn(60, 1)]) == "5m" # dense + assert pol([_turn(0, 1), _turn(1800, 1)]) == "1h" # sparse + assert pol([_turn(0, 1), _turn(7200, 1)]) == "5m" # beyond TTL + + +# --------------------------------------------------------------------------- +# Live counterfactual (diagnostics guardrail input) +# --------------------------------------------------------------------------- + +class TestLiveTtlDelta: + def test_1h_traffic_savings_hand_computed(self): + # (ts, model, input, reads, w5m, w1h) + rows = [ + (0, FABLE, 0, 0, 0, 100_000), + (1800, FABLE, 0, 100_000, 0, 10_000), + ] + est = estimate_live_ttl_delta(rows) + assert est["actual"] == pytest.approx( + 100_000 * 20 / 1e6 + 100_000 * 1.0 / 1e6 + 10_000 * 20 / 1e6, + ) # 2.3 + assert est["baseline_5m"] == pytest.approx( + 100_000 * 12.5 / 1e6 # turn 1 write @ 5m + + 100_000 * 12.5 / 1e6 # converted read → 5m re-write + + 10_000 * 12.5 / 1e6 # suffix write @ 5m + ) # 2.625 + assert est["savings"] == pytest.approx(0.325) + + def test_pure_5m_traffic_has_zero_delta(self): + rows = [ + (0, FABLE, 0, 0, 100_000, 0), + (1800, FABLE, 0, 50_000, 120_000, 0), + ] + est = estimate_live_ttl_delta(rows) + assert est["savings"] == pytest.approx(0.0) + + def test_report_flags_regression(self): + # A lone 1h write with no follow-up costs the 2x premium for + # nothing → negative savings → guardrail flags the source. + rows = [ + ("s1", "cron", 0.0, FABLE, 0, 0, 0, 1_000_000), + ] + report = build_ttl_report(rows) + assert report["by_source"]["cron"]["savings"] < 0 + assert "cron" in report["regressions"] + + def test_report_aggregates_by_source(self): + rows = [ + ("s1", "web", 0.0, FABLE, 0, 0, 0, 100_000), + ("s1", "web", 1800.0, FABLE, 0, 100_000, 0, 10_000), + ("s2", "cron", 0.0, FABLE, 0, 0, 50_000, 0), + ] + report = build_ttl_report(rows) + assert report["by_source"]["web"]["savings"] == pytest.approx(0.325) + assert report["by_source"]["cron"]["savings"] == pytest.approx(0.0) + assert report["total"]["savings"] == pytest.approx(0.325) + assert report["regressions"] == []