dashboards/tests/store_perf.py

r"""The production-shaped store phase 14 measures, and the measurements of its hot paths.

`TODO.md:7` is a complaint about speed, and a speed claim needs a "before" that is measured
rather than remembered. This module is that "before": it seeds a SQLite file with the **shape
and retention semantics of the deployed store** and times every hot read and write path of
`store.py`/`api.py` against it, so tasks 02-04 change a number they can point at.

Why a synthetic seed and not the dev DB
---------------------------------------
`AGENTS.md`'s testing policy keeps `uv run pytest` hermetic: no test opens
`data/dashboards.sqlite3`, because a test that reads the operator's live history changes
behaviour when the operator's live history changes (and a fixture assertion that passes
today and fails next month is not a test). The honest substitute is a file of the same
*shape*: the same `SampleStore` defaults for retention (`raw_keep_hours=48`,
`minute_keep_days=30`, `hour_keep_days=400` — i.e. what a 90 d chart actually reads, 1m/1h,
never raw beyond 48 h), the fleet's real metric names, the real 10 s cadence, and the same
`meta`/`event` rows the app writes.

The seeded numbers are recorded in `data/logs/NOTES-store-perf.md` together with the
dev DB's real inventory, so a reader can see where the reproduction is faithful and where it
is a deliberate subset (per-core `cpu.usage_pct@coreN` series and the `llm`/`ups`/`disks`
families are excluded — see `HOST_SERIES`).

Where the file lives
--------------------
`cache_path()` (default `data/perf/`, overridable with `DASH_PERF_STORE_DIR`) rather than a
pytest tmp dir, for two reasons that are both about the *measurement*, not convenience:

* the same physical file has to be timed before task 02 and after task 04, or the two numbers
  describe two different databases; and
* the repo filesystem is a real block device, while pytest's `tmp_path` here is `tmpfs` —
  RAM. Timing a store on RAM is a flattering number nobody deploys.

Write-side measurements (`measure_maintain`, the two-connection races) run against a
`scratch_copy()` of that file rather than the file itself: a catch-up rollup really does roll ten
more minutes of buckets and the pre-fix control rewrites 330 000 rows, and a baseline that moves
under the next task is not a baseline. The copy is in the same directory, so it is the same device.

The file is a cache, keyed by `SEED_VERSION` and by its own age (`MAX_CACHE_AGE_HOURS`): the
deep 1m/1h history is anchored to the time it was built, so a cache older than half a day no
longer covers "the last 90 days" and gets rebuilt. Builds are serialised with `flock` and
publish atomically, so two sessions never read a half-written store.

Nothing here is a production module: `src/dashboards` must stay free of benchmark furniture,
and coverage counts `src/dashboards` only.
"""

from __future__ import annotations

import fcntl
import json
import math
import os
import platform
import shutil
import sqlite3
import statistics
import sys
import threading
import time
from collections.abc import Callable, Iterator
from contextlib import contextmanager
from dataclasses import asdict, dataclass, field, replace
from pathlib import Path
from typing import Any

from dashboards.store import (
    HOUR,
    JOURNAL_SIZE_LIMIT_BYTES,
    LAST_MAINTAIN_KEY,
    MAINTAIN_SLACK_SECONDS,
    MINUTE,
    RES_COUNT_BRANCH,
    RES_HOUR,
    RES_MIN,
    RES_RAW,
    ROLLUP_INSERT,
    Event,
    Sample,
    SampleStore,
)

ROOT = Path(__file__).resolve().parents[1]

#: Bump when the shape below changes: the cached file is named after it, so an old cache can
#: never be mistaken for the new shape.
SEED_VERSION = "phase14-v1"

#: The seed's clock, aligned to a minute so every rollup bucket boundary is exact. Stored in
#: `meta` (`perf.seed_now`) and re-read by the verifier.
MAX_CACHE_AGE_HOURS = 12.0

#: The store's own retention defaults, restated here so the fixture cannot silently seed a
#: shape the app does not keep (`test_the_seed_uses_the_stores_own_retention` asserts it).
RAW_KEEP_HOURS, MINUTE_KEEP_DAYS, HOUR_KEEP_DAYS = 48.0, 30.0, 400.0

#: The `system` poll cadence (`HOST_INTERVAL_SECONDS` default) — the raw window's spacing.
CADENCE_SECONDS = 10.0

#: Event rows: the dev DB held ~2 800 after a few days of transitions and logins; 2 000 is
#: the same order and keeps `event` big enough for `stats()` to notice it.
EVENT_ROWS = 2000

#: Samples written by the measurements themselves (`race_*`) use this source so they can never
#: be confused with the seeded fleet, and one `delete_series` cleans them up.
RACE_SOURCE, RACE_HOST = "perf", "race-target"

#: The pass line for "did this insert wait": one poll of 30 samples costs ~1 ms, so anything
#: past this is a wait behind someone else's transaction, not the cost of the insert.
WAIT_THRESHOLD_MS = 100.0

#: `busy_timeout` in `store.py`. Production value; the race reports waits against it.
BUSY_TIMEOUT_MS = 5000

#: Wall-clock budget per race pass. One `stats()` on this store costs seconds, so "100 inserts"
#: is minutes of deliberate waiting; the report always says how many were attempted.
MAX_RACE_SECONDS = 30.0

#: Spacing between the raced inserts. The app does not insert 100 polls back to back: the hub
#: has ~9 sources writing every 2-300 s. Un-paced, the inserts finish before the contending
#: transaction even opens, and a race that never happened reports "no lock errors".
RACE_PACE_SECONDS = 0.1

#: Idle time between the maintainer's passes in the cadence-shaped race. The app maintains once
#: every 60 s, so its write transaction holds the WAL lock for a couple of seconds per minute at
#: the very most; the race compresses that minute to this. The saturated variant (`gap=0`, which is
#: what task 01 ran and what its numbers stay comparable to) holds the lock ~100% of the wall
#: clock, because the loop restarts the transaction the instant it commits — no window fixes that,
#: so it answers "can a writer starve forever", not "what does a poller wait during a rollup".
RACE_MAINTAIN_GAP_SECONDS = 1.5

META_SEED_NOW = "perf.seed_now"
META_SEED_VERSION = "perf.seed_version"


# --------------------------------------------------------------------------- #
# The shape: real hosts, real metric names, real cadence
# --------------------------------------------------------------------------- #
#: The `host`-family series one agent publishes, minus the per-core fan-out.
#:
#: Every name here is a name the deployed store actually holds (`SELECT DISTINCT metric FROM
#: sample WHERE source='host'` on the dev DB), because `series_names()`/`history()` cost
#: depends on the number of distinct names and on row width. Deliberately excluded, with the
#: consequence recorded in the NOTES rather than hidden:
#:
#: * `cpu.usage_pct@coreN` — 48 more series on driveripper. `system` stopped plotting them in
#:   phase 03 (a locked decision), so no hot path of this phase reads them; including them
#:   would double the raw window for the family the phase optimises.
#: * the `llm`/`ups`/`disks` families — a *mature* store holds them too (17 + 24 + 4 series at
#:   2 s / 5 s / 300 s), and the extrapolation is in the NOTES. They are out of the seed
#:   because the prescribed 90-series x 48 h x 30 d x 400 d shape already lands in the dev
#:   DB's order of magnitude, and a bigger fixture buys nothing this phase must decide.
BASE_SERIES: tuple[str, ...] = (
    "cpu.context_switches_per_s",
    "cpu.freq_mhz",
    "cpu.iowait_pct",
    "cpu.load1",
    "cpu.load15",
    "cpu.load5",
    "cpu.system_pct",
    "cpu.usage_pct",
    "cpu.user_pct",
    "disk.read_bytes_per_s",
    "disk.reads_per_s",
    "disk.used_bytes@/",
    "disk.used_pct",
    "disk.used_pct@/",
    "disk.util_pct",
    "disk.write_bytes_per_s",
    "disk.writes_per_s",
    "host.processes",
    "host.uptime_seconds",
    "mem.available_bytes",
    "mem.cached_bytes",
    "mem.swap_used_bytes",
    "mem.total_bytes",
    "mem.used_bytes",
    "mem.used_pct",
    "net.rx_bytes_per_s",
    "net.rx_bytes_per_s@enp3s0",
    "net.tx_bytes_per_s",
)

#: The four GPU series, on the two hosts that have a card (`gateway` has none — the shape the
#: `/system` contract calls "no GPU at all", so a plan that assumes every host has a stream
#: fails here rather than in a browser).
GPU_SERIES: tuple[str, ...] = (
    "gpu.busy_pct@0",
    "gpu.mem_used_pct@0",
    "gpu.power_watts@0",
    "gpu.temp_c@0",
)

#: host -> its series. Names are the fleet's; the values are synthetic (`_value`).
HOST_SERIES: dict[str, tuple[str, ...]] = {
    "deskwork": (*BASE_SERIES, *GPU_SERIES),
    "driveripper": (*BASE_SERIES, *GPU_SERIES),
    "gateway": BASE_SERIES,
}

#: The nine metrics the task-01 `/api/system/history` request asks for, per host: the three
#: the utilisation chart plots, the two network rates, the worst mount, the CPU temp, and the
#: GPU stream. 3 hosts x 9 = the 27 `_history` calls the endpoint makes for one page load.
HISTORY_METRICS: tuple[str, ...] = (
    "cpu.usage_pct",
    "mem.used_pct",
    "disk.used_pct",
    "net.rx_bytes_per_s",
    "net.tx_bytes_per_s",
    "disk.used_pct@/",
    "cpu.temp_c",
    "gpu.busy_pct@0",
    "gpu.temp_c@0",
)

#: The metrics `/system` actually asks a *page* for (UTIL_SERIES + NET_SERIES + one GPU), i.e.
#: the request shape `page-system.js` really makes, one per host per refresh.
PAGE_HISTORY_METRICS: tuple[str, ...] = (
    "cpu.usage_pct",
    "mem.used_pct",
    "disk.used_pct",
    "net.rx_bytes_per_s",
    "net.tx_bytes_per_s",
    "gpu.busy_pct@0",
)


