akadmin/headroom
1
0
Fork 0
mirror of https://github.com/headroomlabs-ai/headroom.git synced 2026-08-27 14:17:10 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions

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

Browse source 
## 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)
This commit is contained in:
Tejas Chopra 2026-08-05 12:24:49 -07:00 • committed by GitHub
parent 64e203931b
commit 6ec3e3478a
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
7 changed files with 457 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

85
headroom/cli/extensions.py Normal file
View file

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

9
headroom/cli/main.py
View file

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

17
headroom/pricing/__init__.py
View file

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

93
headroom/pricing/cache_ttl.py Normal file
View file

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

24
headroom/pricing/litellm_pricing.py
View file

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

140
tests/test_cli_extension_seam.py Normal file
View file

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

89
tests/test_pricing_cache_ttl.py Normal file
View file

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