feat(cli,pricing): add CLI extension seam and prompt-cache TTL pricing (#2802)

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

@@ -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 <name>`` 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

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()
 

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",

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)

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``.
 

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"}

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()