dashboards/tests/test_store.py

"""History store: rollups, retention, and the honesty of an empty chart.

The rollup queries are plain SQL with positional bindings, and a mismatch there used to
raise on every maintenance tick while the poller swallowed it — so every window past the
raw retention was silently empty. These tests fail loudly if that ever happens again.
"""
from __future__ import annotations

import logging
import math
import re
import sqlite3
import time
from collections.abc import Iterator
from contextlib import contextmanager
from dataclasses import replace
from pathlib import Path
from typing import Any

import pytest

import store_perf
from dashboards.store import (
    COMPOUND_CHUNK,
    HOUR,
    JOURNAL_SIZE_LIMIT_BYTES,
    LAST_MAINTAIN_KEY,
    MAINTAIN_SLACK_SECONDS,
    MINUTE,
    RES_HOUR,
    RES_MIN,
    RES_RAW,
    ROLLUP_INSERT,
    SERIES_CACHE_MAX_ENTRIES,
    WAL_AUTOCHECKPOINT_PAGES,
    Event,
    Sample,
    SampleStore,
    samples_from_flat,
    select_window,
)


def add_series(db, source: str, host: str, metric: str, values, *, start: float, step: float,
               res: str = RES_RAW) -> None:
    for i, value in enumerate(values):
        # the private handle is the point: writing at an explicit resolution is what
        # the resolution-choice tests need
        db._db.execute(
            "INSERT INTO sample (source, host, metric, res, ts, value) VALUES (?,?,?,?,?,?)",
            (source, host, metric, res, start + i * step, float(value)),
        )
    db._db.commit()


def test_store_starts_in_wal_mode(db) -> None:
    """Readers poll while the writer commits; without WAL the API would hit locked errors."""
    assert db._db.execute("PRAGMA journal_mode").fetchone()[0].lower() == "wal"


def test_add_and_query_roundtrip(db) -> None:
    now = time.time()
    db.add("host", "driveripper", "cpu.usage_pct", 42.5, ts=now)
    points = db.query("host", "driveripper", "cpu.usage_pct", start=now - 60)
    assert points == [(now, 42.5)]


def test_query_is_bounded_and_ascending(db) -> None:
    base = 1_700_000_000.0
    add_series(db, "host", "h", "m", [3, 1, 2], start=base, step=10)
    assert db.query("host", "h", "m", start=base - 1) == [(base, 3.0), (base + 10, 1.0), (base + 20, 2.0)]
    assert len(db.query("host", "h", "m", start=base - 1, limit=2)) == 2


def test_query_tail_keeps_the_newest_rows(db) -> None:
    base = 1_700_000_000.0
    add_series(db, "host", "h", "m", range(100), start=base, step=10)
    head = db.query("host", "h", "m", start=base - 1, limit=3)
    tail = db.query("host", "h", "m", start=base - 1, limit=3, tail=True)
    assert [v for _, v in head] == [0.0, 1.0, 2.0]
    assert [v for _, v in tail] == [97.0, 98.0, 99.0]
    assert tail == sorted(tail), "tail must still come back oldest-first"


def test_samples_from_flat_drops_unknown_and_none() -> None:
    payload: dict[str, Any] = {"a": 1.0, "b": None, "c": "nope"}
    samples = samples_from_flat("host", "gw", payload, ts=5.0)
    assert [(s.metric, s.value) for s in samples] == [("a", 1.0)]