def seed_value(metric: str, index: int) -> float:
    """Deterministic pseudo-signal for one sample: finite, plausible, never random.

    Determinism is the point (`test_the_seed_is_deterministic_and_the_rollups_are_idempotent`
    recomputes a rollup from this function), so there is no RNG: a phase of the row index plus
    a per-metric offset. Values are plausible rather than realistic — a percentage between 0
    and 100, a byte counter that grows — because a chart of them is never read; only the row
    count, the row width, and the arithmetic of the rollups matter.

    Public because `tests/test_store.py` recomputes stored rows from it: that is how the fixture
    proves a rebuild is byte-identical to the store it measured.
    """
    offset = (len(metric) * 31 + sum(ord(ch) for ch in metric[:6])) % 997
    wave = math.sin((index + offset) / 360.0)
    if metric.startswith(("mem.", "disk.used_bytes", "net.")):
        # byte counters: hundreds of MB to tens of GB, and monotonic for the ones that are
        # odometers
        base = 1e8 + (offset % 40) * 5e8
        return base + wave * 5e7 + index * 128.0
    if metric.endswith("_bytes") or metric.endswith("_bytes_per_s"):
        return 1e6 + wave * 5e5 + index * 4.0
    if "temp_c" in metric:
        return 45.0 + wave * 12.0
    if metric.startswith("gpu.power"):
        return 90.0 + wave * 40.0
    if metric.startswith(("cpu.load",)):
        return 2.0 + wave * 2.0
    if metric.endswith("_pct") or metric.endswith("_percent"):
        return 40.0 + wave * 30.0
    if metric.endswith("_per_s"):
        return 5_000.0 + wave * 2_000.0
    if metric == "host.uptime_seconds":
        return index * CADENCE_SECONDS
    return 10.0 + wave * 5.0


@dataclass(frozen=True)
class SeedPlan:
    """What a build *should* contain, computed before (and independently of) the insert.

    `*_steps` are rows **per series** (what the insert loops count); `*_rows` are the totals
    the verifier compares against `COUNT(*)`. Confusing the two once already made this fixture
    try to seed 75 million rows, which is why both are spelled out instead of derived at the
    call site.
    """

    now: float
    hosts: tuple[str, ...]
    series_total: int
    raw_steps: int
    minute_steps: int
    hour_steps: int
    events: int

    @property
    def raw_rows(self) -> int:
        return self.series_total * self.raw_steps

    @property
    def minute_rows_seeded(self) -> int:
        return self.series_total * self.minute_steps

    @property
    def hour_rows_seeded(self) -> int:
        return self.series_total * self.hour_steps

    @property
    def sample_rows_seeded(self) -> int:
        return self.raw_rows + self.minute_rows_seeded + self.hour_rows_seeded

    def as_dict(self) -> dict[str, Any]:
        """The plan plus its derived totals — `dataclasses.asdict` alone would drop them."""
        return {**asdict(self), "raw_rows": self.raw_rows,
                "minute_rows_seeded": self.minute_rows_seeded,
                "hour_rows_seeded": self.hour_rows_seeded,
                "sample_rows_seeded": self.sample_rows_seeded}


