dashboards/scripts/watch-llm-window.py

#!/usr/bin/env python3
"""Phase 23 task 03: watch the running dev server prove the LLM Δ window is one poll interval.

Sampled, not exhaustive: every ``LLM_INTERVAL_SECONDS`` (read back from ``/api/config``, so the
watch polls at the cadence the app itself uses) a signed-in pass reads ``/api/llm/snapshot`` and
records, per engine, the decode/prompt rate **with the ``Δ<window>`` the payload carried**, the
lifetime token counter (so the same samples can be re-diffed over the pre-phase 10 s window for
a counterfactual), and how many slots the engine says are working.

Nothing here writes. The only POST is the dashboard's own sign-in; no engine endpoint is called,
no ``/api/ups/command``, no SET. The workload is whatever the operator's engines are already
doing — loading a shared engine to make a prettier graph is not a test.

What it judges, and the numbers it judges them against:

* **the ramp** — within two polls of a slot *starting* the rate is within 8x of the steady rate of
  that busy period, and within two polls of it *ending* the rate has fallen away by 8x (the
  estimator's own ``JUMP_CONFIRM_RATIO``). Read that on the **ungated** Δ — the same samples
  re-diffed over the estimator's span — because the phase-13 jump gate legitimately holds a correct
  number for a poll, and on edges that are *isolated*: an edge followed by the next request inside
  the examined window is not a slot ending (the counter never stopped), and a slot's first polls
  decode ~0 because the engine is prefilling. Both are reported with their reason and not judged.
  Beside every edge is the identical re-diff over the **10 s** span this phase replaced, so the
  shape that was removed is in the record rather than in a claim: five polls of 1/5, 2/5, 3/5 of
  the truth, up and back down.
* **the label** — zero published values without their ``Δ<window>`` tag, and every tag naming
  the window actually used (``Δ2.0s`` on a healthy cadence, the honest sub-window when the
  floor or a stall engages, never a bare number).
* **the rollups** — the stored 1 m averages of ``llm.decode_tokens_per_second`` inside the watch
  window stay in this fleet's measured band (gap 9: the real maxima are 119.9 gufo / 171.7 vLLM;
  above ~300 tok/s is arithmetic, not work), read straight out of the dev store on a read-only
  connection.

Run it while the dev server keeps polling the fleet::

    set -a; . ./.env.dev; set +a
    DASH_LLM_WATCH_SECONDS=720 uv run python scripts/watch-llm-window.py

Writes ``data/logs/llm-window-watch.tsv`` (the series), ``llm-window-watch.json`` (the verdict)
and ``llm-window-watch.log`` (the same verdict in prose). ``data/`` is gitignored: local evidence
for ``AGENTS.md`` gap 9, never a tracked artifact.
"""

from __future__ import annotations

import json
import math
import os
import sqlite3
import statistics
import sys
import time
from pathlib import Path
from typing import Any, cast

import httpx

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

BASE_URL = os.environ.get("DASH_WATCH_URL", "http://127.0.0.1:8090")
ADMIN_PASSWORD = os.environ.get("DASH_ADMIN_PASSWORD", "")
DB_PATH = Path(os.environ.get("DASH_WATCH_DB", ROOT / "data" / "dashboards.sqlite3"))
OUT = Path(os.environ.get("DASH_LLM_WATCH_OUT", ROOT / "data" / "logs"))
DURATION_SECONDS = float(os.environ.get("DASH_LLM_WATCH_SECONDS", "720"))

#: Gap 9's pass line: this fleet's real maxima are 119.9 (gufo) / 171.7 (vLLM) tok/s, so a
#: value above this is arithmetic on a collapsed window, not work the engine did.
CEILING_TPS = 300.0

#: The estimator's own jump ratio (`rates.JUMP_CONFIRM_RATIO`), used here as the "is this
#: number still ramping?" bound so the pass line and the hold behaviour are one number.
JUMP_RATIO = 8.0

#: How many polls after a transition are examined, and which one has to be settled.
POLLS_AFTER_TRANSITION = 6
SETTLED_BY_POLL = 2

#: The window the phase-23 defect was made of, for the counterfactual re-diff.
PRE_PHASE_SPAN = 10.0

#: `rates.MAX_WINDOW_RATIO` — the reconstruction below mirrors the estimator's fallback.
MAX_WINDOW_RATIO = 2.0

