dashboards/tests/test_llm_source.py

"""LLM engine parsing: real captures, provenance on every number, absent never 0.

The engines do not export the names people expect, so these tests are pinned to captures of
the deployed servers (tests/fixtures/metrics_*.txt), not to upstream documentation.
"""
from __future__ import annotations

import json
import re
from collections.abc import Callable, Mapping, Sequence
from pathlib import Path
from typing import Any

import pytest

from dashboards.prom import Prom
from dashboards.rates import MIN_WINDOW_SECONDS, RateEstimator
from dashboards.severity import classify
from dashboards.sources.llm import (
    RADIANCE_HISTOGRAMS,
    RADIANCE_OPTIONAL,
    RADIANCE_SERIES,
    engine_result,
    fleet_totals,
    parse_gufo_metrics,
    parse_radiance_metrics,
    parse_vllm_metrics,
    rate_key,
    series_names,
)
from rate_window_contract import LLM_WINDOW, assert_window, delta_label
from slots_contract import load_slots_capture


def bump_metric(text: str, name: str, delta: float) -> str:
    """Return the exposition with one counter increased — a scrape `dt` later."""
    pattern = re.compile(rf"^({re.escape(name)}(?:\{{[^}}]*\}})?) (\S+)$", re.M)

    def replace(match: re.Match[str]) -> str:
        return f"{match.group(1)} {float(match.group(2)) + delta}"

    bumped, count = pattern.subn(replace, text)
    assert count, f"{name} not present in the capture"
    return bumped


def poll_engine(metrics_text: Any, engine: str, span: float, stamps: Sequence[float],
                growth: Mapping[str, Callable[[float], float]]) -> list[dict[str, Any]]:
    """Re-scrape one capture at `stamps` with counters growing as f(t), through one estimator.

    The production shape in miniature: **one** `RateEstimator(span=span)` for every poll (the
    object `app.build_hub` hands that source), each poll stamped with its own `now`, the
    exposition carrying the tokens produced since the last scrape. Returns the parsed metrics per
    poll so a pin reads the very cells `/api/llm/snapshot` serves — including their `window`.
    """
    parse = parse_gufo_metrics if engine == "gufo" else parse_vllm_metrics
    rates = RateEstimator(span=span)
    body = metrics_text(engine)
    carried = dict.fromkeys(growth, 0.0)
    scrapes: list[dict[str, Any]] = []
    for now in stamps:
        for name, produced in growth.items():
            value = produced(now)
            body = bump_metric(body, name, value - carried[name])
            carried[name] = value
        scrapes.append(parse(body, rates, now))
    return scrapes


# ---- gufo / llama.cpp ----------------------------------------------------- #

def test_gufo_counters_come_straight_from_the_capture(metrics_text) -> None:
    """The cell's value is the number the exposition declares, verbatim: no
    conversion, no scaling, and the series' own name is the source. Compared against
    the raw `Prom` parse of the same capture — never a value frozen at capture time
    (the counters reset on every engine restart)."""
    text = metrics_text("gufo")
    prom = Prom(text)
    metrics = parse_gufo_metrics(text)
    for key, name in (
        ("tokens_generated_total", "llamacpp:tokens_predicted_total"),
        ("tokens_prompt_total", "llamacpp:prompt_tokens_total"),
        ("tokens_prompt_cached_total", "llamacpp:prompt_tokens_cached_total"),
        ("requests_processing", "llamacpp:requests_processing"),
        ("device_lost_total", "gufo_device_lost_total"),
        ("kv_cache_usage_ratio", "llamacpp:kv_cache_usage_ratio"),
    ):
        assert metrics[key]["value"] == prom.scalar([name]), key
        assert metrics[key]["source"] == name, "the counter's own name is the source"


def test_gufo_speculative_accept_rate_is_accepted_over_draft(metrics_text) -> None:
    metrics = parse_gufo_metrics(metrics_text("gufo"))
    accept = metrics["spec_accept_rate"]["value"]
    accepted = metrics["spec_accepted_tokens_total"]["value"]
    draft_tokens = metrics["spec_draft_tokens_total"]["value"]
    assert accept == pytest.approx(accepted / draft_tokens)
    assert str(metrics["spec_accept_rate"]["source"]).startswith("computed")


def test_gufo_lifetime_tokens_include_cache_hits(metrics_text) -> None:
    """gufo's prompt counter excludes cache hits; the lifetime total must not."""
    metrics = parse_gufo_metrics(metrics_text("gufo"))
    total = metrics["tokens_total"]["value"]
    prompt = metrics["tokens_prompt_total"]["value"]
    cached = metrics["tokens_prompt_cached_total"]["value"] or 0.0
    generated = metrics["tokens_generated_total"]["value"]
    assert total == pytest.approx(prompt + cached + generated)


def test_gufo_throughput_prefers_measured_delta_over_the_native_gauge(metrics_text) -> None:
    """llamacpp:predicted_tokens_seconds is ONE request's last pass — never the headline."""
    text = metrics_text("gufo")
    rates = RateEstimator(span=10)
    now = 1_000.0
    first = parse_gufo_metrics(text, rates, now)
    assert "decode_tokens_per_second" not in first or first["decode_tokens_per_second"]["value"] is None

    text = bump_metric(text, "llamacpp:tokens_predicted_total", 400.0)
    metrics = parse_gufo_metrics(text, rates, now + 10.0)
    decode = metrics["decode_tokens_per_second"]
    assert decode["value"] == pytest.approx(40.0)
    assert decode["window"] == pytest.approx(10.0)
    assert "Δ" in decode["source"] and "tokens_predicted_total" in decode["source"]
    # the native gauge is still there, labelled, because it answers a different question —
    # verbatim the exposition's own number (a volatile gauge: no frozen value)
    assert metrics["decode_tokens_per_second_native"]["value"] == \
        Prom(text).scalar(["llamacpp:predicted_tokens_seconds"])
    assert metrics["decode_tokens_per_second_native"]["source"] == "llamacpp:predicted_tokens_seconds"


def test_gufo_the_window_the_llm_jobs_are_spanned_by_is_one_that_fills(metrics_text) -> None:
    """Phase 23: a 2 s span is a real window at the LLM cadence, and it kills the ramp.

    `app.build_hub` spans the LLM estimator by `LLM_INTERVAL_SECONDS` (the wiring is pinned in
    `tests/test_api.py`); this is what that span buys. A slot that starts mid-window is spread
    over the whole span by the wider window, so the hero climbs one fifth of the true rate per
    poll — the "ramp up/ramp down" of `TODO.md:8` — while a window of one poll interval reads
    the engine's speed on the first poll after it changes and 0 again on the first after it ends.

    The engine here is the capture plus a steady 50 tok/s that begins at t=1006, polled every
    2 s, which is the production cadence rather than a fast scrape: the estimator's
    `MIN_WINDOW_SECONDS` floor is 4x below this window, so nothing is being refused here.
    """
    text = metrics_text("gufo")
    started = 1_006.0                     # the slot starts here, at a steady 50 tok/s
    polls = [1_000.0, 1_002.0, 1_004.0, 1_006.0, 1_008.0, 1_010.0, 1_012.0]

    def series(span: float) -> list[tuple[float | None, float | None]]:
        rates = RateEstimator(span=span)
        body = text
        cells: list[tuple[float | None, float | None]] = []
        previous = 0.0
        for now in polls:
            generated = 50.0 * max(0.0, now - started)
            body = bump_metric(body, "llamacpp:tokens_predicted_total", generated - previous)
            previous = generated
            cell = parse_gufo_metrics(body, rates, now)["decode_tokens_per_second"]
            cells.append((cell["value"], cell.get("window")))
        return cells

    narrow = series(2.0)
    wide = series(10.0)
    # the new window: n/a on the first poll (no window yet), then the engine's real speed, and
    # every one of them labelled with the 2 s window it actually measured
    assert narrow[0] == (None, None)
    for _value, window in narrow[1:]:
        assert window == pytest.approx(2.0), "the label must name the window the value used"
    assert [v for v, _ in narrow[4:]] == [pytest.approx(50.0)] * 3, (
        f"a 2 s window reads the slot's speed at once, got {narrow}")
    # the old one: the same work, averaged across the polls the operator is watching
    assert [v for v, _ in wide[4:]] != [pytest.approx(50.0)] * 3, \
        f"the 10 s window no longer smooths, so this test's premise moved: {wide}"
    assert wide[4][0] is not None and wide[4][0] < 30.0, \
        f"expected the ramp to print well under the true rate, got {wide}"


