From 40d35e2e73880c4c97c8f4ab20871c8c791de201 Mon Sep 17 00:00:00 2001 From: AddisonTech Date: Sat, 24 Oct 2026 16:18:37 -0400 Subject: [PATCH 1/2] Cache the system prompt and price cached tokens Every model call now sends the system prompt as one text block with a cache_control ephemeral breakpoint (default 5 minute TTL). The system prompt depends only on the envelope config and the building names, so it is byte-identical every hour of a run; the hour's observation follows it in the user message. The manifest records prompt_cache, and replay reads it (absent means off), so runs recorded before caching rebuild the exact request they sent and still replay. Cost accounting reads cache_read_input_tokens and cache_creation_input_tokens from every response. input_tokens is now uncached input only. The model table carries cache write (1.25x input) and cache read (0.1x input) prices, and the run budget charges the priced cost. The worst case check before each call prices the cached prefix as a cache write. calls.jsonl records both cache counts per call, result.json records token totals and the cache hit rate, and the run summary, replay, compare output and live page show the hit rate. --- hyperbolic/cli.py | 14 ++- hyperbolic/live.py | 1 + hyperbolic/live_assets/live.js | 3 +- hyperbolic/model.py | 79 ++++++++++----- hyperbolic/prompt.py | 15 ++- hyperbolic/runner.py | 33 ++++++- hyperbolic/viewer.py | 2 + tests/test_dryrun.py | 5 +- tests/test_prompt_cache.py | 173 +++++++++++++++++++++++++++++++++ tests/test_prompt_v7.py | 3 +- 10 files changed, 290 insertions(+), 38 deletions(-) create mode 100644 tests/test_prompt_cache.py diff --git a/hyperbolic/cli.py b/hyperbolic/cli.py index c54d24c..d6085fb 100644 --- a/hyperbolic/cli.py +++ b/hyperbolic/cli.py @@ -122,7 +122,8 @@ def cmd_replay(args) -> int: rep = replay_run(args.run_dir, out_dir=args.out) print(f"replayed {args.run_dir}: {rep.steps} hours, {rep.calls} recorded calls, {rep.api_calls} API calls") print(f" identical decisions and actions: {rep.identical_steps}/{rep.steps}") - print(f" clamped hours {rep.clamped_steps}, fallback hours {rep.fallback_steps}, recorded cost ${rep.cost_usd:.4f}") + print(f" clamped hours {rep.clamped_steps}, fallback hours {rep.fallback_steps}, recorded cost ${rep.cost_usd:.4f}, " + f"cache hit rate {_hit_rate(rep.cache_hit_rate)}") if rep.deterministic: print(" deterministic: yes") return 0 @@ -165,7 +166,8 @@ def cmd_compare(args) -> int: print(f" {label}: {d}") r = s["agent_run"] print(f" model calls {r['calls']}, clamped hours {r['clamped_hours']}, fallback hours {r['fallback_hours']}, " - f"cost ${r['cost_usd']:.4f} of ${r['budget_usd']:.2f}{' (budget stop hit)' if r['budget_stopped'] else ''}") + f"cost ${r['cost_usd']:.4f} of ${r['budget_usd']:.2f}{' (budget stop hit)' if r['budget_stopped'] else ''}" + + (f", cache hit rate {_hit_rate(r['cache_hit_rate'])}" if "cache_hit_rate" in r else "")) print(f"\nviewer data written to {args.out}") return 0 @@ -245,6 +247,10 @@ def _fmt(v): return "n/a" if v is None else f"{v:.4f}" +def _hit_rate(v) -> str: + return "n/a" if v is None else f"{100 * v:.1f}%" + + def _print_result(res: dict) -> None: k = res["district_kpis"] parts = [f"{key}={k[key]:.4f}" for key in dc.HEADLINE_KPIS if k.get(key) is not None] @@ -252,6 +258,10 @@ def _print_result(res: dict) -> None: if res.get("calls"): print(f" model calls {res['calls']}, clamped {res['clamped_steps']}, fallback {res['fallback_steps']}, " f"cost ${res['cost_usd']:.4f}{'; ' + res['budget_stop_reason'] if res['budget_stopped'] else ''}") + t = res.get("tokens") + if t: + print(f" input tokens {t['input']} uncached, {t['cache_read']} cache reads, {t['cache_write']} cache " + f"writes; output {t['output']}; cache hit rate {_hit_rate(res.get('cache_hit_rate'))}") def build_parser() -> argparse.ArgumentParser: diff --git a/hyperbolic/live.py b/hyperbolic/live.py index 71513c6..02dd5ff 100644 --- a/hyperbolic/live.py +++ b/hyperbolic/live.py @@ -408,6 +408,7 @@ def on_step(rec: dict) -> None: "stop_reason": run.result.get("stop_reason"), "steps": run.result["steps"], "hours": run.hours, "agent_run": str(run.agent_dir), "baseline_run": str(run.baseline_dir), "kpis": rows, "cost_usd": run.result["cost_usd"], "calls": run.result["calls"], + "cache_hit_rate": run.result.get("cache_hit_rate"), "clamped": run.result["clamped_steps"], "fallback": run.result["fallback_steps"], "budget_stopped": run.result["budget_stopped"], "budget_stop_reason": run.result["budget_stop_reason"], }) diff --git a/hyperbolic/live_assets/live.js b/hyperbolic/live_assets/live.js index 6c8f657..d596b64 100644 --- a/hyperbolic/live_assets/live.js +++ b/hyperbolic/live_assets/live.js @@ -315,7 +315,8 @@ table.appendChild(tb); box.appendChild(table); var facts = el("p", "live-note", "API spend " + money(ev.cost_usd) + ", " + ev.calls + " calls, " + ev.clamped + " clamped hours, " + - ev.fallback + " fallback hours" + (ev.budget_stopped ? ", budget stop hit" : "") + ". Recorded in " + ev.agent_run + "; replay it with `hyperbolic replay`."); + ev.fallback + " fallback hours" + (ev.cache_hit_rate != null ? ", cache hit rate " + fmt(100 * ev.cache_hit_rate, 1) + "%" : "") + + (ev.budget_stopped ? ", budget stop hit" : "") + ". Recorded in " + ev.agent_run + "; replay it with `hyperbolic replay`."); box.appendChild(facts); var pub = button("live-publish", "Publish to docs/data", "primary"); var note = el("p", "live-note"); note.id = "live-published"; diff --git a/hyperbolic/model.py b/hyperbolic/model.py index 4f0f8d3..1c79c30 100644 --- a/hyperbolic/model.py +++ b/hyperbolic/model.py @@ -25,10 +25,13 @@ KEY_ENV_VARS = ("HYPERBOLIC_ANTHROPIC_KEY", "TRIAGE_ANTHROPIC_KEY") -# USD per million tokens (input, output). +# USD per million tokens. input is uncached input; a cache write (5 minute TTL) bills at 1.25x +# the input price and a cache read at 0.1x. MODELS = { - "haiku": {"id": "claude-haiku-5-5", "input_per_mtok": 0.10, "output_per_mtok": 0.50}, - "sonnet": {"id": "claude-sonnet-5-5", "input_per_mtok": 2.00, "output_per_mtok": 10.00}, + "haiku": {"id": "claude-haiku-5-5", "input_per_mtok": 0.10, "cache_write_per_mtok": 0.125, + "cache_read_per_mtok": 0.01, "output_per_mtok": 0.50}, + "sonnet": {"id": "claude-sonnet-5-5", "input_per_mtok": 2.00, "cache_write_per_mtok": 2.50, + "cache_read_per_mtok": 0.20, "output_per_mtok": 10.00}, } DEFAULT_MODEL = "haiku" DEFAULT_BUDGET_USD = 2.00 @@ -55,15 +58,35 @@ def request_sha256(request: dict) -> str: return hashlib.sha256(blob.encode()).hexdigest() -def call_cost_usd(model_key: str, input_tokens: int, output_tokens: int) -> float: +def call_cost_usd(model_key: str, input_tokens: int, output_tokens: int, + cache_read_tokens: int = 0, cache_write_tokens: int = 0) -> float: + """input_tokens is uncached input only, as the API reports it beside the two cache counts.""" p = MODELS[model_key] - return (input_tokens * p["input_per_mtok"] + output_tokens * p["output_per_mtok"]) / 1_000_000 + return (input_tokens * p["input_per_mtok"] + cache_write_tokens * p["cache_write_per_mtok"] + + cache_read_tokens * p["cache_read_per_mtok"] + output_tokens * p["output_per_mtok"]) / 1_000_000 + + +def cache_hit_rate(input_tokens: int, cache_read_tokens: int, cache_write_tokens: int) -> float | None: + """Share of all input tokens served from the cache; None when there was no input.""" + total = input_tokens + cache_read_tokens + cache_write_tokens + return None if total == 0 else cache_read_tokens / total + + +def _cached_prefix_chars(request: dict) -> int: + """Characters up to and including the last block marked for caching (tools render before system).""" + system = request.get("system") + blocks = [*request.get("tools", []), *(system if isinstance(system, list) else [])] + marked = [i for i, b in enumerate(blocks) if b.get("cache_control")] + return len(json.dumps(blocks[: marked[-1] + 1], ensure_ascii=False)) if marked else 0 def worst_case_cost_usd(model_key: str, request: dict) -> float: - """Upper bound for one call: about 3 characters per input token, full max_tokens output.""" + """Upper bound for one call: about 3 characters per input token, full max_tokens output, and a + cached prefix priced as a cache write, the dearest way it can bill.""" chars = len(json.dumps(request, ensure_ascii=False)) - return call_cost_usd(model_key, math.ceil(chars / 3), int(request.get("max_tokens", DEFAULT_MAX_TOKENS))) + prefix = _cached_prefix_chars(request) + return call_cost_usd(model_key, math.ceil((chars - prefix) / 3), int(request.get("max_tokens", DEFAULT_MAX_TOKENS)), + cache_write_tokens=math.ceil(prefix / 3)) @dataclass @@ -99,6 +122,9 @@ class CallResult: model: str | None = None latency_s: float = 0.0 extra: dict[str, Any] = field(default_factory=dict) + # input_tokens counts uncached input only; these are the cached prefix read or written. + cache_read_input_tokens: int = 0 + cache_creation_input_tokens: int = 0 class Supervisor(Protocol): @@ -135,12 +161,13 @@ def call(self, step: int, request: dict) -> CallResult: usage = resp.usage result = CallResult( text=None, - input_tokens=int(usage.input_tokens or 0) + int(getattr(usage, "cache_read_input_tokens", 0) or 0) - + int(getattr(usage, "cache_creation_input_tokens", 0) or 0), + input_tokens=int(usage.input_tokens or 0), output_tokens=int(usage.output_tokens or 0), stop_reason=resp.stop_reason, model=resp.model, latency_s=latency, + cache_read_input_tokens=int(getattr(usage, "cache_read_input_tokens", 0) or 0), + cache_creation_input_tokens=int(getattr(usage, "cache_creation_input_tokens", 0) or 0), ) if resp.stop_reason == "refusal": result.error = "model declined the request" @@ -156,6 +183,20 @@ class ReplayMismatch(RuntimeError): pass +def _recorded_result(reply: dict) -> CallResult: + # Runs recorded before prompt caching carry no cache counts; their input_tokens is all input. + return CallResult( + text=reply.get("text"), + input_tokens=reply.get("input_tokens", 0), + output_tokens=reply.get("output_tokens", 0), + stop_reason=reply.get("stop_reason"), + error=reply.get("error"), + model=reply.get("model"), + cache_read_input_tokens=reply.get("cache_read_input_tokens", 0), + cache_creation_input_tokens=reply.get("cache_creation_input_tokens", 0), + ) + + class ReplaySupervisor: """Serves recorded replies. Fails loudly if the request differs from the recording.""" @@ -177,15 +218,7 @@ def call(self, step: int, request: dict) -> CallResult: sha = request_sha256(request) if sha != rec["request_sha256"]: raise ReplayMismatch(f"step {step}: request differs from recording") - reply = rec["reply"] - return CallResult( - text=reply.get("text"), - input_tokens=reply.get("input_tokens", 0), - output_tokens=reply.get("output_tokens", 0), - stop_reason=reply.get("stop_reason"), - error=reply.get("error"), - model=reply.get("model"), - ) + return _recorded_result(rec["reply"]) class RecordedReplySupervisor: @@ -203,15 +236,7 @@ def call(self, step: int, request: dict) -> CallResult: rec = self._inner._by_step.get(step) if rec is None: return CallResult(None, error="no recorded reply for this hour") - reply = rec["reply"] - return CallResult( - text=reply.get("text"), - input_tokens=reply.get("input_tokens", 0), - output_tokens=reply.get("output_tokens", 0), - stop_reason=reply.get("stop_reason"), - error=reply.get("error"), - model=reply.get("model"), - ) + return _recorded_result(rec["reply"]) @dataclass(frozen=True) diff --git a/hyperbolic/prompt.py b/hyperbolic/prompt.py index 06aa0f4..4cadf66 100644 --- a/hyperbolic/prompt.py +++ b/hyperbolic/prompt.py @@ -1105,12 +1105,25 @@ def user_message(obs: dict) -> str: ) +def system_text(request: dict) -> str: + """The system prompt's text, whether it was sent as a plain string or as cached blocks.""" + system = request["system"] + return system if isinstance(system, str) else "".join(b["text"] for b in system) + + def build_request(obs: dict, cfg: EnvelopeConfig, buildings: tuple[str, ...], model_key: str, max_tokens: int = DEFAULT_MAX_TOKENS, effort: str = DEFAULT_EFFORT, - prompt_version: int = 1) -> dict: + prompt_version: int = 1, prompt_cache: bool = False) -> dict: + """prompt_cache marks the system prompt as the cached prefix. It depends only on the envelope + config and the building names, so it is the same bytes every hour of a run; everything that + changes by the hour is in the user message after it. Off by default so runs recorded before + prompt caching rebuild the request they sent.""" system = {1: system_prompt, 2: system_prompt_v2, 3: system_prompt_v3, 4: system_prompt_v4, 5: system_prompt_v5, 6: system_prompt_v6, 7: system_prompt_v7, 8: system_prompt_v8, 9: system_prompt_v9}[prompt_version](cfg, buildings) + if prompt_cache: + # Default 5 minute TTL: hourly calls run back to back, seconds apart. + system = [{"type": "text", "text": system, "cache_control": {"type": "ephemeral"}}] return { "model": MODELS[model_key]["id"], "max_tokens": max_tokens, diff --git a/hyperbolic/runner.py b/hyperbolic/runner.py index 66f41da..6dffa45 100644 --- a/hyperbolic/runner.py +++ b/hyperbolic/runner.py @@ -3,8 +3,10 @@ A run directory holds everything needed to replay it with zero model or simulator calls: manifest.json settings, dataset, envelope config steps.jsonl per hour: district state, actions applied, outcome, decision - calls.jsonl per model call: full request, reply, tokens, cost - result.json CityLearn KPIs and counts + calls.jsonl per model call: full request, reply, tokens, cost. The reply's input_tokens is + uncached input, beside cache_read_input_tokens and cache_creation_input_tokens + (runs from before prompt caching record no cache counts and no cache reads). + result.json CityLearn KPIs and counts, token totals and the cache hit rate """ from __future__ import annotations @@ -25,6 +27,7 @@ ReplayMismatch, ReplaySupervisor, Supervisor, + cache_hit_rate, call_cost_usd, request_sha256, worst_case_cost_usd, @@ -170,11 +173,14 @@ class AgentController: max_tokens: int = 2000 effort: str = "low" prompt_version: int = PROMPT_VERSION + # New runs cache the system prompt; replay passes the manifest's value (absent before caching). + prompt_cache: bool = True last: Decision | None = None lookahead: Lookahead = field(default_factory=Lookahead) peaks: PeakTracker | None = None mode: str = "agent" counts: dict = field(default_factory=lambda: {"calls": 0, "clamped": 0, "fallback": 0, "budget_skipped": 0}) + tokens: dict = field(default_factory=lambda: {"input": 0, "cache_read": 0, "cache_write": 0, "output": 0}) def describe(self) -> dict: return { @@ -185,6 +191,7 @@ def describe(self) -> dict: "max_tokens": self.max_tokens, "effort": self.effort, "prompt_version": self.prompt_version, + "prompt_cache": self.prompt_cache, } def decide(self, step: int, state: dict, writer: RunWriter) -> tuple[Decision, dict | None]: @@ -198,7 +205,7 @@ def decide(self, step: int, state: dict, writer: RunWriter) -> tuple[Decision, d buildings = tuple(state["buildings"]) obs = observe(self.prompt_version, state, ctx, self.last, self.lookahead, self.envelope) request = build_request(obs, self.envelope.config, buildings, self.supervisor.model_key, - self.max_tokens, self.effort, self.prompt_version) + self.max_tokens, self.effort, self.prompt_version, self.prompt_cache) worst = worst_case_cost_usd(self.supervisor.model_key, request) call_info = None if not self.budget.allows(worst): @@ -211,9 +218,13 @@ def decide(self, step: int, state: dict, writer: RunWriter) -> tuple[Decision, d raise except Exception as exc: # any client failure is a missing reply, never a crash res = CallResult(None, error=f"client error: {type(exc).__name__}") - cost = call_cost_usd(self.supervisor.model_key, res.input_tokens, res.output_tokens) + cost = call_cost_usd(self.supervisor.model_key, res.input_tokens, res.output_tokens, + res.cache_read_input_tokens, res.cache_creation_input_tokens) self.budget.charge(cost) self.counts["calls"] += 1 + for k, v in (("input", res.input_tokens), ("cache_read", res.cache_read_input_tokens), + ("cache_write", res.cache_creation_input_tokens), ("output", res.output_tokens)): + self.tokens[k] += v sha = request_sha256(request) writer.call({ "step": step, @@ -222,6 +233,8 @@ def decide(self, step: int, state: dict, writer: RunWriter) -> tuple[Decision, d "reply": { "text": res.text, "input_tokens": res.input_tokens, + "cache_read_input_tokens": res.cache_read_input_tokens, + "cache_creation_input_tokens": res.cache_creation_input_tokens, "output_tokens": res.output_tokens, "stop_reason": res.stop_reason, "error": res.error, @@ -351,6 +364,11 @@ def run_session( "budget_stop_reason": budget.stop_reason if budget else None, "finished_at": dt.datetime.now(dt.timezone.utc).isoformat(timespec="seconds"), } + tokens = getattr(controller, "tokens", None) + if tokens is not None: + result["tokens"] = dict(tokens) + rate = cache_hit_rate(tokens["input"], tokens["cache_read"], tokens["cache_write"]) + result["cache_hit_rate"] = None if rate is None else round(rate, 4) if stopped: result["stopped"] = True result["stop_reason"] = f"stopped by the user after {step} of {episode_hours} hours" @@ -410,6 +428,7 @@ class ReplayReport: clamped_steps: int fallback_steps: int error: str | None = None + cache_hit_rate: float | None = None @property def deterministic(self) -> bool: @@ -432,6 +451,8 @@ def replay_run(run_dir: Path, out_dir: Path | None = None, log=lambda *_: None) effort=sup["effort"], # Runs recorded before prompt versions existed used version 1. prompt_version=int(sup.get("prompt_version", 1)), + # Runs recorded before prompt caching sent the system prompt as a plain string. + prompt_cache=bool(sup.get("prompt_cache", False)), ) writer: RunWriter = RunWriter(out_dir) if out_dir is not None else NullWriter() error = None @@ -454,6 +475,9 @@ def replay_run(run_dir: Path, out_dir: Path | None = None, log=lambda *_: None) mismatched.append(min(len(replayed), len(sim.steps))) calls = load_jsonl(run_dir / "calls.jsonl") counts = getattr(controller, "counts", {}) + replies = [c.get("reply") or {} for c in calls] + hit_rate = cache_hit_rate(*(sum(r.get(k, 0) or 0 for r in replies) for k in + ("input_tokens", "cache_read_input_tokens", "cache_creation_input_tokens"))) return ReplayReport( steps=len(sim.steps), identical_steps=identical, @@ -464,4 +488,5 @@ def replay_run(run_dir: Path, out_dir: Path | None = None, log=lambda *_: None) clamped_steps=counts.get("clamped", 0), fallback_steps=counts.get("fallback", 0), error=error, + cache_hit_rate=hit_rate, ) diff --git a/hyperbolic/viewer.py b/hyperbolic/viewer.py index 94a18e8..3172acb 100644 --- a/hyperbolic/viewer.py +++ b/hyperbolic/viewer.py @@ -197,6 +197,8 @@ def build_scenario(baseline_dir: Path, agent_dir: Path) -> dict: "clamped_hours": sum(1 for d in decisions if d["status"] == "clamped"), "fallback_hours": sum(1 for d in decisions if d["status"] == "fallback"), "budget_stopped": bool(ar.get("budget_stopped")), + # Only runs recorded with prompt caching carry it, so older viewer data is unchanged. + **({"cache_hit_rate": ar["cache_hit_rate"]} if "cache_hit_rate" in ar else {}), "envelope": am.get("envelope"), **_trim_end_count(am, decisions), }, diff --git a/tests/test_dryrun.py b/tests/test_dryrun.py index fc4e7c2..bd0ff01 100644 --- a/tests/test_dryrun.py +++ b/tests/test_dryrun.py @@ -10,6 +10,7 @@ from hyperbolic.cli import main from hyperbolic.envelope import Envelope, EnvelopeConfig from hyperbolic.model import Budget, CallResult, ScriptedSupervisor, observation_from_request +from hyperbolic.prompt import system_text from hyperbolic.runner import ( AgentController, BaselineController, @@ -163,7 +164,7 @@ def test_scripted_strategy_dry_run_and_replay(tmp_path, version): calls = load_jsonl(run_dir / "calls.jsonl") obs = observation_from_request(calls[5]["request"]) assert {"reference_action", "indoor_vs_setpoint_c"} <= set(obs["buildings"]["Building_1"]) - assert "reference_action" in calls[5]["request"]["system"] + assert "reference_action" in system_text(calls[5]["request"]) assert ("smoothing" in obs) == (version == 5) rep = replay_run(run_dir) assert rep.deterministic and rep.api_calls == 0 and rep.identical_steps == HOURS @@ -202,7 +203,7 @@ def test_v6_dry_run_with_the_trim_hold_and_replay(tmp_path): calls = load_jsonl(run_dir / "calls.jsonl") obs = observation_from_request(calls[5]["request"]) assert {"strategy_battery", "cooling_trim_held"} <= set(obs["buildings"]["Building_1"]) - assert "reference_action" not in json.dumps(obs) and "end_trim" in calls[5]["request"]["system"] + assert "reference_action" not in json.dumps(obs) and "end_trim" in system_text(calls[5]["request"]) rep = replay_run(run_dir) assert rep.deterministic and rep.api_calls == 0 and rep.identical_steps == HOURS diff --git a/tests/test_prompt_cache.py b/tests/test_prompt_cache.py new file mode 100644 index 0000000..e9a01ba --- /dev/null +++ b/tests/test_prompt_cache.py @@ -0,0 +1,173 @@ +"""Prompt caching: the system prompt is a byte-identical cached prefix, the hour's state follows it, +and cache reads and writes are priced and recorded.""" + +import json +import re +from types import SimpleNamespace + +import pytest +from fakes import BUILDINGS, FakeDistrict + +from hyperbolic.envelope import Envelope +from hyperbolic.model import ( + MODELS, + Budget, + CallResult, + LiveSupervisor, + ScriptedSupervisor, + cache_hit_rate, + call_cost_usd, + observation_from_request, + worst_case_cost_usd, +) +from hyperbolic.prompt import PROMPT_VERSION, PROMPT_VERSIONS, system_text +from hyperbolic.runner import AgentController, NullWriter, RunWriter, controller_settings, load_jsonl, replay_run, run_session + +quiet = lambda *_: None # noqa: E731 + + +class CapturingSupervisor: + """A fake client: keeps every request and answers with no actions.""" + + kind = "fake" + model_key = "haiku" + + def __init__(self): + self.requests = [] + + def call(self, step, request): + self.requests.append(json.loads(json.dumps(request))) + reply = {"reason": "hold", **{b: {"cooling": None, "dhw": None, "battery": None} for b in BUILDINGS}} + return CallResult(json.dumps(reply), stop_reason="end_turn") + + +def _requests(version, hours=2): + version, cfg = controller_settings(False, version) + sup = CapturingSupervisor() + ctl = AgentController(supervisor=sup, envelope=Envelope(cfg), budget=Budget(1.0), prompt_version=version) + run_session(FakeDistrict(), ctl, NullWriter(), hours=hours, log=quiet) + return sup.requests + + +@pytest.mark.parametrize("version", PROMPT_VERSIONS) +def test_cache_control_is_on_the_last_block_of_the_system_prompt(version): + req = _requests(version, hours=1)[0] + blocks = req["system"] + assert isinstance(blocks, list) and blocks[-1]["type"] == "text" + assert blocks[-1]["cache_control"] == {"type": "ephemeral"} # default 5 minute TTL + assert all("cache_control" not in b for b in blocks[:-1]) + assert "tools" not in req # no tools, so the system prompt ends the stable prefix + assert "cache_control" not in json.dumps(req["messages"]) # nothing after the prefix is marked + + +@pytest.mark.parametrize("version", PROMPT_VERSIONS) +def test_the_hours_state_comes_after_the_prefix(version): + req = _requests(version, hours=1)[0] + keys = list(req) + assert keys.index("system") < keys.index("messages") + assert [m["role"] for m in req["messages"]] == ["user"] + obs = observation_from_request(req) # the hour's state is in the last user message + text = system_text(req) + assert "OBSERVATION_JSON" not in text and json.dumps(obs) not in text + assert not re.search(r"\d{4}-\d{2}-\d{2}|\d{2}:\d{2}:\d{2}", text) # no dates or clock times + assert not re.search(r"[0-9a-f]{8}-[0-9a-f]{4}-", text) # no run ids + + +@pytest.mark.parametrize("version", PROMPT_VERSIONS) +def test_prefix_is_byte_identical_across_consecutive_calls(version): + first, second = _requests(version, hours=2) + o1, o2 = observation_from_request(first), observation_from_request(second) + assert o1["time_step"] != o2["time_step"] and o1 != o2 # different state each hour + for key in ("model", "system", "output_config"): + assert json.dumps(first[key]) == json.dumps(second[key]) + assert first["messages"] != second["messages"] + + +def test_current_prompt_is_long_enough_to_cache(): + # Claude Haiku 5.5 caches a prefix of 512 tokens or more. Plain English runs about 4 characters + # a token; 6 is a conservative bound, so this many characters is at least 512 tokens. + text = system_text(_requests(PROMPT_VERSION, hours=1)[0]) + assert len(text) >= 6 * 512 + + +def test_prices_in_the_table(): + for p in MODELS.values(): + assert p["cache_write_per_mtok"] == pytest.approx(1.25 * p["input_per_mtok"]) + assert p["cache_read_per_mtok"] == pytest.approx(0.1 * p["input_per_mtok"]) + + +def test_cost_of_a_call_with_cache_reads_writes_and_plain_input(): + got = call_cost_usd("haiku", 1_000, 300, cache_read_tokens=5_000, cache_write_tokens=2_000) + want = (1_000 * 0.10 + 2_000 * 0.125 + 5_000 * 0.01 + 300 * 0.50) / 1_000_000 + assert got == pytest.approx(want) + assert call_cost_usd("sonnet", 1_000, 300, 5_000, 2_000) == pytest.approx( + (1_000 * 2.00 + 2_000 * 2.50 + 5_000 * 0.20 + 300 * 10.00) / 1_000_000) + assert call_cost_usd("haiku", 1_000, 300) == pytest.approx((1_000 * 0.10 + 300 * 0.50) / 1_000_000) + assert cache_hit_rate(1_000, 5_000, 2_000) == pytest.approx(5_000 / 8_000) + assert cache_hit_rate(0, 0, 0) is None + + +def test_worst_case_prices_the_cached_prefix_as_a_write(): + req = _requests(PROMPT_VERSION, hours=1)[0] + plain = {**req, "system": system_text(req)} + assert worst_case_cost_usd("haiku", req) > worst_case_cost_usd("haiku", plain) + + +def _response(text, inp, read, write, out=200): + return SimpleNamespace( + content=[SimpleNamespace(type="text", text=text)], + usage=SimpleNamespace(input_tokens=inp, output_tokens=out, cache_read_input_tokens=read, + cache_creation_input_tokens=write), + stop_reason="end_turn", + model="claude-haiku-5-5", + ) + + +def test_live_run_records_and_prices_cache_tokens(tmp_path, monkeypatch): + monkeypatch.setenv("HYPERBOLIC_ANTHROPIC_KEY", "sk-test") + sup = LiveSupervisor("haiku") + reply = json.dumps({"reason": "hold", **{b: {"cooling": None, "dhw": None, "battery": None} for b in BUILDINGS}}) + # The first hour writes the prefix, the next two read it. + seq = [_response(reply, 900, 0, 1_200), _response(reply, 950, 1_200, 0), _response(reply, 980, 1_200, 0)] + sent = [] + + def create(**kwargs): + sent.append(kwargs) + return seq[len(sent) - 1] + + monkeypatch.setattr(sup._client.messages, "create", create) + version, cfg = controller_settings(False) + ctl = AgentController(supervisor=sup, envelope=Envelope(cfg), budget=Budget(1.0), prompt_version=version) + res = run_session(FakeDistrict(), ctl, RunWriter(tmp_path / "a"), hours=3, log=quiet) + + assert sent[0]["system"][-1]["cache_control"] == {"type": "ephemeral"} + want = (call_cost_usd("haiku", 900, 200, 0, 1_200) + call_cost_usd("haiku", 950, 200, 1_200, 0) + + call_cost_usd("haiku", 980, 200, 1_200, 0)) + assert res["cost_usd"] == pytest.approx(want, abs=1e-6) + assert ctl.budget.spent_usd == pytest.approx(want) # the cost cap charges the priced cost + assert res["tokens"] == {"input": 2_830, "cache_read": 2_400, "cache_write": 1_200, "output": 600} + assert res["cache_hit_rate"] == pytest.approx(2_400 / (2_830 + 2_400 + 1_200), abs=1e-4) + + calls = load_jsonl(tmp_path / "a" / "calls.jsonl") + assert [(c["reply"]["input_tokens"], c["reply"]["cache_read_input_tokens"], + c["reply"]["cache_creation_input_tokens"]) for c in calls] == [(900, 0, 1_200), (950, 1_200, 0), + (980, 1_200, 0)] + assert json.loads((tmp_path / "a" / "manifest.json").read_text(encoding="utf-8"))["supervisor"]["prompt_cache"] + + rep = replay_run(tmp_path / "a") # replays with zero calls, cache counts and all + assert rep.deterministic and rep.api_calls == 0 + assert rep.cost_usd == pytest.approx(want, abs=1e-6) + assert rep.cache_hit_rate == pytest.approx(2_400 / 6_430) + + +def test_runs_recorded_before_caching_still_replay(tmp_path): + ctl = AgentController(supervisor=ScriptedSupervisor(), envelope=Envelope(controller_settings(True)[1]), + budget=Budget(1.0), prompt_version=controller_settings(True)[0], prompt_cache=False) + run_session(FakeDistrict(), ctl, RunWriter(tmp_path / "old"), hours=3, log=quiet) + m_path = tmp_path / "old" / "manifest.json" + m = json.loads(m_path.read_text(encoding="utf-8")) + del m["supervisor"]["prompt_cache"] # as a manifest written before caching existed + m_path.write_text(json.dumps(m), encoding="utf-8") + calls = load_jsonl(tmp_path / "old" / "calls.jsonl") + assert isinstance(calls[0]["request"]["system"], str) + assert replay_run(tmp_path / "old").deterministic diff --git a/tests/test_prompt_v7.py b/tests/test_prompt_v7.py index d9a862f..38e0cbc 100644 --- a/tests/test_prompt_v7.py +++ b/tests/test_prompt_v7.py @@ -28,6 +28,7 @@ observe, system_prompt_v6, system_prompt_v7, + system_text, trim_view, ) from hyperbolic.runner import ( @@ -329,7 +330,7 @@ def test_v7_dry_run_and_replay(tmp_path): sim_info={"kind": "citylearn", "control": True}) assert res["steps"] == 24 and res["fallback_steps"] == 0 calls = load_jsonl(run_dir / "calls.jsonl") - assert "cooling_range" in calls[5]["request"]["system"] + assert "cooling_range" in system_text(calls[5]["request"]) rep = replay_run(run_dir) assert rep.deterministic and rep.api_calls == 0 and rep.identical_steps == 24 From 58b22e8908551afbe01ad97ce141538aecb67dad Mon Sep 17 00:00:00 2001 From: AddisonTech Date: Sat, 24 Oct 2026 16:34:38 -0400 Subject: [PATCH 2/2] Cover prompt version 9 in the caching tests The prefix tests now run over an explicit list of every prompt version, version 9 included, and a test fails if a new version is not added to it. The length check that keeps the cached prefix above the 512 token minimum for Claude Haiku 5.5 runs on versions 8 and 9. --- tests/test_prompt_cache.py | 17 ++++++++++++----- 1 file changed, 12 insertions(+), 5 deletions(-) diff --git a/tests/test_prompt_cache.py b/tests/test_prompt_cache.py index e9a01ba..5ec08df 100644 --- a/tests/test_prompt_cache.py +++ b/tests/test_prompt_cache.py @@ -24,6 +24,8 @@ from hyperbolic.runner import AgentController, NullWriter, RunWriter, controller_settings, load_jsonl, replay_run, run_session quiet = lambda *_: None # noqa: E731 +# Every prompt version a run can use, written out so a new version has to be added here. +CACHED_VERSIONS = (1, 2, 3, 4, 5, 6, 7, 8, 9) class CapturingSupervisor: @@ -49,7 +51,7 @@ def _requests(version, hours=2): return sup.requests -@pytest.mark.parametrize("version", PROMPT_VERSIONS) +@pytest.mark.parametrize("version", CACHED_VERSIONS) def test_cache_control_is_on_the_last_block_of_the_system_prompt(version): req = _requests(version, hours=1)[0] blocks = req["system"] @@ -60,7 +62,7 @@ def test_cache_control_is_on_the_last_block_of_the_system_prompt(version): assert "cache_control" not in json.dumps(req["messages"]) # nothing after the prefix is marked -@pytest.mark.parametrize("version", PROMPT_VERSIONS) +@pytest.mark.parametrize("version", CACHED_VERSIONS) def test_the_hours_state_comes_after_the_prefix(version): req = _requests(version, hours=1)[0] keys = list(req) @@ -73,7 +75,7 @@ def test_the_hours_state_comes_after_the_prefix(version): assert not re.search(r"[0-9a-f]{8}-[0-9a-f]{4}-", text) # no run ids -@pytest.mark.parametrize("version", PROMPT_VERSIONS) +@pytest.mark.parametrize("version", CACHED_VERSIONS) def test_prefix_is_byte_identical_across_consecutive_calls(version): first, second = _requests(version, hours=2) o1, o2 = observation_from_request(first), observation_from_request(second) @@ -83,10 +85,15 @@ def test_prefix_is_byte_identical_across_consecutive_calls(version): assert first["messages"] != second["messages"] -def test_current_prompt_is_long_enough_to_cache(): +def test_every_prompt_version_is_covered(): + assert CACHED_VERSIONS == PROMPT_VERSIONS + + +@pytest.mark.parametrize("version", sorted({8, 9, PROMPT_VERSION})) +def test_prompt_is_long_enough_to_cache(version): # Claude Haiku 5.5 caches a prefix of 512 tokens or more. Plain English runs about 4 characters # a token; 6 is a conservative bound, so this many characters is at least 512 tokens. - text = system_text(_requests(PROMPT_VERSION, hours=1)[0]) + text = system_text(_requests(version, hours=1)[0]) assert len(text) >= 6 * 512