#: A busy run shorter than this cannot have a "steady" to settle to, so it is reported and not
#: judged (a one-poll request is a burst, not a ramp).
MIN_RUN_TO_JUDGE = 3

METRIC = "llm.decode_tokens_per_second"


def _num(value: object) -> float | None:
    return float(value) if isinstance(value, (int, float)) and not isinstance(value, bool) else None


def load_rows(path: Path) -> list[dict]:
    """Re-read a previous watch's TSV so a verdict can be re-derived without re-watching.

    The file carries the value, the window it was measured over, the engine's own busy count and
    the raw counter, which is everything the judgement needs — so re-running the pass line on
    captured evidence costs no new traffic and stays the *same* measurement.
    """
    rows: list[dict] = []
    with path.open() as fh:
        header = fh.readline().rstrip("\n").split("\t")
        for line in fh:
            fields = dict(zip(header, line.rstrip("\n").split("\t"), strict=True))
            rows.append({
                "ts": _parse(fields.get("ts")),
                "wall": _parse(fields.get("wall")) or 0.0,
                "engine": fields.get("engine", "?"),
                "decode": _parse(fields.get("decode")),
                "window": _parse(fields.get("window")),
                "source": fields.get("source", ""),
                "prompt": _parse(fields.get("prompt")),
                "prompt_window": _parse(fields.get("prompt_window")),
                "busy": _parse(fields.get("busy")) or 0.0,
                "tokens_total": _parse(fields.get("tokens_total")),
                "generated_total": _parse(fields.get("generated_total")),
            })
    return rows


def _parse(text: str | None) -> float | None:
    """The TSV writes `n/a` for a missing value; everything else is a float."""
    if text is None or text == "n/a":
        return None
    try:
        return float(text)
    except ValueError:
        return None


def sign_in(client: httpx.Client) -> None:
    """One sign-in on the client we use for the whole watch; `testpass` in dev."""
    response = client.post("/api/auth/login", json={"password": ADMIN_PASSWORD})
    response.raise_for_status()
    if "dash_session" not in response.headers.get("set-cookie", ""):
        raise SystemExit("sign-in returned no session cookie")


def llm_interval(client: httpx.Client) -> float:
    """The app's own LLM cadence — the watch polls at it, not at a guessed 2 s."""
    config = client.get("/api/config").json()
    return float(config["llm"]["interval_seconds"])


def sample_row(payload: dict, engine: dict) -> dict[str, object]:
    """One (poll, engine) row: value, the Δ it was measured over, activity, raw counters."""
    metrics = engine.get("metrics") or {}
    decode = metrics.get("decode_tokens_per_second") or {}
    prompt = metrics.get("prompt_tokens_per_second") or {}
    sessions = engine.get("sessions") or {}
    busy = _num(sessions.get("active")) or 0.0
    processing = _num((metrics.get("requests_processing") or {}).get("value")) or 0.0
    return {
        "ts": _num(payload.get("ts")),
        "wall": time.time(),
        "engine": engine.get("engine"),
        "decode": _num(decode.get("value")),
        "window": _num(decode.get("window")),
        "source": decode.get("source") or "",
        "prompt": _num(prompt.get("value")),
        "prompt_window": _num(prompt.get("window")),
        "busy": max(busy, processing),
        "tokens_total": _num((metrics.get("tokens_total") or {}).get("value")),
        # the counter this engine's decode rate is actually diffed from: gufo publishes
        # `tokens_generated_total`, vLLM `generation_tokens_total` (sources/llm.py's `measured()`)
        "generated_total": (_num((metrics.get("tokens_generated_total") or {}).get("value"))
                            if (metrics.get("tokens_generated_total") or {}).get("value") is not None
                            else _num((metrics.get("generation_tokens_total") or {}).get("value"))),
    }


def take_samples(client: httpx.Client, interval: float) -> tuple[list[dict], int, float, float]:
    """Poll until the clock runs out; a failed read is counted and the watch goes on."""
    rows: list[dict[str, object]] = []
    errors = 0
    started = time.time()
    deadline = started + DURATION_SECONDS
    while time.time() < deadline:
        loop_start = time.time()
        try:
            response = client.get("/api/llm/snapshot")
            response.raise_for_status()
            payload = response.json()
            for engine in payload.get("engines") or []:
                rows.append(sample_row(payload, engine))
        except (httpx.HTTPError, ValueError, KeyError) as exc:  # a watcher reports, it does not fail
            errors += 1
            print(f"poll error: {type(exc).__name__}: {exc}", file=sys.stderr)
        time.sleep(max(0.05, interval - (time.time() - loop_start)))
    return rows, errors, started, time.time()