def test_gufo_rate_cells_name_the_window_the_llm_jobs_are_measured_over(metrics_text) -> None:
    """Phase 23's payload pin: an LLM rate cell's `window` is the LLM cadence, label and all.

    `build_hub` spans the `llm` estimator by `LLM_INTERVAL_SECONDS` (the object-level pin is
    `tests/test_api.py`'s); this is the same span where the browser reads it — the `window` beside
    the value, the `computed:Δ<series>/Δt` provenance printed with it, and the `Δ2.0s` label
    `provenance()` builds from that number. **Both** throughput cells are checked, on the
    production 2 s cadence. `tests/test_host_source.py` is the other direction of the same pin:
    the host cells keep 10 s in the same payload shape.
    """
    started = 1_006.0
    stamps = [1_000.0, 1_002.0, 1_004.0, 1_006.0, 1_008.0, 1_010.0]
    scrapes = poll_engine(
        metrics_text, "gufo", LLM_WINDOW, stamps,
        {"llamacpp:tokens_predicted_total": lambda t: 50.0 * max(0.0, t - started),
         "llamacpp:prompt_tokens_total": lambda t: 200.0 * max(0.0, t - started)},
    )
    assert "window" not in scrapes[0]["decode_tokens_per_second"], \
        "one sample is not a window, so it must not grow a Δ label"
    for key, series, speed in (
        ("decode_tokens_per_second", "llamacpp:tokens_predicted_total", 50.0),
        ("prompt_tokens_per_second", "llamacpp:prompt_tokens_total", 200.0),
    ):
        cells = [scrape[key] for scrape in scrapes[1:]]
        for cell in cells:
            assert cell["value"] is not None, f"{key} went n/a on a healthy cadence: {cell}"
            assert_window(cell.get("window"), LLM_WINDOW, where=f"gufo {key}")
            assert cell["derived"] is True
            assert cell["source"] == f"computed:Δ{series}/Δt", \
                "the provenance string travels with the value whatever window it used"
        assert cells[-1]["value"] == pytest.approx(speed), (
            f"a one-interval window reads the engine's speed on the next poll, got {cells}")


def test_gufo_the_floor_still_refuses_a_catch_up_gap_at_the_llm_window(metrics_text) -> None:
    """The phase-13 floor belongs to the estimator, not to the old 10 s span.

    Shrinking the window to one poll interval must not shrink the guard: `poller.Job._loop` still
    sleeps `max(0.2, interval - elapsed)`, so a slow fetch is still followed by a catch-up poll
    ~0.2 s later, and 0.2 s of a 2 s window is the same lie a 0.2 s slice of a 10 s one was (the
    3284.59 tok/s row for a 53 tok/s engine). `test_rates.py` pins the gate at both spans; this
    is the refusal through the production parse path, where the `n/a` keeps the window it refused
    **and** its `source` so the cell still says which Δ it is talking about.
    """
    scrapes = poll_engine(
        metrics_text, "gufo", LLM_WINDOW,
        [1_000.0, 1_002.0, 1_004.0, 1_014.25, 1_014.47, 1_016.47],
        {"llamacpp:tokens_predicted_total": lambda t: 50.0 * (t - 1_000.0)},
    )
    stalled = scrapes[3]["decode_tokens_per_second"]
    assert stalled["value"] is None, \
        "a 10 s gap is past a 2 s span's ceiling (2 x span): n/a, not a diluted rate"
    assert "window" not in stalled, "the ceiling answers no window because it measured none"
    refused = scrapes[4]["decode_tokens_per_second"]
    assert refused["value"] is None, f"0.22 s is the loop's catch-up gap, not a rate: {refused}"
    assert refused["window"] == pytest.approx(0.22)
    assert refused["window"] < MIN_WINDOW_SECONDS
    assert refused["source"] == "computed:Δllamacpp:tokens_predicted_total/Δt"
    assert delta_label(refused["window"]) == "Δ0.2s", "the label names the window it refused"
    recovered = scrapes[5]["decode_tokens_per_second"]
    assert recovered["value"] == pytest.approx(50.0), (
        f"the refusal re-bases, so the next honest poll reads the engine again: {recovered}")
    assert_window(recovered.get("window"), LLM_WINDOW, where="gufo decode")


def test_gufo_a_poll_gap_a_hair_wider_than_the_span_is_still_the_2s_claim(metrics_text) -> None:
    """Live reality: a poll lands slightly after its interval, and that is still `Δ2.0s`.

    With the span equal to the cadence (phase 23) only the current sample is strictly *inside* the
    window, so the base comes from the estimator's window-empty fallback — the previous sample,
    one poll back — and the published window is the real gap. Read off the running dev server the
    LLM cells report 2.009-2.026 s and the host payload's `window_seconds` 9.978-10.002 s, so a
    pin demanding `window == span` exactly would fail against working software. This drives the
    jittered shape and asserts what the operator sees: the same `Δ2.0s` label and the engine's real
    speed. A gap of a whole extra interval is the other story — the `2 x span` ceiling refuses it,
    which is what the test above pins.
    """
    started = 1_006.0
    stamps = [1_000.0, 1_002.03, 1_004.07, 1_006.1, 1_008.13, 1_010.16]
    scrapes = poll_engine(
        metrics_text, "gufo", LLM_WINDOW, stamps,
        {"llamacpp:tokens_predicted_total": lambda t: 50.0 * max(0.0, t - started)},
    )
    cells = [scrape["decode_tokens_per_second"] for scrape in scrapes[1:]]
    for cell in cells:
        assert cell["value"] is not None
        assert_window(cell.get("window"), LLM_WINDOW, where="gufo decode on a jittered cadence")
        assert delta_label(cell["window"]) == "Δ2.0s", \
            "a 2.03 s gap still prints the window the panel claims"
    assert cells[-1]["value"] == pytest.approx(50.0, rel=0.02)


def test_gufo_prompt_rate_excludes_cached_tokens(metrics_text) -> None:
    text = metrics_text("gufo")
    rates = RateEstimator(span=10)
    parse_gufo_metrics(text, rates, 1_000.0)
    text = bump_metric(text, "llamacpp:prompt_tokens_total", 1_000.0)
    metrics = parse_gufo_metrics(text, rates, 1_010.0)
    assert metrics["prompt_tokens_per_second"]["value"] == pytest.approx(100.0)


def test_gufo_counter_reset_produces_n_a_not_a_spike(metrics_text) -> None:
    text = metrics_text("gufo")
    rates = RateEstimator(span=60)
    parse_gufo_metrics(text, rates, 1_000.0)
    restarted = bump_metric(text, "llamacpp:tokens_predicted_total", -800_000.0)
    metrics = parse_gufo_metrics(restarted, rates, 1_010.0)
    assert metrics.get("decode_tokens_per_second", {}).get("value") is None


def test_gufo_a_stalled_poll_then_the_catch_up_is_not_a_spike(metrics_text) -> None:
    """The exact stored incident, through the production parse path.

    2026-10-18 09:16:31 on deskwork: a 15 s gap in every source, a first result carrying 114
    tokens (its body was generated ~2 s into the stall), then the poll loop's 0.2 s catch-up
    printing **3284.59 tok/s** for an engine doing 53. `test_rates.py` holds the estimator-level
    pair; this is the same defect where the operator sees it.
    """
    text = metrics_text("gufo")
    rates = RateEstimator(span=10.0)
    parse_gufo_metrics(text, rates, 1_000.0)
    text = bump_metric(text, "llamacpp:tokens_predicted_total", 114.0)
    parse_gufo_metrics(text, rates, 1_015.25)
    text = bump_metric(text, "llamacpp:tokens_predicted_total", 713.0)
    metrics = parse_gufo_metrics(text, rates, 1_015.47)
    cell = metrics["decode_tokens_per_second"]
    assert cell["window"] == pytest.approx(0.22)
    assert cell["value"] is None or cell["value"] < 100.0, (
        f"53 tok/s of real work printed as {cell['value']:.0f} tok/s")


def test_gufo_a_stale_scrape_body_never_becomes_a_rate_above_the_engine(metrics_text) -> None:
    """The other half of the stall: the reply is stamped when the poll *finished*, not when the
    engine rendered it, so its counter is the stall's length older than its own timestamp.

    That sample then anchors every window for the rest of the span and each poll re-diffs the
    same skewed delta over a longer one — the live 2026-10-19 12:42 incident printed 232 tok/s
    for one request running at ~50, then a decaying 146 / 114 / 95 tail. The estimator's jump
    gate holds the first of those and now refuses to call a window anchored on the same sample a
    confirmation, so nothing above the engine's real rate reaches the payload.
    """
    text = metrics_text("gufo")
    rates = RateEstimator(span=10.0)
    parse_gufo_metrics(text, rates, 1_000.0)
    # (stamp, tokens the engine had really produced by then): the 1011 reply is the stalled one —
    # its body was rendered ~9 s before the poll stamped it — and from there the counter advances
    # at a steady 45 tok/s, so 45 is the only rate that is true to print.
    timeline = [(1_011.0, 90.0), (1_012.0, 540.0), (1_014.0, 630.0), (1_016.0, 720.0),
                (1_018.0, 810.0), (1_022.0, 990.0), (1_026.0, 1_170.0)]
    produced = 0.0
    cells = []
    for stamp, total in timeline:
        text = bump_metric(text, "llamacpp:tokens_predicted_total", total - produced)
        produced = total
        cells.append(parse_gufo_metrics(text, rates, stamp)["decode_tokens_per_second"])
    for cell in cells:
        assert cell["value"] is None or cell["value"] <= 100.0, \
            f"a 45 tok/s engine printed {cell['value']:.0f} tok/s ({cells})"
    assert cells[-1]["value"] == pytest.approx(45.0), "the stall ends in the truth, not a tail"
    assert cells[-1]["window"] == pytest.approx(10.0)