def test_maintain_rolls_raw_into_minutes(db) -> None:
    """Regression: the rollup statement had 7 placeholders and 6 bindings."""
    now = time.time()
    # two full, aligned minute buckets — see the hour test for why alignment matters
    minute = int(now // MINUTE) * MINUTE
    add_series(db, "host", "h", "cpu.usage_pct", [10, 20, 30, 40], start=minute - 4 * MINUTE,
               step=1)
    add_series(db, "host", "h", "cpu.usage_pct", [1, 1, 1, 1], start=minute - 3 * MINUTE, step=1)
    counts = db.maintain(now=now)
    assert counts["to_minute"] > 0
    minutes = db.query("host", "h", "cpu.usage_pct", res=RES_MIN, start=now - 3600)
    assert minutes, "no 1m rollup was written"
    assert [round(value) for _ts, value in minutes] == [25, 1]


def test_maintain_is_idempotent(db) -> None:
    """Rollups run on a timer: running twice must not double-count or duplicate rows.

    The second pass is asked for the whole window again (the `last_maintain` row is deleted
    first), because since phase 14 task 03 a pass that has recorded its own clock has nothing
    left to do — the window is where the work got bounded, and this test is about the rollup being
    an `INSERT OR REPLACE` rewrite rather than a duplicate row. `test_a_second_maintain_reads_
    only_the_window_since_the_last_run` is the other half: the pass that does run again at a later
    clock re-reads only what became due since.
    """
    now = time.time()
    minute = int(now // MINUTE) * MINUTE
    add_series(db, "host", "h", "m", [10, 20], start=minute - 4 * MINUTE, step=1)
    first = db.maintain(now=now)
    rows = db.query("host", "h", "m", res=RES_MIN, start=now - 3600)
    count = db.count()
    assert db.delete_meta("last_maintain") == 1, "the whole-window re-run needs no previous pass"
    second = db.maintain(now=now)
    assert rows == db.query("host", "h", "m", res=RES_MIN, start=now - 3600)
    assert first["to_minute"] == second["to_minute"] > 0
    assert db.count() == count, "re-running maintenance duplicated samples"


def test_maintain_rolls_minutes_into_hours(db) -> None:
    now = time.time()
    # buckets are aligned to the epoch (`ts/3600*3600`), not to "hours from now", so the
    # fixtures start on a boundary or the two hours mix into a third, averaged bucket
    hour = int(now // HOUR) * HOUR
    # two *complete* hours of minute buckets: the current and previous hour are still
    # filling in, and maintenance must not touch them (see the partial-bucket test)
    add_series(db, "host", "h", "m", [1] * 60, start=hour - 5 * HOUR, step=60, res=RES_MIN)
    add_series(db, "host", "h", "m", [3] * 60, start=hour - 4 * HOUR, step=60, res=RES_MIN)
    counts = db.maintain(now=now)
    assert counts["to_hour"] > 0
    hours = db.query("host", "h", "m", res=RES_HOUR, start=now - 86_400)
    assert sorted(round(v) for _, v in hours) == [1.0, 3.0]



def test_maintain_prunes_expired_raw(db) -> None:
    now = time.time()
    old = now - (db.raw_keep_seconds + 7200)
    add_series(db, "host", "h", "m", [1, 2], start=old, step=60)
    add_series(db, "host", "h", "fresh", [1], start=now - 60, step=1)
    counts = db.maintain(now=now)
    assert counts["pruned"] >= 2
    assert db.query("host", "h", "m", start=0) == []


def test_maintain_records_timestamp_in_stats(db) -> None:
    db.maintain()
    stats = db.stats()
    assert stats["last_maintain"] is not None
    assert "samples_by_res" in stats and "path" in stats


def test_pick_res_prefers_the_finest_resolution_that_fits(db) -> None:
    now = time.time()
    add_series(db, "host", "h", "m", list(range(500)), start=now - 3600, step=1)
    assert db.pick_res("host", "h", "m", 3600, 600) == RES_RAW
    add_series(db, "host", "h", "m", list(range(4000)), start=now - 7200, step=1)
    assert db.pick_res("host", "h", "m", 7200, 600) == RES_RAW, "raw still fits"
    add_series(db, "host", "h", "m", list(range(120)), start=now - 7200, step=60,
               res=RES_MIN)
    # 4500 raw points blow the 600 budget, so the minute rollup takes over
    assert db.pick_res("host", "h", "m", 7200, 600) == RES_MIN


def test_pick_res_never_picks_an_empty_resolution(db) -> None:
    """A fresh install has raw only: choosing 1h because it 'fits' would chart nothing."""
    now = time.time()
    add_series(db, "host", "h", "m", [1] * 2000, start=now - 7200, step=1)
    assert db.pick_res("host", "h", "m", 86_400, 600) == RES_RAW


def test_history_falls_back_to_raw_when_rollups_lag(db) -> None:
    now = time.time()
    add_series(db, "host", "h", "m", list(range(1500)), start=now - 1500, step=1)
    points = db.history("host", "h", "m", span=3600, max_points=600)
    assert len(points) > 0, "empty chart while raw data exists"
    assert len(points) <= 600


def test_history_always_keeps_the_newest_point(db) -> None:
    now = time.time()
    values = list(range(2000))
    add_series(db, "host", "h", "m", values, start=now - 2000, step=1)
    points = db.history("host", "h", "m", span=3600, max_points=50)
    assert points[-1][1] == pytest.approx(values[-1])


def test_series_names_and_hosts(db) -> None:
    db.add("host", "a", "cpu.usage_pct", 1.0)
    db.add("host", "a", "mem.used_pct", 2.0)
    db.add("host", "b", "cpu.usage_pct", 3.0)
    db.add("llm", "gufo", "llm.decode_tokens_per_second", 4.0)
    assert db.series_names("host", "a") == ["cpu.usage_pct", "mem.used_pct"]
    assert db.hosts("host") == ["a", "b"]
    assert db.series_names("llm") == ["llm.decode_tokens_per_second"]


def test_delete_series_cleans_up(db) -> None:
    db.add("host", "a", "gone.metric", 1.0)
    db.add("host", "a", "keep.metric", 2.0)
    assert db.delete_series("host", "a", "gone.metric") == 1
    assert db.series_names("host", "a") == ["keep.metric"]


def test_events_roundtrip_and_filter(db) -> None:
    now = time.time()
    db.log_event(Event(ts=now, source="host", host="a", kind="down", severity="error",
                       message="a is not answering", data={"fails": 3}))
    db.log_event(Event(ts=now + 1, source="nut", host="cyberpower", kind="state",
                       severity="warn", message="OL → OB"))
    everything = db.recent_events(10)
    assert [e.host for e in everything] == ["cyberpower", "a"]
    assert everything[-1].data == {"fails": 3}
    assert [e.source for e in db.recent_events(10, source="nut")] == ["nut"]
    assert len(db.recent_events(1)) == 1
    # the API hands out plain JSON, so the boundary has to be serialisable
    assert everything[0].as_json()["message"] == "OL → OB"


def test_event_as_json_is_serialisable(db) -> None:
    event = Event(ts=1.0, source="host", host="a", kind="up", severity="info", message="ok")
    payload = event.as_json()
    assert payload["severity"] == "info" and payload["data"] is None


def test_count_filters_by_resolution(db) -> None:
    now = time.time()
    db.add("host", "h", "m", 1.0, ts=now)
    add_series(db, "host", "h", "m", [2.0], start=now, step=1, res=RES_MIN)
    assert db.count() == 2
    assert db.count(res=RES_RAW) == 1
    assert db.count(res=RES_HOUR) == 0


def test_add_many_reports_inserted_rows(db) -> None:
    now = time.time()
    rows = [Sample("host", "h", f"m{i}", float(i), now) for i in range(5)]
    assert db.add_many(rows) == 5
    assert db.count() == 5


def test_store_reopens_without_losing_history(tmp_path) -> None:
    path = tmp_path / "reopen.sqlite3"
    first = SampleStore(path)
    now = time.time()
    first.add("host", "h", "m", 7.0, ts=now)
    first.close()
    second = SampleStore(path)
    try:
        assert second.query("host", "h", "m", start=now - 5) == [(now, 7.0)]
    finally:
        second.close()


def test_select_window_maps_range_onto_presets() -> None:
    assert select_window(60, [300, 3600, 86_400]) == 300
    assert select_window(3600, [300, 3600, 86_400]) == 3600
    assert select_window(86_401, [300, 3600, 86_400]) == 86_400


def test_rollup_does_not_average_a_partial_minute(db) -> None:
    """A bucket is only rolled once its whole window is in the past, or it gets rewritten."""
    now = time.time()
    add_series(db, "host", "h", "m", [1, 1, 1], start=now - 5, step=1)  # current minute
    db.maintain(now=now)
    assert db.query("host", "h", "m", res=RES_MIN, start=now - 600) == []


def test_hour_constant_is_used_in_retention(db) -> None:
    assert db.raw_keep_seconds == pytest.approx(48 * HOUR)


# ---- a 300-second source (phase 09: `disks` samples SMART every 5 minutes) -- #
#
# The rollup SQL groups by minute and hour buckets, so a source slower than a minute
# puts one raw row in every fifth bucket and nothing in the other four. Two things had to
# be true and are asserted rather than assumed: an empty bucket produces *no* rollup row
# (inventing a value for a minute nothing was sampled is rule 2), and the hour bucket
# still averages the samples it did get.

SMART_STEP = 300.0


def test_a_300s_source_rolls_into_minutes_without_inventing_the_empty_ones(db) -> None:
    now = time.time()
    # five SMART polls at exactly the phase's cadence (-30, -25, -20, -15, -10 minutes),
    # so every bucket is fully in the past (+60 s slack) and maintenance may roll all five
    start = int(now // MINUTE) * MINUTE - 30 * MINUTE
    add_series(db, "disks", "driveripper", "disk.temp_c@nvme0n1",
               [38, 39, 38, 40, 37], start=start, step=SMART_STEP)
    counts = db.maintain(now=now)
    minutes = db.query("disks", "driveripper", "disk.temp_c@nvme0n1", res=RES_MIN, start=0)
    assert counts["to_minute"] == 5
    assert [value for _ts, value in minutes] == [38.0, 39.0, 38.0, 40.0, 37.0], (
        "a bucket holding one sample rolls up that sample, unchanged")
    assert [int(ts) for ts, _value in minutes] == [
        int(start + index * SMART_STEP // MINUTE * MINUTE) for index in range(5)], (
        "every fifth minute bucket has a row; the empty ones must have none")
    assert db.count(res=RES_MIN) == 5, "empty buckets became fabricated rows"


def test_a_300s_source_averages_into_the_hour_bucket_it_lands_in(db) -> None:
    now = time.time()
    # aligned to the epoch, and old enough that the hour bucket is complete: 12 samples
    # five minutes apart inside one hour average to their mean, not to a sample count
    hour = int(now // HOUR) * HOUR - 3 * HOUR
    add_series(db, "disks", "driveripper", "disk.power_on_hours@nvme0n1",
               [100, 100, 100, 100, 100, 100, 100, 100, 100, 100, 100, 124],
               start=hour, step=SMART_STEP)
    db.maintain(now=now)
    db.maintain(now=now)  # the 1h pass reads 1m rows, so the first pass only makes them
    hours = db.query("disks", "driveripper", "disk.power_on_hours@nvme0n1", res=RES_HOUR,
                     start=0)
    assert len(hours) == 1
    assert hours[0][0] == hour
    assert hours[0][1] == pytest.approx((11 * 100 + 124) / 12)


def test_a_300s_series_is_queryable_at_every_resolution(db) -> None:
    """A 5-minute series over a 1h window is 12 points; `pick_res` must find them raw."""
    now = time.time()
    add_series(db, "disks", "driveripper", "disk.reallocated@nvme0n1",
               [8] * 12, start=now - 3600 + 60, step=SMART_STEP)
    assert db.pick_res("disks", "driveripper", "disk.reallocated@nvme0n1", 3600, 600) == RES_RAW
    points = db.history("disks", "driveripper", "disk.reallocated@nvme0n1", span=3600)
    assert len(points) == 12, "the chart is empty while raw history exists"
    db.maintain(now=now)
    db.maintain(now=now)
    # with rollups present, a long window reads the rolled-up series, not 288 raw rows
    assert db.pick_res("disks", "driveripper", "disk.reallocated@nvme0n1",
                       24 * 3600, 600) in (RES_MIN, RES_RAW)


def test_a_300s_source_is_pruned_by_the_existing_retention(db) -> None:
    """288 rows a day per series is fine for 48h and not fine forever — rule 9."""
    now = time.time()
    expired = now - db.raw_keep_seconds - 6 * SMART_STEP
    add_series(db, "disks", "h", "disk.temp_c@sda", [30] * 4, start=expired, step=SMART_STEP)
    add_series(db, "disks", "h", "disk.temp_c@sdb", [30], start=now - SMART_STEP,
               step=SMART_STEP)
    counts = db.maintain(now=now)
    assert counts["pruned"] >= 4
    assert db.query("disks", "h", "disk.temp_c@sda", start=0) == []
    assert len(db.query("disks", "h", "disk.temp_c@sdb", start=0)) == 1


# ---- meta: the durable key/value state (phase 11) --------------------------- #


def test_get_meta_answers_none_for_a_key_nobody_wrote(db) -> None:
    """`None` and `""` are different sentences here.

    `auth.verifier` missing means "this box has never been seeded" (fail closed at startup),
    while a value means "the credential lives here". A store that returned `""` for an absent
    row would read as a corrupt one instead.
    """
    assert db.get_meta("no.such.key") is None
    db.set_meta("empty.on.purpose", "")
    assert db.get_meta("empty.on.purpose") == ""


def test_meta_round_trips_through_a_reopen(db) -> None:
    db.set_meta("auth.session_secret", "ab" * 32)
    assert db.get_meta("auth.session_secret") == "ab" * 32
    db.close()
    reopened = SampleStore(db.path)
    try:
        assert reopened.get_meta("auth.session_secret") == "ab" * 32
    finally:
        reopened.close()


def test_set_meta_replaces_instead_of_adding(db) -> None:
    """One row per key (`INSERT OR REPLACE`), because `key` is the primary key.

    An `INSERT` alone would raise on the second write, which is the change-password path
    (`auth.update_verifier`) failing after the user pressed Save.
    """
    db.set_meta("auth.verifier", "first")
    db.set_meta("auth.verifier", "second")
    assert db.get_meta("auth.verifier") == "second"
    (rows,) = db._db.execute(  # the private handle is the point: exactly one row, not two
        "SELECT COUNT(*) FROM meta WHERE key='auth.verifier'").fetchone()
    assert rows == 1


def test_delete_meta_reports_whether_a_row_was_there(db) -> None:
    """The rowcount is the whole point: the password-reset procedure has to say what it did."""
    db.set_meta("auth.verifier", "x")
    assert db.delete_meta("auth.verifier") == 1
    assert db.delete_meta("auth.verifier") == 0
    assert db.get_meta("auth.verifier") is None


def test_delete_meta_touches_only_the_key_it_names(db) -> None:
    """A reset that deletes one row too many deletes the operator's only credential."""
    db.set_meta("auth.verifier", "the-verifier")
    db.set_meta("auth.session_secret", "the-secret")
    db.set_meta("last_maintain", "1700000000.0")

    assert db.delete_meta("auth.session_secret") == 1

    assert db.get_meta("auth.session_secret") is None
    assert db.get_meta("auth.verifier") == "the-verifier"
    assert db.get_meta("last_maintain") == "1700000000.0"


def test_stats_reads_last_maintain_through_get_meta(db) -> None:
    """One path to the table.

    `stats()` reported `None` before any maintenance ran and the timestamp afterwards; that
    behaviour is pinned here while the read goes through the same helper everything else uses,
    and it is why the helper is called outside `stats()`'s own lock (`threading.Lock` is not
    reentrant — call it inside and the dashboard deadlocks on the first `/api/health`).
    """
    assert db.stats()["last_maintain"] is None
    db.maintain(now=1_700_000_000.0)
    assert db.stats()["last_maintain"] == 1_700_000_000.0


# ---- phase 14 task 02: nothing on a per-request path may scan the table ---------------- #
#
# `TODO.md:7` is "pages take forever to load and seem to lock up the database", and the half of
# it a browser feels was two whole-table answers served on every request (`series_names`,
# `stats`) plus three `COUNT`s per series to pick a chart's resolution. The pins below assert the
# two things that make a cache trustworthy rather than fast: it is *invalidated* by the write that
# would make it a lie, and it *expires* so a missed invalidation is a 60-second bug rather than a
# permanent one.
#
# They count SQL statements rather than wall time. `sqlite3`'s trace callback is the query counter
# the task asks for, and it is the right instrument: "the second call did not scan" is a fact
# about statements, and a wall-time assertion here would be a budget that a noisy laptop fails for
# the wrong reason (task 04 owns the budgets).


@contextmanager
def query_log(store: SampleStore) -> Iterator[list[str]]:
    """Every statement the store executes on this handle while the context is open, in order."""
    seen: list[str] = []
    store._db.set_trace_callback(seen.append)
    try:
        yield seen
    finally:
        store._db.set_trace_callback(None)


def plan_of(store: SampleStore, sql: str, args: tuple[Any, ...] = ()) -> str:
    """`EXPLAIN QUERY PLAN` as one line, so a test can say "seek, not scan" in words."""
    with store._lock:
        return " | ".join(row[3] for row in store._db.execute("EXPLAIN QUERY PLAN " + sql, args))


def test_series_names_answers_from_cache_and_a_write_invalidates_it(db) -> None:
    """The `/api/metrics` answer is cached — and a write ends the cache, not the clock.

    A cache that outlives a write answers "why is this chart empty" with yesterday's truth, so
    the invalidation is the load-bearing half: a new metric name must be visible on the very next
    call, and a deleted one must be gone.
    """
    now = time.time()
    db.add("host", "driveripper", "cpu.usage_pct", 1.0, ts=now)
    with query_log(db) as seen:
        first = db.series_names("host")
        assert [s for s in seen if "DISTINCT metric" in s], "the first call pays the scan once"

        seen.clear()
        assert db.series_names("host") == first
        assert seen == [], "a cached series_names() must not touch the table at all"

        db.add("host", "driveripper", "gpu.busy_pct@0", 2.0, ts=now + 1)
        seen.clear()
        assert db.series_names("host") == ["cpu.usage_pct", "gpu.busy_pct@0"], (
            "a metric that appeared after the cached read has to be in the next answer")
        assert [s for s in seen if "DISTINCT metric" in s], "the write must have invalidated it"


def test_every_write_that_can_change_the_series_list_invalidates_it(db) -> None:
    """`add`, `add_many` and `delete_series` — the three writes that move the name set."""
    db.add("host", "a", "cpu.usage_pct", 1.0)
    assert db.series_names("host") == ["cpu.usage_pct"]
    db.add_many([Sample("host", "a", "mem.used_pct", 2.0, time.time())])
    assert "mem.used_pct" in db.series_names("host")
    assert db.delete_series("host", "a", "cpu.usage_pct") == 1
    assert db.series_names("host") == ["mem.used_pct"], "a deleted series must not linger in the list"


def test_maintaining_the_store_invalidates_the_series_list(db) -> None:
    """Pruning can delete the last row of a name; the list must not promise it afterwards."""
    old = time.time() - (db.raw_keep_seconds + 7200)
    add_series(db, "host", "h", "gone.soon", [1, 2], start=old, step=60)
    assert db.series_names("host") == ["gone.soon"]
    db.maintain(now=time.time())
    assert db.series_names("host") == []


def test_a_write_of_an_already_written_series_leaves_the_series_cache_warm(db) -> None:
    """The live defect the task-04 watch caught, pinned so it cannot come back quietly.

    Task 02 invalidated the series cache on *every* sample write. On a store the hub writes to
    every 2-300 s that meant the cache never served: `/api/metrics` on the running dev server was
    answering with its full `SELECT DISTINCT` scan (535 ms of store mutex) on nearly every call —
    the same per-request scan the task existed to remove. A poll of a series the store has already
    written cannot change the *set of names*, so it must not cost a scan; only a new name can.
    """
    now = time.time()
    db.add("host", "driveripper", "cpu.usage_pct", 1.0, ts=now)
    assert db.series_names("host") == ["cpu.usage_pct"]

    db.add("host", "deskwork", "cpu.usage_pct", 1.0, ts=now)  # the second host of the same name
    assert db.series_names("host") == ["cpu.usage_pct"]

    # the next nine polls, of series the store already knows: the answer costs no statement
    with query_log(db) as seen:
        for poll in range(9):
            db.add_many([Sample("host", "driveripper", "cpu.usage_pct", float(poll), now + poll),
                         Sample("host", "deskwork", "cpu.usage_pct", float(poll), now + poll)])
        # the transaction's own BEGIN/COMMIT ride in the trace; what must not be in there is a
        # read, and above all not the `SELECT DISTINCT` this whole task is about
        assert not [s for s in seen if s.lstrip().upper().startswith("SELECT")], (
            "a poll should cost its inserts and nothing else")
        seen.clear()
        assert db.series_names("host") == ["cpu.usage_pct"]
        assert seen == [], f"a write of known series must not re-scan the name list: {seen}"


def test_a_write_of_a_series_nobody_wrote_yet_still_invalidates_immediately(db) -> None:
    """The other half of the rule, in the direction the cache must never be lazy about.

    A new metric name is exactly the event `/api/metrics` exists to report, and a source gaining
    one is a deploy — so "eventually" is not good enough and the next call has to see it.
    """
    now = time.time()
    db.add("host", "driveripper", "cpu.usage_pct", 1.0, ts=now)
    assert db.series_names("host") == ["cpu.usage_pct"]
    with query_log(db) as seen:
        db.add("host", "deskwork", "a.brand.new.metric", 2.0, ts=now + 1)
        seen.clear()
        assert db.series_names("host") == ["a.brand.new.metric", "cpu.usage_pct"], (
            "a metric nobody has written since the last read must be in the very next answer")
    assert "a.brand.new.metric" in db.series_names("host")
    # …and a per-device stream on a name the list already carries is a new triple too
    db.series_names("host")
    db.add_many([Sample("host", "driveripper", "gpu.busy_pct@3", 1.0, now + 2)])
    assert db.series_names("host", "driveripper") == ["cpu.usage_pct", "gpu.busy_pct@3"]


def test_a_deleted_series_that_comes_back_is_a_new_series_again(db) -> None:
    """`delete_series` empties rows, so the triples it emptied have to be forgotten too.

    Without that, re-creating a deleted series would leave `/api/metrics` answering without it for
    a whole TTL window: the write matches a triple the store still believes exists.
    """
    db.add("host", "deskwork", "gpu.busy_pct@0", 1.0)
    assert db.series_names("host") == ["gpu.busy_pct@0"]
    assert db.delete_series("host", "deskwork", "gpu.busy_pct@0") == 1
    assert db.series_names("host") == []
    db.add("host", "deskwork", "gpu.busy_pct@0", 2.0)  # the same triple, now a live series again
    assert db.series_names("host") == ["gpu.busy_pct@0"]


def test_the_series_write_tracking_is_capped(db, monkeypatch: pytest.MonkeyPatch) -> None:
    """The existence set is memory, so it has a ceiling and a safe fallback below it.

    Past the cap the store invalidates on every write — task 02's behaviour, correct and slower.
    The cap is what keeps a namespace a caller can influence from becoming an unbounded structure.
    """
    monkeypatch.setattr("dashboards.store.SERIES_TRACKED_SERIES_MAX", 4)
    for index in range(30):
        db.add("host", "deskwork", f"metric{index}", 1.0)
        assert f"metric{index}" in db.series_names("host"), (
            "giving up the tracking must never cost a name from the list")
    assert len(db._known_series) <= 4, f"the tracking grew past its cap: {len(db._known_series)}"
    with query_log(db) as seen:
        db.add("host", "deskwork", "metric99", 1.0)
        db.series_names("host")
    assert [s for s in seen if "DISTINCT metric" in s], (
        "past the cap the store must go back to deriving the list from the table")


def test_the_series_cache_is_bounded_however_many_names_a_caller_asks_for(db) -> None:
    """`/api/metrics?source=&host=` is free text, so the cache is capped, not grown.

    The key space belongs to whoever sends the request; an unbounded dict keyed by it is a slow
      memory leak with a query string on the front.
    """
    for index in range(SERIES_CACHE_MAX_ENTRIES + 40):
        db.series_names("host", f"host{index}")
    assert len(db._series_cache) <= SERIES_CACHE_MAX_ENTRIES


def test_series_names_expires_on_the_clock(db) -> None:
    """The TTL is the ceiling on staleness when nothing wrote."""
    db.add("host", "a", "cpu.usage_pct", 1.0)
    db.series_names("host")
    db.series_ttl_seconds = 0.0
    with query_log(db) as seen:
        db.series_names("host")
    assert [s for s in seen if "DISTINCT metric" in s], "an expired entry must be re-derived"


def test_invalidate_read_caches_forces_both_answers_back_to_the_table(db) -> None:
    """The blunt instrument the cold measurement needs, pinned so it stays blunt.

    `store_perf` asks for a cold `stats()`/`series_names()` against a store whose caches are warm;
    if either survived that call, the phase's before/after would be timing a cache hit and calling
    it a speed-up.
    """
    db.add("host", "a", "cpu.usage_pct", 1.0)
    assert db.series_names("host") == ["cpu.usage_pct"]
    assert db.stats()["samples_total"] == 1
    with query_log(db) as seen:
        db.invalidate_read_caches()
        assert seen == [], "invalidating is bookkeeping, not a query"
        db.series_names("host")
        db.stats()
    assert [s for s in seen if "DISTINCT metric" in s], "the name list came back from the scan"
    assert [s for s in seen if "COUNT(*) FROM sample" in s], "the counts did too"


def test_stats_answers_from_cache_until_the_ttl_expires(db) -> None:
    """`/api/health` is polled by every open tab; its two `COUNT` scans may run once per TTL.

    The payload keys are a contract with deployment (`/api/health` is public), so this asserts
    the same dict twice and the statements in between: nothing about the answer changes, only how
    often the table is asked.
    """
    db.add("host", "a", "cpu.usage_pct", 1.0)
    db.log_event(Event(time.time(), "system", "a", "recording", "info", "up again"))
    with query_log(db) as seen:
        first = db.stats()
        assert first["samples_total"] == 1 and first["events"] == 1
        assert [s for s in seen if "COUNT(*) FROM sample" in s]
        assert [s for s in seen if "COUNT(*) FROM event" in s]

        seen.clear()
        second = db.stats()
        assert not [s for s in seen if "COUNT(*)" in s], (
            "a cached stats() must not run a COUNT — that 1.3 s is the page load")
        assert second == first

        first["samples_by_res"][RES_RAW] = 999  # a caller must not be able to poison the cache
        assert db.stats()["samples_by_res"] == second["samples_by_res"]

        db.stats_ttl_seconds = 0.0
        seen.clear()
        db.add("host", "a", "cpu.usage_pct", 2.0)
        third = db.stats()
        assert [s for s in seen if "COUNT(*) FROM sample" in s], "the TTL, not a write, ages counts"
        assert third["samples_total"] == 2


def test_stats_caches_the_counts_but_never_the_maintenance_timestamp(db) -> None:
    """`last_maintain` is how an operator spots a stalled rollup loop; caching it hides the bug.

    The counts behind `/api/health` may be 30 s old. A `last_maintain` that never moves while the
    page says `ok` is exactly the frozen-source failure AGENTS.md invariant 4 calls the one worse
    than an outage, so it is read on every call and this test keeps it that way.
    """
    db.add("host", "a", "cpu.usage_pct", 1.0)
    assert db.stats()["last_maintain"] is None
    with query_log(db) as seen:
        db.maintain(now=1_700_000_000.0)
        seen.clear()
        stats = db.stats()
        assert stats["last_maintain"] == 1_700_000_000.0
        assert not [s for s in seen if "COUNT(*)" in s], "the counts stay cached; only meta is re-read"
        assert [s for s in seen if "FROM meta" in s]


def test_pick_res_still_picks_the_finest_resolution_that_fits(db) -> None:
    """The capped count must decide exactly what a full `COUNT(*)` decided.

    Cap at `max_points + 1` keeps both predicates — `count == 0` and `count <= max_points` —
    identical, so this is a behaviour pin and not a tolerance.
    """
    now = time.time()
    # raw over budget, minutes under it: the answer is 1m whichever way the rows are counted
    add_series(db, "host", "h", "m", range(900), start=now - 900, step=1)
    add_series(db, "host", "h", "m", range(30), start=now - 3_600, step=60, res=RES_MIN)
    assert db.pick_res("host", "h", "m", span=3_600.0, max_points=600) == RES_MIN
    # nothing fits, so the finest resolution that has *any* row wins — an empty one reads as a
    # dead sensor, which is the rule the pre-cap implementation had and the cap must not move
    assert db.pick_res("host", "h", "m", span=3_600.0, max_points=20) == RES_RAW
    assert db.pick_res("host", "h", "absent", span=3_600.0, max_points=600) == RES_HOUR


def test_the_resolution_count_is_one_statement_of_index_seeks(db) -> None:
    """One statement for all three resolutions, and its plan is a seek, not a scan."""
    now = time.time()
    for host in ("a", "b"):
        add_series(db, "host", host, "m", range(50), start=now - 300, step=1)
    with query_log(db) as seen:
        db.pick_res_many("host", ["a", "b"], ["m", "absent"], span=3_600.0, max_points=600)
    statements = [s for s in seen if "UNION ALL" in s]
    assert len(statements) == 1, (
        f"4 series x 3 resolutions must be one grouped round trip, saw {seen}")
    # the trace callback hands back the statement with its bindings expanded, which is the exact
    # text that ran — so the plan below is the plan the request paid, not a plan for a lookalike
    plan = plan_of(db, statements[0])
    assert "COVERING INDEX" in plan, plan
    assert "SCAN sample" not in plan, f"the count fell back to a table scan: {plan}"


def test_pick_res_many_agrees_with_pick_res_for_every_pair(db) -> None:
    now = time.time()
    hosts = ["driveripper", "deskwork", "gateway"]
    metrics = ["cpu.usage_pct", "mem.used_pct", "gpu.busy_pct@0", "never.written"]
    for host in hosts:
        if host == "gateway":
            continue  # the fleet's GPU-less box: a whole host with no gpu.* rows
        for metric in metrics[:3]:
            add_series(db, "host", host, metric, range(40), start=now - 400, step=1)
    batched = db.pick_res_many("host", hosts, metrics, span=3_600.0, max_points=600)
    assert batched == {(host, metric): db.pick_res("host", host, metric, 3_600.0, 600)
                       for host in hosts for metric in metrics}
    assert batched[("gateway", "gpu.busy_pct@0")] == RES_HOUR, "no rows anywhere: the documented default"
    assert db.pick_res_many("host", [], metrics, span=3_600.0, max_points=600) == {}
    assert db.pick_res_many("host", hosts, [], span=3_600.0, max_points=600) == {}


def test_history_many_answers_exactly_what_history_answers_per_series(db) -> None:
    """The batched door and the single-series door are the same answer, series for series.

    The shapes a chart depends on are in here too: the raw fallback while rollups lag, the
    thinning of an over-budget window, and the newest point surviving both.
    """
    now = time.time()
    hosts = ["driveripper", "deskwork", "gateway"]
    metrics = ["cpu.usage_pct", "mem.used_pct", "never.written"]
    for host in hosts:
        for index, metric in enumerate(metrics):
            add_series(db, "host", host, metric, range(30 + index), start=now - 900, step=30)
    # one series with rollups as well, so the resolution decision (not just the fetch) is compared
    add_series(db, "host", "driveripper", "cpu.usage_pct", range(20), start=now - 7_200,
               step=300, res=RES_MIN)
    # one crowded series, for the thinning contract below
    add_series(db, "host", "driveripper", "crowded.series", range(100), start=now - 1_000, step=10)

    for span, max_points in ((3_600.0, 600), (3_600.0, 12), (30.0, 600), (90.0, 600)):
        batched = db.history_many("host", hosts, [*metrics, "crowded.series"], span=span,
                                  max_points=max_points)
        per_series = {(host, metric): db.history("host", host, metric, span=span,
                                                 max_points=max_points)
                      for host in hosts for metric in [*metrics, "crowded.series"]}
        assert batched == per_series, f"the batch changed a series (span={span}, points={max_points})"

    # the tail contract survives batching: a thinned series is ascending, inside the budget, and
    # still ends on the newest row the window holds — the point a chart cannot afford to lose
    thinned = db.history_many("host", ["driveripper"], ["crowded.series"], span=3_600.0,
                              max_points=12)[("driveripper", "crowded.series")]
    full = db.history("host", "driveripper", "crowded.series", span=3_600.0, max_points=600)
    # `<= max_points + 1`, not `<= max_points`: the thinning has always been allowed to append the
    # newest point back after striding, so a series can land one over budget. That quirk is part
    # of the answer the endpoints must keep byte-identically, not something this task fixes.
    assert len(full) == 100 and 12 <= len(thinned) <= 13
    assert [ts for ts, _ in thinned] == sorted(ts for ts, _ in thinned)
    assert thinned[-1] == full[-1], "the batch must not lose 'now' any more than the loop did"
    assert thinned[0] > full[0], "the head is tail-truncated — the documented `tail=True` semantic"

    assert db.history_many("host", [], ["cpu.usage_pct"]) == {}
    # a series nobody ever wrote answers as an empty list, per pair — the contract every history
    # endpoint shows the browser as an empty plot rather than a 404
    assert db.history_many("host", ["gateway"], ["still.unwritten"]) == {
        ("gateway", "still.unwritten"): []}
    # duplicate names in a request cost one query each, and still answer once per pair
    assert db.history_many("host", ["gateway", "gateway"], ["cpu.usage_pct"]) == db.history_many(
        "host", ["gateway"], ["cpu.usage_pct"])


def test_history_many_batches_the_decision_and_not_the_rows(db) -> None:
    """One grouped statement for the resolution of the whole batch, whatever its size.

    Recorded deliberately: folding the *rows* into one statement (a window function over the
    range, or a compound of the per-series tail selects) measured slower on the seeded store, so
    the batch is the decision and the fetch stays a per-series index seek. This pins the half
    that did get batched, so a future edit cannot quietly return to three counts per series.
    """
    now = time.time()
    hosts = [f"host{index}" for index in range(20)]
    for host in hosts:
        add_series(db, "host", host, "cpu.usage_pct", range(10), start=now - 300, step=30)
    with query_log(db) as seen:
        db.history_many("host", hosts, ["cpu.usage_pct", "mem.used_pct"], span=3_600.0,
                        max_points=600)
    grouped = [s for s in seen if "UNION ALL" in s]
    # 20 hosts x 2 metrics x 3 resolutions = 120 branches = two chunks of COMPOUND_CHUNK
    assert len(grouped) == -(-120 // COMPOUND_CHUNK), (
        f"expected one grouped count per {COMPOUND_CHUNK} branches, saw {grouped}")


# ---- phase 14 task 03: the rollup window, and the WAL it writes into ------------------- #
#
# `maintain()` used to average every row its source resolution retains, on every tick, inside one
# write transaction: 3.97 M rows of 1-minute history per minute. Task 01 measured that at 5.4 s,
# and at 5 of 9 concurrent poller inserts outliving `busy_timeout=5000` — the
# `store-error: database is locked` line in AGENTS.md known gap 4. The fix is a window per step,
# and the tests below are the two halves of "a window that is only cheaper": the bound is read out
# of the statements that actually ran (never recomputed from the same arithmetic under test), and
# the values come out identical to a whole-table rollup of the same rows.

#: The `(ceiling, floor)` a rollup statement ran with, pulled out of `sqlite3`'s trace text.
ROLLUP_WINDOW = re.compile(
    r"FROM sample WHERE res='(?P<res>\w+)' AND ts < (?P<ceiling>[\d.]+) AND ts >= (?P<floor>[\d.]+)")


def rollup_windows(statements: list[str]) -> dict[str, tuple[float, float]]:
    """`(ceiling, floor)` per rollup step, keyed by the resolution it read from.

    The trace callback hands back the statement with its bindings expanded, so this is the window
    the statement ran with — the task's "assert via the EXPLAIN/row count hook, or a sqlite3
    trace", taken literally. Keying on the source resolution is what tells the minute step
    (`res='raw'`) from the hour step (`res='1m'`).
    """
    found: dict[str, tuple[float, float]] = {}
    for statement in statements:
        match = ROLLUP_WINDOW.search(statement)
        if match:
            found[match["res"]] = (float(match["ceiling"]), float(match["floor"]))
    return found


def bucket_start(ts: float, bucket: float = MINUTE) -> float:
    """The aligned bucket a clock falls in — the shape every window floor is snapped to."""
    return math.floor(ts / bucket) * bucket


#: Wall clocks for the window-bound tests. `1_800_000_000` is exactly on an hour *and* a minute
#: boundary, so these walk the interesting cases: on it, one second past, half an hour past, and one
#: second short of the next hour. A minute-aligned bound and an hour-aligned one are the same number
#: at offset 0 and 1 980 s apart at offset 1 799, which is exactly the luck this test used to run on
#: (`pytest.approx`'s relative tolerance hid up to 1 800 s of misalignment and not more) — an
#: explicit clock makes every edge a case instead of a coin flip.
WINDOW_CLOCK_OFFSETS = (0, 1, 30, 59, 1_800, 3_539, 3_599)


def test_the_window_floor_is_always_aligned_to_its_own_bucket(db) -> None:
    """`_rollup_floor` snaps to the step's bucket, at every clock, and never above its bound.

    The alignment is not cosmetic: the ceiling of a rollup step is a *row* bound, so the newest
    bucket is a partial one that later passes re-roll, and a floor that lands inside a bucket makes
    that re-roll average its own tail rows. Asserting the property over offsets — rather than one
    number at one instant — is what keeps a future edit from re-fitting the test to whichever clock
    it was written on.
    """
    for bucket in (MINUTE, HOUR):
        for offset in WINDOW_CLOCK_OFFSETS:
            clock = 1_800_000_000.0 + offset
            floor = db._rollup_floor(clock, 0.0, bucket)
            assert floor % bucket == 0, f"bucket={bucket:g} offset={offset} floor={floor}"
            assert floor <= clock - bucket - MAINTAIN_SLACK_SECONDS, "never above the bound"
            assert floor > clock - (2 * bucket) - MAINTAIN_SLACK_SECONDS, "never a whole window back"
    # …and the two steps really do use different alignments, or the test above proves nothing
    clock = 1_800_000_000.0 + 1_799
    assert db._rollup_floor(clock, 0.0, HOUR) != db._rollup_floor(clock, 0.0, MINUTE)


@pytest.mark.parametrize("hour_offset", WINDOW_CLOCK_OFFSETS)
def test_the_first_rollup_pass_scans_the_retention_it_has(db, hour_offset: int) -> None:
    """No `last_maintain` row means nothing is known, so the window is the whole retention.

    This is the "a first run on a fresh DB scans the full retention window exactly as today"
    half of the task, and it is also the restore-from-backup case: a store opened on history it
    has never rolled has to be able to reach 47-hour-old raw rows, or its charts above the last
    minute stay empty forever.
    """
    now = 1_800_000_000.0 + hour_offset
    deep = bucket_start(now - 47 * HOUR)
    add_series(db, "host", "h", "deep.metric", [7], start=deep, step=1)
    with query_log(db) as seen:
        db.maintain(now=now)
    windows = rollup_windows(seen)
    assert windows[RES_RAW] == pytest.approx(
        (now - MINUTE - MAINTAIN_SLACK_SECONDS, now - db.raw_keep_seconds))
    assert windows[RES_MIN] == pytest.approx(
        (now - HOUR - MAINTAIN_SLACK_SECONDS, now - db.minute_keep_seconds))
    assert db.query("host", "h", "deep.metric", res=RES_MIN, start=deep - 1,
                    end=deep + MINUTE) == [(deep, 7.0)], (
        "a first pass cannot reach the deep history it retains")


@pytest.mark.parametrize("hour_offset", WINDOW_CLOCK_OFFSETS)
def test_a_second_maintain_reads_only_the_window_since_the_last_run(db, hour_offset: int) -> None:
    """The cost change, pinned as a bound and as behaviour.

    A bucket that filled after the previous pass is inside the next window and gets rolled. A raw
    row that arrives for a bucket ten minutes behind it is not, and that bucket keeps the average
    the earlier pass computed — the documented cost of a window, stated here as a passing test
    rather than as a surprise. Reaching it in production costs a poller stall measured in minutes
    (the poller stamps samples with its own wall clock); the alternative it replaced was
    re-averaging 3.97 M rows every minute.
    """
    now = 1_800_000_000.0 + hour_offset
    bucket_a = bucket_start(now - 10 * MINUTE)
    add_series(db, "host", "h", "m", [10, 20], start=bucket_a, step=1)
    db.maintain(now=now)
    assert db.query("host", "h", "m", res=RES_MIN, start=bucket_a - 1,
                    end=bucket_a + MINUTE) == [(bucket_a, 15.0)]

    second = now + 60
    # Two full minutes behind the ceiling, so both rows of the pair are always inside the next
    # window: with the bucket only one step behind, a clock that lands within a second of the
    # boundary puts the pair's second row above the ceiling and the bucket rolls 40.0 instead.
    bucket_b = bucket_start(second - 3 * MINUTE)
    add_series(db, "host", "h", "m", [40, 50], start=bucket_b, step=1)
    add_series(db, "host", "h", "m", [1000], start=bucket_a + 30, step=1)
    with query_log(db) as seen:
        counts = db.maintain(now=second)

    windows = rollup_windows(seen)
    # The bound: one minute (one hour) of rows plus the overlap, not 48 h of raw (not 30 d of 1m).
    # Each floor is snapped to *its own step's* bucket — the minute step to a minute, the hour step
    # to an hour — which is what `hour_offset` walks, and both are far inside retention.
    assert windows[RES_RAW] == pytest.approx(
        (second - MINUTE - MAINTAIN_SLACK_SECONDS,
         bucket_start(now - MINUTE - MAINTAIN_SLACK_SECONDS, MINUTE)))
    assert windows[RES_MIN] == pytest.approx(
        (second - HOUR - MAINTAIN_SLACK_SECONDS,
         bucket_start(now - HOUR - MAINTAIN_SLACK_SECONDS, HOUR)))
    assert windows[RES_RAW][1] > second - db.raw_keep_seconds
    assert windows[RES_MIN][1] > second - db.minute_keep_seconds
    # …and the behaviour that bound implies, in both directions
    assert db.query("host", "h", "m", res=RES_MIN, start=bucket_b - 1,
                    end=bucket_b + MINUTE) == [(bucket_b, 45.0)], "the due bucket must be rolled"
    assert db.query("host", "h", "m", res=RES_MIN, start=bucket_a - 1,
                    end=bucket_a + MINUTE) == [(bucket_a, 15.0)], (
        "the row behind the window must not rewrite an older bucket")
    assert counts["to_minute"] == 1, f"only the due bucket was written: {counts}"
    assert db.get_meta(LAST_MAINTAIN_KEY) == str(second)


@pytest.mark.parametrize("hour_offset", WINDOW_CLOCK_OFFSETS)
def test_a_missed_interval_is_a_catch_up_scan_not_a_gap(db, hour_offset: int) -> None:
    """Twelve minutes with no pass, then one: every bucket in between is rolled.

    The window reaches one bucket plus slack *behind* the previous pass's ceiling rather than
    starting at it, and that overlap is the whole difference between "the app was stopped for
    twelve minutes" and "twelve minutes of charts are silently empty". One row is placed
    deliberately inside the overlap: above the first pass's ceiling, below the clock the first pass
    recorded, so a bound of exactly `last_maintain` would skip it for ever.
    """
    # An explicit clock, for the same reason as the test above: the overlap row's bucket and the
    # window's floor are both snapped, and the count that proves the catch-up happened depends on
    # which side of a boundary they land on.
    first_at = 1_800_000_000.0 + hour_offset - 20 * MINUTE
    db.maintain(now=first_at)  # an otherwise empty store: this only records the clock
    aligned = bucket_start(first_at)
    for index in range(10):
        add_series(db, "host", "h", "m", [float(index)], start=aligned + index * MINUTE + 5, step=1)
    inside_the_overlap = first_at - 100
    add_series(db, "host", "h", "m", [99], start=inside_the_overlap, step=1)

    second = first_at + 12 * MINUTE
    counts = db.maintain(now=second)
    # `end` is explicit because `query` defaults it to the wall clock, and this test's clock is its
    # own: a default would silently answer "no rows" for buckets that are in the store.
    minutes = db.query("host", "h", "m", res=RES_MIN, start=0, end=second)
    assert counts["to_minute"] == 11, f"ten buckets plus the overlap row: {counts}"
    assert [value for _ts, value in minutes] == [99.0, *[float(step) for step in range(10)]], (
        "a missed interval must be caught up, not skipped")
    assert minutes[0][0] <= inside_the_overlap, "the overlap row has its bucket"


@pytest.mark.parametrize("sub_second", [0.0, 0.5, 30.0, 59.5])
@pytest.mark.parametrize("hour_offset", [0, 600, 1800, 3_540])
def test_windowed_rollups_answer_exactly_what_one_full_window_rollup_answers(
        tmp_path, hour_offset: int, sub_second: float) -> None:
    """The equality the task asks for: the cost moves, not one averaged value.

    Two stores, the same rows. One is maintained twice with an hour between the passes, so its
    second pass reads a window; the other gets every row up front and is maintained once over the
    whole retention at the later clock — the shape task 01 measured. Every row of every resolution
    must match, `ts` and `value` exactly. `INSERT OR REPLACE` over a complete bucket is what makes
    that safe, and the bucket-aligned floor (`_rollup_floor`) is what makes it true: a floor that
    cut into a bucket would instead leave the *tail* rows of a split minute as its "average".
    """
    # `first_at` is parametrised across an hour on purpose: the windows land on bucket boundaries
    # or in the middle of them depending on where the clock fell, and the boundary cases are the
    # ones a partial bucket can hide in.
    first_at = bucket_start(time.time() - 6 * HOUR, HOUR) + hour_offset + sub_second
    second_at = first_at + HOUR + 90.0  # an hour and a half: straddles an hour bucket boundary

    def seed(store: SampleStore, *, late: bool) -> None:
        for series, base in (("cpu.usage_pct", 0.0), ("mem.used_pct", 100.0)):
            # three hours of raw at the fleet's 10 s cadence, all of it complete by `first_at`
            add_series(store, "host", "h", series,
                       [base + step for step in range(1_080)],
                       start=first_at - 4 * HOUR, step=10)
        if late:
            # the buckets that fill in the hour between the two passes, unaligned on purpose
            for series, base in (("cpu.usage_pct", 0.0), ("mem.used_pct", 100.0)):
                add_series(store, "host", "h", series,
                           [base + step for step in range(180)],
                           start=first_at + 95.0, step=20)

    def dump(store: SampleStore) -> list[tuple[str, str, str, str, float, float]]:
        with store._lock, store._db:
            return store._db.execute(
                "SELECT source, host, metric, res, ts, value FROM sample"
                " ORDER BY source, host, metric, res, ts").fetchall()

    windowed = SampleStore(tmp_path / "windowed.sqlite3")
    reference = SampleStore(tmp_path / "reference.sqlite3")
    try:
        seed(windowed, late=False)
        windowed.maintain(now=first_at)
        seed(windowed, late=True)
        windowed.maintain(now=second_at)
        windowed.maintain(now=second_at + 60)  # a third tick, the steady state the app lives in

        seed(reference, late=True)
        assert reference.get_meta(LAST_MAINTAIN_KEY) is None, "a fresh store is a full window"
        reference.maintain(now=second_at + 60)

        windowed_rows, reference_rows = dump(windowed), dump(reference)
        assert {row[3] for row in windowed_rows} == {RES_RAW, RES_MIN, RES_HOUR}, (
            f"the comparison needs rolled rows on both steps: {sorted({r[3] for r in windowed_rows})}")
        assert len([row for row in windowed_rows if row[3] == RES_MIN]) > 200, (
            "the equality has to cover a lot of buckets, not a token few")
        assert windowed_rows == reference_rows, (
            "the window changed a rollup value (or a bucket a full scan would have written)")
    finally:
        windowed.close()
        reference.close()


def test_a_re_rolled_bucket_only_ever_gains_rows(db) -> None:
    """The progressive correction the whole-table scan used to provide still happens.

    A pass whose ceiling lands mid-bucket writes a partial average — as it always did — and the
    next pass must re-read that bucket **from its own start** and publish the full one. A window
    floor that cut into the bucket instead would re-read its tail and overwrite 2.5 with 4.5, which
    is an invented number wearing a real chart's clothes (AGENTS.md rule 2).
    """
    minute = bucket_start(time.time() - 10 * MINUTE)
    add_series(db, "host", "h", "m", [0, 1, 2, 3, 4, 5], start=minute, step=10)
    first = minute + 155.0  # ceiling = minute + 35: four of the six rows are in the window
    db.maintain(now=first)
    partial = db.query("host", "h", "m", res=RES_MIN, start=minute - 1, end=minute + MINUTE)
    assert partial == [(minute, 1.5)], "a boundary bucket is rolled from its start, partially"

    db.maintain(now=first + 60)
    rolled = db.query("host", "h", "m", res=RES_MIN, start=minute - 1, end=minute + MINUTE)
    assert rolled == [(minute, 2.5)], (
        f"the re-roll must add the remaining rows, not replace them: {rolled}")
    db.maintain(now=first + 120)
    assert db.query("host", "h", "m", res=RES_MIN, start=minute - 1,
                    end=minute + MINUTE) == rolled, "a settled bucket keeps its value"


def test_a_corrupt_last_maintain_row_falls_back_to_the_full_window(db, caplog) -> None:
    """A bound the store cannot read is the expensive answer, said once, never a crash.

    `maintain()` is the only writer of that row, so a non-number in it means something outside the
    app edited the database. Rolling the whole window again rewrites identical rows; guessing a
    window from a value that is not a timestamp is how a rollup starts skipping buckets in
    silence, and the maintenance loop swallowing an exception is worse still (invariant 4).
    """
    now = time.time()
    db.set_meta(LAST_MAINTAIN_KEY, "not-a-timestamp")
    with caplog.at_level(logging.WARNING, logger="dashboards.store"), query_log(db) as seen:
        db.maintain(now=now)
    assert rollup_windows(seen)[RES_RAW] == pytest.approx(
        (now - MINUTE - MAINTAIN_SLACK_SECONDS, now - db.raw_keep_seconds))
    assert [rec for rec in caplog.records if "last_maintain" in rec.message], (
        "drift in the row that bounds every rollup is worth one line")
    assert "not-a-timestamp" not in caplog.text, "the log quotes the key, not the stranger's value"
    assert db.get_meta(LAST_MAINTAIN_KEY) == str(now), "the pass repairs the row it reads"


def test_a_failing_optimize_is_a_log_line_and_not_a_failed_tick(db, monkeypatch, caplog) -> None:
    """Statistics are bookkeeping: `maintain()` may not turn them into an outage.

    `_optimize` runs outside the rollup transaction, so the only way it can hurt is by raising — and
    a `store-error` event for a planner hint is worse than a slightly slower query (AGENTS.md
    invariant 4: a source that stops answering is the failure, and the maintenance loop's health is
    reported the same way). The stub quotes a message of its own; the log line names the pragma and
    the consequence, nothing else.
    """
    class _Boom:
        def execute(self, _sql: str) -> None:
            raise sqlite3.Error("database is locked")

    monkeypatch.setattr(db, "_db", _Boom())
    with caplog.at_level(logging.WARNING, logger="dashboards.store"):
        db._optimize()  # must not raise
    assert "PRAGMA optimize failed" in caplog.text
    assert "database is locked" not in caplog.text, "the tick's own words, not the driver's"


def test_the_perf_probe_explains_the_statement_that_runs(db) -> None:
    """The probe's rollup `EXPLAIN` is derived from `ROLLUP_INSERT`, so its bindings must match.

    A hand-copied lookalike is how a NOTES table ends up describing a statement nobody runs — and
    when the rollup grew its seventh binding (the target resolution), the copy in the probe was the
    thing that broke. Preparing the derived statement with the probe's own arity is the pin: retyping
    it, or letting the two drift, raises here instead of in the middle of a measurement run.
    """
    whole_select = "SELECT" + ROLLUP_INSERT.split("SELECT", 1)[1]
    assert whole_select == store_perf.ROLLUP_SELECT
    assert ROLLUP_INSERT.count("?") == store_perf.ROLLUP_SELECT.count("?") == 7
    with db._lock:
        db._db.execute("EXPLAIN QUERY PLAN " + store_perf.ROLLUP_SELECT,
                       (RES_MIN, MINUTE, MINUTE, RES_RAW, 1.0, 0.0, MINUTE)).fetchall()


def test_the_wal_pragmas_are_set_on_every_connection(db) -> None:
    """The WAL half of task 03, pinned as the settings that ship.

    `journal_size_limit` is the lever: a WAL that grows past it is truncated to it at the next
    checkpoint instead of staying at a 140 MB high-water mark (AGENTS.md known gap 4).
    `wal_autocheckpoint` is SQLite's own default and is set so the knob is visible beside it, and
    `busy_timeout` stays at 5 000 ms because the fix is a shorter write, not a longer wait.
    All the pragmas are per connection, so this asserts them on a handle built by `SampleStore` and
    not on whatever SQLite happens to have guessed.
    """
    with db._lock:
        assert db._db.execute("PRAGMA journal_mode").fetchone()[0].lower() == "wal"
        assert db._db.execute("PRAGMA synchronous").fetchone()[0] == 1
        assert db._db.execute("PRAGMA busy_timeout").fetchone()[0] == 5000
        assert db._db.execute("PRAGMA wal_autocheckpoint").fetchone()[0] == WAL_AUTOCHECKPOINT_PAGES
        assert db._db.execute("PRAGMA journal_size_limit").fetchone()[0] == JOURNAL_SIZE_LIMIT_BYTES
    assert JOURNAL_SIZE_LIMIT_BYTES == 64 * 1024 * 1024 == 67108864
    # `automatic_analyze` (off, so no poller's commit pays for a statistics refresh) is set-only in
    # this build — reading it back returns no row — so its effect is pinned where it is observable,
    # in `test_maintain_leaves_query_statistics`.


def test_maintain_leaves_query_statistics(db) -> None:
    """`maintain()` is the app's only statistics refresh, and the windowed rollup needs them.

    Without `sqlite_stat1` the planner cannot price a range and reads the table for every window:
    on the seeded store the windowed rollup cost 328 ms of `SCAN sample` matching no rows at all,
    and 1 ms with `SEARCH sample USING INDEX sqlite_autoindex_sample_1` after one analyse. So the
    statistics are half of "cheaper for the same rows", and they are refreshed here rather than by
    SQLite's own automatic ANALYZE — which would run inside whichever poller's commit tripped its
    change counter, the exact path this task is shortening. On a database this size a scan is free,
    so the assertion is that the statistics exist, not what plan they produce; the plan is pinned at
    production scale in the `store_perf` report.
    """
    def stat_rows() -> int:
        with db._lock:
            if db._db.execute("SELECT 1 FROM sqlite_master WHERE name='sqlite_stat1'").fetchone() is None:
                return 0
            return int(db._db.execute("SELECT COUNT(*) FROM sqlite_stat1").fetchone()[0])

    assert stat_rows() == 0, "a store that has never been maintained has no statistics"
    db.add("host", "gateway", "cpu.usage_pct", 1.0)
    db.maintain(now=time.time())
    assert stat_rows() > 0, "maintenance is the one place the planner's statistics get refreshed"


# ---- phase 14: the production-shaped store the performance work measures ---- #
#
# `TODO.md:7` is "pages take forever to load and seem to lock up the database", and a claim
# about speed needs a measured "before". The fixture below is that before: an on-disk store with
# the deployed retention semantics (`raw_keep_hours=48`, `minute_keep_days=30`,
# `hour_keep_days=400` — the `SampleStore` defaults, i.e. what a 90 d chart really reads is 1m
# and 1h, never raw beyond 48 h), the fleet's real metric names, the real 10 s cadence, ~2 000
# events and the `meta` rows the app writes. ~6.4 M rows in ~700 MB, which is the dev DB's order
# of magnitude (`AGENTS.md` gap 4 records 636 MB / 3.4 M rows for a store only days old).
#
# Why the fixture is not the dev DB: the testing policy keeps `uv run pytest` hermetic, and a
# fixture that read the operator's live history would change behaviour when that history does.
# Why it is not an in-memory store: `:memory:` hides the WAL and the lock behaviour, which are
# half of what this phase measures. Why it is not `tmp_path`: pytest's tmp dir on this box is
# `tmpfs`, i.e. RAM, and a database timed on RAM is a flattering number nobody deploys —
# `store_perf.cache_dir()` defaults to a real block device and says so in its report.
#
# The default-run tests here assert **shape and determinism**; the wall-time budgets task 04 pins
# are behind `-m store_perf --store-perf` (and in `scripts/store_perf_probe.py`, which writes the
# NOTES tables) because a budget asserted on first writing is a number fitted to the machine it was
# written on. The budgets themselves live in `store_perf.BUDGETS` — one table, priced by the probe,
# the report, and the two budget tests below, because three copies of a budget drift the moment one
# is relaxed. The `perf_store`/`perf_store_path` fixtures are in `tests/conftest.py`: the HTTP
# budgets in `tests/test_api.py` measure the same file.




def test_the_seed_seeds_the_retention_the_app_actually_keeps() -> None:
    """The fixture's retention must be `SampleStore`'s own defaults, not a story about them.

    A budget measured on a store seeded to a *different* retention is a budget for a database
    that does not exist: `pick_res`/`history` costs depend on how deep each resolution goes.
    """
    fresh = SampleStore(":memory:")
    try:
        assert fresh.raw_keep_seconds == pytest.approx(store_perf.RAW_KEEP_HOURS * HOUR)
        assert fresh.minute_keep_seconds == pytest.approx(store_perf.MINUTE_KEEP_DAYS * 24 * HOUR)
        assert fresh.hour_keep_seconds == pytest.approx(store_perf.HOUR_KEEP_DAYS * 24 * HOUR)
    finally:
        fresh.close()
    assert store_perf.CADENCE_SECONDS == 10.0, "the raw window's cadence is HOST_INTERVAL_SECONDS"


def test_the_seeded_store_is_the_shape_the_budget_tests_measure(perf_store: SampleStore) -> None:
    """Row counts, series, retention windows, events, meta — asserted, never timed."""
    now = store_perf.seed_now(perf_store)
    expected = store_perf.plan(now)

    with perf_store._lock, perf_store._db:  # one pass over the table, not three
        rows = {row[0]: row for row in perf_store._db.execute(
            "SELECT res, COUNT(*), MIN(ts), MAX(ts) FROM sample WHERE source='host' GROUP BY res")}

    assert expected.raw_rows == expected.series_total * expected.raw_steps
    assert expected.minute_rows_seeded == expected.series_total * expected.minute_steps
    assert expected.hour_rows_seeded == expected.series_total * expected.hour_steps
    assert sum(len(store_perf.HOST_SERIES[name]) for name in expected.hosts) == (
        expected.series_total)
    assert rows[RES_RAW][1] == expected.raw_rows, (
        "the raw window is inserted, not rolled: its count is the one number the seed owns "
        "outright, so a mismatch here means the shape drifted from the plan")
    assert rows[RES_MIN][1] >= expected.minute_rows_seeded
    assert rows[RES_HOUR][1] >= expected.hour_rows_seeded
    # …and the rollups `maintain()` added inside the windows are there too, or the fixture
    # would be a store whose charts draw nothing above the raw window
    assert rows[RES_MIN][1] > expected.minute_rows_seeded
    assert rows[RES_HOUR][1] > expected.hour_rows_seeded

    # Retention: nothing lives past its resolution's window, and each window is deep enough to
    # be the one the pages read (a 90 d chart reads 1m/1h; a 5 m chart reads raw).
    assert rows[RES_RAW][2] >= now - store_perf.RAW_KEEP_HOURS * HOUR - 1.0
    assert rows[RES_RAW][3] <= now + 1.0
    assert rows[RES_MIN][2] >= now - store_perf.MINUTE_KEEP_DAYS * 24 * HOUR - 1.0
    assert rows[RES_HOUR][2] >= now - store_perf.HOUR_KEEP_DAYS * 24 * HOUR - 1.0

    assert perf_store.hosts("host") == list(expected.hosts)
    assert len(perf_store.series_names("host")) == len(store_perf.BASE_SERIES) + len(
        store_perf.GPU_SERIES), "the union: gateway has no gpu.* series, the other two do"
    assert len(perf_store.series_names("host", "gateway")) == len(store_perf.BASE_SERIES)
    assert len(perf_store.series_names("host", "driveripper")) == (
        len(store_perf.BASE_SERIES) + len(store_perf.GPU_SERIES))

    assert perf_store.count() == rows[RES_RAW][1] + rows[RES_MIN][1] + rows[RES_HOUR][1]

    events = perf_store.recent_events(store_perf.EVENT_ROWS)
    assert len(events) == store_perf.EVENT_ROWS
    assert {event.source for event in events} == {"system"}

    # The `meta` rows the app writes: `last_maintain` from the build's own `maintain()` call,
    # plus this fixture's own provenance. Phase 11's auth rows are not seeded — the app writes
    # them the first time a test boots a server on this store.
    assert perf_store.get_meta("last_maintain") is not None
    assert perf_store.get_meta(store_perf.META_SEED_VERSION) == store_perf.SEED_VERSION

    # Magnitude, not exact bytes: a fixture two orders of magnitude smaller would not catch a
    # full-table scan, and the number that matters is in the NOTES (dev DB 636 MB, this ~700 MB).
    size = Path(perf_store.path).stat().st_size
    assert size > 100 * 1024 * 1024, f"seeded store is {size / 1e6:.0f} MB — too small to measure"


def test_the_rollups_in_the_seed_came_from_the_raw_rows_and_run_twice_the_same(
    perf_store: SampleStore,
) -> None:
    """Determinism, in both directions the phase depends on.

    1. the seed is a pure function of `(shape, seed clock)`, so a rebuild is byte-identical and
       a "before/after" comparison is comparing the same data — asserted by recomputing stored
       raw values from `store_perf._value`;
    2. the rollup rows are the real `maintain()` arithmetic over those raw values (the mean of
       the raw samples in the bucket), and running maintenance again changes nothing.
    """
    now = store_perf.seed_now(perf_store)
    host, metric = "driveripper", "cpu.usage_pct"

    first_rows = perf_store.query("host", host, metric, res=RES_RAW,
                                  start=now - store_perf.RAW_KEEP_HOURS * HOUR,
                                  limit=60, tail=False)
    assert first_rows, "the raw window is empty"
    offset = round((first_rows[0][0] - (now - store_perf.RAW_KEEP_HOURS * HOUR))
                   / store_perf.CADENCE_SECONDS)
    assert [value for _ts, value in first_rows] == [
        store_perf.seed_value(metric, offset + index) for index in range(len(first_rows))], (
        "stored raw values are not the seed function's output — the seed is no longer a pure "
        "function of the shape, so before/after numbers would not be comparable")

    # one complete bucket inside the raw window, recomputed by hand
    bucket_ts = first_rows[0][0] + 30 * store_perf.CADENCE_SECONDS
    bucket_ts = int(bucket_ts // MINUTE) * MINUTE
    raw_in_bucket = perf_store.query("host", host, metric, res=RES_RAW, start=bucket_ts,
                                     end=bucket_ts + MINUTE - 0.001)
    assert len(raw_in_bucket) == 6, raw_in_bucket
    rolled = perf_store.query("host", host, metric, res=RES_MIN, start=bucket_ts,
                              end=bucket_ts + MINUTE - 0.001)
    assert rolled == [(float(bucket_ts), pytest.approx(
        sum(value for _ts, value in raw_in_bucket) / len(raw_in_bucket)))]

    def shape() -> dict[str, tuple[int, float, float]]:
        """Rows, and the time window each resolution holds — the exact half of the checksum.

        The total of `value` is deliberately *not* in here. `maintain()` writes rollups with
        `INSERT OR REPLACE`, which deletes and re-inserts the row, so a pass that changes no value
        still moves every rowid it touched, and a `SUM(value)` scanned in rowid order accumulates
        those doubles in a different sequence — a correctly idempotent pass then prints a
        different total (2.8e5 out of 2.5e15 was how this test failed once the seeded file had
        been maintained a few times). Counts and windows are exact; the sum is compared to a
        tolerance below, and the *values* are compared exactly, in primary-key order, by
        `values_slice()`.
        """
        with perf_store._lock, perf_store._db:
            return {row[0]: (row[1], row[2], row[3]) for row in perf_store._db.execute(
                "SELECT res, COUNT(*), COALESCE(MIN(ts),0), COALESCE(MAX(ts),0)"
                " FROM sample GROUP BY res")}

    def totals() -> dict[str, float]:
        with perf_store._lock, perf_store._db:
            return {row[0]: row[1] for row in perf_store._db.execute(
                "SELECT res, COALESCE(SUM(value),0) FROM sample GROUP BY res")}

    def values_slice(res: str, limit: int = 4000) -> list[tuple[str, str, str, float, float]]:
        """A deterministic slice of the table, in primary-key order, compared exactly.

        `ORDER BY source, host, metric, ts` walks the primary-key index rather than the rowids,
        so this is the same list of rows before and after regardless of where maintenance moved
        them — which is the assertion the aggregate cannot make: not one stored value rewritten.
        """
        with perf_store._lock, perf_store._db:
            return perf_store._db.execute(
                "SELECT source, host, metric, ts, value FROM sample WHERE res=?"
                " ORDER BY source, host, metric, ts LIMIT ?", (res, limit)).fetchall()

    before, before_totals = shape(), totals()
    slices = {res: values_slice(res) for res in (RES_RAW, RES_MIN, RES_HOUR)}
    again = perf_store.maintain(now=now)
    after, after_totals = shape(), totals()

    assert after == before, (
        "maintenance is on a timer: a second pass over the same data may not add or drop a row, "
        "or move a resolution's time window")
    for res, total in after_totals.items():
        assert total == pytest.approx(before_totals[res], rel=1e-9), (
            f"{res}: the stored total moved more than float accumulation can explain — a rollup "
            f"re-averaged a bucket it had already rolled")
    for res, rows in slices.items():
        assert values_slice(res) == rows, (
            f"{res}: a row's value changed on the second pass — the rollup is not idempotent")
    assert again["pruned"] == 0, "nothing in this store is past its retention"
    assert perf_store.count() == sum(row[0] for row in before.values())


def test_the_race_measurements_leave_the_seeded_shape_alone(perf_store: SampleStore) -> None:
    """A measurement that dirtied the baseline would move the next one — so this is a test.

    The race inserts 30-sample polls (`store_perf.RACE_SOURCE`, never a seeded series) and the
    task-04 comparison depends on the seeded counts being what they were.
    """
    before = store_perf.counts(perf_store)
    result = store_perf.race_single_connection(perf_store, iterations=5, max_seconds=15.0)
    after = store_perf.counts(perf_store)

    assert result.attempted >= 1 and result.inserts == result.attempted
    assert after == before, f"the race left rows behind: {before} → {after}"
    with perf_store._lock, perf_store._db:
        leftovers = int(perf_store._db.execute(
            "SELECT COUNT(*) FROM sample WHERE source=?",
            (store_perf.RACE_SOURCE,)).fetchone()[0])
    assert leftovers == 0


def test_the_batched_answers_match_the_per_series_answers_on_the_real_shaped_store(
        perf_store: SampleStore) -> None:
    """The equality that lets the endpoints switch doors: same series, same order, same cost class.

    On the store this phase measures its budgets on, at the two windows where the resolution
    decision actually differs (24 h reads 1m, 90 d reads 1h), for the 27-series request
    `/api/system/history` answers for a whole page. A capped count that decided a resolution
    differently from a full `COUNT(*)` would move a chart onto the wrong resolution — the same
    points averaged together, which is the kind of wrong that still renders.
    """
    hosts = list(store_perf.HOST_SERIES)
    metrics = list(store_perf.HISTORY_METRICS)
    for span in (24 * 3_600.0, 90 * 24 * 3_600.0):
        res_map = perf_store.pick_res_many("host", hosts, metrics, span, 600)
        assert res_map == {(host, metric): perf_store.pick_res("host", host, metric, span, 600)
                           for host in hosts for metric in metrics}
        batched = perf_store.history_many("host", hosts, metrics, span=span, max_points=600)
        assert batched == {(host, metric): perf_store.history("host", host, metric,
                                                              span=span, max_points=600)
                           for host in hosts for metric in metrics}, (
            f"the batched path changed a series at span={span:.0f}")
        # and the series themselves are still the shape the charts parse: ascending, budgeted,
        # ending at the newest row the window holds. `HISTORY_METRICS` is the *request* the page
        # makes and the seed holds a subset of it (`cpu.temp_c` is not one of the seeded names),
        # so the absent ones are asserted empty rather than skipped — an empty series is an answer.
        held = {(host, metric) for host, names in store_perf.HOST_SERIES.items() for metric in names}
        for (host, metric), points in batched.items():
            if (host, metric) not in held:
                assert points == [], f"{host}/{metric} is not in the seed: it must stay empty"
                continue
            assert points, f"{host}/{metric} at {span} came back empty on a store full of it"
            assert 0 < len(points) <= 601, "the budget plus the point the thinning appends back"
            assert [ts for ts, _value in points] == sorted(ts for ts, _value in points)


def test_two_maintain_cycles_leave_the_seeded_wal_under_the_cap(perf_store: SampleStore) -> None:
    """Criterion 3, on the store this phase measures its numbers on.

    `journal_size_limit` is the ceiling; the windowed rollup is the reason the file has hardly
    anything to grow with. The assertion is on the cap rather than a byte count because the cap is
    the promise — a WAL that stays under 64 MB across ticks is what AGENTS.md known gap 4 needed,
    and a run that left a bigger file behind would have to be explained by something other than
    `maintain()`.
    """
    path = Path(perf_store.path)
    now = store_perf.seed_now(perf_store)
    with perf_store._lock, perf_store._db:  # start from a checkpointed file, so this measures the cycles
        perf_store._db.execute("PRAGMA wal_checkpoint(TRUNCATE)")
    wal_before = store_perf.wal_bytes(path)
    for _ in range(2):
        perf_store.maintain(now=now)
    wal_after = store_perf.wal_bytes(path)
    # 8 MB of headroom is generous for a tick that writes one bucket per series: it is there so the
    # test says something about the cycles rather than about the filesystem's rounding.
    assert wal_after <= JOURNAL_SIZE_LIMIT_BYTES, (
        f"WAL {wal_after / 1e6:,.1f} MB after two cycles, cap "
        f"{JOURNAL_SIZE_LIMIT_BYTES / 1e6:,.0f} MB")
    assert wal_after <= wal_before + 8 * 1024 * 1024, (
        f"two steady ticks grew the WAL {wal_before:,} → {wal_after:,} bytes")


# --------------------------------------------------------------------------- #
# task 04: the budgets
#
# Two kinds of test, because the phase's own rule is that a cost claim is pinned as a shape *and*
# priced, and the two need different stores. The shape half runs on the little `db` fixture in
# every `uv run pytest`: the budget table has to be internally sound and every budget has to name
# a measurement that still exists, so a budget cannot rot into a comment by pointing at a report
# key somebody renamed. The price half runs on the seeded store behind `--store-perf`, where the
# numbers mean something; a budget priced on a 40-row store would be a budget for a database
# nobody has, which is the failure AGENTS.md warns about.
# --------------------------------------------------------------------------- #


def test_the_budget_table_is_internally_coherent() -> None:
    """Every pinned budget is a checkable claim about a real report path.

    Cheap, and it is the test that keeps the rest honest: a budget with a limit of 0 is always
    crossed (or never, if the statistic is a count), a ratio pointing at a key in another section
    silently measures nothing, and two rows on the same measurement with different ceilings is the
    moment a table stops meaning one thing.
    """
    assert store_perf.BUDGETS, "there are no budgets"
    seen: set[tuple[str, str, str]] = set()
    for budget in store_perf.BUDGETS:
        assert budget.section in store_perf.BUDGET_SECTIONS, f"unknown section {budget.section}"
        assert budget.key, "a budget must name the report key it prices"
        assert budget.limit >= 0.0, f"{budget.label()} has a negative ceiling"
        assert budget.unit in {"ms", "count"}, f"{budget.label()} has an unknown unit {budget.unit}"
        assert budget.basis, f"{budget.label()} records no reasoning for its ceiling"
        identity = (budget.section, budget.key, budget.statistic)
        assert identity not in seen, f"{identity} is pinned twice, with different ceilings possible"
        seen.add(identity)
        # every budget's key has to be a key the pass that prices it actually produces
        measured = (store_perf.READ_MEASURED_KEYS if budget.section == "paths"
                    else store_perf.HTTP_MEASURED_KEYS if budget.section == "http"
                    else store_perf.MAINTAIN_MEASURED_KEYS if budget.section == "maintain" else None)
        if measured is not None:
            assert budget.key in measured, (f"{budget.label()} names {budget.key}, which "
                                            f"{budget.section} does not measure")
        if budget.ratio_key is not None:
            # `_budget_value` swaps only the key, so the partner has to live in the same section
            # and be priced the same way, or the ratio compares two different things. The partner
            # is usually *not* itself a budget — nobody promises the cold scan a ceiling; it is the
            # denominator that says the cached answer is not a scan in disguise.
            assert measured is not None and budget.ratio_key in measured, (
                f"{budget.label()} is relative to {budget.ratio_key}, which {budget.section} "
                "does not measure")
            assert budget.ratio_limit is not None and 0.0 < budget.ratio_limit < 1.0, (
                f"{budget.label()} must claim a *fraction* of its partner")
    # the cached reads are the ones the phase claims are no longer scans: that claim is the ratio,
    # so both cached paths carry one rather than a bare 10 ms that a fast laptop always beats
    ratios = {budget.key for budget in store_perf.BUDGETS if budget.ratio_key is not None}
    assert {"series_names_host_cached", "stats_cached"} <= ratios


def test_every_read_budget_names_a_measurement_that_still_exists(db) -> None:
    """The stale-pointer check, on the little store, in the default run.

    `measure_budget_reads` is what the priced test calls; if a key there stops matching a key in
    `BUDGETS` the priced test would quietly price nothing and pass. Crossings are not asserted
    here — a 40-row store makes every ceiling trivially true and every ratio a coin flip.
    """
    report = store_perf.measure_budget_reads(db, runs=2)
    assert set(report["paths"]) == set(store_perf.READ_MEASURED_KEYS), (
        "the read pass and the budget table no longer describe the same measurements")
    assert store_perf.missing_budget_keys(report, ("paths",)) == []
    rows = store_perf.budget_rows(report, ("paths",))
    assert len(rows) == sum(1 for budget in store_perf.BUDGETS
                            if budget.section == "paths")
    assert all(row.measured is not None and row.measured >= 0.0 for row in rows)
    # the ratio partner is looked up by swapping the key, so it has to resolve on the same report
    for row in rows:
        if row.budget.ratio_key is not None:
            partner = store_perf._budget_value(
                report, replace(row.budget, key=row.budget.ratio_key))
            assert partner is not None, f"{row.budget.label()} has an unpriceable partner"


@pytest.mark.store_perf
def test_the_seeded_store_reads_fit_their_pinned_budgets(perf_store: SampleStore) -> None:
    """The read budgets, priced on the store this phase's numbers were taken from.

    Ceiling 2-3x the measured "after" (see each row's `basis`), so a slower deployment disk does
    not turn red for the wrong reason — but the *scan* paths are what these catch: put the
    whole-table `COUNT(*)` back in `stats()` and the cached row crosses on the ratio alone, which
    is the half a generous absolute ceiling cannot see.
    """
    report = store_perf.measure_budget_reads(perf_store)
    print()
    print(store_perf.render_budgets_markdown(store_perf.budget_rows(report, ("paths",))))
    assert store_perf.budget_violations(report, ("paths",)) == []


@pytest.mark.store_perf
def test_the_write_side_fits_its_pinned_budgets(perf_store: SampleStore) -> None:
    """`maintain()`'s write transaction and the poller racing it — the gap-4 budgets.

    Priced on a scratch copy (`measure_maintain`'s rule: a rollup really writes, and the cached
    baseline has to come back unchanged), at the seed's own clock. The race needs two connections,
    so it runs on its own copy too; the whole-window control is *not* budgeted here — it is what
    `test_the_baseline_report_measures_every_hot_path` uses to prove the race can still see a
    failure, and it is supposed to be slow.
    """
    path = Path(perf_store.path)
    now = store_perf.seed_now(perf_store)
    maintain = store_perf.measure_maintain(path, now=now)
    report = {"maintain": maintain}
    print()
    print(store_perf.render_budgets_markdown(store_perf.budget_rows(report, ("maintain",))))
    assert store_perf.budget_violations(report, ("maintain",)) == []

    with store_perf.scratch_copy(path, name="budget-race") as arena:
        race = store_perf.race_two_connections(arena, iterations=100, now=now,
                                               max_seconds=60.0,
                                               maintain_gap_seconds=store_perf
                                               .RACE_MAINTAIN_GAP_SECONDS)
    report = {"race": {"poller_during_windowed_maintain": race.as_dict()}}
    print()
    print(store_perf.render_budgets_markdown(store_perf.budget_rows(report, ("race",))))
    assert store_perf.budget_violations(report, ("race",)) == []


@pytest.mark.store_perf
def test_the_baseline_report_measures_every_hot_path(perf_store: SampleStore) -> None:
    """The measurement itself, inside pytest: every path task 01 names, with its EXPLAIN shape.

    Opt-in (`-m store_perf --store-perf`) because it takes minutes, and skipped by default so
    the shape tests above still run in every `uv run pytest`. The numbers that go in
    `data/logs/NOTES-store-perf.md` come from `scripts/store_perf_probe.py` with 100 race
    iterations; this runs the same code with a short race so the *report's shape* is guarded.
    """
    report = store_perf.measure(perf_store, race_iterations=10, race_max_seconds=10.0,
                                with_http=True, log=lambda line: print(line, flush=True))

    assert {"machine", "store", "paths", "plans", "race", "http", "maintain"} <= report.keys()
    for key in ("series_names_all", "series_names_host", "stats", "pick_res_90d",
                "history_5m", "history_24h", "history_90d",
                # task 02: both halves of a cached answer, so the before/after compares cold to
                # cold instead of calling a cache hit a 1 000x speed-up of a full scan
                "series_names_all_cached", "series_names_host_cached", "stats_cached",
                # task 02: the batched door against the loop it replaced, at both widths
                "pick_res_x27_24h", "pick_res_many_x27_24h", "history_x27_24h",
                "history_many_x27_24h", "pick_res_x27_90d", "pick_res_many_x27_90d",
                "history_x27_90d", "history_many_x27_90d"):
        assert report["paths"][key]["runs"] >= 1 and report["paths"][key]["max"] > 0.0
    for key in store_perf.HTTP_MEASURED_KEYS:
        assert key in report["http"], f"{key} is a hot path the operator's browser pays"
    assert set(report["http"]) == set(store_perf.HTTP_MEASURED_KEYS)
    # task 04: the report is the budget table's other customer. A budget crossed here is the phase
    # saying "the regression is back", and a budget this report cannot price is a budget nobody
    # checks any more — which is why `budget_rows` reports the missing ones as failures.
    assert store_perf.missing_budget_keys(report) == []
    assert report["budgets_ok"], "a pinned budget is crossed:\n" + "\n".join(
        store_perf.budget_violations(report))
    for key, plan_rows in report["plans"].items():
        assert plan_rows, f"{key}: no EXPLAIN QUERY PLAN row — the statement was not explained"

    # one connection: the store lock serialises, so an insert never sees SQLITE_BUSY here
    assert report["race"]["single_connection"]["busy_errors"] == 0
    # two connections: the gap-4 failure mode is a *raised* error, and a short-race pass here still
    # has to have attempted inserts for that to mean anything. The saturated rows (task 01's shape)
    # are reported for comparability only; the pass line is the cadence-shaped one, and the
    # whole-window control beside it is what proves the race can still see the failure.
    for key in ("two_connections_prod_timeout", "two_connections_no_timeout",
                "poller_during_windowed_maintain", "poller_during_full_window_maintain"):
        entry = report["race"][key]
        assert entry["attempted"] >= 1 and entry["attempted"] <= 10
        assert entry["inserts"] + entry["busy_errors"] == entry["attempted"]
        assert entry["notes"], "every race says what it actually ran"
    windowed = report["race"]["poller_during_windowed_maintain"]
    assert windowed["busy_errors"] == 0, (
        f"a poller insert lost to a windowed rollup: {windowed}")
    assert report["race"]["poller_during_full_window_maintain"]["waiters"] >= 1, (
        "the control pass must contend, or the windowed zero proves nothing")
    # the write transaction, per window, on a scratch copy — windowing has to show up as a shorter
    # write for the same rows, which is the whole point of task 03
    passes = report["maintain"]["passes"]
    assert set(passes) == {label for label, _offset, _reset in store_perf.MAINTAIN_PASSES}
    assert all(entry["seconds"] > 0.0 for entry in passes.values())
    assert passes["full_window"]["seconds"] > passes["steady"]["seconds"], (
        "the control measured no slower than the windowed pass — the copy or the window is wrong")
    assert passes["full_window"]["rows_written"] > passes["steady"]["rows_written"]
    # criterion 1's cost bound. Ten minutes of missed buckets is ten times the rollup work of one
    # minute and costs under twice the wall clock at this size because the fixed costs dominate; the
    # slack is the prune, which is deliberately **not** windowed and therefore deletes as many rows
    # as aged out during the jump whatever the rollup window was.
    assert passes["catch_up_10m"]["seconds"] <= 2 * passes["steady"]["seconds"] + 0.10, (
        f"catch-up {passes['catch_up_10m']['seconds']:.3f} s vs steady "
        f"{passes['steady']['seconds']:.3f} s")
    # criterion 2's shape, at production scale: a window the planner can seek. Without statistics it
    # prices a range as a fraction of the table and reads the table for it (`SCAN sample`) — which is
    # how a "windowed" rollup came to cost 328 ms while matching nothing.
    for key, plan_rows in report["maintain"]["plans"].items():
        assert plan_rows, f"{key}: nothing was explained"
        assert not any(row.startswith("SCAN") for row in plan_rows), (
            f"{key} still reads the whole table: {plan_rows}")
    assert report["maintain"]["wal_within_cap"], (
        f"the WAL escaped journal_size_limit across two ticks: {report['maintain']['two_steady_cycles']}")
    assert report["store_after"]["integrity"] == "unchanged", (
        f"the measurements changed the baseline: {report['store_after']['integrity']}")

    print()
    print(store_perf.render_markdown(report))