def rollup_window(db: sqlite3.Connection, since: float, until: float) -> dict[str, Any]:
    """The stored 1 m averages inside the watch window, per engine (read-only connection)."""
    out: dict[str, Any] = {"rows": 0, "max": None, "per_engine": {}}
    found = db.execute(
        "SELECT host, ts, value FROM sample WHERE metric=? AND res='1m' AND ts>=? AND ts<=? "
        "ORDER BY ts",
        (METRIC, since, until),
    ).fetchall()
    per_engine: dict[str, list[float]] = {}
    for host, _ts, value in found:
        number = _num(value)
        if number is None:
            continue
        per_engine.setdefault(str(host), []).append(number)
    values = [n for numbers in per_engine.values() for n in numbers]
    newest = db.execute(
        "SELECT MAX(ts) FROM sample WHERE metric=? AND res='1m'", (METRIC,)
    ).fetchone()[0]
    out["rows"] = len(found)
    out["max"] = max(values) if values else None
    out["per_engine"] = {
        host: {"rows": len(nums), "max": max(nums), "median": statistics.median(nums)}
        for host, nums in per_engine.items()
    }
    out["newest_1m_ts"] = _num(newest)
    out["lag_seconds"] = (until - float(newest)) if newest else None
    return out


def raw_band(db: sqlite3.Connection, since: float, until: float) -> dict[str, object]:
    """The raw samples in the same window — the row-level view under the 1 m averages."""
    found = db.execute(
        "SELECT host, MAX(value) FROM sample WHERE metric=? AND res='raw' AND ts>=? AND ts<=? "
        "GROUP BY host",
        (METRIC, since, until),
    ).fetchall()
    return {str(host): value for host, value in found if value is not None}


def rediff(series: list[dict], span: float) -> list[float | None]:
    """What a `span`-window estimator would have printed, from the counters in the payload.

    Same samples, same arithmetic as `rates.RateEstimator.rate` (base = the oldest sample inside
    the window), without the phase-13 gates: this is the counterfactual that shows the ramp the
    10 s window used to draw, not a second measurement of the engine. The index of the returned
    list matches `series`, so a transition can point at the row it happened on.
    """
    out: list[float | None] = []
    for index, row in enumerate(series):
        counter = row["generated_total"]
        if counter is None or row["ts"] is None:
            out.append(None)
            continue
        prior = [(p["ts"], float(p["generated_total"])) for p in series[:index]
                 if p["ts"] is not None and p["generated_total"] is not None]
        if not prior:
            out.append(None)
            continue
        inside = [point for point in prior if row["ts"] - point[0] <= span]
        if inside:
            base = inside[0]
        else:
            # The estimator's own window-empty fallback (`MAX_WINDOW_RATIO x span`): a cadence
            # of exactly `span` leaves the previous sample a hair *outside* the window, and
            # without this the reconstruction would answer n/a where the app answers a rate.
            fallback = prior[-1]
            if row["ts"] - fallback[0] > MAX_WINDOW_RATIO * span:
                out.append(None)
                continue
            base = fallback
        dt = row["ts"] - base[0]
        dv = float(counter) - base[1]
        out.append(None if dt <= 0 or dv < 0 else dv / dt)
    return out