def test_absent_gufo_metrics_are_none_with_no_source() -> None:
    metrics = parse_gufo_metrics("# nothing here\n")
    assert metrics["requests_processing"]["value"] is None
    assert metrics["requests_processing"]["source"] is None
    assert metrics.get("spec_accept_rate", {}).get("value") is None


# ---- severity: KV pressure arrives with a verdict ------------------------- #

def test_gufo_kv_cache_cell_keeps_provenance_and_gains_severity(metrics_text) -> None:
    metrics = parse_gufo_metrics(metrics_text("gufo"))
    cell = metrics["kv_cache_usage_ratio"]
    assert cell["value"] == Prom(metrics_text("gufo")).scalar(["llamacpp:kv_cache_usage_ratio"]), \
        "verbatim the exposition's own ratio — the cell stays what it was"
    assert cell["source"] == "llamacpp:kv_cache_usage_ratio"
    assert cell["derived"] is False, "the verdict is additive; the cell still says what it was"
    assert cell["severity"] == classify("llm.kv_cache_usage_pct", cell["value"] * 100.0), \
        "the verdict is the table's for this value (percent scale)"


@pytest.mark.parametrize("ratio,expected", [
    ("0.74", "ok"), ("0.75", "warn"), ("0.89", "warn"), ("0.9", "fail"), ("1.0", "fail"),
])
def test_kv_ratio_is_classified_on_a_percent_scale(metrics_text, ratio, expected) -> None:
    """0.9 of a KV cache is 90 %, not 0.9 % — classifying the raw ratio is a permanent green."""
    text = re.sub(r"^(llamacpp:kv_cache_usage_ratio) \S+$", rf"\g<1> {ratio}",
                  metrics_text("gufo"), count=1, flags=re.M)
    metrics = parse_gufo_metrics(text)
    assert metrics["kv_cache_usage_ratio"]["severity"] == expected
    assert metrics["kv_cache_usage_ratio"]["value"] == pytest.approx(float(ratio)), \
        "the scale is for the verdict only; the published value is untouched"


def test_absent_kv_cell_gets_no_verdict() -> None:
    metrics = parse_gufo_metrics("# nothing here\n")
    assert metrics["kv_cache_usage_ratio"]["severity"] is None


# ---- vLLM ----------------------------------------------------------------- #

def test_vllm_counters_and_gauges(metrics_text) -> None:
    metrics = parse_vllm_metrics(metrics_text("vllm"))
    assert metrics["prompt_tokens_total"]["value"] == pytest.approx(1.8389909e7)
    assert metrics["generation_tokens_total"]["value"] == pytest.approx(611_163.0)
    usage = metrics["gpu_cache_usage"]["value"]
    assert usage is None or 0.0 <= usage <= 1.0, "a ratio summed over label sets would exceed 1"
    assert metrics["num_requests_running"]["value"] is not None
    assert metrics["num_preemptions_total"]["value"] is not None


def test_vllm_kv_cache_cell_carries_a_severity(metrics_text) -> None:
    assert parse_vllm_metrics(metrics_text("vllm"))["gpu_cache_usage"]["severity"] == "ok"
    pressed = bump_metric(metrics_text("vllm"), "vllm:kv_cache_usage_perc", 0.95)
    cell = parse_vllm_metrics(pressed)["gpu_cache_usage"]
    assert cell["value"] == pytest.approx(0.95)
    assert cell["severity"] == "fail"
    assert cell["source"] == "vllm:kv_cache_usage_perc"


def test_vllm_absent_kv_cell_gets_no_verdict(metrics_text) -> None:
    stripped = "\n".join(line for line in metrics_text("vllm").splitlines()
                         if "kv_cache_usage" not in line and "gpu_cache_usage" not in line)
    metrics = parse_vllm_metrics(stripped)
    assert metrics["gpu_cache_usage"]["value"] is None
    assert metrics["gpu_cache_usage"]["severity"] is None


def test_vllm_latency_histograms(metrics_text) -> None:
    metrics = parse_vllm_metrics(metrics_text("vllm"))
    ttft = metrics["time_to_first_token_seconds"]
    assert ttft and ttft["count"] > 0 and ttft["mean"] > 0
    assert metrics["time_to_first_token_seconds__p95"]["value"] >= ttft["mean"]
    assert metrics["time_per_output_token_seconds"]["mean"] > 0


def test_vllm_per_request_speed_is_the_reciprocal_of_tpot(metrics_text) -> None:
    metrics = parse_vllm_metrics(metrics_text("vllm"))
    tpot = metrics["time_per_output_token_seconds"]["mean"]
    per_request = metrics["per_request_decode_tokens_per_second"]
    assert per_request["value"] == pytest.approx(1.0 / tpot)
    assert per_request["derived"] is True


def test_vllm_prefix_cache_hit_ratio(metrics_text) -> None:
    metrics = parse_vllm_metrics(metrics_text("vllm"))
    ratio = metrics["prefix_cache_hit_ratio"]["value"]
    assert ratio is not None and 0.0 <= ratio <= 1.0
    assert "hits" in metrics["prefix_cache_hit_ratio"]["source"]


def test_vllm_measured_rate_when_the_build_has_no_throughput_gauge(metrics_text) -> None:
    text = metrics_text("vllm")
    rates = RateEstimator(span=10)
    parse_vllm_metrics(text, rates, 2_000.0)
    text = bump_metric(text, "vllm:generation_tokens_total", 250.0)
    metrics = parse_vllm_metrics(text, rates, 2_010.0)
    assert metrics["decode_tokens_per_second"]["value"] == pytest.approx(25.0)


def test_vllm_rate_cells_name_the_same_window_the_llm_jobs_are_measured_over(
        metrics_text) -> None:
    """The pin is the **source's** span, so both engines answer it the same way.

    The capture exports no throughput gauge, which is what makes these two cells measured (see
    `test_vllm_measured_rate_when_the_build_has_no_throughput_gauge`), and vLLM's counters are
    labelled — the estimator keys each label set separately while the cell's `source` names the
    series. Same two rules as the gufo pin: the `window` inside the LLM span, and
    `computed:Δ<series>/Δt` beside the value.
    """
    started = 2_006.0
    scrapes = poll_engine(
        metrics_text, "vllm", LLM_WINDOW,
        [2_000.0, 2_002.0, 2_004.0, 2_006.0, 2_008.0, 2_010.0],
        {"vllm:generation_tokens_total": lambda t: 40.0 * max(0.0, t - started),
         "vllm:prompt_tokens_total": lambda t: 300.0 * max(0.0, t - started)},
    )
    for key, series, speed in (
        ("decode_tokens_per_second", "vllm:generation_tokens_total", 40.0),
        ("prompt_tokens_per_second", "vllm:prompt_tokens_total", 300.0),
    ):
        cells = [scrape[key] for scrape in scrapes[1:]]
        for cell in cells:
            assert cell["value"] is not None, f"{key} went n/a on a healthy cadence: {cell}"
            assert_window(cell.get("window"), LLM_WINDOW, where=f"vllm {key}")
            assert cell["source"] == f"computed:Δ{series}/Δt"
        assert cells[-1]["value"] == pytest.approx(speed), (
            f"the labelled series still reads its own speed over one interval, got {cells}")


def test_vllm_a_joining_label_set_does_not_lend_its_carried_counter(metrics_text) -> None:
    """A second `engine=`/`model_name=` series appearing mid-window must not be a rate.

    vLLM's counters are labelled, so `scalar(agg="sum")` added whatever label sets exist right
    now. A rank joining (or a reloaded model whose old series is still exported) carries its own
    cumulative counter, and `value < samples[-1]` never fires for an *increase*, so the carried
    value was diffed. Confirmed as a defect, not as this fleet's trigger: `generation_tokens_total`
    has exactly one label set live (`{engine="0",model_name="juggernaut"}`). Fixed in task 03 by
    diffing each label set under its own estimator key — the joining series has no baseline and
    contributes 0 tokens until it has one.
    """
    text = metrics_text("vllm")
    rates = RateEstimator(span=10.0)
    parse_vllm_metrics(text, rates, 2_000.0)
    gained = text + 'vllm:generation_tokens_total{engine="1",model_name="juggernaut"} 500000.0\n'
    metrics = parse_vllm_metrics(gained, rates, 2_010.0)
    cell = metrics["decode_tokens_per_second"]
    assert cell["value"] is None or cell["value"] < 100.0, (
        f"a joining label set printed {cell['value']:.0f} tok/s")


