"""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 re
from collections.abc import Callable, Mapping, Sequence
from typing import Any

import pytest

from dashboards.rates import MIN_WINDOW_SECONDS, RateEstimator
from dashboards.sources.llm import (
    engine_result,
    fleet_totals,
    parse_gufo_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:
    metrics = parse_gufo_metrics(metrics_text("gufo"))
    assert metrics["tokens_generated_total"]["value"] == 877_336.0
    assert metrics["tokens_prompt_total"]["value"] == 1_350_666.0
    assert metrics["tokens_prompt_cached_total"]["value"] == 124_529_105.0
    assert metrics["requests_processing"]["value"] == 0.0
    assert metrics["device_lost_total"]["value"] == 0.0
    assert metrics["kv_cache_usage_ratio"]["value"] == 0.0


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"]
    assert accept == pytest.approx(578_514 / 789_028)
    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"]
    assert total == 1_350_666 + 124_529_105 + 877_336


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
    assert metrics["decode_tokens_per_second_native"]["value"] == pytest.approx(42.621031668072597)


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"] == 0.0
    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"] == "ok"


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