def transitions(rows: list[dict], engine: str) -> list[dict[str, object]]:
    """Every busy-edge on this engine and the polls on each side of it.

    The busy signal is the engine's own (`sessions.active` for gufo, the running count for
    vLLM, `requests_processing` as the fallback), never an inference from the rate: judging the
    rate against itself would call a held poll a slot ending. No verdict here — `judge_edges()`
    makes those, because a verdict needs the re-diffed Δ as well as the published value.
    """
    series = [r for r in rows if r["engine"] == engine]
    found: list[dict[str, object]] = []
    for index in range(1, len(series)):
        before, after = series[index - 1]["busy"], series[index]["busy"]
        if before == after:
            continue
        kind = "start" if after > before else "end"
        run = _run(series, index if kind == "start" else index - 1,
                   forward=kind == "start")
        examined = (run[:POLLS_AFTER_TRANSITION] if kind == "start"
                    else series[index:index + POLLS_AFTER_TRANSITION])
        found.append({
            "kind": kind,
            "engine": engine,
            "ts": series[index]["ts"],
            "index": index,
            "busy_before": before,
            "busy_after": after,
            "run_polls": len(run),
            "values": [r["decode"] for r in examined],
            "windows": [r["window"] for r in examined],
            # the busy state flickering back inside the examined window means the engine never
            # went idle (one request ended and the next started within a poll or two) — an edge
            # that is not an edge, so decay cannot be judged from it
            "flips_back": any((series[j]["busy"] > 0) != (after > 0)
                              for j in range(index, min(index + POLLS_AFTER_TRANSITION, len(series)))),
            "truncated": len(examined) < SETTLED_BY_POLL if kind == "end" else index + len(run) >= len(series),
        })
    return found


def judge_edges(edges: list[dict], series: list[dict], delta: list[float | None],
                pre_phase: list[float | None], app_span: float) -> None:
    """Fill in each edge's steady, ratios and verdict, in place.

    Two columns are judged, and the difference between them is the whole point of the task:

    * **`published`** — what `/api/llm/snapshot` answered, which carries the phase-13 jump gate
      as well as the window (a value held one poll is designed behaviour, not a ramp);
    * **`delta`** — the same samples re-diffed over the estimator's own span with no gate, which
      is the *window* on its own. This is the column the pass line is read from, because the
      gate can hold a correct number for a poll and no window can make it settle sooner.

    `pre_phase` is the identical re-diff over the 10 s span this phase replaced; it is reported
    beside the other two so the reader can see the shape that was removed rather than take the
    claim on trust.
    """
    for edge in edges:
        index = int(edge["index"])
        kind = str(edge["kind"])
        if kind == "start":
            run = series[index:index + int(edge["run_polls"])]
            pool = [r["decode"] for r in run if r["decode"]]
        else:
            run = _run(series, index - 1, forward=False)
            pool = [r["decode"] for r in run if r["decode"]]
        steady = statistics.median(pool) if len(pool) >= MIN_RUN_TO_JUDGE else None
        edge["steady"] = steady
        edge["delta_values"] = delta[index:index + POLLS_AFTER_TRANSITION]
        edge["pre_phase_values"] = pre_phase[index:index + POLLS_AFTER_TRANSITION]
        edge["pre_phase_span"] = PRE_PHASE_SPAN
        edge["app_span"] = app_span
        edge["ratios"] = _ratios(cast("list[float | None]", edge["values"]), steady, kind)
        edge["delta_ratios"] = _ratios(cast("list[float | None]", edge["delta_values"]), steady, kind)
        edge["pre_phase_ratios"] = _ratios(cast("list[float | None]", edge["pre_phase_values"]),
                                           steady, kind)
        edge["settled_by_poll"] = _settled_by(cast("list[float | None]", edge["ratios"]), kind)
        edge["delta_settled_by_poll"] = _settled_by(cast("list[float | None]", edge["delta_ratios"]),
                                                    kind)
        edge["pre_phase_settled_by_poll"] = _settled_by(
            cast("list[float | None]", edge["pre_phase_ratios"]), kind)
        # a published value that repeats while the ungated Δ moved is the jump gate holding
        held = any(v is not None and i > 0 and v == edge["values"][i - 1]
                   for i, v in enumerate(edge["values"]) if i <= SETTLED_BY_POLL)
        edge["jump_gate_held"] = bool(held and edge["settled_by_poll"] != edge["delta_settled_by_poll"])
        if edge["flips_back"]:
            edge["judgable"] = False
            edge["reason"] = "busy again within the examined window — the engine never went idle"
        elif edge["truncated"]:
            edge["judgable"] = False
            edge["reason"] = "no polls after the edge inside the watch"
        elif steady is None:
            edge["judgable"] = False
            edge["reason"] = f"busy run of {edge['run_polls']} poll(s) has no steady to settle to"
        else:
            edge["judgable"] = True
            edge["reason"] = None