def test_vllm_a_label_set_that_survives_starts_contributing_its_own_delta(metrics_text) -> None:
    """Fresh series contribute nothing on their first poll and their real Δ on the second."""
    text = metrics_text("vllm")
    rates = RateEstimator(span=10.0)
    parse_vllm_metrics(text, rates, 2_000.0)
    gained = text + 'vllm:generation_tokens_total{engine="1",model_name="juggernaut"} 500000.0\n'
    metrics = parse_vllm_metrics(gained, rates, 2_010.0)
    assert metrics["decode_tokens_per_second"]["value"] == pytest.approx(0.0), \
        "the joining series is baseline-less: 500 000 carried, 0 tokens diffed"
    grown = bump_metric(gained, "vllm:generation_tokens_total", 100.0)  # bumps every label set
    metrics = parse_vllm_metrics(grown, rates, 2_020.0)
    assert metrics["decode_tokens_per_second"]["value"] == pytest.approx(20.0), \
        "two series x 100 tokens / 10 s: the rate is the sum of the per-series rates"
    assert metrics["decode_tokens_per_second"]["window"] == pytest.approx(10.0)


def test_gufo_measured_cell_keeps_the_provenance_shape(metrics_text) -> None:
    """Rule 2 after the per-series rewrite: value, `computed:Δ<series>/Δt`, `Δ<window>`."""
    text = metrics_text("gufo")
    rates = RateEstimator(span=10)
    parse_gufo_metrics(text, rates, 1_000.0)
    text = bump_metric(text, "llamacpp:tokens_predicted_total", 400.0)
    decode = parse_gufo_metrics(text, rates, 1_010.0)["decode_tokens_per_second"]
    assert decode["value"] == pytest.approx(40.0)
    assert decode["derived"] is True
    assert decode["source"] == "computed:Δllamacpp:tokens_predicted_total/Δt"
    assert decode["window"] == pytest.approx(10.0), "the widest window a contributing used"


def test_gufo_a_refused_window_is_n_a_and_names_the_window_it_refused(metrics_text) -> None:
    """The stall refusal is an answer: value None, but the series and Δ still travel."""
    text = metrics_text("gufo")
    rates = RateEstimator(span=10)
    parse_gufo_metrics(text, rates, 1_000.0)
    text = bump_metric(text, "llamacpp:tokens_predicted_total", 50.0)
    parse_gufo_metrics(text, rates, 1_010.0)
    text = bump_metric(text, "llamacpp:tokens_predicted_total", 60.0)
    decode = parse_gufo_metrics(text, rates, 1_010.3)["decode_tokens_per_second"]
    assert decode["value"] is None, "0.3 s is the loop's catch-up gap, not a rate"
    assert decode["window"] == pytest.approx(0.3)
    assert decode["source"] == "computed:Δllamacpp:tokens_predicted_total/Δt"


def test_rate_key_puts_the_label_set_in_the_estimator_key() -> None:
    """The key is the resolved name plus its labels, label order included-stable."""
    assert rate_key("llamacpp:tokens_predicted_total", {}) == "llamacpp:tokens_predicted_total"
    assert rate_key("vllm:generation_tokens_total", {"model_name": "m", "engine": "0"}) == \
        'vllm:generation_tokens_total{engine="0",model_name="m"}'
    assert rate_key("m", {"weird": 'a"b\\c'}) == 'm{weird="a\\"b\\\\c"}', \
        "a quote in a label value must not splice two keys into one"


def test_vllm_waiting_requests_are_broken_down_by_reason(metrics_text) -> None:
    metrics = parse_vllm_metrics(metrics_text("vllm"))
    reasons = metrics["waiting_by_reason"]["value"]
    assert isinstance(reasons, dict)


def test_vllm_empty_exposition_is_all_none() -> None:
    metrics = parse_vllm_metrics("")
    assert metrics["num_requests_running"]["value"] is None
    assert metrics["time_to_first_token_seconds"] is None
    assert metrics["per_request_decode_tokens_per_second"]["value"] is None


# ---- Result assembly ------------------------------------------------------ #

def test_engine_result_payload_and_samples(metrics_text) -> None:
    rates = RateEstimator(span=10)
    result = engine_result("gufo", "http://gufo.internal:8731", metrics_text("gufo"), rates,
                           now=3_000.0)
    payload = result.payload
    assert payload["engine"] == "gufo"
    assert payload["upstream"] == "http://gufo.internal:8731"
    assert payload["series_count"] > 10
    assert payload["unparsed"] == 0
    assert payload["sessions"]["detail"], "a missing /slots must be explained, not blank"
    for sample in result.samples:
        assert sample.source == "llm" and sample.host == "gufo"
        assert sample.metric.startswith("llm.")
        assert sample.value == sample.value and sample.value is not None  # no NaN


def test_engine_result_charts_a_measured_rate_after_the_second_poll(metrics_text) -> None:
    rates = RateEstimator(span=10)
    engine_result("gufo", "http://g:1", metrics_text("gufo"), rates, now=3_000.0)
    bumped = bump_metric(metrics_text("gufo"), "llamacpp:tokens_predicted_total", 100.0)
    result = engine_result("gufo", "http://g:1", bumped, rates, now=3_010.0)
    metrics = result.payload["metrics"]
    assert metrics["decode_tokens_per_second"]["value"] == pytest.approx(10.0)
    assert any(s.metric == "llm.decode_tokens_per_second" for s in result.samples)


def test_vllm_stores_the_tpot_series_the_cell_sparkline_draws(metrics_text) -> None:
    """Phase 04 added `llm.tpot_seconds` so the vLLM TPOT row has a shape, not a dead slot.

    Hermetic twin of the live check: the captured exposition must resolve the histogram the
    page charts, or `/llm` shows a permanent "no history" next to a real mean.
    """
    result = engine_result("vllm", "http://v:1", metrics_text("vllm"), RateEstimator(), now=1.0)
    mean = result.payload["metrics"]["time_per_output_token_seconds__mean"]["value"]
    assert mean is not None, "the vLLM capture lost its TPOT histogram — re-capture fixtures"
    stored = {s.metric: s.value for s in result.samples}
    assert stored["llm.tpot_seconds"] == pytest.approx(mean)
    # gufo publishes no output-token histogram, so it stores no TPOT series (never a 0)
    gufo = engine_result("gufo", "http://g:1", metrics_text("gufo"), RateEstimator(), now=1.0)
    assert "llm.tpot_seconds" not in {s.metric for s in gufo.samples}


def test_gufo_stores_the_spec_accept_series_the_cell_sparkline_draws(metrics_text) -> None:
    """`llm.spec_accept_rate` is the gufo-only half of the phase-04 history addition."""
    result = engine_result("gufo", "http://g:1", metrics_text("gufo"), RateEstimator(), now=1.0)
    ratio = result.payload["metrics"]["spec_accept_rate"]["value"]
    assert ratio is not None and 0.0 < ratio <= 1.0
    stored = {s.metric: s.value for s in result.samples}
    assert stored["llm.spec_accept_rate"] == pytest.approx(ratio)
    vllm = engine_result("vllm", "http://v:1", metrics_text("vllm"), RateEstimator(), now=1.0)
    assert "llm.spec_accept_rate" not in {s.metric for s in vllm.samples}


def test_sessions_pass_through_slots_for_gufo() -> None:
    """`sessions.list` is the raw gufo rows — the grid, not the source, reads slot fields.

    The rows are phase 07's live capture (`tests/fixtures/slots_gufo.json`), not a guessed
    shape: the old table's `idle`/`n_prompt`/`n_predict` names were invented and the server
    never answered them. `_sessions()` must pass the rows through untouched — the shape is
    contracted separately in `tests/slots_contract.py`.
    """
    slots = load_slots_capture()
    result = engine_result("gufo", "http://g:1", "", RateEstimator(), slots=slots, now=1.0)
    sessions = result.payload["sessions"]
    assert sessions["list"] == slots
    assert sessions["list"][0]["is_processing"] is False, "the passthrough re-shaped the rows"
    assert sessions["source"] == "gufo /slots"


def test_sessions_say_vllm_has_no_per_session_view() -> None:
    result = engine_result("vllm", "http://v:1", "", RateEstimator(), now=1.0)
    sessions = result.payload["sessions"]
    assert sessions["list"] == []
    assert "vLLM" in sessions["detail"]


# ---- fleet totals --------------------------------------------------------- #

def _engine(name: str, **values) -> dict:
    metrics = {key: {"value": value} for key, value in values.items()}
    return {"engine": name, "metrics": metrics}