def plan(now: float) -> SeedPlan:
    """The row counts a build of this shape produces, from the shape alone.

    The deep 1m history starts where the raw window ends (that is what a store with 48 h of
    raw retention looks like: anything older than 48 h exists only as a rollup) and the deep
    1h history starts where the 1m window ends, for the same reason. The rows inside those
    windows are produced by the real `maintain()`, so the rollup path is exercised rather
    than faked.

    Every count is in *its own* unit: a step is one row of that resolution, so the deep window
    is `retention - the window the finer resolution already covers`, both converted to that
    resolution's period.
    """
    raw_steps = int(RAW_KEEP_HOURS * HOUR // CADENCE_SECONDS)
    minute_steps = int((MINUTE_KEEP_DAYS * 24 * HOUR - RAW_KEEP_HOURS * HOUR) // MINUTE)
    hour_steps = int((HOUR_KEEP_DAYS * 24 * HOUR - MINUTE_KEEP_DAYS * 24 * HOUR) // HOUR)
    series = sum(len(metrics) for metrics in HOST_SERIES.values())
    return SeedPlan(
        now=now,
        hosts=tuple(sorted(HOST_SERIES)),
        series_total=series,
        raw_steps=raw_steps,
        minute_steps=minute_steps,
        hour_steps=hour_steps,
        events=EVENT_ROWS,
    )


# --------------------------------------------------------------------------- #
# building
# --------------------------------------------------------------------------- #
def open_store(path: str | Path, *, extra_retention_hours: float = 0.0) -> SampleStore:
    """A `SampleStore` on `path` with the retention the seed promises.

    `extra_retention_hours` is for handles that must **not** prune: the seeded store is anchored
    to the time it was built, so a `maintain()` at the wall clock while the cache is six hours
    old prunes six hours of raw rows, and the next session would rebuild the whole thing over a
    measurement artefact. The cache's own age is the only thing past the window, and the read
    paths this phase optimises (`history`, `pick_res`, `series_names`, `stats`) do not read
    retention at all — so widening it for a measurement handle changes no number and keeps the
    baseline byte-stable. The store under test is still opened with the real defaults.
    """
    margin = extra_retention_hours
    return SampleStore(path, raw_keep_hours=RAW_KEEP_HOURS + margin,
                       minute_keep_days=MINUTE_KEEP_DAYS + margin / 24.0,
                       hour_keep_days=HOUR_KEEP_DAYS + margin / 24.0)


def build(path: str | Path, *, now: float | None = None,
          log: Callable[[str], None] | None = None) -> dict[str, Any]:
    """Seed `path` production-shaped and return a build report.

    Insert order is `(source, host, metric, res, ts)` ascending — the primary key, and
    `"1h" < "1m" < "raw"` sorts the resolutions the same way) — so the writes append instead
    of splitting index pages. That is not a trick to hide cost: it is how the app itself writes
    (a series' samples always arrive in time order), and the rollup inserts of `maintain()` are
    the ones that scatter, which is what task 03 times.
    """
    say = log or (lambda _line: None)
    now = float(now if now is not None else math.floor(time.time() / MINUTE) * MINUTE)
    path = Path(path)
    path.parent.mkdir(parents=True, exist_ok=True)
    expected = plan(now)
    store = open_store(path)
    started = time.perf_counter()
    try:
        buffer: list[tuple[str, str, str, str, float, float]] = []

        def flush() -> None:
            if not buffer:
                return
            with store._lock, store._db:
                store._db.executemany("INSERT OR REPLACE INTO sample VALUES (?,?,?,?,?,?)", buffer)
            buffer.clear()

        def emit(source: str, host: str, metric: str, res: str, ts: float, value: float) -> None:
            buffer.append((source, host, metric, res, ts, value))
            if len(buffer) >= 100_000:
                flush()

        # One pass per series, deepest resolution first: that is primary-key order, and it is
        # the only order in which a 6-million-row load does not rewrite the index behind itself.
        t0 = time.perf_counter()
        hour_start = now - HOUR_KEEP_DAYS * 24 * HOUR
        minute_start = now - MINUTE_KEEP_DAYS * 24 * HOUR
        raw_start = now - RAW_KEEP_HOURS * HOUR
        window = ((RES_HOUR, hour_start, HOUR, expected.hour_steps),
                  (RES_MIN, minute_start, MINUTE, expected.minute_steps),
                  (RES_RAW, raw_start, CADENCE_SECONDS, expected.raw_steps))
        for host in sorted(HOST_SERIES):
            for metric in sorted(HOST_SERIES[host]):
                for res, start, step, steps in window:
                    for index in range(steps):
                        emit("host", host, metric, res, start + index * step, seed_value(metric, index))
        flush()
        inserted_seconds = time.perf_counter() - t0
        # Row counts by resolution are re-read after the fact (`counts` below); the per-
        # resolution totals the NOTES quotes come from `expected`, and the verifier compares
        # the two, so this one loop cannot drift from the shape it claims to seed.

        # events + the meta rows the app writes (`last_maintain` comes from maintain below).
        t0 = time.perf_counter()
        for index in range(EVENT_ROWS):
            store.log_event(Event(ts=now - (index + 1) * 600.0, source="system",
                                  host=expected.hosts[index % len(expected.hosts)],
                                  kind="recording" if index % 7 else "reachable",
                                  severity="info" if index % 7 else "warn",
                                  message=f"seeded event {index}"))
        store.set_meta(META_SEED_VERSION, SEED_VERSION)
        store.set_meta(META_SEED_NOW, str(now))
        event_seconds = time.perf_counter() - t0

        # the rollups inside the retention windows, from the real maintenance path.
        t0 = time.perf_counter()
        first = store.maintain(now=now)
        maintain_seconds = time.perf_counter() - t0

        say(f"seeded {path.name}: {expected.sample_rows_seeded:,} sample rows in "
            f"{inserted_seconds:.1f}s · events {event_seconds:.1f}s · "
            f"maintain {maintain_seconds:.1f}s {first}")
        report = {
            "path": str(path),
            "now": now,
            "expected": expected.as_dict(),
            "maintain": first,
            "seconds": {
                "sample_rows": round(inserted_seconds, 3),
                "events": round(event_seconds, 3),
                "first_maintain": round(maintain_seconds, 3),
                "total": round(time.perf_counter() - started, 3),
            },
        }
        report["counts"] = counts(store)
        return report
    finally:
        store.close()


def seed_now(store: SampleStore) -> float:
    """The seed's own clock, read back out of `meta`.

    Every window in this module is relative to it rather than to `time.time()`: the deep history
    is anchored to the moment the file was built, so a test or a pass that used the wall clock
    would measure a different slice of the store on every run (and prune the baseline). Lives
    here rather than in `tests/test_store.py` because the HTTP budget pass in
    `tests/test_api.py` needs it too, and two copies of "what time is this store" is how a
    before/after stops being a before/after.
    """
    raw = store.get_meta(META_SEED_NOW)
    assert raw is not None, "the seeded store lost its clock row"
    return float(raw)


def counts(store: SampleStore) -> dict[str, int]:
    """Row counts by resolution, plus series and event totals."""
    with store._lock, store._db:
        by_res = {row[0]: int(row[1]) for row in store._db.execute(
            "SELECT res, COUNT(*) FROM sample WHERE source='host' GROUP BY res")}
        series = int(store._db.execute(
            "SELECT COUNT(*) FROM (SELECT DISTINCT source, host, metric FROM sample"
            " WHERE source='host')").fetchone()[0])
        events = int(store._db.execute("SELECT COUNT(*) FROM event").fetchone()[0])
    return {"raw": by_res.get(RES_RAW, 0), "1m": by_res.get(RES_MIN, 0),
            "1h": by_res.get(RES_HOUR, 0), "series": series, "events": events}


# --------------------------------------------------------------------------- #
# the cached store: one file, built once, verified on reuse
# --------------------------------------------------------------------------- #
def cache_dir(explicit: str | Path | None = None) -> Path:
    """Where the cached seeded store lives: `DASH_PERF_STORE_DIR`, else `data/perf`.

    A directory inside the repo (and inside the gitignored `data/`) on purpose: it is the same
    *kind* of filesystem the container gets, unlike pytest's `tmp_path` on this box, which is
    `tmpfs`. Set `DASH_PERF_STORE_DIR=/tmp/…` to measure on RAM instead, and know that you
    then measured the faster machine.
    """
    if explicit is not None:
        return Path(explicit)
    env = os.environ.get("DASH_PERF_STORE_DIR", "").strip()
    return Path(env) if env else ROOT / "data" / "perf"


def cache_path(directory: str | Path | None = None) -> Path:
    return cache_dir(directory) / f"seeded-{SEED_VERSION}.sqlite3"


def _close_and_remove(path: Path) -> None:
    for suffix in ("", "-wal", "-shm", "-journal"):
        candidate = Path(str(path) + suffix)
        if candidate.exists():
            candidate.unlink()


@contextmanager
def scratch_copy(path: Path, *, name: str = "scratch") -> Iterator[Path]:
    """A private copy of the seeded store, for the measurements that **write**.

    The write-side passes this phase times are not all idempotent-at-a-clock: a rollup run at
    `seed_now + 600` really does roll ten more minutes of buckets and prune ten minutes of raw, and
    the pre-fix whole-window rollup rewrites 330 000 rows. Doing that to the cached file would move
    the baseline every later task compares against, so it happens here instead — same directory
    (same device, same page-cache behaviour), same rows, deleted afterwards. `Connection.backup`
    rather than `copyfile` because the seed has a WAL of its own and a byte copy would miss it; the
    copy is 720 MB in ~1.3 s on the probe's machine, which is cheap next to the passes it buys.
    """
    dest = path.with_name(f"{path.stem}.{name}-{os.getpid()}.sqlite3")
    # A run killed mid-pass leaves a 720 MB copy behind; sweep the old ones (never a live one:
    # an hour of mtime is several passes) before adding to the pile.
    for stale in path.parent.glob(f"{path.stem}.{name}-*.sqlite3"):
        if time.time() - stale.stat().st_mtime > 3_600:
            _close_and_remove(stale)
    source = sqlite3.connect(str(path))
    try:
        target = sqlite3.connect(str(dest))
        try:
            source.backup(target)
        finally:
            target.close()
    finally:
        source.close()
    try:
        yield dest
    finally:
        _close_and_remove(dest)


def wal_bytes(path: Path) -> int:
    """Public spelling of `_wal_bytes`, for the notes and the live watch."""
    return _wal_bytes(path)


def _verify(path: Path) -> str | None:
    """`None` when the cached file matches the shape, else the reason it does not."""
    if not path.exists():
        return "not built yet"
    store = open_store(path)
    try:
        version = store.get_meta(META_SEED_VERSION)
        if version != SEED_VERSION:
            return f"seed version {version!r} != {SEED_VERSION!r}"
        now = float(store.get_meta(META_SEED_NOW) or "0")
        if not now:
            return "no seed clock recorded"
        age_hours = (time.time() - now) / 3600.0
        if age_hours > MAX_CACHE_AGE_HOURS:
            return f"seeded {age_hours:.1f} h ago (max {MAX_CACHE_AGE_HOURS:.0f} h)"
        if age_hours < -1.0:
            return "seeded in the future (the clock moved backwards)"
        expected = plan(now)
        got = counts(store)
        if got["raw"] != expected.raw_rows:
            return f"raw rows {got['raw']} != {expected.raw_rows}"
        for res, label in ((RES_MIN, "1m"), (RES_HOUR, "1h")):
            seeded = {"1m": expected.minute_rows_seeded, "1h": expected.hour_rows_seeded}[label]
            if got[res] < seeded:
                return f"{label} rows {got[res]} < the {seeded} the shape seeds"
        if got["events"] < expected.events:
            return f"{got['events']} events < {expected.events}"
        if got["series"] != expected.series_total:
            return f"{got['series']} series != {expected.series_total}"
        return None
    finally:
        store.close()


def ensure_seeded(directory: str | Path | None = None, *, force: bool = False,
                  log: Callable[[str], None] | None = None) -> Path:
    """The cached seeded store's path, building (or rebuilding) it when needed.

    `flock` because a half-built store is worse than a slow one: a session that read a partial
    file would report timings for a database nobody has. The build writes to a sibling
    `.building` path and renames, so readers only ever see a finished store.
    """
    target = cache_path(directory)
    target.parent.mkdir(parents=True, exist_ok=True)
    reason = None if force else _verify(target)
    if reason is None:
        if log:
            log(f"reusing seeded store {target}")
        return target
    say = log or (lambda _line: None)
    say(f"building seeded store ({reason}) — this takes a minute or two")
    lock_path = Path(str(target) + ".lock")
    with open(lock_path, "w") as handle:
        fcntl.flock(handle, fcntl.LOCK_EX)
        try:
            # re-check: another session may have finished the same build while we waited
            reason = None if force else _verify(target)
            if reason is not None:
                _close_and_remove(target)
                staging = Path(str(target) + ".building")
                _close_and_remove(staging)
                build(staging, log=log)
                staging.replace(target)
                for suffix in ("-wal", "-shm"):
                    leftover = Path(str(staging) + suffix)
                    if leftover.exists():
                        leftover.replace(Path(str(target) + suffix))
        finally:
            fcntl.flock(handle, fcntl.LOCK_UN)
    return target


# --------------------------------------------------------------------------- #
# measurement
# --------------------------------------------------------------------------- #
def timed(fn: Callable[[], Any], *, runs: int = 3) -> dict[str, float]:
    """Wall-clock seconds for `fn`, run `runs` times: mean/median/max of the individual runs."""
    samples: list[float] = []
    for _ in range(runs):
        started = time.perf_counter()
        fn()
        samples.append(time.perf_counter() - started)
    return {"mean": statistics.fmean(samples), "median": statistics.median(samples),
            "max": max(samples), "min": min(samples), "runs": len(samples)}


def explain(store: SampleStore, sql: str, args: tuple[Any, ...] = ()) -> list[str]:
    """`EXPLAIN QUERY PLAN` for one statement, as the one-line shapes the NOTES quote."""
    with store._lock, store._db:
        rows = store._db.execute(f"EXPLAIN QUERY PLAN {sql}", args).fetchall()
    return [str(row[3]) for row in rows]


def _machine() -> dict[str, Any]:
    cpu = platform.uname()
    model = "unknown"
    try:
        for line in Path("/proc/cpuinfo").read_text().splitlines():
            if line.lower().startswith("model name"):
                model = line.split(":", 1)[1].strip()
                break
    except OSError:  # pragma: no cover - /proc is always there on Linux
        pass
    usage = shutil.disk_usage(str(cache_dir()))
    return {
        "cpu": model,
        "cores": os.cpu_count(),
        "machine": cpu.machine,
        "kernel": cpu.release,
        "python": platform.python_version(),
        "sqlite": sqlite3.sqlite_version,
        "store_dir": str(cache_dir()),
        "store_dir_free_gb": round(usage.free / 1e9, 1),
        "store_dir_on_tmpfs": _is_tmpfs(str(cache_dir())),
    }


def _is_tmpfs(path: str) -> bool:
    """Is the store's directory RAM? A caveat that changes every number in this report."""
    try:
        lines = Path("/proc/mounts").read_text()
    except OSError:  # pragma: no cover
        return False
    best, best_len, best_type = "", 0, "unknown"
    for line in lines.splitlines():
        parts = line.split()
        if len(parts) < 3:
            continue
        mount, kind = parts[1], parts[2]
        if path.startswith(mount) and len(mount) > best_len:
            best, best_len, best_type = mount, len(mount), kind
    del best
    return best_type == "tmpfs"


def _main_bytes(path: Path) -> int:
    """The database file alone — the WAL is reported separately, because *that* is the number
    task 03 tunes and folding the two together hides which one moved."""
    return path.stat().st_size if path.exists() else 0


def _size_bytes(path: Path) -> int:
    total = 0
    for suffix in ("", "-wal", "-shm"):
        candidate = Path(str(path) + suffix)
        if candidate.exists():
            total += candidate.stat().st_size
    return total


def _wal_bytes(path: Path) -> int:
    wal = Path(str(path) + "-wal")
    return wal.stat().st_size if wal.exists() else 0


@dataclass
class RaceResult:
    """One round of "the poller writes while a page reads" — the known-gap-4 race, as numbers."""

    target: int = 0        # inserts asked for
    attempted: int = 0     # inserts actually run (a deadline can stop it early — see MAX_RACE_SECONDS)
    inserts: int = 0       # inserts that succeeded
    seconds: float = 0.0
    waiters: int = 0       # inserts slower than WAIT_THRESHOLD_MS
    max_wait_ms: float = 0.0
    median_wait_ms: float = 0.0
    busy_errors: int = 0   # OperationalError "database is locked" — outlived busy_timeout
    notes: list[str] = field(default_factory=list)

    def as_dict(self) -> dict[str, Any]:
        return {**asdict(self), "busy_timeout_ms": BUSY_TIMEOUT_MS,
                "wait_threshold_ms": WAIT_THRESHOLD_MS}


def _record_waits(result: RaceResult, waits: list[float]) -> None:
    if not waits:
        return
    result.waiters = sum(1 for wait in waits if wait > WAIT_THRESHOLD_MS)
    result.max_wait_ms = max(waits)
    result.median_wait_ms = statistics.median(waits)


def race_single_connection(store: SampleStore, *, iterations: int = 100,
                           max_seconds: float = MAX_RACE_SECONDS) -> RaceResult:
    """The one-process race: the poller's `add_many` while a page runs `stats()`.

    Everything in the app shares `SampleStore._lock`, so this cannot produce a SQLite lock
    error — it produces the *other* half of the complaint, a poll that waits behind a read (and
    a page that waits behind a poll). The number that matters is how long an insert waits for the
    mutex, because on today's store that is the read's own full-table cost: one `/api/health`
    poll can hold every writer out for as long as it takes to count the table.

    The loop stops at `max_seconds` rather than running all 100 inserts, because a read here
    costs seconds and 100 of them would be a five-minute benchmark; the report says how many
    were attempted, so the sample size is part of the number.
    """
    result = RaceResult(target=iterations)
    stop = threading.Event()
    reads: list[float] = []

    def reader() -> None:
        while not stop.is_set():
            started = time.perf_counter()
            store.stats()
            store.series_names("host")
            reads.append(time.perf_counter() - started)

    thread = threading.Thread(target=reader, name="perf-reader", daemon=True)
    thread.start()
    waits: list[float] = []
    started_at = time.perf_counter()
    deadline = started_at + max_seconds
    try:
        while result.attempted < iterations and time.perf_counter() < deadline:
            index = result.attempted
            result.attempted += 1
            ts = time.time()
            samples = [Sample(RACE_SOURCE, RACE_HOST, f"race.metric{column}", float(index), ts)
                       for column in range(30)]
            started = time.perf_counter()
            try:
                store.add_many(samples)
                result.inserts += 1
            except sqlite3.OperationalError as exc:  # pragma: no cover - the mutex prevents it
                result.busy_errors += 1
                result.notes.append(f"add_many raised {exc}")
            waits.append((time.perf_counter() - started) * 1000.0)
    finally:
        stop.set()
        thread.join(timeout=120.0)
        store.delete_series(RACE_SOURCE, RACE_HOST)
    result.seconds = time.perf_counter() - started_at
    _record_waits(result, waits)
    if reads:
        result.notes.append(
            f"{len(reads)} `stats()`+`series_names()` passes in {result.seconds:.1f} s, median "
            f"{statistics.median(reads) * 1000:,.0f} ms — the store mutex serialises them, so an "
            f"insert above {WAIT_THRESHOLD_MS:.0f} ms waited behind a read, not behind a SQLite "
            "lock (one process cannot get one)")
    return result


def race_two_connections(path: Path, *, iterations: int = 100, busy_timeout_ms: int | None = None,
                         now: float | None = None,
                         max_seconds: float = MAX_RACE_SECONDS,
                         pace_seconds: float = RACE_PACE_SECONDS,
                         reset_window: bool = False,
                         maintain_gap_seconds: float = 0.0) -> RaceResult:
    """The two-process race: an insert losing to `maintain()`'s write transaction.

    A second `SampleStore` on the same file is the cheapest faithful stand-in for the case
    `AGENTS.md` gap 4 proved for this repo — *two servers on one database* — because one
    process's mutex hides the SQLite-level contention entirely. `busy_timeout_ms=0` counts raw
    `SQLITE_BUSY` replies (what the poller would see with no timeout at all, and the measure of
    how much of the window the write lock was held); the production 5 000 ms answers the question
    the gap actually asks: does an insert ever outlive the timeout and surface as `store-error`?

    `max_seconds` bounds the pass for the same reason as the other race: with a timeout of five
    seconds and a `maintain()` of several, "100 inserts" would be minutes of waiting on purpose.

    `reset_window=True` deletes `last_maintain` before every pass, which puts the maintainer back
    into the shape task 01 measured — a whole-retention rollup on every tick — and it is run as the
    **control**. A windowed `maintain()` writes almost nothing per minute, so "zero `database is
    locked`" would also be what a race that never contended reports; the control is the number that
    says this race can still see the failure, and therefore that the zero means the fix.

    `maintain_gap_seconds` is the app's 60 s cadence, compressed: the maintainer idles between
    passes instead of restarting the transaction the instant it commits. `measure()` runs both, and
    the gap is where the honest pass line lives — see `RACE_MAINTAIN_GAP_SECONDS`.
    """
    timeout = BUSY_TIMEOUT_MS if busy_timeout_ms is None else busy_timeout_ms
    # prune-safe: the point of this race is the lock, not the retention sweep (see open_store)
    writer = open_store(path, extra_retention_hours=2 * MAX_CACHE_AGE_HOURS)
    maintainer = open_store(path, extra_retention_hours=2 * MAX_CACHE_AGE_HOURS)
    writer._db.execute(f"PRAGMA busy_timeout={timeout}")
    result = RaceResult(target=iterations)
    stop = threading.Event()
    writing = threading.Event()
    maintains: list[float] = []

    def maintainer_loop() -> None:
        while not stop.is_set():
            if reset_window:
                # outside `writing` on purpose: this is not part of the transaction being raced
                maintainer.delete_meta(LAST_MAINTAIN_KEY)
            started = time.perf_counter()
            writing.set()
            try:
                # the seed's own clock, so the loop repeats the same write instead of ageing the
                # store it is measuring
                maintainer.maintain(now=now)
            finally:
                maintains.append(time.perf_counter() - started)
                writing.clear()
            if maintain_gap_seconds:
                time.sleep(maintain_gap_seconds)

    thread = threading.Thread(target=maintainer_loop, name="perf-maintain", daemon=True)
    thread.start()
    # Don't start inserting until a write transaction is actually open: un-paced, 100 inserts
    # finish in ~50 ms, before the maintainer's first `BEGIN`, and the pass reports "no
    # contention" for a race that never happened. `writing` is set just before `maintain()`.
    if not writing.wait(timeout=max_seconds):  # pragma: no cover - a dead maintainer thread
        result.notes.append("the maintainer thread never started a transaction")
    time.sleep(0.05)  # the flag is set just before `BEGIN`; give the transaction a moment
    waits: list[float] = []
    started_at = time.perf_counter()
    deadline = started_at + max_seconds
    try:
        while result.attempted < iterations and time.perf_counter() < deadline:
            index = result.attempted
            result.attempted += 1
            ts = time.time()
            samples = [Sample(RACE_SOURCE, RACE_HOST, f"race.metric{column}", float(index), ts)
                       for column in range(30)]
            started = time.perf_counter()
            try:
                writer.add_many(samples)
                result.inserts += 1
            except sqlite3.OperationalError as exc:
                result.busy_errors += 1
                if len(result.notes) < 2:
                    result.notes.append(f"add_many raised: {exc}")
            waits.append((time.perf_counter() - started) * 1000.0)
            # pace it: this is the poller's cadence, not a tight loop — 9 sources writing every
            # few seconds, so inserts land at every offset inside the write transaction
            time.sleep(pace_seconds)
    finally:
        stop.set()
        thread.join(timeout=180.0)
        try:
            writer.delete_series(RACE_SOURCE, RACE_HOST)
        except sqlite3.OperationalError as exc:  # pragma: no cover - cleanup only
            result.notes.append(f"cleanup skipped: {exc}")
        writer.close()
        maintainer.close()
    result.seconds = time.perf_counter() - started_at
    _record_waits(result, waits)
    result.notes.insert(0, (
        f"busy_timeout={timeout} ms · {'whole-window (pre-fix control) ' if reset_window else ''}"
        f"{'cadence ' + format(maintain_gap_seconds, '.1f') + ' s gap' if maintain_gap_seconds else 'saturated (no gap between passes)'}"
        f" · {len(maintains)} `maintain()` write transactions "
        f"(median {statistics.median(maintains) * 1000:,.0f} ms) held the WAL write lock across "
        f"{result.seconds:.1f} s · {result.busy_errors} of {result.attempted} inserts answered "
        f"`database is locked`"))
    return result


#: The write-side passes, as `(label, seconds after the seed clock, drop `last_maintain` first)`.
#: All three run on the same copy of the same rows, so they differ only in the window:
#:
#: * `steady` is the shape of nearly every real tick — the previous pass recorded the same clock,
#:   so each step re-reads its overlap and writes the buckets that filled since;
#: * `catch_up_10m` is the restart-after-a-quiet-interval case, ten minutes of buckets due;
#: * `full_window` is the **pre-fix control**: with no `last_maintain` row the steps scan their whole
#:   retention, which is the transaction task 01 measured at 5.4 s. Reproducing it in the same run
#:   on the same file is what turns "the race is clean now" into a measurement instead of a hope.
MAINTAIN_PASSES: tuple[tuple[str, float, bool], ...] = (
    ("steady", 0.0, False),
    ("catch_up_10m", 600.0, False),
    ("full_window", 1_200.0, True),
)

#: The SELECT half of `store.ROLLUP_INSERT`, derived rather than retyped so the `EXPLAIN QUERY PLAN`
#: the NOTES record is the plan the rollup actually runs (the same reason the count branch is a
#: constant). `SELECT` occurs once in that statement.
ROLLUP_SELECT = "SELECT" + ROLLUP_INSERT.split("SELECT", 1)[1]


def measure_maintain(path: Path, *, now: float, log: Callable[[str], None] | None = None) -> dict[str, Any]:
    """The write transaction, per window, on a scratch copy — with the WAL around each pass.

    `now` is the seed's own clock. Offsets from it are the only way to ask "what did the next ten
    minutes cost?" of a store whose history is anchored to when it was seeded, and the copy keeps
    those writes away from the file every later measurement in the phase compares against.
    """
    say = log or (lambda _line: None)
    out: dict[str, Any] = {
        "journal_size_limit_bytes": JOURNAL_SIZE_LIMIT_BYTES,
        "seed_now": now,
        "passes": {},
    }
    with scratch_copy(path) as copy:
        out["scratch_bytes"] = _main_bytes(copy)
        store = open_store(copy)
        try:
            for label, offset, reset in MAINTAIN_PASSES:
                if reset:
                    store.delete_meta(LAST_MAINTAIN_KEY)
                written: dict[str, int] = {}

                def run_pass(_offset: float = offset, _into: dict[str, int] = written) -> None:
                    _into.update(store.maintain(now=now + _offset))

                wal_before = _wal_bytes(copy)
                run = timed(run_pass, runs=1)
                wal_after = _wal_bytes(copy)
                out["passes"][label] = {
                    "seconds": run["mean"],
                    "clock": now + offset,
                    "counts": written,
                    "rows_written": written["to_minute"] + written["to_hour"],
                    "wal_before_bytes": wal_before,
                    "wal_after_bytes": wal_after,
                }
                say(f"  {label}: {run['mean'] * 1000:,.0f} ms, "
                    f"{written} -> WAL {wal_after / 1e6:,.1f} MB")
            # Two more ticks for the WAL criterion: the cap is what the file must stay under, and
            # the steady-state growth per tick is what the 140 MB in AGENTS.md gap 4 adds up from.
            cycles: list[dict[str, int]] = []
            for _ in range(2):
                before = _wal_bytes(copy)
                store.maintain(now=now + 1_260.0)
                after = _wal_bytes(copy)
                cycles.append({"wal_before_bytes": before, "wal_after_bytes": after})
            out["two_steady_cycles"] = cycles
            out["wal_within_cap"] = all(
                cycle["wal_after_bytes"] <= JOURNAL_SIZE_LIMIT_BYTES for cycle in cycles)
            out["store_bytes_after"] = _main_bytes(copy)
            # The plans, from the copy, after the first pass put statistics in the file and against
            # the windows just timed. The SELECT half of a rollup is what costs and the DELETE half
            # is what a planner gets wrong: `SCAN sample` here means the window bought nothing, no
            # matter how few rows are inside it.
            out["plans"] = {
                "rollup_minute_steady": explain(store, ROLLUP_SELECT, (
                    RES_MIN, MINUTE, MINUTE, RES_RAW, now - MINUTE - MAINTAIN_SLACK_SECONDS,
                    now - MINUTE - MAINTAIN_SLACK_SECONDS, MINUTE)),
                "rollup_minute_catch_up": explain(store, ROLLUP_SELECT, (
                    RES_MIN, MINUTE, MINUTE, RES_RAW, now - MINUTE - MAINTAIN_SLACK_SECONDS,
                    now - MINUTE - MAINTAIN_SLACK_SECONDS - 600.0, MINUTE)),
                "rollup_hour_steady": explain(store, ROLLUP_SELECT, (
                    RES_HOUR, HOUR, HOUR, RES_MIN, now - HOUR - MAINTAIN_SLACK_SECONDS,
                    now - HOUR - MAINTAIN_SLACK_SECONDS, HOUR)),
                "prune_raw": explain(store, "DELETE FROM sample WHERE res = ? AND ts < ?",
                                     (RES_RAW, now - RAW_KEEP_HOURS * HOUR)),
            }
        finally:
            store.close()
    return out


def measure_http(path: Path, *, hosts: tuple[str, ...],
                 metrics: tuple[str, ...]) -> dict[str, Any]:
    """The browser's own numbers, through the real ASGI gate (a session is required).

    `TestClient` is in-process, so this is the handler's cost plus the store's, with no network
    — which is the half the operator feels as "the page took forever" once the response is big.
    """
    sys_path_fixup()
    from fastapi.testclient import TestClient  # noqa: PLC0415 - only the HTTP numbers need it

    # tests/ is on sys.path via sys_path_fixup(); these two are the suite's own door in.
    from auth_client import ADMIN_PASSWORD, sign_in  # noqa: PLC0415
    from auth_contract import FAST_PBKDF2  # noqa: PLC0415
    from dashboards.app import create_app  # noqa: PLC0415
    from dashboards.config import Settings  # noqa: PLC0415

    settings = Settings(db_path=str(path), admin_password=ADMIN_PASSWORD,
                        pbkdf2_iterations=FAST_PBKDF2)
    # The lifespan starts the hub's maintenance loop, and that loop calls `maintain()` at the
    # wall clock the moment it starts — so this handle gets the prune-safe margin, or booting a
    # server to time a read would prune the baseline it is timing.
    store = open_store(path, extra_retention_hours=2 * MAX_CACHE_AGE_HOURS)
    out: dict[str, Any] = {}
    with TestClient(create_app(settings, store)) as client:
        sign_in(client)
        joined_hosts = ",".join(hosts)
        joined_metrics = ",".join(metrics)
        page = ",".join(PAGE_HISTORY_METRICS)
        out["health"] = timed(lambda: client.get("/api/health"), runs=5)
        out["metrics_list"] = timed(lambda: client.get("/api/metrics?source=host"), runs=3)
        out["system_history_9x3_1h"] = timed(
            lambda: client.get(f"/api/system/history?hosts={joined_hosts}"
                               f"&metrics={joined_metrics}&range=1h"), runs=5)
        out["system_history_9x3_24h"] = timed(
            lambda: client.get(f"/api/system/history?hosts={joined_hosts}"
                               f"&metrics={joined_metrics}&range=24h"), runs=3)
        out["system_history_9x3_90d"] = timed(
            lambda: client.get(f"/api/system/history?hosts={joined_hosts}"
                               f"&metrics={joined_metrics}&range=90d"), runs=3)
        out["system_history_page_shape_1h"] = timed(
            lambda: client.get(f"/api/system/history?hosts={hosts[0]}"
                               f"&metrics={page}&range=1h"), runs=5)
        out["summary"] = timed(lambda: client.get("/api/summary"), runs=3)
    store.close()
    return out


def sys_path_fixup() -> None:
    """Make `tests/` importable from a bare script (pytest does it for free via conftest)."""
    here = str(Path(__file__).resolve().parent)
    if here not in sys.path:
        sys.path.insert(0, here)


def measure(store: SampleStore, *, race_iterations: int = 100, with_http: bool = True,
            race_max_seconds: float = MAX_RACE_SECONDS,
            log: Callable[[str], None] | None = None) -> dict[str, Any]:
    """Every number task 01 asks for, in one dict.

    Read-only except the two race helpers, which clean up after themselves (the seeded store is
    reused by later tasks, so a measurement that leaves 3 000 rows behind would move the next
    one's baseline).
    """
    say = log or (lambda _line: None)
    path = Path(store.path)
    now = float(store.get_meta(META_SEED_NOW) or time.time())
    expected = plan(now)
    report: dict[str, Any] = {
        "generated_at": time.time(),
        "seed_version": SEED_VERSION,
        "machine": _machine(),
        "store": {
            "path": str(path),
            "bytes": _size_bytes(path),
            "wal_bytes": _wal_bytes(path),
            "seed_now": now,
            "expected": expected.as_dict(),
            "counts": counts(store),
        },
        "paths": {},
        "plans": {},
        "race": {},
    }

    def cold(fn: Callable[[], Any]) -> Callable[[], Any]:
        """Time the *uncached* answer, whatever the short-TTL read caches are holding.

        Task 02 put a cache in front of the two scanned answers, so a report that timed them
        naively would compare a cache hit against the baseline's full scan and call it a 1 000x
        win. Both numbers belong in the NOTES: the cold one is what the first poll after a process
        start pays, the warm one is what the browser's next poll pays.
        """
        def run() -> Any:
            store.invalidate_read_caches()
            return fn()
        return run

    say("series_names() — cold (cache invalidated) and cached")
    report["paths"]["series_names_all"] = timed(cold(lambda: store.series_names()), runs=3)
    report["paths"]["series_names_all_cached"] = timed(lambda: store.series_names(), runs=5)
    report["paths"]["series_names_host"] = timed(cold(lambda: store.series_names("host")), runs=3)
    report["paths"]["series_names_host_cached"] = timed(lambda: store.series_names("host"), runs=5)
    report["paths"]["series_names_host_driveripper"] = timed(
        lambda: store.series_names("host", "driveripper"), runs=3)
    report["plans"]["series_names_all"] = explain(store, "SELECT DISTINCT metric FROM sample"
                                                        " ORDER BY metric")
    report["plans"]["series_names_host"] = explain(
        store, "SELECT DISTINCT metric FROM sample WHERE source=? ORDER BY metric", ("host",))

    say("stats() — what the browser-polled /api/health runs, cold and cached")
    report["paths"]["stats"] = timed(cold(store.stats), runs=5)
    report["paths"]["stats_cached"] = timed(store.stats, runs=5)
    report["plans"]["stats_by_res"] = explain(
        store, "SELECT res, COUNT(*) FROM sample GROUP BY res")
    report["plans"]["stats_events"] = explain(store, "SELECT COUNT(*) FROM event")

    say("pick_res() + history() over 90 d")
    host, metric = "driveripper", "cpu.usage_pct"
    report["paths"]["pick_res_90d"] = timed(
        lambda: [store.pick_res("host", host, metric, span, 600)
                 for span in (3600.0, 24 * 3600.0, 90 * 24 * 3600.0)], runs=3)
    report["paths"]["history_90d"] = timed(
        lambda: store.history("host", host, metric, span=90 * 24 * 3600.0, max_points=600),
        runs=3)
    report["paths"]["history_24h"] = timed(
        lambda: store.history("host", host, metric, span=24 * 3600.0, max_points=600), runs=3)
    report["paths"]["history_5m"] = timed(
        lambda: store.history("host", host, metric, span=300.0, max_points=600), runs=3)
    # The statement behind `pick_res` since task 02: three capped counts in one compound, one
    # index seek each. EXPLAINed here rather than the old per-`res` COUNT because the NOTES
    # compare shapes across the phase, and the shape that runs is the one worth recording.
    res_count = " UNION ALL ".join([RES_COUNT_BRANCH] * 3)
    res_args: list[object] = []
    for res in (RES_RAW, RES_MIN, RES_HOUR):
        # the branch's own parameter order: three label columns, then the seek itself
        res_args += [host, metric, res, "host", host, metric, res, now - 90 * 24 * 3600.0, 601]
    report["plans"]["pick_res_count"] = explain(store, res_count, tuple(res_args))

    say("the N x M batch (27 series, the /system page's request) — grouped vs per-series")
    pair_hosts = list(HOST_SERIES)
    pair_metrics = list(HISTORY_METRICS)
    for label, span in (("24h", 24 * 3600.0), ("90d", 90 * 24 * 3600.0)):
        report["paths"][f"pick_res_x27_{label}"] = timed(
            lambda hs=pair_hosts, ms=pair_metrics, s=span: [
                store.pick_res("host", h, m, s, 600) for h in hs for m in ms], runs=3)
        report["paths"][f"pick_res_many_x27_{label}"] = timed(
            lambda hs=pair_hosts, ms=pair_metrics, s=span: store.pick_res_many(
                "host", hs, ms, s, 600), runs=3)
        report["paths"][f"history_x27_{label}"] = timed(
            lambda hs=pair_hosts, ms=pair_metrics, s=span: [
                store.history("host", h, m, span=s, max_points=600)
                for h in hs for m in ms], runs=3)
        report["paths"][f"history_many_x27_{label}"] = timed(
            lambda hs=pair_hosts, ms=pair_metrics, s=span: store.history_many(
                "host", hs, ms, span=s, max_points=600), runs=3)
    # The batched statement is that same branch repeated per `(series, res)` and chunked at
    # `COMPOUND_CHUNK`; EXPLAINing three of them is the shape without a 300-line plan in the NOTES.
    grouped_args: list[object] = []
    for (h, m), res in zip((("driveripper", "cpu.usage_pct"), ("deskwork", "mem.used_pct"),
                            ("gateway", "disk.used_pct")), (RES_RAW, RES_MIN, RES_HOUR),
                           strict=True):
        grouped_args += [h, m, res, "host", h, m, res, now - 90 * 24 * 3600.0, 601]
    report["plans"]["pick_res_many_branch"] = explain(
        store, " UNION ALL ".join([RES_COUNT_BRANCH] * 3), tuple(grouped_args))
    report["plans"]["history_tail"] = explain(
        store, "SELECT ts, value FROM (SELECT ts, value FROM sample WHERE source=? AND host=?"
               " AND metric=? AND res=? AND ts>=? AND ts<=? ORDER BY ts DESC LIMIT ?)"
               " ORDER BY ts ASC",
        ("host", host, metric, RES_MIN, now - 90 * 24 * 3600.0, now, 2400))

    say("maintain() — the write transaction per window, and the WAL around each pass")
    # Seeded-clock maintenance on a scratch copy: the store's newest row is at `now`, so timing a
    # rollup at the wall clock would measure a slightly different job (and prune the baseline) every
    # time the probe runs. The copy is where the passes that *write* belong — a catch-up pass rolls
    # ten more minutes of buckets, and the pre-fix whole-window control rewrites 330 000 rows.
    report["maintain"] = measure_maintain(path, now=now, log=say)
    # The write-side plans (`maintain.plans`) are explained inside `measure_maintain`, on the copy:
    # a plan is only worth recording next to the timings that produced it, and only after that
    # pass has put query statistics in the file — without statistics the planner reads the table for
    # every window, so explaining it beforehand would record a shape production no longer runs.

    say(f"the race: a 30-sample add_many against {race_iterations} concurrent readers/maintains")
    single = race_single_connection(store, iterations=race_iterations,
                                    max_seconds=race_max_seconds)
    report["race"]["single_connection"] = single.as_dict()
    # The two-connection races write (rollups on the maintainer's side, 3 000 raced samples on the
    # writer's), so they run against one scratch copy of the same rows: the seed the next task
    # measures has to come back exactly as it went in.
    with scratch_copy(path, name="race") as arena:
        # (1) the saturated harness, unchanged from task 01 so its numbers stay comparable, and
        # (2) the same race at the app's cadence, which is the one with a pass line: a poller's
        # insert landing inside a rollup waits that rollup out, and windowing is what made the
        # wait short enough to never reach `busy_timeout`.
        report["race"]["two_connections_prod_timeout"] = race_two_connections(
            arena, iterations=race_iterations, now=now, max_seconds=race_max_seconds).as_dict()
        report["race"]["two_connections_no_timeout"] = race_two_connections(
            arena, iterations=race_iterations, busy_timeout_ms=0, now=now,
            max_seconds=race_max_seconds).as_dict()
        report["race"]["poller_during_windowed_maintain"] = race_two_connections(
            arena, iterations=race_iterations, now=now, max_seconds=race_max_seconds,
            maintain_gap_seconds=RACE_MAINTAIN_GAP_SECONDS).as_dict()
        report["race"]["poller_during_full_window_maintain"] = race_two_connections(
            arena, iterations=race_iterations, now=now, max_seconds=race_max_seconds,
            reset_window=True, maintain_gap_seconds=RACE_MAINTAIN_GAP_SECONDS).as_dict()
    for key in ("two_connections_prod_timeout", "two_connections_no_timeout"):
        report["race"][key]["notes"].append(
            "saturated harness (task 01's shape): the maintainer restarts `maintain()` the instant "
            "it commits, so the WAL write lock is held ~100% of the wall clock and a writer can be "
            "starved however short each pass is. The app maintains once per 60 s — the "
            "`poller_during_*` rows are that shape, and they are the ones with a pass line.")

    if with_http:
        say("through the app (signed in, TestClient, no network)")
        report["http"] = measure_http(path, hosts=expected.hosts, metrics=HISTORY_METRICS)

    # Task 04's budgets, priced against everything above. Computed last on purpose: a budget row
    # whose path was never measured is a failing row, so it has to see the finished report.
    rows = budget_rows(report)
    report["budgets"] = [row.as_dict() for row in rows]
    report["budgets_ok"] = all(row.ok for row in rows)

    # `_verify` is cheap and proves the measurements left the seeded shape alone
    problem = _verify(path)
    report["store_after"] = {"counts": counts(store), "integrity": problem or "unchanged"}
    return report


# --------------------------------------------------------------------------- #
# the budgets (task 04): the measured "after", promoted to a red test
# --------------------------------------------------------------------------- #
#: Why a table and not `assert elapsed < 0.05` sprinkled through the tests: the same numbers have
#: to appear in three places — the pytest budgets, the probe's report, and
#: `data/logs/NOTES-store-perf.md` — and three copies of a budget drift the moment one is relaxed.
#: One table, one verdict function, and the notes are rendered from it.
#:
#: Every limit is **2-3x the measured "after"** on the seeded store (`AGENTS.md`: a budget asserted
#: on first writing is a number fitted to the machine it was written on), and each row records that
#: measured figure in `basis` so a reader can see the margin instead of trusting it. They are cost
#: *ceilings*, not targets: a path that used to cost a whole-table pass and now costs 1 ms still
#: gets a 10 ms ceiling, because the claim under test is "bounded", not "1 ms".
@dataclass(frozen=True)
class Budget:
    """One pinned cost: a report path, the ceiling it may not cross, and where the ceiling came from."""

    #: which section of a `measure()` report to read: `paths`, `http`, `maintain`, `race`
    section: str
    #: the key inside that section
    key: str
    #: the ceiling, in `unit`
    limit: float
    #: what the limit is compared against: a `timed()` field (`mean`/`median`/`max`) for the
    #: `paths`/`http` sections, `seconds` for a `maintain` pass, and the `RaceResult` field name
    #: (`max_wait_ms`, `busy_errors`) for a race
    statistic: str = "median"
    unit: str = "ms"
    #: prose: the measured "after" this ceiling was set from, and the margin on it
    basis: str = ""
    #: an optional *relative* claim — this path must also cost at most `ratio_limit` of the named
    #: key in the same section. That is the assertion a cache needs: "0.0 ms" on a machine that
    #: got faster for an unrelated reason is uninteresting, "50x cheaper than the scan it replaced"
    #: is the scan being gone.
    ratio_key: str | None = None
    ratio_limit: float | None = None

    def label(self) -> str:
        return f"{self.section}.{self.key}[{self.statistic}]"

    def as_dict(self) -> dict[str, Any]:
        return asdict(self)


#: The budgets themselves, in the order the NOTES table prints them. Keys are report keys, so
#: `test_every_pinned_budget_names_a_path_the_probe_measures` can fail a budget that points at a
#: measurement nobody takes any more.
BUDGETS: tuple[Budget, ...] = (
    # --- the two reads that used to be a scan per request -------------------------------
    Budget("paths", "series_names_host_cached", 10.0, ratio_key="series_names_host",
           ratio_limit=0.02, basis="measured ~0.0 ms cached vs 1,440 ms cold (task 03); the claim "
           "is 'the scan is gone', so the ceiling comes with a 50x-relative claim beside it"),
    Budget("paths", "stats_cached", 10.0, ratio_key="stats", ratio_limit=0.02,
           basis="measured ~0.0 ms cached vs 1,336 ms cold (task 03); `/api/health` runs this "
           "every 15 s per tab, which is why both halves are pinned"),
    Budget("paths", "history_90d", 25.0, basis="measured 1.9-2.6 ms; task 01's 'do not regress it' "
           "line, 10x the after and 6x under the 4.1 ms baseline"),
    Budget("paths", "pick_res_many_x27_90d", 50.0, basis="measured 2.7-3.0 ms for 27 series; "
           "one grouped statement, 16x headroom"),
    Budget("paths", "history_many_x27_90d", 150.0, basis="measured ~30 ms for 27 series at the "
           "widest window; 5x headroom on the slowest request the pages can make"),
    # --- the same paths over HTTP, gate included -----------------------------------------
    Budget("http", "health", 150.0, basis="measured 0.8-1.4 ms median (task 01: 1,272 ms); "
           "task 01's own suggested ceiling, kept because a deployment disk is slower than "
           "the dev box"),
    Budget("http", "metrics_list", 150.0, basis="measured 1.6-2.0 ms median (task 01: 1,415 ms)"),
    Budget("http", "system_history_9x3_1h", 100.0, basis="the raw-resolution case: measured ~6 ms"),
    Budget("http", "system_history_9x3_90d", 200.0, basis="measured ~50 ms mean (task 01: 100 ms); "
           "4x the after, and under the baseline the phase started from"),
    Budget("http", "summary", 150.0, basis="measured ~1 ms; the overview's single call"),
    # --- the write side: the transaction that held the WAL write lock --------------------
    Budget("maintain", "steady", 1_000.0, statistic="seconds", basis="measured 2.7 ms steady "
           "(task 01: 5,368 ms); task 01's ceiling, 5x under the baseline's *mean* so a "
           "catch-up tick cannot cross it either"),
    Budget("maintain", "catch_up_10m", 2_000.0, statistic="seconds",
           basis="measured 48 ms after ten quiet minutes; the ceiling is the promise that a "
           "restart after a quiet spell is still well inside a poller's 5 s busy_timeout"),
    Budget("race", "poller_during_windowed_maintain", 1_000.0, statistic="max_wait_ms",
           basis="measured 21 ms worst wait (task 01: 5,005 ms median); the wait that matters "
           "is the worst one, since that is the one that raises"),
    Budget("race", "poller_during_windowed_maintain", 0.0, statistic="busy_errors", unit="count",
           basis="measured 0 of 100 (task 01: 5 of 9) — AGENTS.md known gap 4's failure mode, "
           "pinned at zero"),
)

#: Which budgets the *default* test run can price without the seeded store (see
#: `measure_budget_reads`): the cheap-store pass produces these sections and no others.
BUDGET_SECTIONS = ("paths", "http", "maintain", "race")

#: The keys the two cheap measurement passes produce, pinned by tests against the reports they
#: build. A budget (or the ratio partner of one) has to name a key from the right one of these,
#: because the failure mode of a budget table is a row that quietly stops being priced: rename the
#: measurement, and the ceiling still "passes".
READ_MEASURED_KEYS: tuple[str, ...] = (
    "series_names_host", "series_names_host_cached", "stats", "stats_cached", "history_90d",
    "pick_res_many_x27_90d", "history_many_x27_90d",
)
HTTP_MEASURED_KEYS: tuple[str, ...] = (
    "health", "metrics_list", "system_history_9x3_1h", "system_history_9x3_24h",
    "system_history_9x3_90d", "system_history_page_shape_1h", "summary",
)
#: The report keys the write-side passes produce, for the same reason. `full_window` is the
#: pre-fix control and deliberately has no budget: a thing is only a control if it is slow.
MAINTAIN_MEASURED_KEYS: tuple[str, ...] = tuple(label for label, _o, _r in MAINTAIN_PASSES)


@dataclass(frozen=True)
class BudgetRow:
    """One budget, the number it was priced against, and the verdict."""

    budget: Budget
    measured: float | None      # in the budget's unit; `None` = the report did not measure it
    ok: bool
    detail: str = ""

    @property
    def margin(self) -> float | None:
        """How many times inside the ceiling the measurement is (`None` when unpricables make that
        meaningless: a 0.0 ms measurement against a 10 ms ceiling has no finite margin)."""
        if self.measured is None or self.measured <= 0.0 or self.budget.limit <= 0.0:
            return None
        return self.budget.limit / self.measured

    def as_dict(self) -> dict[str, Any]:
        return {**self.budget.as_dict(), "measured": self.measured, "ok": self.ok,
                "margin": self.margin, "detail": self.detail}


def _budget_value(report: dict[str, Any], budget: Budget) -> float | None:
    """The measured value for one budget, in its unit, or `None` when the report lacks it."""
    section = report.get(budget.section) or {}
    if budget.section == "maintain":
        entry = (section.get("passes") or {}).get(budget.key)
        return None if entry is None else float(entry["seconds"]) * 1000.0
    entry = section.get(budget.key)
    if entry is None:
        return None
    if budget.section == "race":
        # a RaceResult: the fields are already ms / counts, no unit conversion
        return None if entry.get(budget.statistic) is None else float(entry[budget.statistic])
    value = entry.get(budget.statistic)
    return None if value is None else float(value) * 1000.0


def budget_rows(report: dict[str, Any],
                sections: tuple[str, ...] | None = None) -> list[BudgetRow]:
    """Price every pinned budget against `report` — the table the NOTES print and the tests assert.

    `sections` narrows the pass to the parts a given report actually contains (the cheap read-only
    pass prices `paths` alone, the HTTP pass `http` alone); without it a partial report would report
    every other budget as unmeasured, which is true but not what that caller is claiming.

    A budget whose path the report does not measure comes back as a failing row rather than a
    missing one: a budget that silently stops being checked is the same failure as a budget that
    is crossed, and it is the one a refactor causes.
    """
    rows: list[BudgetRow] = []
    for budget in BUDGETS:
        if sections is not None and budget.section not in sections:
            continue
        measured = _budget_value(report, budget)
        if measured is None:
            rows.append(BudgetRow(budget, None, False, "not measured by this report"))
            continue
        if measured > budget.limit:
            rows.append(BudgetRow(budget, measured, False,
                                  f"over the {fmt_budget(budget)} ceiling"))
            continue
        if budget.ratio_key is not None and budget.ratio_limit is not None:
            other = _budget_value(report, replace(budget, key=budget.ratio_key))
            if other is None:
                rows.append(BudgetRow(budget, measured, False,
                                      f"ratio partner {budget.ratio_key} was not measured"))
                continue
            ratio = (measured / other) if other > 0 else math.inf
            if ratio > budget.ratio_limit:
                rows.append(BudgetRow(budget, measured, False, f"{ratio:.3f} of the {budget.ratio_key} "
                                      f"cost, must be <= {budget.ratio_limit:.3f}"))
                continue
        rows.append(BudgetRow(budget, measured, True))
    return rows


def budget_violations(report: dict[str, Any],
                      sections: tuple[str, ...] | None = None) -> list[str]:
    """The failing rows as sentences — what a budget test asserts away."""
    return [f"{row.budget.label()} = {fmt_budget_value(row.budget, row.measured)} "
            f"(limit {fmt_budget(row.budget)}): {row.detail}"
            for row in budget_rows(report, sections) if not row.ok]


def missing_budget_keys(report: dict[str, Any],
                        sections: tuple[str, ...] | None = None) -> list[str]:
    """Budgets this report cannot price at all — the stale-pointer check, not the cost check.

    The default suite runs this against a *small* store, where a budget may legitimately be
    crossed (a ratio against a 20 us scan is luck) but must never be unmeasurable.
    """
    return [row.budget.label() for row in budget_rows(report, sections)
            if row.measured is None]


def fmt_budget_value(budget: Budget, value: float | None) -> str:
    if value is None:
        return "n/a"
    return _ms(value / 1000.0) if budget.unit == "ms" else f"{value:,.0f}"


def fmt_budget(budget: Budget) -> str:
    return _ms(budget.limit / 1000.0) if budget.unit == "ms" else f"<= {budget.limit:,.0f}"


def render_budgets_markdown(rows: list[BudgetRow]) -> str:
    """The budget table: measured, ceiling, margin, verdict — one row per pinned budget."""
    lines = ["| budget | path | measured | ceiling | margin | verdict |",
             "| --- | --- | --- | --- | --- | --- |"]
    for row in rows:
        margin = "—" if row.margin is None else f"{row.margin:,.0f}x"
        lines.append(f"| `{row.budget.section}` | `{row.budget.key}` "
                     f"({row.budget.statistic}) | {fmt_budget_value(row.budget, row.measured)} | "
                     f"{fmt_budget(row.budget)} | {margin} | "
                     + ("**inside**" if row.ok else f"**OVER** — {row.detail}") + " |")
    return "\n".join(lines)


def measure_budget_reads(store: SampleStore, *, runs: int = 5) -> dict[str, Any]:
    """Only the paths with a pinned budget, timed — the cheap half of `measure()`.

    Task 04's budget tests must not each pay for the races (minutes) or the write passes, so this
    produces the same-shaped `paths` document for the budget keys and nothing else. It is a
    *measurement*, not a baseline: the numbers it prints go in the notes next to the full probe
    report, and both are on the same seeded file.
    """
    host, metric = "driveripper", "cpu.usage_pct"
    hosts = list(HOST_SERIES)
    metrics = list(HISTORY_METRICS)
    paths: dict[str, Any] = {}
    paths["series_names_host"] = timed(lambda: (store.invalidate_read_caches(),
                                                store.series_names("host"))[1], runs=runs)
    paths["series_names_host_cached"] = timed(lambda: store.series_names("host"), runs=runs)
    paths["stats"] = timed(lambda: (store.invalidate_read_caches(), store.stats())[1], runs=runs)
    paths["stats_cached"] = timed(store.stats, runs=runs)
    paths["history_90d"] = timed(
        lambda: store.history("host", host, metric, span=90 * 24 * 3600.0, max_points=600),
        runs=runs)
    paths["pick_res_many_x27_90d"] = timed(
        lambda: store.pick_res_many("host", hosts, metrics, 90 * 24 * 3600.0, 600), runs=runs)
    paths["history_many_x27_90d"] = timed(
        lambda: store.history_many("host", hosts, metrics, span=90 * 24 * 3600.0, max_points=600),
        runs=runs)
    return {"budget_runs": runs, "paths": paths}


def measure_budget_http(path: str | Path, *, hosts: tuple[str, ...] | None = None,
                        metrics: tuple[str, ...] | None = None) -> dict[str, Any]:
    """The HTTP half of the budgets, through the same signed-in `TestClient` as `measure()`.

    Split out so `tests/test_api.py` can price the endpoint budgets without also paying for the
    races and the write passes. `hosts`/`metrics` default to the seeded shape so the request is
    the one `/system` makes; a caller passing them explicitly is pricing a *different* request and
    should say so in the notes rather than reuse these budgets.
    """
    return {"http": measure_http(Path(path),
                                 hosts=hosts or tuple(sorted(HOST_SERIES)),
                                 metrics=metrics or HISTORY_METRICS)}


def measure_dev_db(path: str | Path, *, runs: int = 3) -> dict[str, Any]:
    """The same three hot reads against a *real* store, opened read-only.

    The answer to "is your synthetic seed representative?" is better as a measurement than as an
    argument. `mode=ro` and nothing else: no DDL, no PRAGMA that writes, no `SampleStore`
    (constructing one runs the schema and `journal_mode=WAL`, which need write access). Tests
    never pass this path — `AGENTS.md`'s testing policy keeps the suite off the operator's
    history — but the probe is run by a human, exactly like `data/logs/analyze-watch.py` did in
    phase 13, and the numbers it prints belong in the NOTES either way.
    """
    uri = f"file:{Path(path)}?mode=ro"
    db = sqlite3.connect(uri, uri=True)
    out: dict[str, Any] = {"path": str(path), "bytes": Path(path).stat().st_size}
    try:
        out["counts"] = {row[0]: int(row[1]) for row in db.execute(
            "SELECT res, COUNT(*) FROM sample GROUP BY res")}
        out["events"] = int(db.execute("SELECT COUNT(*) FROM event").fetchone()[0])
        out["series"] = {row[0]: int(row[1]) for row in db.execute(
            "SELECT source, COUNT(*) FROM (SELECT DISTINCT source, host, metric FROM sample)"
            " GROUP BY source")}

        def run(sql: str, args: tuple[Any, ...] = ()) -> dict[str, float]:
            def call() -> None:
                db.execute(sql, args).fetchall()

            return timed(call, runs=runs)

        out["paths"] = {
            "series_names_all": run("SELECT DISTINCT metric FROM sample ORDER BY metric"),
            "series_names_host": run("SELECT DISTINCT metric FROM sample WHERE source=?"
                                     " ORDER BY metric", ("host",)),
            "stats_by_res": run("SELECT res, COUNT(*) FROM sample GROUP BY res"),
            "stats_events": run("SELECT COUNT(*) FROM event"),
        }
        out["plans"] = {key: [str(row[3]) for row in db.execute(
            f"EXPLAIN QUERY PLAN {sql}", args).fetchall()]
            for key, sql, args in (
                ("series_names_all", "SELECT DISTINCT metric FROM sample ORDER BY metric", ()),
                ("stats_by_res", "SELECT res, COUNT(*) FROM sample GROUP BY res", ()),
            )}
    finally:
        db.close()
    return out


# --------------------------------------------------------------------------- #
# rendering
# --------------------------------------------------------------------------- #
def _ms(seconds: float) -> str:
    if seconds >= 1.0:
        return f"{seconds * 1000:,.0f} ms"
    return f"{seconds * 1000:.1f} ms"


def _mb(value: float) -> str:
    return f"{value / 1e6:,.1f} MB"


def render_markdown(report: dict[str, Any]) -> str:
    """The NOTES tables, as markdown, from one report.

    Machine-written so the "before" and the "after" of this phase are the same tables with
    different numbers, which is the only form of the comparison that survives a week.
    """
    machine = report["machine"]
    store = report["store"]
    counts_ = store["counts"]
    lines: list[str] = []
    push = lines.append
    push(f"- seed version: `{report['seed_version']}` · generated "
         f"{time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(report['generated_at']))}")
    push(f"- cpu: {machine['cpu']} ({machine['cores']} cores, {machine['machine']}, "
         f"kernel {machine['kernel']})")
    push(f"- python {machine['python']} · sqlite {machine['sqlite']}")
    push(f"- store directory: `{machine['store_dir']}` "
         f"({'tmpfs — RAM, the faster machine' if machine['store_dir_on_tmpfs'] else 'a real block device'})")
    push("")
    push("| seeded store | rows | bytes |")
    push("| --- | --- | --- |")
    push(f"| `{Path(store['path']).name}` | raw {counts_['raw']:,} · 1m {counts_['1m']:,} · "
         f"1h {counts_['1h']:,} · events {counts_['events']:,} · "
         f"{counts_['series']} series | {_mb(store['bytes'])} (+{_mb(store['wal_bytes'])} WAL) |")
    push("")
    push("| hot path | mean | median | max |")
    push("| --- | --- | --- | --- |")
    labels = {
        "series_names_all": "`series_names()` — no filter, **cold**",
        "series_names_all_cached": "`series_names()` — cached",
        "series_names_host": "`series_names('host')` — `/api/metrics?source=host`, **cold**",
        "series_names_host_cached": "`series_names('host')` — cached",
        "series_names_host_driveripper": "`series_names('host','driveripper')`",
        "stats": "`stats()` — `/api/health`, **cold**",
        "stats_cached": "`stats()` — cached (what the next `/api/health` poll pays)",
        "pick_res_90d": "`pick_res()` x 3 (1 h, 24 h, 90 d) — one series",
        "pick_res_x27_24h": "`pick_res()` x 27, one call per series — 24 h",
        "pick_res_many_x27_24h": "`pick_res_many()` — the same 27 in one grouped call — 24 h",
        "pick_res_x27_90d": "`pick_res()` x 27, one call per series — 90 d",
        "pick_res_many_x27_90d": "`pick_res_many()` — the same 27 in one grouped call — 90 d",
        "history_x27_24h": "`history()` x 27, one call per series — 24 h",
        "history_many_x27_24h": "`history_many()` — the same 27 batched — 24 h",
        "history_x27_90d": "`history()` x 27, one call per series — 90 d",
        "history_many_x27_90d": "`history_many()` — the same 27 batched — 90 d",
        "history_5m": "`history()` — 5 m window",
        "history_24h": "`history()` — 24 h window",
        "history_90d": "`history()` — 90 d window",
    }
    for key, label in labels.items():
        entry = report["paths"].get(key)
        if entry is None:
            continue
        push(f"| {label} | {_ms(entry['mean'])} | {_ms(entry['median'])} | {_ms(entry['max'])} |")
    maintain = report["maintain"]
    for label, entry in maintain["passes"].items():
        push(f"| `maintain()` — {label} | {_ms(entry['seconds'])} | — | — |")
    for key, entry in report.get("http", {}).items():
        push(f"| HTTP `{key}` | {_ms(entry['mean'])} | {_ms(entry['median'])} | "
             f"{_ms(entry['max'])} |")
    push("")
    push("`maintain()` passes on a scratch copy of the seeded store — same rows, same device, and "
         "the writes kept off the file every later measurement compares against:")
    push("")
    push("| pass | write transaction | rows written | WAL before → after |")
    push("| --- | --- | --- | --- |")
    for label, entry in maintain["passes"].items():
        counts_ = entry["counts"]
        push(f"| `{label}` | {_ms(entry['seconds'])} | {entry['rows_written']:,} "
             f"(1m {counts_['to_minute']:,} · 1h {counts_['to_hour']:,}) | "
             f"{_mb(entry['wal_before_bytes'])} → {_mb(entry['wal_after_bytes'])} |")
    cycles = maintain["two_steady_cycles"]
    push("")
    push("Two more steady ticks, WAL after each: "
         + " → ".join(_mb(cycle["wal_after_bytes"]) for cycle in cycles)
         + f", against the {_mb(maintain['journal_size_limit_bytes'])} `journal_size_limit` cap: "
         + ("inside it" if maintain["wal_within_cap"] else "OVER IT")
         + f". Scratch file {_mb(maintain['scratch_bytes'])} → "
         f"{_mb(maintain['store_bytes_after'])} (the main file does not move; the WAL is the "
         "number that grows between checkpoints).")
    push("")
    push("| race | inserts | waited > "
         f"{WAIT_THRESHOLD_MS:.0f} ms | median | max | `database is locked` |")
    push("| --- | --- | --- | --- | --- | --- |")
    for key, label in (("single_connection", "one connection, `stats()`+`series_names()` looping"),
                       ("two_connections_prod_timeout",
                        f"two connections, windowed `maintain()` looping, "
                        f"busy_timeout={BUSY_TIMEOUT_MS}"),
                       ("two_connections_no_timeout",
                        "two connections, windowed `maintain()` looping, busy_timeout=0"),
                       ("poller_during_windowed_maintain",
                        f"two connections, windowed `maintain()` at a "
                        f"{RACE_MAINTAIN_GAP_SECONDS:.1f} s cadence, "
                        f"busy_timeout={BUSY_TIMEOUT_MS} — **the pass line**"),
                       ("poller_during_full_window_maintain",
                        "CONTROL: two connections, whole-window (pre-fix) `maintain()` at the same "
                        f"cadence, busy_timeout={BUSY_TIMEOUT_MS}")):
        entry = report["race"].get(key)
        if entry is None:
            continue
        push(f"| {label} | {entry['inserts']} of {entry['attempted']} attempted "
             f"(target {entry['target']}, {entry['seconds']:.1f} s) | {entry['waiters']} | "
             f"{entry['median_wait_ms']:,.1f} ms | {entry['max_wait_ms']:,.0f} ms | "
             f"{entry['busy_errors']} |")
        for note in entry.get("notes", []):
            push(f"| ↳ | {note} | | | | |")
    push("")
    rows = budget_rows(report)
    if any(row.measured is not None for row in rows):
        push("The task-04 budgets, priced against the run above. Each ceiling is 2-3x the measured "
             "\"after\" (and the `basis` column of `store_perf.BUDGETS` says which figure and how "
             "much margin), so this table is the regression gate: a path that drifts back toward "
             "its baseline turns red here instead of just feeling slow.")
        push("")
        push(render_budgets_markdown(rows))
        push("")
        verdict = "all inside" if all(row.ok for row in rows) else "OVER"
        push(f"Budget verdict: **{verdict}** "
             f"({sum(1 for row in rows if row.ok)} of {len(rows)} inside).")
        push("")
    push("EXPLAIN QUERY PLAN shapes (read paths):")
    push("")
    for key, plan_rows in report["plans"].items():
        push(f"- `{key}`: " + " · ".join(f"`{row}`" for row in plan_rows))
    push("")
    push("EXPLAIN QUERY PLAN shapes (`maintain()`, on the copy, after its own statistics landed):")
    push("")
    for key, plan_rows in maintain.get("plans", {}).items():
        push(f"- `{key}`: " + " · ".join(f"`{row}`" for row in plan_rows))
    push("")
    dev = report.get("dev_db")
    if dev:
        dev_counts = dev["counts"]
        dev_total = sum(dev_counts.values())
        dev_series = sum(dev["series"].values())
        dev_by_res = " ".join(f"{key}={value:,}" for key, value in sorted(dev_counts.items()))
        push(f"The same reads against the operator's **real** store "
             f"(`{dev['path']}`, {_mb(dev['bytes'])}, opened read-only, no test ever does this):")
        push("")
        push("| read | what is in that store | mean |")
        push("| --- | --- | --- |")
        push(f"| `series_names()` | {dev_total:,} samples · {dev_series:,} distinct series · "
             f"{dev['events']:,} events | {_ms(dev['paths']['series_names_all']['mean'])} |")
        host_series = dev["series"].get("host", 0)
        push(f"| `series_names('host')` | {host_series:,} host series | "
             f"{_ms(dev['paths']['series_names_host']['mean'])} |")
        push(f"| `stats()` counts | `{dev_by_res}` | "
             f"{_ms(dev['paths']['stats_by_res']['mean'])} + "
             f"{_ms(dev['paths']['stats_events']['mean'])} |")
        push("")
        for key, plan_rows in dev["plans"].items():
            push(f"- `{key}`: " + " · ".join(f"`{row}`" for row in plan_rows))
        push("")
    integrity = report.get("store_after", {}).get("integrity", "?")
    push(f"Seeded shape after the measurements: `{integrity}` "
         f"(raw {report['store_after']['counts']['raw']:,}).")
    return "\n".join(lines)


def write_report(report: dict[str, Any], out_prefix: Path) -> dict[str, str]:
    """Write `<prefix>.json` and `<prefix>.md` so the numbers survive the terminal."""
    out_prefix.parent.mkdir(parents=True, exist_ok=True)
    json_path = out_prefix.with_suffix(".json")
    md_path = out_prefix.with_suffix(".md")
    json_path.write_text(json.dumps(report, indent=2, sort_keys=True) + "\n")
    md_path.write_text(render_markdown(report) + "\n")
    return {"json": str(json_path), "md": str(md_path)}