def _run(series: list[dict], index: int, forward: bool) -> list[dict]:
    """The consecutive polls on the busy side of an edge, starting at `index`."""
    if series[index]["busy"] <= 0 and forward:
        return []
    out = [series[index]]
    step = 1 if forward else -1
    cursor = index + step
    while 0 <= cursor < len(series) and series[cursor]["busy"] > 0:
        out.append(series[cursor])
        cursor += step
    return out


#: The cap on "the rate is now effectively zero", so the report stays JSON and not `Infinity`.
DECAY_CAP = 999.0


def _ratios(values: list[float | None], steady: float | None, kind: str) -> list[float | None]:
    """How far each poll after the edge still is from the steady rate of the other side.

    A **start** is settled when the value reaches the busy steady (ratio ~1); an **end** is
    settled when it has fallen away from the steady it used to have (ratio >= 8, capped), so one
    number reads the same way on both edges: bigger is closer to done on an end, closer to 1 on
    a start.
    """
    if not steady or steady <= 0:
        return [None] * len(values)
    out: list[float | None] = []
    for value in values:
        if value is None:
            out.append(None)
        elif kind == "start":
            out.append(value / steady)
        else:
            out.append(DECAY_CAP if value <= 0 else min(DECAY_CAP, steady / value))
    return out


def judge(rows: list[dict], edges: list[dict], rollups: dict, app_span: float) -> dict[str, Any]:
    """The four pass lines of the task, as numbers."""
    verdict: dict[str, Any] = {"ceiling_tps": CEILING_TPS, "jump_ratio": JUMP_RATIO,
                                  "polls": len(rows), "app_span": app_span,
                                  "engines": {}, "transitions": edges, "counterfactual": {}}
    problems: list[str] = []
    for engine in sorted({str(r["engine"]) for r in rows}):
        series = [r for r in rows if r["engine"] == engine]
        values = [r["decode"] for r in series if r["decode"] is not None]
        unlabelled = [r for r in series if r["decode"] is not None and r["window"] is None]
        windows = [r["window"] for r in series if r["window"] is not None]
        over = [v for v in values if v > CEILING_TPS]
        verdict["engines"][engine] = {
            "rows": len(series),
            "null_values": sum(1 for r in series if r["decode"] is None),
            "max": max(values) if values else None,
            "median": statistics.median(values) if values else None,
            "window_min": min(windows) if windows else None,
            "window_max": max(windows) if windows else None,
            "values_without_delta_label": len(unlabelled),
            "values_over_ceiling": len(over),
            "over_ceiling_examples": over[:5],
        }
        if unlabelled:
            problems.append(f"{engine}: {len(unlabelled)} values printed without a Δ label")
        if over:
            problems.append(f"{engine}: {len(over)} values above {CEILING_TPS} tok/s")
        cf_app = rediff(series, app_span)
        cf10 = rediff(series, PRE_PHASE_SPAN)
        verdict["counterfactual"][engine] = {"app_span": _shape(cf_app),
                                             "pre_phase_span": _shape(cf10)}
        judge_edges([e for e in edges if e["engine"] == engine], series, cf_app, cf10, app_span)
    settled = [e for e in edges if e["kind"] == "start" and e["judgable"]]
    decayed = [e for e in edges if e["kind"] == "end" and e["judgable"]]
    verdict["transitions_checked"] = {
        "starts": len(settled), "ends": len(decayed),
        "pass_line": f"the ungated Δ over the app's own {app_span}s span is inside "
                     f"{JUMP_RATIO}x of steady by poll {SETTLED_BY_POLL}",
        "unsettled_by_poll_2": [
            {"kind": e["kind"], "ts": e["ts"], "delta_ratios": e["delta_ratios"]}
            for e in settled + decayed
            if not _settled(e["delta_ratios"], SETTLED_BY_POLL, e["kind"])],
        # how much of steady the first two polls delivered, this phase's window and the one it
        # replaced: the pre-phase 10 s window sat at ~1/5 and climbed, which is the complaint
        "poll_1_of_steady_at_a_start": _at_poll(settled, "ratios", 1),
        "poll_2_of_steady_at_a_start": _at_poll(settled, "ratios", 2),
        "poll_1_of_steady_at_a_start_ungated": _at_poll(settled, "delta_ratios", 1),
        "poll_2_of_steady_at_a_start_ungated": _at_poll(settled, "delta_ratios", 2),
        "poll_1_of_steady_at_a_start_pre_phase": _at_poll(settled, "pre_phase_ratios", 1),
        "poll_2_of_steady_at_a_start_pre_phase": _at_poll(settled, "pre_phase_ratios", 2),
        # the headline: which poll settles, this span vs the 10 s span, over the same samples
        "settled_by_poll_median": {
            "app_span": _median_of(settled + decayed, "delta_settled_by_poll"),
            "pre_phase_span": _median_of(settled + decayed, "pre_phase_settled_by_poll"),
            "published": _median_of(settled + decayed, "settled_by_poll"),
        },
        # and how far from steady the second poll still is, as a median |ln(ratio)|: the 8x line
        # alone is satisfied by a 10 s window at 0.25 of steady, so this is the number that says
        # the ramp is gone rather than merely inside the tolerance
        "median_log_distance_from_steady_at_poll_2": {
            "app_span": _median_log_distance(settled, "delta_ratios", 2),
            "pre_phase_span": _median_log_distance(settled, "pre_phase_ratios", 2),
        },
        "jump_gate_holds": sum(1 for e in edges if e.get("jump_gate_held")),
        "reported_not_judged": [{"kind": e["kind"], "ts": e["ts"], "reason": e["reason"]}
                                for e in edges if not e["judgable"]],
    }
    for e in settled + decayed:
        if not _settled(e["delta_ratios"], SETTLED_BY_POLL, e["kind"]):
            problems.append(f"{e['kind']} at ts={e['ts']} still ramping at poll "
                            f"{SETTLED_BY_POLL}: ungated Δ ratios={e['delta_ratios']}, "
                            f"published ratios={e['ratios']}")
    rollup_max = rollups.get("max")
    verdict["rollups"] = rollups
    verdict["raw_max_in_window"] = rollups.get("raw_max_in_window")
    if rollup_max is not None and rollup_max > CEILING_TPS:
        problems.append(f"1 m rollups peaked at {rollup_max:.1f} tok/s, above the pass line")
    # An empty pass is not a pass: "no transition happened" must never read as "the ramp is
    # gone", and a watch that recorded nothing cannot speak about the engines at all.
    if not rows:
        problems.append("no rows were recorded — the watch proves nothing")
    elif not any(r["decode"] for r in rows):
        problems.append("no engine decoded anything during the watch — no rate to judge")
    elif not (settled or decayed):
        problems.append("no busy-edge held long enough to judge (a slot start or end of "
                        f"{MIN_RUN_TO_JUDGE}+ polls, not followed by the next request) — "
                        "run the watch again; the ramp claim is unproven without one")
    verdict["problems"] = problems
    verdict["pass"] = not problems
    return verdict