def test_fleet_totals_sum_only_engines_that_answered() -> None:
    engines = [
        _engine("gufo", decode_tokens_per_second=40.0, prompt_tokens_per_second=10.0,
                kv_cache_usage_ratio=0.25, requests_processing=1.0, requests_deferred=0.0,
                tokens_total=100.0),
        _engine("vllm", decode_tokens_per_second=10.0, prompt_tokens_per_second=5.0,
                gpu_cache_usage=0.75, num_requests_running=2.0, num_requests_waiting=3.0,
                tokens_total=200.0),
        {},  # an engine whose poll failed
    ]
    totals = fleet_totals(engines)
    assert totals["engines_total"] == 3
    assert totals["engines_live"] == 2
    assert totals["decode_tokens_per_second"] == 50.0
    assert totals["prompt_tokens_per_second"] == 15.0
    assert totals["requests_active"] == 3.0
    assert totals["requests_waiting"] == 3.0
    assert totals["kv_cache_usage_max"] == 0.75, "cache pressure is a max, never a sum"
    assert totals["tokens_total"] == 300.0


def test_fleet_totals_report_the_gufo_vllm_seam() -> None:
    totals = fleet_totals([_engine("gufo"), _engine("vllm")])
    assert totals["seams"], "the two engines count prompt tokens differently; say so"


def test_fleet_totals_of_nothing_is_none_not_zero() -> None:
    totals = fleet_totals([])
    assert totals["decode_tokens_per_second"] is None
    assert totals["tokens_total"] is None
    assert totals["engines_live"] == 0


def test_series_names_are_the_parse_catalog() -> None:
    assert "tokens_generated_total" in series_names("gufo")
    assert "generation_tokens_total" in series_names("vllm")
    assert "generation_tokens_total" not in series_names("gufo")


def test_gufo_a_measured_zero_is_not_replaced_by_the_gauge(metrics_text) -> None:
    """`measured or gauge` made an idle engine advertise its last request's speed."""
    text = metrics_text("gufo")
    rates = RateEstimator(span=10)
    parse_gufo_metrics(text, rates, 1_000.0)
    metrics = parse_gufo_metrics(text, rates, 1_010.0)   # counters unchanged: truly idle
    decode = metrics["decode_tokens_per_second"]
    assert decode["value"] == 0.0
    assert "predicted_tokens_seconds" not in str(decode["source"]), "native gauge must not leak in"


def test_vllm_a_measured_zero_beats_the_running_estimate(metrics_text) -> None:
    text = metrics_text("vllm")
    rates = RateEstimator(span=10)
    parse_vllm_metrics(text, rates, 2_000.0)
    metrics = parse_vllm_metrics(text, rates, 2_010.0)
    decode = metrics["decode_tokens_per_second"]
    assert decode["value"] == 0.0
    assert "running/tpot" not in str(decode["source"])


# ---- radiance: the ground-truth fixtures (phase 34, task 01) ---------------- #
#
# Radiance is a vLLM derivative: its /metrics carries the `radiance:` native series plus
# a `vllm:` compatibility family, and the capture records /stats and /v1/models in the
# same pass. These tests pin the *shape and the name set* of those captures, never the
# numbers (the nut_*.vars precedent): a metric rename or a new build's series fails here
# before a dashboard does.


def _load_radiance_json(name: str) -> Any:
    """Load a radiance JSON fixture, skipping with the capture command like the others."""
    path = Path(__file__).resolve().parent / "fixtures" / name
    if not path.exists():
        pytest.skip(f"missing fixture {path.name} — run scripts/capture-fixtures.sh --with-llm")
    # the capture script writes its verbatim header as `#` comment lines; the document
    # is what follows them
    body = "\n".join(line for line in path.read_text().splitlines() if not line.startswith("#"))
    return json.loads(body)


def _radiance_map_names() -> set[str]:
    return ({name for names, _ in RADIANCE_SERIES.values() for name in names}
            | {name for names in RADIANCE_HISTOGRAMS.values() for name in names})


def test_radiance_fixture_name_set_is_pinned_by_the_map_and_optionals(metrics_text) -> None:
    """The drift pin: every name in the capture's radiance:/vllm: families is either a
    payload cell (the map) or a declared non-cell (RADIANCE_OPTIONAL), and vice versa —
    so a rename, a dropped series, or a typo in a map name fails the suite before a
    dashboard does."""
    names = set()
    for line in metrics_text("radiance").splitlines():
        if line and not line.startswith("#"):
            names.add(line.split("{")[0].split()[0])
    contract = {name for name in names if name.startswith(("radiance:", "vllm:"))}
    assert contract, "the capture lost its series — re-run the capture with the engine up"
    map_names = _radiance_map_names()
    unexplained = contract - map_names - RADIANCE_OPTIONAL
    assert not unexplained, (
        f"names in the capture that are neither a payload cell nor declared optional: "
        f"{sorted(unexplained)}")
    absent_everywhere = map_names - contract - RADIANCE_OPTIONAL
    assert not absent_everywhere, (
        f"map names in neither the capture nor RADIANCE_OPTIONAL (a rename or a typo): "
        f"{sorted(absent_everywhere)}")
    stale = RADIANCE_OPTIONAL - contract - map_names
    assert not stale, f"RADIANCE_OPTIONAL entries that are neither real names: {sorted(stale)}"


def test_radiance_fixture_carries_both_families(metrics_text) -> None:
    """The vLLM-derived shape is part of the contract: the capture must hold both the
    native series and the compatibility family, or the name set has drifted."""
    text = metrics_text("radiance")
    native = sum(1 for line in text.splitlines() if line.startswith("radiance:"))
    compat = sum(1 for line in text.splitlines() if line.startswith("vllm:"))
    assert native > 0, "the capture lost its radiance: native series"
    assert compat > 0, "the capture lost its vllm: compatibility family"


def test_radiance_per_position_series_has_the_seven_positions_the_phase_pins(
    metrics_text,
) -> None:
    """The per-position spec-acceptance series is the 7-bar curve: its label set is
    position="0".."6", each position a series of its own."""
    positions = set()
    for line in metrics_text("radiance").splitlines():
        if line.startswith("radiance:draft_position_acceptance_ratio{"):
            m = re.search(r'position="(\d+)"', line)
            assert m, f"a per-position line without a position label: {line}"
            positions.add(m.group(1))
    assert positions == {str(i) for i in range(7)}, (
        f"the phase pins seven draft positions; the capture has {sorted(positions)}")


STATS_TOP_LEVEL_KEYS = (
    "cards", "checkpoints_written", "ckpt_evictions", "ckpt_refused", "decode_tps",
    "draft_accept", "draft_positions", "hit_cut_tokens", "hit_cuts", "in_flight",
    "kv_bytes_carved", "kv_bytes_total", "kv_bytes_used", "kv_tokens_cached",
    "kv_tokens_max", "kv_tokens_total", "kv_tokens_used", "kv_util",
    "linear_cached_tokens", "linear_hit_rate", "link", "model", "now_ms", "passes",
    "preempted", "preemptions", "prefill_chunks_capped", "prefill_tps",
    "prefix_evictions", "prefix_hit_rate", "queue_cap", "released", "requests",
    "rows_unspec", "running", "started_at", "startup_s", "step_ns", "step_ns_decode",
    "step_ns_decode_best", "step_ns_decode_recent", "steps", "steps_decode", "steps_held",
    "steps_mixed", "steps_prefill", "total_requests", "uptime_s", "version", "waiting",
)


def test_radiance_stats_fixture_shape() -> None:
    """The /stats document: the top-level key set, the per-card VRAM rows, the requests
    list and the per-position spec-acceptance list — shape, not numbers. A capture at
    rest records `requests: []` (the header comment says so); a row shape is recorded
    by re-running the capture with a request in flight."""
    stats = _load_radiance_json("stats_radiance.json")
    assert isinstance(stats, dict)
    assert tuple(sorted(stats.keys())) == STATS_TOP_LEVEL_KEYS, (
        "the /stats top-level key set changed — the shape the page reads drifted; "
        "re-capture and re-decide the payload set")
    cards = stats["cards"]
    assert isinstance(cards, list) and cards, "cards[] is the per-card VRAM panel's source"
    for card in cards:
        assert isinstance(card, dict)
        assert isinstance(card["index"], int)
        assert isinstance(card["name"], str) and card["name"]
        assert isinstance(card["vram_used"], int)
        assert isinstance(card["vram_total"], int)
        assert 0 < card["vram_used"] <= card["vram_total"]
    assert isinstance(stats["requests"], list)
    assert isinstance(stats["queue_cap"], int)
    assert isinstance(stats["in_flight"], int)
    positions = stats["draft_positions"]
    assert isinstance(positions, list)
    assert len(positions) == 7, (
        "the phase pins seven draft positions; the capture has "
        f"{len(positions)} — the 7-bar curve needs re-deciding")
    for pos in positions:
        assert isinstance(pos, dict)
        assert isinstance(pos["reached"], int)
        assert isinstance(pos["accepted"], int)
        assert isinstance(pos["rate"], float)


