From 6ec3e3478abf058fe1460f91342bcdadf54a1ba8 Mon Sep 17 00:00:00 2001 From: Tejas Chopra Date: Wed, 5 Aug 2026 12:24:49 -0700 Subject: [PATCH] feat(cli,pricing): add CLI extension seam and prompt-cache TTL pricing (#2802) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Description Two small, independent additions. Both exist because an out-of-tree package needed them and neither had a home in the current API. 1. **`headroom.cli_extension`** — an entry-point group so a package can add a `headroom` subcommand. `headroom.proxy_extension` requires a running FastAPI app, so it cannot carry a read-only CLI tool, and `_register_commands()` was a hardcoded import list with no discovery. 2. **`headroom/pricing/cache_ttl.py`** — the prompt-cache TTL price structure (read `0.10x`, 5m write `1.25x`, 1h write `2.00x` of base input) plus the break-even share above which the 1h TTL is cheaper. `ModelPricing` carries `cached_input_per_1m` for reads but has no field for the *write* side. Closes # ## Type of Change - [ ] Bug fix (non-breaking change that fixes an issue) - [x] New feature (non-breaking change that adds functionality) - [ ] Breaking change (fix or feature that would cause existing functionality to change) - [ ] Documentation update - [ ] Performance improvement - [ ] Code refactoring (no functional changes) ## Changes Made - `headroom/cli/extensions.py` (new) — `register_all(main)` discovers the `headroom.cli_extension` group. Contract: `register(main: click.Group) -> None`. Invoked from `_register_commands()` **last**, so built-ins are already attached. - Deliberately **not** opt-in gated, unlike proxy extensions: installing the package is the opt-in, because adding a subcommand cannot silently change what an existing command does. What *would* be a silent change is shadowing a built-in, so that is detected and rolled back — a stale plugin can never quietly take over `headroom proxy`. Load failures and partial registrations roll back too, and one bad plugin never blocks another. - `headroom/pricing/cache_ttl.py` (new) — `CACHE_READ_MULTIPLIER`, `CACHE_WRITE_MULTIPLIERS`, `cache_write_multiplier()`, `cache_rates_per_1m()`, `ttl_breakeven_share()`. Ratios are derived from base input rather than transcribed into a per-model table that would triple its columns and drift. - `cache_write_multiplier()` raises on an unknown TTL rather than falling back to the cheaper 5m rate, which would understate cost. - `headroom/pricing/litellm_pricing.py` — three additive optional fields on `LiteLLMModelPricing` exposing LiteLLM's own `cache_read_input_token_cost`, `cache_creation_input_token_cost` and `cache_creation_input_token_cost_above_1hr` (present for 212 and 123 models respectively). Published rates should win over derived ones. All default to `None`, and `None` means "not published" — distinct from `0.0` meaning "free" — so every existing caller is unaffected. - `headroom/pricing/__init__.py` — re-exports. ### Why `ttl_breakeven_share()` exists The TTL trade has two terms and both must be counted: moving to 1h turns idle-gap rewrites into cheap reads **and** raises the price of every write that still happens. Modelling only the recovery overstates the saving. On a real 531-transcript corpus that error was **1.9x** — $1,021 claimed against $538 real. `test_write_premium_is_not_forgotten` pins those exact figures so the mistake cannot be reintroduced quietly. ## Testing - [x] Unit tests pass (`pytest`) - [x] Linting passes (`ruff check .`) - [x] Type checking passes (`mypy headroom`) — see note below - [x] New tests added for new functionality - [x] Manual testing performed ### Test Output ```text $ python -m pytest tests/test_cli_extension_seam.py tests/test_pricing_cache_ttl.py tests/test_pricing_from_litellm.py -q tests/test_cli_extension_seam.py ....... [ 25%] tests/test_pricing_cache_ttl.py .......... [ 62%] tests/test_pricing_from_litellm.py .......... [100%] ======================== 27 passed, 1 warning in 3.54s ========================= $ .venv/bin/ruff check headroom/cli/extensions.py headroom/cli/main.py headroom/pricing/ tests/test_cli_extension_seam.py tests/test_pricing_cache_ttl.py All checks passed! $ .venv/bin/mypy --python-version 3.12 headroom/cli/extensions.py headroom/pricing/cache_ttl.py headroom/pricing/litellm_pricing.py Success: no issues found in 3 source files ``` `tests/test_pricing_from_litellm.py` is the **pre-existing** pricing suite, included to show the `LiteLLMModelPricing` change is non-breaking. **mypy note.** With the repo's configured `python_version = "3.10"`, mypy fails on numpy's own stubs for any file that transitively reaches numpy: ```text $ .venv/bin/mypy headroom/pricing/cache_ttl.py .venv/lib/python3.12/site-packages/numpy/__init__.pyi:737: error: Type statement is only supported in Python 3.12 and greater [syntax] Found 1 error in 1 file (errors prevented further checking) ``` This is pre-existing and unrelated — untouched files reproduce it identically (`mypy headroom/cli/doctor.py`, `mypy headroom/pricing/registry.py`). Hence the `--python-version 3.12` run above, which matches the interpreter actually in use. Worth fixing separately; not addressed here. ## Real Behavior Proof - **Environment:** macOS 15 (darwin 25.4.0), Python 3.12.6, `headroom-ai` 0.34.0 working tree, branch off `upstream/main`. - **Exact command / steps:** 1. Built a separate out-of-tree package declaring `[project.entry-points."headroom.cli_extension"] fleet = "headroom_fleet.cli:register"`. 2. `pip install --no-deps headroom_fleet-0.1.0-py3-none-any.whl` 3. `headroom econ --help` - **Observed result:** the subcommand registers with no configuration and appears in `headroom --help`: ```text $ python -c "import importlib.metadata as m; print([e.name+' -> '+e.value for e in m.entry_points(group='headroom.cli_extension')])" ['fleet -> headroom_fleet.cli:register'] $ headroom --help | grep econ econ Report where local AI-coding token spend goes, and what... $ headroom econ --help Usage: headroom econ [OPTIONS] [COMMAND] [ARGS]... Commands: fix Write the recommended cache-TTL and compaction settings. unfix Restore every setting ``econ fix`` changed, exactly as it was. ``` `headroom --help` and every built-in still work with the plugin installed and after it is uninstalled. Verified per-model cache rates resolve from LiteLLM for `claude-opus-4-8`, `claude-opus-5`, `claude-sonnet-5` and `claude-haiku-4-5-20251001` — all exactly `1.250x` / `2.000x` / `0.100x` of base input, matching the derived fallback. - **Not tested:** Windows; a plugin that raises at *import* time rather than in `register()` (covered by unit test with a stubbed entry point, not a real package); the `--python-version 3.10` mypy path, which is blocked by the pre-existing numpy stub issue above. ## Review Readiness - [x] I have performed a self-review - [x] This PR is ready for human review ## Checklist - [x] My code follows the project's style guidelines ## Note for reviewers This PR contains **only** the two additions above. Unrelated `plugins/opencode/src/transport.ts` changes in my working tree are deliberately excluded and will follow separately. 🤖 Generated with [Claude Code](https://claude.com/claude-code) --- headroom/cli/extensions.py | 85 +++++++++++++++++ headroom/cli/main.py | 9 ++ headroom/pricing/__init__.py | 17 ++++ headroom/pricing/cache_ttl.py | 93 ++++++++++++++++++ headroom/pricing/litellm_pricing.py | 24 +++++ tests/test_cli_extension_seam.py | 140 ++++++++++++++++++++++++++++ tests/test_pricing_cache_ttl.py | 89 ++++++++++++++++++ 7 files changed, 457 insertions(+) create mode 100644 headroom/cli/extensions.py create mode 100644 headroom/pricing/cache_ttl.py create mode 100644 tests/test_cli_extension_seam.py create mode 100644 tests/test_pricing_cache_ttl.py diff --git a/headroom/cli/extensions.py b/headroom/cli/extensions.py new file mode 100644 index 000000000..2d66ec0c7 --- /dev/null +++ b/headroom/cli/extensions.py @@ -0,0 +1,85 @@ +"""Third-party CLI subcommand extension point. + +External packages add subcommands to the ``headroom`` CLI by declaring an entry +point in the ``headroom.cli_extension`` group in their ``pyproject.toml``: + + [project.entry-points."headroom.cli_extension"] + my_commands = "my_pkg.cli:register" + +Each ``register`` callable is invoked with the root ``click.Group`` and is free +to attach commands or groups to it:: + + def register(main: click.Group) -> None: + main.add_command(my_command) + +Unlike :mod:`headroom.proxy.extensions`, this seam is **not** opt-in gated. +Installing the package is the opt-in: adding a subcommand cannot silently +change what an existing command does, so requiring an extra env var before +``headroom `` even appears in ``--help`` would buy no safety. + +What *would* be a silent behavior change is shadowing a built-in command, so +that is refused: a registrant may only add names the core CLI does not already +define. Everything else is the extension's business — OSS makes no assumptions +about what the commands do. + +Stability contract: this module is load-bearing for any third-party CLI +extensions. Changes to the signature of ``register(main)`` or the entry-point +group name require a deprecation cycle. +""" + +from __future__ import annotations + +import importlib.metadata +import logging +from collections.abc import Callable +from typing import Any + +import click + +log = logging.getLogger(__name__) + +ENTRY_POINT_GROUP = "headroom.cli_extension" + +CliExtension = Callable[[Any], None] +"""Signature: ``register(main: click.Group) -> None``.""" + + +def register_all(main: click.Group) -> list[str]: + """Invoke every registered ``register(main)`` and return the names installed. + + Failures are isolated: a broken third-party package is logged and skipped so + the rest of the CLI still works. Registrants that try to shadow an existing + command are rolled back to the built-in and reported, so a stale plugin can + never quietly take over ``headroom proxy``. + """ + try: + entries = importlib.metadata.entry_points(group=ENTRY_POINT_GROUP) + except Exception as exc: # noqa: BLE001 — importlib.metadata can raise varied types + log.debug("cli extensions: entry-point enumeration failed: %s", exc) + return [] + + installed: list[str] = [] + for entry in entries: + before = dict(main.commands) + try: + register = entry.load() + register(main) + except Exception as exc: # noqa: BLE001 — one bad plugin must not brick the CLI + log.warning("cli extension %r failed to register and was skipped: %s", entry.name, exc) + main.commands = before + continue + + shadowed = [n for n, cmd in before.items() if main.commands.get(n) is not cmd] + if shadowed: + log.warning( + "cli extension %r tried to override built-in command(s) %s and was skipped", + entry.name, + ",".join(sorted(shadowed)), + ) + main.commands = before + continue + + installed.append(entry.name) + log.debug("cli extension registered: %s", entry.name) + + return installed diff --git a/headroom/cli/main.py b/headroom/cli/main.py index b55c96e1e..1b74e22df 100644 --- a/headroom/cli/main.py +++ b/headroom/cli/main.py @@ -88,6 +88,15 @@ def _register_commands() -> None: except ImportError: pass + # Third-party subcommands (headroom.cli_extension entry points). Runs last so + # built-ins are already attached and a plugin cannot shadow one. + try: + from .extensions import register_all + + register_all(main) + except Exception: # noqa: BLE001 — extension discovery must never break the CLI + pass + _register_commands() diff --git a/headroom/pricing/__init__.py b/headroom/pricing/__init__.py index 60ce24f30..b96b01522 100644 --- a/headroom/pricing/__init__.py +++ b/headroom/pricing/__init__.py @@ -13,6 +13,15 @@ from .anthropic_prices import ( from .anthropic_prices import ( LAST_UPDATED as ANTHROPIC_LAST_UPDATED, ) +from .cache_ttl import ( + CACHE_READ_MULTIPLIER, + CACHE_WRITE_MULTIPLIERS, + DEFAULT_CACHE_TTL, + CacheTTL, + cache_rates_per_1m, + cache_write_multiplier, + ttl_breakeven_share, +) from .deepseek_prices import ( DEEPSEEK_PRICES, get_deepseek_registry, @@ -37,6 +46,14 @@ from .openai_prices import ( from .registry import CostEstimate, ModelPricing, PricingRegistry __all__ = [ + # Prompt-cache TTL structure (read / 5m write / 1h write) + "CACHE_READ_MULTIPLIER", + "CACHE_WRITE_MULTIPLIERS", + "DEFAULT_CACHE_TTL", + "CacheTTL", + "cache_rates_per_1m", + "cache_write_multiplier", + "ttl_breakeven_share", # LiteLLM-based pricing (preferred) "LiteLLMModelPricing", "estimate_cost", diff --git a/headroom/pricing/cache_ttl.py b/headroom/pricing/cache_ttl.py new file mode 100644 index 000000000..02e997041 --- /dev/null +++ b/headroom/pricing/cache_ttl.py @@ -0,0 +1,93 @@ +"""Prompt-cache TTL pricing structure. + +Anthropic prices prompt-cache traffic as fixed multiples of a model's base input +rate rather than as independent per-model numbers: + +=============== ========== ================================================== +Traffic Multiplier Notes +=============== ========== ================================================== +cache read ``0.10x`` Any TTL — reads are the same price either way. +cache write 5m ``1.25x`` The default TTL. +cache write 1h ``2.00x`` Survives 12x longer for 1.6x the write price. +=============== ========== ================================================== + +Because the ratios are structural, they are derived from the base input price +here instead of being transcribed into every row of +:mod:`headroom.pricing.anthropic_prices` — a per-model table would be 3x the +columns, would drift, and would let one tier be mis-typed without anything +noticing. :class:`~headroom.pricing.registry.ModelPricing` already carries +``cached_input_per_1m`` for reads; this module adds the *write* side, which the +registry has no field for. + +Scope: these ratios are Anthropic's. Other providers differ structurally, not +just numerically — OpenAI, for instance, discounts cached input but does not +bill for cache writes at all, so there is no 5m/1h trade to make there. + +The TTL trade has two terms and both must be counted. Switching a workload to +the 1h TTL turns idle-gap rewrites into cheap reads, but it also raises the +price of *every* write that still happens. Modelling only the recovery +overstates the saving — measured at 1.9x on a real 531-transcript corpus — so +:func:`ttl_breakeven_share` exists to make the threshold explicit. +""" + +from __future__ import annotations + +from typing import Literal + +CacheTTL = Literal["5m", "1h"] + +#: Cache reads cost this multiple of the base input rate, at either TTL. +CACHE_READ_MULTIPLIER = 0.10 + +#: Cache writes cost this multiple of the base input rate, by TTL. +CACHE_WRITE_MULTIPLIERS: dict[str, float] = {"5m": 1.25, "1h": 2.00} + +#: Anthropic's default when a request does not ask for the extended TTL. +DEFAULT_CACHE_TTL: CacheTTL = "5m" + + +def cache_write_multiplier(ttl: str) -> float: + """Return the base-input multiplier for a cache write at ``ttl``. + + Raises ``ValueError`` on an unknown TTL rather than silently falling back to + the cheaper 5m rate, which would understate cost. + """ + try: + return CACHE_WRITE_MULTIPLIERS[ttl] + except KeyError: + known = ", ".join(sorted(CACHE_WRITE_MULTIPLIERS)) + raise ValueError(f"unknown cache TTL {ttl!r} (known: {known})") from None + + +def cache_rates_per_1m(input_per_1m: float) -> dict[str, float]: + """Derive per-1M-token cache rates from a model's base input rate. + + Returns ``{"read": …, "write_5m": …, "write_1h": …}`` in the same units as + ``input_per_1m`` (USD per 1M tokens). + """ + return { + "read": input_per_1m * CACHE_READ_MULTIPLIER, + "write_5m": input_per_1m * CACHE_WRITE_MULTIPLIERS["5m"], + "write_1h": input_per_1m * CACHE_WRITE_MULTIPLIERS["1h"], + } + + +def ttl_breakeven_share() -> float: + """Return the share of cache writes above which the 1h TTL is cheaper. + + Let ``W`` be total cache-write tokens and ``G`` the subset written again + after an idle gap longer than the 5m TTL but shorter than 1h. Under the 5m + default the workload pays ``W * write_5m``. Under the 1h TTL the ``G`` + tokens become reads while the remaining ``W - G`` writes pay the higher + rate:: + + (W - G) * write_1h + G * read < W * write_5m + => G / W > (write_1h - write_5m) / (write_1h - read) + + The threshold is scale- and model-independent because every term is a + multiple of the same base input rate: ``(2.00 - 1.25) / (2.00 - 0.10)``, + i.e. **39.5%**. + """ + w5 = CACHE_WRITE_MULTIPLIERS["5m"] + w1h = CACHE_WRITE_MULTIPLIERS["1h"] + return (w1h - w5) / (w1h - CACHE_READ_MULTIPLIER) diff --git a/headroom/pricing/litellm_pricing.py b/headroom/pricing/litellm_pricing.py index 094873d39..9a4252689 100644 --- a/headroom/pricing/litellm_pricing.py +++ b/headroom/pricing/litellm_pricing.py @@ -160,6 +160,13 @@ class LiteLLMModelPricing: max_output_tokens: int | None = None supports_vision: bool = False supports_function_calling: bool = False + # Prompt-cache traffic, where LiteLLM knows it. ``None`` means "not published + # for this model", which is distinct from 0.0 ("free"): the 1h write rate in + # particular is absent for most models, and a caller that needs it must either + # derive it (see :mod:`headroom.pricing.cache_ttl`) or report that it could not. + cache_read_per_1m: float | None = None + cache_write_5m_per_1m: float | None = None + cache_write_1h_per_1m: float | None = None def get_litellm_model_cost() -> dict[str, Any]: @@ -209,9 +216,26 @@ def get_model_pricing(model: str) -> LiteLLMModelPricing | None: max_output_tokens=info.get("max_output_tokens"), supports_vision=info.get("supports_vision", False), supports_function_calling=info.get("supports_function_calling", False), + cache_read_per_1m=_per_1m(info.get("cache_read_input_token_cost")), + cache_write_5m_per_1m=_per_1m(info.get("cache_creation_input_token_cost")), + cache_write_1h_per_1m=_per_1m(info.get("cache_creation_input_token_cost_above_1hr")), ) +def _per_1m(cost_per_token: object) -> float | None: + """Scale a LiteLLM per-token cost to per-1M, preserving "not published". + + ``None`` in, ``None`` out — the absence of a rate is information and must not + collapse into 0.0, which would read as free. + """ + if not isinstance(cost_per_token, (int, float, str)): + return None + try: + return float(cost_per_token) * 1_000_000 + except (TypeError, ValueError): + return None + + def pricing_per_1m(model: str) -> tuple[float, float] | None: """``(input, output)`` USD per 1M tokens from LiteLLM, or ``None``. diff --git a/tests/test_cli_extension_seam.py b/tests/test_cli_extension_seam.py new file mode 100644 index 000000000..64ba1b565 --- /dev/null +++ b/tests/test_cli_extension_seam.py @@ -0,0 +1,140 @@ +"""Tests for the ``headroom.cli_extension`` third-party subcommand seam.""" + +from __future__ import annotations + +import click +import pytest + +from headroom.cli import extensions + + +class _FakeEntry: + """Stand-in for an ``importlib.metadata.EntryPoint``.""" + + def __init__(self, name: str, loaded: object, load_raises: bool = False) -> None: + self.name = name + self._loaded = loaded + self._load_raises = load_raises + + def load(self) -> object: + if self._load_raises: + raise ImportError("boom") + return self._loaded + + +@pytest.fixture +def group() -> click.Group: + @click.group() + def root() -> None: + pass + + @root.command(name="proxy") + def proxy() -> None: + pass + + return root + + +def _patch_entries(monkeypatch: pytest.MonkeyPatch, entries: list[_FakeEntry]) -> None: + monkeypatch.setattr( + extensions.importlib.metadata, + "entry_points", + lambda group: entries, + ) + + +def test_registers_new_command(monkeypatch: pytest.MonkeyPatch, group: click.Group) -> None: + @click.command(name="econ") + def econ() -> None: + pass + + _patch_entries(monkeypatch, [_FakeEntry("fleet", lambda main: main.add_command(econ))]) + + assert extensions.register_all(group) == ["fleet"] + assert "econ" in group.commands + assert "proxy" in group.commands + + +def test_raising_registrant_is_skipped(monkeypatch: pytest.MonkeyPatch, group: click.Group) -> None: + def bad(main: click.Group) -> None: + raise RuntimeError("unlicensed") + + _patch_entries(monkeypatch, [_FakeEntry("broken", bad)]) + + assert extensions.register_all(group) == [] + assert set(group.commands) == {"proxy"} + + +def test_partial_registration_is_rolled_back( + monkeypatch: pytest.MonkeyPatch, group: click.Group +) -> None: + """A registrant that adds a command then raises leaves nothing behind.""" + + @click.command(name="half") + def half() -> None: + pass + + def bad(main: click.Group) -> None: + main.add_command(half) + raise RuntimeError("failed after partial work") + + _patch_entries(monkeypatch, [_FakeEntry("broken", bad)]) + + assert extensions.register_all(group) == [] + assert "half" not in group.commands + + +def test_load_failure_is_skipped(monkeypatch: pytest.MonkeyPatch, group: click.Group) -> None: + _patch_entries(monkeypatch, [_FakeEntry("stale", None, load_raises=True)]) + + assert extensions.register_all(group) == [] + assert set(group.commands) == {"proxy"} + + +def test_cannot_shadow_builtin_command(monkeypatch: pytest.MonkeyPatch, group: click.Group) -> None: + """Overriding a built-in IS a silent behavior change, so it is refused.""" + builtin = group.commands["proxy"] + + @click.command(name="proxy") + def evil_proxy() -> None: + pass + + _patch_entries(monkeypatch, [_FakeEntry("evil", lambda main: main.add_command(evil_proxy))]) + + assert extensions.register_all(group) == [] + assert group.commands["proxy"] is builtin + + +def test_shadowing_registrant_does_not_block_others( + monkeypatch: pytest.MonkeyPatch, group: click.Group +) -> None: + @click.command(name="proxy") + def evil_proxy() -> None: + pass + + @click.command(name="econ") + def econ() -> None: + pass + + _patch_entries( + monkeypatch, + [ + _FakeEntry("evil", lambda main: main.add_command(evil_proxy)), + _FakeEntry("fleet", lambda main: main.add_command(econ)), + ], + ) + + assert extensions.register_all(group) == ["fleet"] + assert "econ" in group.commands + + +def test_enumeration_failure_is_survivable( + monkeypatch: pytest.MonkeyPatch, group: click.Group +) -> None: + def boom(group: str) -> list[_FakeEntry]: + raise RuntimeError("no metadata") + + monkeypatch.setattr(extensions.importlib.metadata, "entry_points", boom) + + assert extensions.register_all(group) == [] + assert set(group.commands) == {"proxy"} diff --git a/tests/test_pricing_cache_ttl.py b/tests/test_pricing_cache_ttl.py new file mode 100644 index 000000000..8ea964d34 --- /dev/null +++ b/tests/test_pricing_cache_ttl.py @@ -0,0 +1,89 @@ +"""Tests for prompt-cache TTL pricing structure.""" + +from __future__ import annotations + +import pytest + +from headroom.pricing import cache_ttl + + +def test_multipliers_match_anthropic_structure() -> None: + assert cache_ttl.CACHE_READ_MULTIPLIER == 0.10 + assert cache_ttl.CACHE_WRITE_MULTIPLIERS == {"5m": 1.25, "1h": 2.00} + assert cache_ttl.DEFAULT_CACHE_TTL == "5m" + + +def test_cache_write_multiplier() -> None: + assert cache_ttl.cache_write_multiplier("5m") == 1.25 + assert cache_ttl.cache_write_multiplier("1h") == 2.00 + + +def test_unknown_ttl_raises_rather_than_defaulting_cheap() -> None: + """Silently returning the 5m rate would understate cost.""" + with pytest.raises(ValueError, match="unknown cache TTL"): + cache_ttl.cache_write_multiplier("30m") + + +def test_rates_derive_from_base_input() -> None: + rates = cache_ttl.cache_rates_per_1m(5.00) # opus-class base input + assert rates == {"read": 0.50, "write_5m": 6.25, "write_1h": 10.00} + + +def test_breakeven_share_is_39_5_percent() -> None: + assert cache_ttl.ttl_breakeven_share() == pytest.approx(0.3947, abs=1e-4) + + +def test_breakeven_is_model_independent() -> None: + """Every term scales with base input, so the threshold is a pure ratio.""" + for base in (1.00, 3.00, 5.00, 15.00): + r = cache_ttl.cache_rates_per_1m(base) + share = (r["write_1h"] - r["write_5m"]) / (r["write_1h"] - r["read"]) + assert share == pytest.approx(cache_ttl.ttl_breakeven_share()) + + +class TestBreakevenDecision: + """The threshold must actually predict which TTL is cheaper.""" + + @staticmethod + def _cost(total_writes: int, idle_gap_writes: int, base: float) -> tuple[float, float]: + r = cache_ttl.cache_rates_per_1m(base) + at_5m = total_writes * r["write_5m"] + at_1h = (total_writes - idle_gap_writes) * r["write_1h"] + idle_gap_writes * r["read"] + return at_5m / 1e6, at_1h / 1e6 + + def test_above_threshold_1h_wins(self) -> None: + at_5m, at_1h = self._cost(1_000_000, 500_000, 5.00) # 50% > 39.5% + assert at_1h < at_5m + + def test_below_threshold_5m_wins(self) -> None: + at_5m, at_1h = self._cost(1_000_000, 300_000, 5.00) # 30% < 39.5% + assert at_5m < at_1h + + def test_at_threshold_costs_are_equal(self) -> None: + share = cache_ttl.ttl_breakeven_share() + at_5m, at_1h = self._cost(1_000_000, int(1_000_000 * share), 5.00) + assert at_1h == pytest.approx(at_5m, rel=1e-5) + + +def test_write_premium_is_not_forgotten() -> None: + """Regression guard for the 1.9x overstatement class of error. + + Figures are the real measured corpus: 306,631,892 cache-write tokens of + which 177,636,344 followed a 5m-1h idle gap, at opus-class $5/1M input. + Counting only the recovered rewrites reports ~$1,021; the honest net after + the write premium on the remaining 128,995,548 writes is ~$538. + """ + total_writes = 306_631_892 + idle_gap = 177_636_344 + r = cache_ttl.cache_rates_per_1m(5.00) + + naive = idle_gap * (r["write_5m"] - r["read"]) / 1e6 + at_5m = total_writes * r["write_5m"] / 1e6 + at_1h = ((total_writes - idle_gap) * r["write_1h"] + idle_gap * r["read"]) / 1e6 + net = at_5m - at_1h + + assert naive == pytest.approx(1021.41, abs=0.5) + assert net == pytest.approx(537.68, abs=0.5) + assert naive / net == pytest.approx(1.9, abs=0.05) + # And this corpus is past the threshold, so the switch is correct here. + assert idle_gap / total_writes > cache_ttl.ttl_breakeven_share()