def _at_poll(edges: list[dict], key: str, poll: int) -> list[float]:
    """The ratio at `poll` for every edge that has one, rounded for the report."""
    return [round(float(e[key][poll - 1]), 3) for e in edges  # type: ignore[index]
            if e.get(key) and len(e[key]) >= poll and e[key][poll - 1] is not None]  # type: ignore[arg-type]


def _median_of(edges: list[dict], key: str) -> float | None:
    values = [e[key] for e in edges if e.get(key) is not None]
    return statistics.median(values) if values else None


def _median_log_distance(edges: list[dict], key: str, poll: int) -> float | None:
    """Median |ln(value / steady)| at `poll` — 0 is exactly steady, ln(8) is the pass line."""
    logs = [abs(math.log(r)) for r in _at_poll(edges, key, poll) if r > 0]
    return round(statistics.median(logs), 3) if logs else None


def _settled_by(ratios: list[float | None], kind: str) -> int | None:
    """Which poll first sits inside the pass line, or None if none of them does."""
    for poll, ratio in enumerate(ratios, start=1):
        if _settled([ratio], 1, kind):
            return poll
    return None


def _settled(ratios: list[float | None], by_poll: int, kind: str) -> bool:
    """A start has *reached* steady by `by_poll`; an end has *left* it by at least 8x.

    A held poll (the jump gate repeating the previous value) counts as settled: the hold is the
    designed behaviour, and the following poll is the honest publish.
    """
    if len(ratios) < by_poll:
        return False
    ratio = ratios[by_poll - 1]
    if ratio is None:
        return False
    if kind == "end":
        return ratio >= JUMP_RATIO
    return 1.0 / JUMP_RATIO <= ratio <= JUMP_RATIO