def test_radiance_models_fixture_shape() -> None:
    """The /v1/models document: one model, and the capabilities / max_model_len it
    publishes — the only place the engine states what it can do."""
    models = _load_radiance_json("models_radiance.json")
    assert isinstance(models, dict)
    assert models["object"] == "list"
    data = models["data"]
    assert isinstance(data, list) and data
    model = data[0]
    assert model["owned_by"] == "radiance", "the engine's own word for its family"
    assert isinstance(model["id"], str) and model["id"]
    assert isinstance(model["max_model_len"], int)
    assert isinstance(model["capabilities"], dict)


# ---- radiance: the field map (phase 34, task 03) ----------------------- #
#
# The map is built from the task-01 fixture (rule 7); the tests below pin the three
# throughput arms (gauge wins / measured fills / n/a on a refused window), the severity
# pass-through on the existing KV row, the /stats session arms, the whitelisted stats
# block and the additive history series. As always: shape and name set, never the
# numbers of a live engine.


def _radiance_stats() -> dict:
    return _load_radiance_json("stats_radiance.json")


# ---- the fixture parse -------------------------------------------------- #

def test_radiance_fixture_parse_every_mapped_key_resolves(metrics_text) -> None:
    """Every mapped key resolves on the verbatim capture, with its pinned source string."""
    metrics = parse_radiance_metrics(metrics_text("radiance"))
    expected_source = {
        "decode_tokens_per_second_native": "radiance:decode_tokens_per_second",
        "prompt_tokens_per_second_native": "radiance:prefill_tokens_per_second",
        "num_requests_running": "vllm:num_requests_running",
        "num_requests_waiting": "vllm:num_requests_waiting",
        "gpu_cache_usage": "vllm:kv_cache_usage_perc",
        "kv_cache_max_tokens": "radiance:kv_cache_max_tokens",
        "kv_cache_max_bytes": "radiance:kv_cache_max_bytes",
        "queue_depth": "radiance:http_queue_depth",
        "admission_rejected_total": "radiance:http_admission_rejected_total",
        "num_preemptions_total": "vllm:num_preemptions_total",
        "prompt_tokens_total": "vllm:prompt_tokens_total",
        "generation_tokens_total": "vllm:generation_tokens_total",
        "gpu_prefix_cache_hit_rate": "vllm:gpu_prefix_cache_hit_rate",
        "spec_accept_rate": "radiance:draft_acceptance_ratio",
        "decode_step_seconds_recent": "radiance:decode_step_seconds_recent",
        "decode_step_seconds_best": "radiance:decode_step_seconds_best",
    }
    for key, source in expected_source.items():
        cell = metrics[key]
        assert set(cell) >= {"value", "source", "derived"}, f"{key} lost its cell shape"
        assert cell["source"] == source, f"{key} resolved to {cell['source']!r}, not {source!r}"
        assert cell["value"] is not None, f"{key} should resolve on the capture"
    # a map name this build does not carry: n/a, never 0
    assert metrics["num_requests_swapped"]["value"] is None
    assert metrics["num_requests_swapped"]["source"] is None


def test_radiance_a_series_removed_from_the_capture_is_n_a_never_zero(metrics_text) -> None:
    """The completion-criteria arm: a line removed from a fixture copy answers
    `{\"value\": null}` — 0 would claim \"the engine published zero\", which it did not."""
    by_key = {
        "radiance:http_queue_depth": "queue_depth",
        "radiance:kv_cache_max_bytes": "kv_cache_max_bytes",
        "radiance:decode_step_seconds_best": "decode_step_seconds_best",
        "vllm:num_preemptions_total": "num_preemptions_total",
        "vllm:gpu_prefix_cache_hit_rate": "gpu_prefix_cache_hit_rate",
        "radiance:draft_acceptance_ratio": "spec_accept_rate",
    }
    for name, key in by_key.items():
        text = "\n".join(line for line in metrics_text("radiance").splitlines()
                         if not line.startswith(name))
        cell = parse_radiance_metrics(text)[key]
        assert cell["value"] is None, f"{key} answered a value for a removed series: {cell}"
        assert cell["source"] is None


def test_radiance_spec_positions_are_the_ordered_seven(metrics_text) -> None:
    metrics = parse_radiance_metrics(metrics_text("radiance"))
    positions = metrics["spec_positions"]
    assert [p["position"] for p in positions] == list(range(7)), \
        "the phase pins seven draft positions, in the order the engine publishes them"
    for p in positions:
        assert 0.0 <= p["ratio"] <= 1.0, "a per-position ratio is a 0-1 gauge"
    # the ratios are a running-window aggregate: volatile between captures, so the
    # pin is the shape (seven, ordered, in-range) — the name set is pinned above
    prom = Prom(metrics_text("radiance"))
    by_position = prom.by_label(["radiance:draft_position_acceptance_ratio"], "position")
    for p in positions:
        assert p["ratio"] == by_position[str(p["position"])], \
            f"position {p['position']} is not the exposition's own number"


def test_radiance_an_absent_position_stays_absent_not_zeroed(metrics_text) -> None:
    text = "\n".join(
        line for line in metrics_text("radiance").splitlines()
        if not ('radiance:draft_position_acceptance_ratio{' in line
                and 'position="6"' in line)
    )
    positions = parse_radiance_metrics(text)["spec_positions"]
    assert [p["position"] for p in positions] == [0, 1, 2, 3, 4, 5], \
        "position 6 is absent from this capture: the list holds what the engine published"
    assert all(p["ratio"] is not None and p["ratio"] > 0.0 for p in positions), \
        "an absent position must not be filled with a 0"


def test_radiance_tokens_total_is_the_two_vllm_counters(metrics_text) -> None:
    metrics = parse_radiance_metrics(metrics_text("radiance"))
    prompt = metrics["prompt_tokens_total"]["value"]
    generated = metrics["generation_tokens_total"]["value"]
    assert metrics["tokens_total"]["value"] == pytest.approx(prompt + generated)
    assert metrics["tokens_total"]["source"] == "computed:prompt+generated"
    # both counters removed: n/a, never 0
    text = metrics_text("radiance")
    for name in ("vllm:prompt_tokens_total", "vllm:generation_tokens_total"):
        text = "\n".join(line for line in text.splitlines()
                         if not line.startswith(name))
    assert parse_radiance_metrics(text)["tokens_total"]["value"] is None


def test_radiance_prefix_hit_ratio_is_the_gauge_source_named_not_derived(metrics_text) -> None:
    """This build publishes no prefix counters, so a hits/queries ratio would be an
    invented number — the gauge is the answer, under the shared UI key."""
    text = metrics_text("radiance")
    metrics = parse_radiance_metrics(text)
    gauge = Prom(text).scalar(["vllm:gpu_prefix_cache_hit_rate"])
    cell = metrics["prefix_cache_hit_ratio"]
    assert cell["value"] == pytest.approx(gauge)
    assert cell["source"] == "vllm:gpu_prefix_cache_hit_rate"
    assert cell["derived"] is False, "the gauge is not a computed ratio"
    # the gauge cell itself is the payload cell too (the map entry)
    assert metrics["gpu_prefix_cache_hit_rate"]["value"] == pytest.approx(gauge)


def test_radiance_latency_histograms_are_the_vllm_shaped_family(metrics_text) -> None:
    metrics = parse_radiance_metrics(metrics_text("radiance"))
    ttft = metrics["time_to_first_token_seconds"]
    assert ttft and ttft["count"] > 0 and ttft["mean"] > 0
    assert metrics["time_to_first_token_seconds__p95"]["value"] >= ttft["mean"]
    assert metrics["time_per_output_token_seconds"]["mean"] > 0
    assert metrics["e2e_request_latency_seconds"]["count"] > 0
    # no request_queue_time on this build: the cell is absent, never back-filled
    assert "request_queue_time_seconds" not in metrics
    assert "request_queue_time_seconds__mean" not in metrics


# ---- rule 2, the three throughput arms ----------------------------------- #

def test_radiance_throughput_arm_gauge_wins(metrics_text) -> None:
    """The native gauge is the scheduler's recent-window aggregate, so it wins."""
    text = metrics_text("radiance")
    metrics = parse_radiance_metrics(text, RateEstimator(span=10), 1_000.0)
    decode = metrics["decode_tokens_per_second"]
    # the gauge's value is the exposition's own number (volatile — the pin is the
    # passthrough, not a frozen value), and it wins the cell
    assert decode["value"] == metrics["decode_tokens_per_second_native"]["value"]
    assert decode["value"] == Prom(text).scalar(["radiance:decode_tokens_per_second"], agg="avg")
    assert decode["source"] == "radiance:decode_tokens_per_second"
    assert decode["derived"] is False, "a scraped gauge is not a computed Δ"
    assert "window" not in decode, "a scraped gauge names no Δ window"
    prompt = metrics["prompt_tokens_per_second"]
    assert prompt["value"] == metrics["prompt_tokens_per_second_native"]["value"]
    assert prompt["source"] == "radiance:prefill_tokens_per_second"


def test_radiance_throughput_arm_measured_delta_when_the_gauge_is_absent(metrics_text) -> None:
    """A gauge this build does not publish falls back to the measured Δ, per label set."""
    text = metrics_text("radiance")
    text = "\n".join(line for line in text.splitlines()
                     if not line.startswith("radiance:decode_tokens_per_second"))
    rates = RateEstimator(span=10)
    parse_radiance_metrics(text, rates, 1_000.0)
    text = bump_metric(text, "vllm:generation_tokens_total", 250.0)
    metrics = parse_radiance_metrics(text, rates, 1_010.0)
    decode = metrics["decode_tokens_per_second"]
    assert decode["value"] == pytest.approx(25.0)
    assert decode["derived"] is True
    assert decode["source"] == "computed:Δvllm:generation_tokens_total/Δt"
    assert decode["window"] == pytest.approx(10.0), \
        "the label must name the window the value used"
    # the arms are independent: the prompt gauge still wins its own cell
    prompt = metrics["prompt_tokens_per_second"]
    assert prompt["value"] == metrics["prompt_tokens_per_second_native"]["value"]
    assert prompt["derived"] is False


def test_radiance_throughput_arm_a_refused_window_is_n_a_with_its_label(metrics_text) -> None:
    """A refused window is still an answer: n/a beside the honest Δ<window> it refused."""
    text = metrics_text("radiance")
    text = "\n".join(line for line in text.splitlines()
                     if not line.startswith("radiance:decode_tokens_per_second"))
    rates = RateEstimator(span=10)
    parse_radiance_metrics(text, rates, 1_000.0)
    text = bump_metric(text, "vllm:generation_tokens_total", 50.0)
    parse_radiance_metrics(text, rates, 1_010.0)
    text = bump_metric(text, "vllm:generation_tokens_total", 60.0)
    decode = parse_radiance_metrics(text, rates, 1_010.3)["decode_tokens_per_second"]
    assert decode["value"] is None, "0.3 s is the loop's catch-up gap, not a rate"
    assert decode["window"] == pytest.approx(0.3)
    assert decode["window"] < MIN_WINDOW_SECONDS
    assert decode["source"] == "computed:Δvllm:generation_tokens_total/Δt"


def test_radiance_throughput_arm_both_absent_is_n_a_never_zero(metrics_text) -> None:
    text = metrics_text("radiance")
    for name in ("radiance:decode_tokens_per_second", "radiance:prefill_tokens_per_second",
                 "vllm:generation_tokens_total", "vllm:prompt_tokens_total"):
        text = "\n".join(line for line in text.splitlines() if not line.startswith(name))
    metrics = parse_radiance_metrics(text)  # no estimator: no measured arm either
    for key in ("decode_tokens_per_second", "prompt_tokens_per_second"):
        cell = metrics[key]
        assert cell["value"] is None, f"{key} printed a value with no source: {cell}"
        assert cell["source"] is None


# ---- severity: the KV cell is rated, the spec cell is not --------------- #

def test_radiance_kv_cell_carries_the_severity_verdict(metrics_text) -> None:
    """Radiance publishes the same 0-1 ratio the vLLM map feeds the table: the
    existing `llm.kv_cache_usage_pct` row rates it, no new threshold anywhere."""
    text = metrics_text("radiance")
    base = Prom(text).scalar(["vllm:kv_cache_usage_perc"], agg="max")
    assert base is not None
    cell = parse_radiance_metrics(text)["gpu_cache_usage"]
    assert cell["value"] == pytest.approx(base)
    assert cell["source"] == "vllm:kv_cache_usage_perc"
    assert cell["derived"] is False, "the verdict is additive; the cell still says what it was"
    assert cell["severity"] == classify("llm.kv_cache_usage_pct", base * 100.0), \
        "the table's verdict for this value (percent scale)"
    # push it above the table's fail bound (90 %): the verdict follows the value
    target = 0.99
    pressed = bump_metric(text, "vllm:kv_cache_usage_perc", target - base)
    cell = parse_radiance_metrics(pressed)["gpu_cache_usage"]
    assert cell["value"] == pytest.approx(target)
    assert cell["severity"] == "fail"


def test_radiance_spec_accept_carries_no_severity(metrics_text) -> None:
    """Decision 4 pinned both ways: the KV cell is rated, the spec cell is not — an
    efficiency number is not a verdict."""
    text = metrics_text("radiance")
    metrics = parse_radiance_metrics(text)
    assert metrics["spec_accept_rate"]["value"] == \
        Prom(text).scalar(["radiance:draft_acceptance_ratio"], agg="avg")
    assert "severity" not in metrics["spec_accept_rate"]
    assert metrics["gpu_cache_usage"]["severity"] is not None, \
        "the other direction: the KV cell is rated"


# ---- /stats as the session source ---------------------------------------- #

def test_radiance_sessions_stats_arm_at_rest_counts_and_cap(metrics_text) -> None:
    """The at-rest capture: requests[] empty → the counts-otherwise sentence, the cap
    from /stats, `active` from the /metrics gauge (one definition in every arm).
    The numbers come from the capture itself — volatile between re-captures."""
    stats = _radiance_stats()
    result = engine_result("radiance", "http://r:8000", metrics_text("radiance"),
                          RateEstimator(), stats=stats, now=1.0)
    sessions = result.payload["sessions"]
    active_expected = Prom(metrics_text("radiance")).scalar(["vllm:num_requests_running"])
    assert sessions["active"] == active_expected, "active stays the /metrics gauge in every arm"
    assert sessions["running"] == stats["running"]
    assert sessions["waiting"] == stats["waiting"]
    assert sessions["queue_cap"] == stats["queue_cap"]
    assert sessions["list"] == stats["requests"]
    assert sessions["source"] == "radiance /stats"
    if not stats["requests"]:
        assert sessions["detail"] == (
            f"radiance reports counts, not sessions: {stats['running']:g} running"
            f" of cap {stats['queue_cap']:g}. Per-request detail appears here while "
            f"requests are in flight.")
    else:
        assert sessions["detail"] is None, "rows in flight: no at-rest sentence"


def test_radiance_sessions_stats_arm_non_empty_rows_verbatim(metrics_text) -> None:
    """A request in flight: the rows are the engine's own shape, verbatim."""
    stats = _radiance_stats()
    rows = [{"id": 42, "model": "juggernaut", "state": "running",
             "prompt_tokens": 1_200, "completed_tokens": 250}]
    stats = {**stats, "requests": rows, "running": 1, "waiting": 2}
    result = engine_result("radiance", "http://r:8000", metrics_text("radiance"),
                          RateEstimator(), stats=stats, now=1.0)
    sessions = result.payload["sessions"]
    assert sessions["list"] == rows, "the rows pass through untouched"
    assert sessions["source"] == "radiance /stats"
    assert sessions["detail"] is None
    assert sessions["running"] == 1
    assert sessions["waiting"] == 2
    assert sessions["queue_cap"] == stats["queue_cap"]  # the capture's cap, untouched


def test_radiance_sessions_missing_stats_fall_back_to_the_gauges(metrics_text) -> None:
    """The sidecar failed: the poll is still healthy, the counts come from the
    /metrics gauges (a different provenance, said so), and the cap is null — never 0."""
    result = engine_result("radiance", "http://r:8000", metrics_text("radiance"),
                          RateEstimator(), now=1.0)  # stats=None: /stats did not answer
    sessions = result.payload["sessions"]
    assert sessions["running"] == 0.0, "the /metrics gauge's value"
    assert sessions["waiting"] == 0.0
    assert sessions["queue_cap"] is None, "null, not 0 — rule 2"
    assert sessions["list"] == []
    assert "did not answer" in sessions["detail"]
    assert "gauges" in sessions["detail"]
    assert "stats" not in result.payload, "no stats block: the sidecar never answered"
    # the metrics passthroughs degrade with it
    assert result.payload["metrics"]["queue_cap"]["value"] is None
    assert result.payload["metrics"]["vram_used_pct"]["value"] is None


# ---- the whitelisted stats block ------------------------------------------ #