def _shape(values: list[float | None]) -> dict[str, object]:
    numbers = [v for v in values if v is not None]
    return {"n": len(numbers), "max": max(numbers) if numbers else None,
            "median": statistics.median(numbers) if numbers else None}


def write_outputs(rows: list[dict], verdict: dict, errors: int,
                  started: float, stopped: float) -> None:
    OUT.mkdir(parents=True, exist_ok=True)
    tsv = OUT / "llm-window-watch.tsv"
    with tsv.open("w") as fh:
        fh.write("ts\twall\tengine\tdecode\twindow\tΔ_label\tprompt\tprompt_window\t"
                 "busy\ttokens_total\tgenerated_total\tsource\n")
        for row in rows:
            label = f"Δ{row['window']:.1f}s" if row["window"] else "n/a"
            fh.write("\t".join([
                _fmt(row["ts"]), _fmt(row["wall"]), str(row["engine"]), _fmt(row["decode"]),
                _fmt(row["window"]), label, _fmt(row["prompt"]), _fmt(row["prompt_window"]),
                _fmt(row["busy"]), _fmt(row["tokens_total"]), _fmt(row["generated_total"]),
                str(row["source"]),
            ]) + "\n")
    (OUT / "llm-window-watch.json").write_text(json.dumps({
        "start": started, "stop": stopped,
        "start_iso": _iso(started), "stop_iso": _iso(stopped),
        "polls": len(rows), "poll_errors": errors,
    } | verdict, indent=2, sort_keys=True))
    with (OUT / "llm-window-watch.log").open("w") as fh:
        fh.write(_report(verdict, errors, started, stopped))


def _report(verdict: dict, errors: int, started: float, stopped: float) -> str:
    lines = [
        f"phase 23 live window watch  {_iso(started)} -> {_iso(stopped)}  "
        f"({stopped - started:.0f}s, {verdict['polls']} rows, {errors} poll errors)",
    ]
    for engine, stats in sorted(verdict["engines"].items()):
        lines.append(
            f"  {engine}: rows={stats['rows']} null={stats['null_values']} "
            f"max={_s(stats['max'])} median={_s(stats['median'])} "
            f"window={_s(stats['window_min'])}-{_s(stats['window_max'])} "
            f"unlabelled={stats['values_without_delta_label']} "
            f"over{CEILING_TPS:.0f}={stats['values_over_ceiling']}"
        )
    tcheck = verdict["transitions_checked"]
    lines.append(f"  transitions: {tcheck['starts']} judgable start(s), {tcheck['ends']} judgable "
                 f"end(s); unsettled by poll {SETTLED_BY_POLL}: "
                 f"{len(tcheck['unsettled_by_poll_2'])}; pass line: {tcheck['pass_line']}")
    for poll in (1, 2):
        got = tcheck[f"poll_{poll}_of_steady_at_a_start"]
        pre = tcheck[f"poll_{poll}_of_steady_at_a_start_pre_phase"]
        if got:
            lines.append(f"  poll {poll} of a start delivered {got} of steady with this phase's "
                         f"Δ, where the 10 s window it replaced delivered {pre}")
    distance = tcheck["median_log_distance_from_steady_at_poll_2"]
    if distance["app_span"] is not None:
        lines.append(f"  median |ln(value/steady)| at poll 2: {distance['app_span']} with this "
                     f"phase's Δ vs {distance['pre_phase_span']} with the 10 s window "
                     f"(ln({JUMP_RATIO:g}) = {math.log(JUMP_RATIO):.2f} is the pass line)")
    settled_by = tcheck["settled_by_poll_median"]
    lines.append(f"  median poll that settles an edge: {settled_by['app_span']} with this "
                 f"phase's window (published: {settled_by['published']}), "
                 f"{settled_by['pre_phase_span']} with the 10 s window")
    lines.append(f"  jump-gate holds inside an examined window: {tcheck['jump_gate_holds']}; "
                 f"edges reported but not judged: {len(tcheck['reported_not_judged'])}")
    for edge in verdict["transitions"]:
        lines.append(
            f"    {edge['engine']}/{edge['kind']:5s} ts={_s(edge['ts'])} busy "
            f"{_s(edge['busy_before'])}->{_s(edge['busy_after'])} run={edge['run_polls']}poll "
            f"steady={_s(edge['steady'])}"
            + ("" if edge["judgable"] else f"  [{edge['reason']}]")
            + ("  [jump gate held a poll]" if edge.get("jump_gate_held") else "")
        )
        lines.append(f"        published      ="
                     f"{[None if v is None else round(v, 2) for v in edge['values']]}")
        lines.append(f"        Δ{edge.get('app_span')}s (this phase)="
                     f"{[None if v is None else round(v, 2) for v in edge['delta_values']]} "
                     f"settled by poll {edge.get('delta_settled_by_poll')}")
        lines.append(f"        Δ{edge.get('pre_phase_span')}s (before)="
                     f"{[None if v is None else round(v, 2) for v in edge['pre_phase_values']]} "
                     f"settled by poll {edge.get('pre_phase_settled_by_poll')}")
    for engine, cf in sorted(verdict["counterfactual"].items()):
        lines.append(f"  whole-window re-diff {engine}: app span {cf['app_span']} | "
                     f"pre-phase 10 s span {cf['pre_phase_span']}")
    roll = verdict["rollups"]
    lines.append(f"  1 m rollups in window: rows={roll['rows']} max={_s(roll['max'])} "
                 f"per_engine={json.dumps(roll['per_engine'])}")
    lines.append(f"  raw max in window: {json.dumps(verdict['raw_max_in_window'])}")
    lines.append("  RESULT: " + ("PASS" if verdict["pass"] else "FAIL"))
    for problem in verdict["problems"]:
        lines.append(f"    ! {problem}")
    return "\n".join(lines) + "\n"