def test_radiance_payload_stats_block_is_whitelisted(metrics_text) -> None:
    stats = _radiance_stats()
    result = engine_result("radiance", "http://r:8000", metrics_text("radiance"),
                          RateEstimator(), stats=stats, now=1.0)
    block = result.payload["stats"]
    assert set(block) == {"model", "version", "uptime_s", "startup_s",
                         "total_requests", "cards"}, \
        "the engine internals (link / passes / steps* / …) stay out of the payload"
    assert block["model"] == "juggernaut"
    assert block["version"] == "1.15.1"
    assert isinstance(block["uptime_s"], int) and block["uptime_s"] >= 0, \
        "uptime is the engine's own seconds — volatile between captures"
    card = block["cards"][0]
    src = stats["cards"][0]
    assert set(card) == {"index", "name", "vram_used", "vram_total", "vram_used_pct"}
    assert card["vram_used"] == src["vram_used"], \
        "the raw bytes travel as the engine published them"
    assert card["vram_total"] == src["vram_total"]
    pct = card["vram_used_pct"]
    assert pct["derived"] is True, "the one computed number in the block"
    assert pct["source"] == "computed:vram_used/vram_total*100"
    assert pct["value"] == pytest.approx(src["vram_used"] / src["vram_total"] * 100.0)


def test_radiance_a_renamed_stats_key_degrades_to_absent(metrics_text) -> None:
    """A radiance upgrade that renames a /stats key degrades it to absent, not a
    crash — the poll still returns its /metrics payload."""
    stats = _radiance_stats()
    stats["engine_version"] = stats.pop("version")
    result = engine_result("radiance", "http://r:8000", metrics_text("radiance"),
                          RateEstimator(), stats=stats, now=1.0)
    block = result.payload["stats"]
    assert "version" not in block, "renamed away: absent, never a guessed value"
    assert "engine_version" not in block, "and not carried under the new name either"
    assert block["model"] == "juggernaut", "the rest of the block is intact"
    assert result.payload["metrics"]["spec_accept_rate"]["value"] is not None, \
        "the poll still returns"
    # and a renamed card key degrades the computed pct, not the whole block
    stats2 = _radiance_stats()
    card = {**stats2["cards"][0], "vram_in_use": stats2["cards"][0]["vram_used"]}
    del card["vram_used"]
    stats2["cards"] = [card]
    result2 = engine_result("radiance", "http://r:8000", metrics_text("radiance"),
                           RateEstimator(), stats=stats2, now=1.0)
    row = result2.payload["stats"]["cards"][0]
    assert "vram_used" not in row and "vram_in_use" not in row
    assert row["vram_used_pct"]["value"] is None, "the computed number degrades to n/a"


# ---- the /stats passthroughs in the metrics dict ------------------------- #

def test_radiance_queue_cap_is_the_stats_passthrough(metrics_text) -> None:
    stats = _radiance_stats()
    metrics = parse_radiance_metrics(metrics_text("radiance"), stats=stats)
    cell = metrics["queue_cap"]
    assert cell["value"] == stats["queue_cap"], "the /stats number, verbatim"
    assert cell["source"] == "radiance /stats"
    assert cell["derived"] is False
    absent = parse_radiance_metrics(metrics_text("radiance"))
    assert absent["queue_cap"]["value"] is None, "null, not 0 — rule 2"
    assert absent["queue_cap"]["source"] is None


# ---- the history samples ------------------------------------------------- #

def test_radiance_history_samples_include_the_new_series(metrics_text) -> None:
    text = metrics_text("radiance")
    stats = _radiance_stats()
    result = engine_result("radiance", "http://r:8000", text,
                          RateEstimator(), stats=stats, now=1.0)
    names = {s.metric for s in result.samples}
    for expected in ("llm.decode_tokens_per_second", "llm.prompt_tokens_per_second",
                     "llm.kv_cache_usage", "llm.requests_running", "llm.requests_waiting",
                     "llm.spec_accept_rate",
                     "llm.prefix_cache_hit_ratio", "llm.tpot_seconds",
                     "llm.ttft_seconds", "llm.ttft_p95_seconds",
                     "llm.queue_depth", "llm.queue_cap",
                     "llm.decode_step_seconds_recent", "llm.vram_used_pct"):
        assert expected in names, f"{expected} missing from the radiance samples"
    # `tokens_total` is a payload key, not a history series (it is not an existing
    # HISTORY_METRICS key, and the task's additions are the four above): the fleet
    # block reads it from the payload, the chart never drew it
    assert "llm.tokens_total" not in names
    # the stored values are the capture's own numbers (volatile — pinned as passthrough,
    # never as frozen values)
    prom = Prom(text)
    stored = {s.metric: s.value for s in result.samples}
    assert stored["llm.decode_tokens_per_second"] == \
        prom.scalar(["radiance:decode_tokens_per_second"], agg="avg")
    assert stored["llm.prompt_tokens_per_second"] == \
        prom.scalar(["radiance:prefill_tokens_per_second"], agg="avg")
    assert stored["llm.queue_cap"] == stats["queue_cap"]
    assert stored["llm.queue_depth"] == prom.scalar(["radiance:http_queue_depth"], agg="max")
    assert stored["llm.decode_step_seconds_recent"] == \
        prom.scalar(["radiance:decode_step_seconds_recent"], agg="max")
    assert stored["llm.kv_cache_usage"] == \
        prom.scalar(["vllm:kv_cache_usage_perc"], agg="max")
    # the per-position ratios are snapshot-only (decision 2): no history series
    assert not any("position" in name for name in names)
    # no request_queue_time on this build: queue_seconds is n/a by absence
    assert "llm.queue_seconds" not in names


def test_radiance_history_does_not_touch_gufo_or_vllm_samples(metrics_text) -> None:
    """The existing keys and their candidate order are untouched: a gufo or vLLM
    payload never grows a radiance series."""
    for engine in ("gufo", "vllm"):
        result = engine_result(engine, "http://x:1", metrics_text(engine),
                              RateEstimator(), now=1.0)
        names = {s.metric for s in result.samples}
        for foreign in ("llm.queue_depth", "llm.queue_cap",
                        "llm.decode_step_seconds_recent", "llm.vram_used_pct"):
            assert foreign not in names, f"{engine} grew a radiance series: {foreign}"


def test_series_names_radiance_is_its_own_map() -> None:
    """`series_names` is the parse catalog — the map's keys, per engine."""
    names = series_names("radiance")
    assert "spec_accept_rate" in names
    assert "gpu_cache_usage" in names
    assert "queue_depth" in names
    assert "tokens_generated_total" not in names, "that is the gufo key"
    assert "spec_accept_rate" not in series_names("vllm")


# ---- the fleet totals ---------------------------------------------------- #

def test_fleet_totals_with_a_gufo_and_radiance_payload(metrics_text) -> None:
    """The code needed no change: the aliases and the KV max already cover a radiance
    payload; the seams sentence says the known asymmetries in its own words.
    The numbers are re-derived from the two captures (volatile between re-captures).
    """
    gufo_text = metrics_text("gufo")
    rad_text = metrics_text("radiance")
    gufo = engine_result("gufo", "http://g:1", gufo_text, RateEstimator(), now=1.0)
    rad = engine_result("radiance", "http://r:8000", rad_text,
                       RateEstimator(), stats=_radiance_stats(), now=1.0)
    totals = fleet_totals([gufo.payload, rad.payload])
    assert totals["engines_total"] == 2
    assert totals["engines_live"] == 2
    # running/waiting: the sum of each capture's own counters (gufo's aliases + the gauges)
    gp = Prom(gufo_text)
    rp = Prom(rad_text)
    assert totals["requests_active"] == pytest.approx(
        (gp.scalar(["llamacpp:requests_processing"]) or 0.0)
        + (rp.scalar(["vllm:num_requests_running"]) or 0.0))
    assert totals["requests_waiting"] == pytest.approx(
        (gp.scalar(["llamacpp:requests_deferred"]) or 0.0)
        + (rp.scalar(["vllm:num_requests_waiting"]) or 0.0))
    gufo_kv = gp.scalar(["llamacpp:kv_cache_usage_ratio"], agg="max") or 0.0
    rad_kv = rp.scalar(["vllm:kv_cache_usage_perc"], agg="max") or 0.0
    assert totals["kv_cache_usage_max"] == pytest.approx(max(gufo_kv, rad_kv)), \
        "the true max: the lower engine's value must not hide the other's pressure"
    assert totals["tokens_total"] == pytest.approx(
        gufo.payload["metrics"]["tokens_total"]["value"]
        + rad.payload["metrics"]["tokens_total"]["value"])
    # a fresh estimator: gufo's decode is n/a on its first poll, radiance's gauge wins
    assert totals["decode_tokens_per_second"] == pytest.approx(
        rad.payload["metrics"]["decode_tokens_per_second"]["value"])
    assert totals["prompt_tokens_per_second"] == pytest.approx(
        rad.payload["metrics"]["prompt_tokens_per_second"]["value"])
    seams = totals["seams"]
    assert "cache-hit" in seams, "gufo's prompt-counter asymmetry stays named"
    assert "provenance" in seams, "the seams name the provenance labelling"
    assert "vllm:prompt_tokens_total" in seams, \
        "and the thing the dashboard deliberately does not claim"