def _fmt(value: object) -> str:
    return "n/a" if value is None else f"{value}"


def _s(value: object) -> str:
    if value is None:
        return "n/a"
    return f"{float(value):.3f}" if isinstance(value, (int, float)) else str(value)


def _iso(ts: float) -> str:
    return time.strftime("%Y-%m-%dT%H:%M:%S%z", time.localtime(ts))


def wait_for_rollups(db_path: Path, since: float, until: float,
                     timeout: float = 150.0) -> dict[str, Any]:
    """1 m rows appear on the maintenance tick (60 s), so give it two ticks, no more."""
    deadline = time.time() + timeout
    rollups: dict[str, Any] = {"rows": 0, "max": None, "per_engine": {}}
    while True:
        with sqlite3.connect(f"file:{db_path}?mode=ro", uri=True, timeout=30) as db:
            rollups = rollup_window(db, since, until)
            rollups["raw_max_in_window"] = raw_band(db, since, until)
        newest = rollups.get("newest_1m_ts")
        if rollups["rows"] and isinstance(newest, (int, float)) and newest >= until - 65:
            return rollups
        if time.time() >= deadline:
            return rollups
        time.sleep(5)


def main() -> int:
    OUT.mkdir(parents=True, exist_ok=True)
    replay = os.environ.get("DASH_LLM_WATCH_REPLAY")
    if replay:
        rows = load_rows(Path(replay))
        if not rows:
            raise SystemExit(f"{replay} has no rows")
        errors, interval = 0, float(os.environ.get("DASH_LLM_WATCH_SPAN", "2"))
        started = float(rows[0]["ts"] or 0)
        stopped = float(rows[-1]["ts"] or 0)
        print(f"replaying {replay}: {len(rows)} rows, "
              f"{_iso(started)} -> {_iso(stopped)} (no traffic sent)")
    else:
        if not ADMIN_PASSWORD:
            raise SystemExit("DASH_ADMIN_PASSWORD is required (it is `testpass` in .env.dev)")
        with httpx.Client(base_url=BASE_URL, timeout=30, follow_redirects=False) as client:
            sign_in(client)
            interval = llm_interval(client)
            print(f"signed in; polling /api/llm/snapshot every {interval:.1f}s "
                  f"for {DURATION_SECONDS:.0f}s (read-only)")
            rows, errors, started, stopped = take_samples(client, interval)
    rollups = wait_for_rollups(DB_PATH, started, stopped)
    edges = [edge for engine in sorted({str(r["engine"]) for r in rows})
             for edge in transitions(rows, engine)]
    verdict = judge(rows, edges, rollups, interval)
    write_outputs(rows, verdict, errors, started, stopped)
    print(_report(verdict, errors, started, stopped))
    return 0 if verdict["pass"] else 1


if __name__ == "__main__":
    raise SystemExit(main())