"""llm_monitor — read-only, real-time dashboard for a gufo server and a vLLM server.

Two upstream engines, one page. The backend exists purely to sidestep CORS: the
browser only ever talks to this origin, and this origin forwards a small set of
read-only GETs to the engines.

    GET /                  -> the dashboard (static/index.html)
    GET /api/config        -> upstream base URLs (no secrets)
    GET /api/gufo/slots    -> proxied passthrough of gufo /slots
    GET /api/gufo/metrics  -> gufo /metrics (Prometheus text) as JSON
    GET /api/vllm/metrics  -> vLLM /metrics (Prometheus text) as JSON

STRICTLY READ-ONLY: this module only ever issues `client.get(...)`. There is no
POST/PUT/PATCH/DELETE route, nothing writes to the engines, and upstream bodies
are parsed and re-emitted as JSON rather than echoed.

An engine only exists when its *_BASE_URL env var is set: unset means this deployment
does not run it, its routes answer 404 "not configured" without dialling anything, and
the page leaves its section out entirely.

Both /metrics endpoints share ONE Prometheus text parser (class `Prom`) — the
gufo (llama.cpp-style `llamacpp:*`) and vLLM (`vllm:*`) expositions are the same
format, so there is exactly one parsing path and per-engine field maps on top.
"""

from __future__ import annotations

import math
import os
import re
import time
from collections import deque
from contextlib import asynccontextmanager
from pathlib import Path
from typing import Any, Sequence

import httpx
from fastapi import Depends, FastAPI, Request
from fastapi.responses import FileResponse, JSONResponse, Response

# --------------------------------------------------------------------------- #
# Configuration (env only — no config file, no CLI flags)
# --------------------------------------------------------------------------- #
# An engine is part of this deployment only when its *_BASE_URL is set. There is
# deliberately NO default host: a built-in 127.0.0.1 guess made a vLLM-only box (or
# anyone who just cloned this repo) render a permanently red gufo panel for a server
# they do not run. Unset -> the endpoint 404s as "not configured" and the page hides
# that section entirely.
#
# Real values live in ~/Deployments/reeseapps/ai/deployments:
#   gufo/quadlets/gufo.container            -> PublishPort=8731:8731/tcp
#   vllm-radiance/quadlets/...container     -> PublishPort=8000:8000/tcp
# Keep them tunnel-friendly on 127.0.0.1 and put real hosts in the env.
def _base_url_from_env(name: str) -> str:
    """`http://host:port/` -> `http://host:port`; unset or blank -> `""` (not configured).

    Split out so the "does this env configure the engine?" rule is testable without
    re-importing the app.
    """
    return os.environ.get(name, "").strip().rstrip("/")


GUFO_BASE_URL = _base_url_from_env("GUFO_BASE_URL")
VLLM_BASE_URL = _base_url_from_env("VLLM_BASE_URL")
GUFO_API_KEY = os.environ.get("GUFO_API_KEY", "").strip()
VLLM_API_KEY = os.environ.get("VLLM_API_KEY", "").strip()
# Short on purpose: the UI polls every second, so a wedged engine must fail fast
# rather than pile up requests.
UPSTREAM_TIMEOUT = float(os.environ.get("UPSTREAM_TIMEOUT_SECONDS", "3"))

STATIC_DIR = Path(__file__).resolve().parent / "static"

POLL_MS = int(os.environ.get("POLL_MS", "1000"))
HISTORY_LEN = int(os.environ.get("HISTORY_LEN", "60"))


def _auth_headers(api_key: str) -> dict[str, str]:
    """Optional bearer token. Empty key -> no Authorization header at all."""
    return {"Authorization": f"Bearer {api_key}"} if api_key else {}


def not_configured(engine: str, env_var: str) -> JSONResponse:
    """Body for an engine this deployment does not run.

    404 rather than 502: nothing is broken upstream, the upstream was never called. The
    `configured: false` flag lets the page keep "absent engine" and "engine is down"
    visibly different instead of showing a red dot for a server nobody installed.
    """
    return JSONResponse(
        {
            "ok": False,
            "engine": engine,
            "configured": False,
            "detail": f"{engine} is not configured — set {env_var} and restart",
        },
        status_code=404,
        headers=NO_STORE,
    )


# --------------------------------------------------------------------------- #
# Prometheus text exposition parser (shared by gufo and vLLM)
# --------------------------------------------------------------------------- #
_NAME_RE = r"[a-zA-Z_:][a-zA-Z0-9_:]*"
# name{label="v",...} value [timestamp]
# - the label block is greedy so a '}' inside a quoted value cannot truncate the
#   match; the anchored value/end-of-line does that instead.
# - whitespace before '{' and a trailing '# ...' comment are not part of the
#   official exposition format, but tolerating them means a quirky exporter shows
#   a number here instead of 'n/a'.
_LINE_RE = re.compile(
    rf"^(?P<name>{_NAME_RE})(?:[ \t]*\{{(?P<labels>.*)\}})?[ \t]+"
    r"(?P<value>[+-]?(\d+\.?\d*(e[+-]?\d+)?|inf|nan))"
    r"(?:[ \t]+(?P<ts>-?\d+))?[ \t]*(?:#.*)?$",
    re.IGNORECASE,
)
_LABEL_RE = re.compile(r'([a-zA-Z_][a-zA-Z0-9_.:]*)="((?:[^"\\]|\\.)*)"')


def _unquote(value: str) -> str:
    return re.sub(r"\\(.)", r"\1", value)


class Prom:
    """Parsed Prometheus exposition: name -> [(labels, value), ...]."""

    def __init__(self, text: str) -> None:
        self.series: dict[str, list[tuple[dict[str, str], float]]] = {}
        self.parse_errors = 0
        for raw in text.splitlines():
            line = raw.strip()
            if not line or line[0] == "#":  # comments / # TYPE / # HELP
                continue
            m = _LINE_RE.match(line)
            if not m:
                self.parse_errors += 1
                continue
            try:
                value = float(m.group("value"))
            except ValueError:  # pragma: no cover - regex keeps this unreachable
                self.parse_errors += 1
                continue
            labels = {
                _unquote(k): _unquote(v) for k, v in _LABEL_RE.findall(m.group("labels") or "")
            }
            self.series.setdefault(m.group("name"), []).append((labels, value))

    # -- lookup helpers ----------------------------------------------------- #
    def resolve(self, names: Sequence[str]) -> str | None:
        """First name from `names` that is actually present."""
        return next((n for n in names if n in self.series), None)

    def resolve_hist(self, names: Sequence[str]) -> str | None:
        """Like resolve(), but for histograms: the bare name never appears in the
        exposition, only <name>_bucket / _sum / _count do."""
        for n in names:
            if any(f"{n}{suffix}" in self.series for suffix in ("_bucket", "_sum", "_count")):
                return n
        return None

    def values(
        self,
        names: Sequence[str],
        label_match: dict[str, str] | None = None,
    ) -> list[float]:
        name = self.resolve(names)
        if name is None:
            return []
        if not label_match:
            return [v for _labels, v in self.series[name]]
        return [v for labels, v in self.series[name] if all(labels.get(k) == x for k, x in label_match.items())]

    def scalar(
        self,
        names: Sequence[str],
        agg: str = "sum",
        label_match: dict[str, str] | None = None,
    ) -> float | None:
        """Aggregate every matching series, or None when the metric is absent.

        agg="sum"  counters and per-engine gauges (running/waiting) — summing
                   across label sets is the right fleet-wide total.
        agg="max"  usage ratios — report the most pressured engine, never a sum
                   that could exceed 1.0.
        agg="avg"  rate gauges — an average across label sets.
        """
        found = self.values(names, label_match)
        finite = [v for v in found if math.isfinite(v)]
        if not finite:
            return None
        if agg == "max":
            return max(finite)
        if agg == "avg":
            return sum(finite) / len(finite)
        return sum(finite)

    def by_label(self, names: Sequence[str], label: str) -> dict[str, float]:
        """Sum series grouped by `label` (e.g. vLLM waiting-by-reason)."""
        name = self.resolve(names)
        if name is None:
            return {}
        out: dict[str, float] = {}
        for labels, value in self.series[name]:
            key = labels.get(label, "")
            if not math.isfinite(value):
                continue
            out[key] = out.get(key, 0.0) + value
        return out

    def histogram(self, names: Sequence[str]) -> dict[str, Any] | None:
        """Mean (+ p95 when interpolable) of a `<name>_sum` / `_count` / `_bucket`."""
        base = self.resolve_hist(names)
        if base is None:
            return None
        total = self.scalar([f"{base}_sum"])
        count = self.scalar([f"{base}_count"])
        buckets = self._buckets(base)
        out: dict[str, Any] = {
            "count": count,
            "sum": total,
            "mean": (total / count) if (total is not None and count) else None,
            "p95": None,
            "source": base,
        }
        if buckets and count:
            # Interpolate only for a single label set: mixing per-DP-rank
            # histograms into one cumulative curve would invent a bogus p95.
            distinct_groups = self._bucket_label_groups(base)
            if distinct_groups <= 1:
                out["p95"] = _quantile(buckets, count, 0.95)
        return out

    def _bucket_label_groups(self, base: str) -> int:
        groups = set()
        for labels, _v in self.series.get(f"{base}_bucket", []):
            groups.add(tuple(sorted((k, v) for k, v in labels.items() if k != "le")))
        return len(groups)

    def _buckets(self, base: str) -> list[tuple[float, float]]:
        """(le, cumulative_count) sorted ascending by bound."""
        out: list[tuple[float, float]] = []
        for labels, value in self.series.get(f"{base}_bucket", []):
            le = labels.get("le")
            if le is None or not math.isfinite(value):
                continue
            bound = math.inf if le in ("+Inf", "Inf", "inf") else float(le)
            out.append((bound, value))
        out.sort(key=lambda item: item[0])
        return out


def _quantile(buckets: list[tuple[float, float]], count: float, q: float) -> float | None:
    """Linear interpolation inside the bucket that contains the `q` quantile."""
    target = q * count
    prev_bound, prev_cum = 0.0, 0.0
    for bound, cum in buckets:
        if cum >= target:
            if math.isinf(bound):
                # Only mass above the last real bucket — best honest guess.
                return prev_bound if math.isfinite(prev_bound) else None
            span = cum - prev_cum
            if span <= 0:
                return bound
            return prev_bound + (bound - prev_bound) * ((target - prev_cum) / span)
        prev_bound, prev_cum = bound, cum
    return None


# --------------------------------------------------------------------------- #
# Per-engine field maps (the only engine-specific knowledge on the backend)
# --------------------------------------------------------------------------- #
def _field(
    value: float | None,
    source: str | None,
    derived: bool = False,
    window: float | None = None,
) -> dict[str, Any]:
    """Uniform metric cell: the UI renders `n/a` whenever value is None.

    `window` is the Δt a rate was measured over, so the UI can say `Δ1.0s` instead of
    implying a scraped gauge it never got.
    """
    cell: dict[str, Any] = {"value": value, "source": source, "derived": derived}
    if window is not None:
        cell["window"] = window
    return cell


class RateEstimator:
    """Δcounter / Δt across consecutive scrapes.

    Stock vLLM exports no throughput gauge at all — only counters — so the only way to
    answer "how many tokens per second RIGHT NOW" is to difference `*_total` between two
    scrapes and label it as exactly that. Deliberately injected rather than global so the
    parsers stay pure and testable; a counter reset (engine restart) yields None, never a
    huge negative spike.

    One instance per process: with several uvicorn workers each keeps its own previous
    sample, which only skews the window, not the meaning.
    """

    def __init__(self, span: float = 10.0, max_samples: int = 512) -> None:
        self.span = span
        self.max_samples = max_samples
        self._samples: dict[str, deque[tuple[float, float]]] = {}

    def rate(
        self, key: str, value: float | None, now: float | None = None
    ) -> tuple[float | None, float | None]:
        """(tokens per second, window actually used) or (None, window|None)."""
        if value is None:
            return None, None
        now = time.time() if now is None else now
        samples = self._samples.setdefault(key, deque())
        if samples and value < samples[-1][1]:
            samples.clear()  # counter reset = engine restart; the old history is a lie
        samples.append((now, value))
        while len(samples) > self.max_samples:
            samples.popleft()
        if len(samples) < 2:
            return None, None  # first scrape after start: nothing to diff against

        # Oldest sample still inside the span; if only the newest one is, fall back to
        # the previous sample so a fast poller still gets a rate (over a shorter window).
        inside = [point for point in samples if now - point[0] <= self.span]
        base = inside[0] if len(inside) > 1 else samples[-2]
        dt, dv = now - base[0], value - base[1]
        # forget anything older than we could ever use again
        cutoff = now - 2 * self.span
        while samples and samples[0][0] < cutoff and len(samples) > 2:
            samples.popleft()
        if dt <= 0 or dv < 0:
            return None, dt if dt > 0 else None
        return dv / dt, dt


# ui key -> (candidate upstream series in preference order, aggregation over label sets).
#
# The first entry of each tuple is what the DEPLOYED gufo actually exports
# (verified against deskwork-ai:8731/metrics). gufo/llama.cpp does not use the names
# people expect: `prompt_tokens_total` not `tokens_prompt_total`, `tokens_predicted_total`
# not `tokens_generated_total`, and the speed gauges are named `*_tokens_seconds` while
# their unit is tokens/second. Later entries are the older `llama:` prefix and the
# spec'd names, kept as fallbacks so an older/other build still renders.
GUFO_SERIES: dict[str, tuple[tuple[str, ...], str]] = {
    "requests_processing": (("llamacpp:requests_processing", "llama:requests_processing"), "sum"),
    "requests_deferred": (("llamacpp:requests_deferred", "llama:requests_deferred"), "sum"),
    "kv_cache_usage_ratio": (("llamacpp:kv_cache_usage_ratio", "llama:kv_cache_usage_ratio"), "max"),
    # gufo counts cache hits separately: prompt_tokens_total EXCLUDES cached tokens.
    "tokens_prompt_total": (
        ("llamacpp:prompt_tokens_total", "llama:prompt_tokens_total",
         "llamacpp:tokens_prompt_total", "llama:tokens_prompt_total"),
        "sum",
    ),
    "tokens_prompt_cached_total": (
        ("llamacpp:prompt_tokens_cached_total", "llama:prompt_tokens_cached_total"),
        "sum",
    ),
    "tokens_generated_total": (
        ("llamacpp:tokens_predicted_total", "llamacpp:tokens_generated_total",
         "llama:tokens_predicted_total", "llama:tokens_generated_total"),
        "sum",
    ),
    # NOT the throughput cell: prefill speed of one request's last pass (see below).
    "prompt_tokens_per_second_native": (
        ("llamacpp:prompt_tokens_seconds", "llamacpp:prompt_tokens_per_second",
         "llama:prompt_tokens_seconds"),
        "avg",
    ),
    # NOT the throughput cell: one unlabeled gauge holding the speed of the most recent
    # decode pass, i.e. a single request. Kept as a fallback + per-request reference.
    "decode_tokens_per_second_native": (
        ("llamacpp:predicted_tokens_seconds", "llamacpp:decode_tokens_per_second",
         "llama:predicted_tokens_seconds"),
        "avg",
    ),
    "n_tokens_max": (("llamacpp:n_tokens_max", "llama:n_tokens_max"), "max"),
    # GPU context losses (gufo-specific): any non-zero value here is an incident.
    "device_lost_total": (("gufo_device_lost_total",), "sum"),
    # speculative decoding (MTP on the fleet): accept rate is the useful number.
    "spec_drafts_total": (("llamacpp:spec_decode_num_drafts_total",), "sum"),
    "spec_draft_tokens_total": (("llamacpp:spec_decode_num_draft_tokens_total",), "sum"),
    "spec_accepted_tokens_total": (("llamacpp:spec_decode_num_accepted_tokens_total",), "sum"),
}


def _ratio(numerator: dict[str, Any], denominator: dict[str, Any], source: str) -> dict[str, Any]:
    num, den = numerator["value"], denominator["value"]
    if num is None or den is None or den == 0:
        return _field(None, None)
    return _field(num / den, source, derived=True)


def parse_gufo_metrics(
    text: str, rates: RateEstimator | None = None, now: float | None = None
) -> dict[str, Any]:
    """`rates` enables the Δ-based aggregate decode rate; omitted, the parse stays pure
    (deterministic) and that cell falls back to the scraped gauge.
    """
    p = Prom(text)
    out: dict[str, Any] = {}

    for key, (names, agg) in GUFO_SERIES.items():
        out[key] = _field(p.scalar(names, agg), p.resolve(names))

    # ------- aggregate throughput: the counters add slots, gufo's gauges cannot ------- #
    # gufo exports both speeds as ONE unlabeled gauge each, holding the last decode (or
    # prefill) pass of ONE request: with two slots busy it still reports a single
    # request's speed, and it holds that number stale once the box goes idle. `avg` over
    # label sets can never sum concurrency — there are no label sets to sum. The token
    # counters advance once per token from every slot, so Δtokens/Δt over the window IS
    # the server-aggregate: two slots at 35 tok/s read ~70. Measured therefore wins over
    # the gauge for both cells, which deliberately inverts the usual precedence (vLLM's
    # native gauge really is an aggregate, gufo's are not). Each gauge stays the fallback
    # for the first scrape and for builds missing the counter, and the UI tags provenance.
    def measured(total_key: str) -> dict[str, Any] | None:
        """Δcounter/Δt for one cumulative counter, keyed by the RESOLVED series name so an
        alias rename starts a fresh window instead of diffing two different series into a
        fabricated rate.
        """
        if rates is None:
            return None
        source = out[total_key]["source"]
        if source is None:
            return None
        rate, window = rates.rate(source, out[total_key]["value"], now)
        if rate is None:
            return None
        return _field(rate, f"computed:Δ{source}/Δt", derived=True, window=window)

    for ui_key, native_key, total_key in (
        ("decode_tokens_per_second", "decode_tokens_per_second_native", "tokens_generated_total"),
        ("prompt_tokens_per_second", "prompt_tokens_per_second_native", "tokens_prompt_total"),
    ):
        gauge = out[native_key]
        out[ui_key] = measured(total_key) or _field(gauge["value"], gauge["source"])

    # gufo counts prompt cache hits separately and prompt_tokens_total EXCLUDES them, so
    # the prompt rate above is newly-computed prefill work, not prompt tokens accepted.
    # That is the number the gauge it replaces was reporting; vLLM's prompt counter
    # includes cached prefix tokens, so a cross-engine prompt-rate sum is close to, not
    # identical to, like-for-like — the UI says so on hover.

    prompt = out["tokens_prompt_total"]["value"]
    cached = out["tokens_prompt_cached_total"]["value"] or 0.0
    generated = out["tokens_generated_total"]["value"]
    parts = [v for v in (prompt, generated) if v is not None]
    out["tokens_total"] = (
        _field(sum(parts) + cached, "computed:prompt+cached+generated")
        if parts
        else _field(None, None)
    )
    # share of prompt tokens served from cache, over everything the model was asked
    if prompt is not None:
        out["prompt_cache_hit_ratio"] = _field(
            cached / (prompt + cached) if (prompt + cached) > 0 else None,
            "computed:cached/(prompt+cached)",
            derived=True,
        )
    else:
        out["prompt_cache_hit_ratio"] = _field(None, None)
    out["spec_accept_rate"] = _ratio(
        out["spec_accepted_tokens_total"], out["spec_draft_tokens_total"],
        "computed:accepted/draft_tokens",
    )
    return out


# Same idea as GUFO_SERIES: deployed vLLM first, older/spec'd names as fallbacks.
# Verified against driveripper-ai:8000/metrics, which exports `kv_cache_usage_perc`
# (NOT `gpu_cache_usage`) and `generation_tokens_total` (NOT `generation_tokens_total`),
# tags every series with engine= / model_name=, and reports TPOT as
# `inter_token_latency_seconds`.
VLLM_SERIES: dict[str, tuple[tuple[str, ...], str]] = {
    "num_requests_running": (("vllm:num_requests_running",), "sum"),
    "num_requests_waiting": (("vllm:num_requests_waiting",), "sum"),
    "gpu_cache_usage": (("vllm:kv_cache_usage_perc", "vllm:gpu_cache_usage"), "max"),
    "prompt_tokens_total": (("vllm:prompt_tokens_total",), "sum"),
    "generation_tokens_total": (("vllm:generation_tokens_total", "vllm:generation_tokens_total"), "sum"),
    # throughput gauges only exist on some vLLM builds; absent -> derived from TPOT
    "prompt_tokens_per_second": (
        ("vllm:prompt_throughput", "vllm:avg_prompt_throughput_toks_per_s"),
        "avg",
    ),
    "decode_tokens_per_second_native": (
        ("vllm:generation_throughput", "vllm:avg_generation_throughput_toks_per_s"),
        "avg",
    ),
    "prefix_cache_queries_total": (("vllm:prefix_cache_queries_total",), "sum"),
    "prefix_cache_hits_total": (("vllm:prefix_cache_hits_total",), "sum"),
    # preemption = a running request kicked back to the queue: the real KV-pressure signal
    "num_preemptions_total": (("vllm:num_preemptions_total",), "sum"),
}

VLLM_HISTOGRAMS = {
    # ui key -> candidate upstream names
    "time_to_first_token_seconds": ("vllm:time_to_first_token_seconds",),
    "time_per_output_token_seconds": (
        "vllm:time_per_output_token_seconds",
        "vllm:inter_token_latency_seconds",
        "vllm:request_time_per_output_token_seconds",
    ),
    "request_queue_time_seconds": ("vllm:request_queue_time_seconds",),
    # not shown directly: mean prompt length / mean prefill time = lifetime prefill
    # speed, which is what makes prompt tok/s non-n/a on a build with no rate gauge
    "request_prefill_time_seconds": ("vllm:request_prefill_time_seconds",),
    "request_prompt_tokens": ("vllm:request_prompt_tokens",),
}


def out_histogram_mean(out: dict[str, Any], key: str) -> float | None:
    """Mean of a histogram already parsed into `out` under `key`."""
    cell = out.get(key)
    mean = cell.get("mean") if isinstance(cell, dict) else None
    return mean if isinstance(mean, float) and math.isfinite(mean) else None


def parse_vllm_metrics(
    text: str, rates: RateEstimator | None = None, now: float | None = None
) -> dict[str, Any]:
    """`rates` enables the Δ-based rates; omitted, the parse stays pure and those cells
    fall back to whatever the engine natively reports (or None)."""
    p = Prom(text)
    out: dict[str, Any] = {}

    for key, (names, agg) in VLLM_SERIES.items():
        out[key] = _field(p.scalar(names, agg), p.resolve(names))

    # Waiting, split by reason (capacity / defragmentation / ...).
    reasons = p.by_label(
        ("vllm:num_requests_waiting_by_reason", "vllm:num_requests_waiting_by_reason_total"),
        "reason",
    )
    out["waiting_by_reason"] = {
        "value": reasons,
        "source": p.resolve(
            ("vllm:num_requests_waiting_by_reason", "vllm:num_requests_waiting_by_reason_total")
        ),
        "derived": False,
    }
    out["waiting_capacity"] = _field(
        reasons.get("capacity"),
        p.resolve(("vllm:num_requests_waiting_by_reason",)),
    )

    # Histograms -> mean (and p95 when interpolable).
    for ui_key, names in VLLM_HISTOGRAMS.items():
        out[ui_key] = p.histogram(names)
        # Flattened scalar form for the simple card renderer.
        out[f"{ui_key}__mean"] = _field(
            out[ui_key]["mean"] if out[ui_key] else None,
            out[ui_key]["source"] if out[ui_key] else None,
        )
        out[f"{ui_key}__p95"] = _field(
            out[ui_key]["p95"] if out[ui_key] else None,
            out[ui_key]["source"] if out[ui_key] else None,
        )

    # Decode throughput: prefer a native gauge, else derive from TPOT so the
    # stats strip is never blank on a stock vLLM.
    tpot = out["time_per_output_token_seconds"]["mean"] if out["time_per_output_token_seconds"] else None
    per_seq = (1.0 / tpot) if tpot else None
    out["per_request_decode_tokens_per_second"] = _field(per_seq, "computed:1/tpot", derived=True)

    # ---- throughput: native gauge > measured Δcounter > estimate, in that order ---- #
    def measured(key: str, total_key: str) -> dict[str, Any] | None:
        if rates is None:
            return None
        rate, window = rates.rate(key, out[total_key]["value"], now)
        if rate is None:
            return None
        return _field(
            rate, f"computed:Δ{VLLM_SERIES[total_key][0][0]}/Δt", derived=True, window=window
        )

    decode = out["decode_tokens_per_second_native"]["value"]
    running = out["num_requests_running"]["value"]
    if decode is not None:
        out["decode_tokens_per_second"] = _field(
            decode, out["decode_tokens_per_second_native"]["source"]
        )
    else:
        # measured aggregate beats the running/tpot estimate: it counts real output, the
        # estimate assumes every running request is decoding this very instant
        out["decode_tokens_per_second"] = (
            measured("vllm:generation_tokens_total", "generation_tokens_total")
            or (
                _field((running or 0) * per_seq, "computed:running/tpot", derived=True)
                if per_seq is not None
                else _field(None, None)
            )
        )

    # prompt tok/s: the deployed build has no rate gauge, so diff the counter. On the
    # first scrape after start there is nothing to diff, and we fall back to the lifetime
    # average prefill speed (mean prompt length / mean prefill time) rather than n/a.
    prefill_avg = None
    prefill_time = out_histogram_mean(out, "request_prefill_time_seconds")
    prompt_len = out_histogram_mean(out, "request_prompt_tokens")
    if prefill_time and prompt_len:
        prefill_avg = _field(
            prompt_len / prefill_time,
            "computed:mean(request_prompt_tokens)/mean(request_prefill_time_seconds)",
            derived=True,
        )
    if out["prompt_tokens_per_second"]["value"] is None:
        out["prompt_tokens_per_second"] = (
            measured("vllm:prompt_tokens_total", "prompt_tokens_total")
            or prefill_avg
            or _field(None, None)
        )
    out["prefill_tokens_per_second_lifetime"] = prefill_avg or _field(None, None)

    # sum only the parts that exist: a partial exposition (one counter renamed away)
    # must degrade to a smaller total, not raise and 500 the endpoint
    parts = [v for v in (out["prompt_tokens_total"]["value"], out["generation_tokens_total"]["value"]) if v is not None]
    out["tokens_total"] = (
        _field(sum(parts), "computed:prompt+generated") if parts else _field(None, None)
    )
    out["prefix_cache_hit_ratio"] = _ratio(
        out["prefix_cache_hits_total"], out["prefix_cache_queries_total"],
        "computed:hits/queries",
    )
    return out


# --------------------------------------------------------------------------- #
# App + upstream fetch helpers
# --------------------------------------------------------------------------- #
@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.client = httpx.AsyncClient(
        timeout=httpx.Timeout(UPSTREAM_TIMEOUT),
        follow_redirects=False,
        limits=httpx.Limits(max_connections=32, max_keepalive_connections=16),
    )
    try:
        yield
    finally:
        await app.state.client.aclose()


app = FastAPI(title="llm_monitor", version="1.0.0", lifespan=lifespan, docs_url=None, redoc_url=None)

# Rolling counters behind every Δcounter/Δt rate. Module-level and deliberately the only
# mutable state in the app: everything else is a pure function of one scrape. The ~10s
# span keeps the strip readable — a bare 1s delta jitters between 0 and big spikes
# because prompts arrive in bursts, not continuously.
#
# ONE instance for both engines: it is keyed by the resolved upstream series name
# (`llamacpp:tokens_predicted_total` vs `vllm:generation_tokens_total`), so the engines
# can never read each other's samples.
RATES = RateEstimator(span=float(os.environ.get("RATE_WINDOW_SECONDS", "10")))

NO_STORE = {"cache-control": "no-store, max-age=0"}


def client_of(request: Request) -> httpx.AsyncClient:
    return request.app.state.client


class UpstreamError(Exception):
    """An upstream problem. Carries the upstream HTTP status when there was one,
    so a caller can branch (gufo /slots -> /v1/slots fallback) before it is
    rendered as a JSON error for the browser."""

    def __init__(self, message: str, *, upstream_status: int | None = None) -> None:
        super().__init__(message)
        self.message = message
        self.upstream_status = upstream_status


@app.exception_handler(UpstreamError)
async def upstream_error_handler(request: Request, exc: UpstreamError) -> JSONResponse:
    # 502 = upstream answered badly / not at all, 504 = upstream too slow.
    status = 504 if exc.upstream_status == 504 else 502
    return JSONResponse({"ok": False, "detail": exc.message}, status_code=status, headers=NO_STORE)


async def fetch_text(client: httpx.AsyncClient, url: str, headers: dict[str, str]) -> str:
    return (await fetch(client, url, headers)).text


async def fetch(client: httpx.AsyncClient, url: str, headers: dict[str, str]) -> httpx.Response:
    """GET only. Any upstream problem becomes an UpstreamError so the UI can flip
    that engine's dot red while the other engine keeps polling."""
    try:
        resp = await client.get(url, headers=headers)
    except httpx.TimeoutException as exc:
        raise UpstreamError(f"{url} timed out after {UPSTREAM_TIMEOUT:g}s", upstream_status=504) from exc
    except httpx.HTTPError as exc:
        raise UpstreamError(f"{url} unreachable: {type(exc).__name__}") from exc
    if resp.status_code >= 400:
        raise UpstreamError(
            f"{url} -> HTTP {resp.status_code} {resp.reason_phrase}".strip(),
            upstream_status=resp.status_code,
        )
    return resp


def ok_payload(engine: str, upstream: str, metrics: Any, parsed: Prom | None = None) -> JSONResponse:
    """`series` / `unparsed` make a metric-name change diagnosable from the browser
    without ssh'ing to the engine: 0 series, or a big `unparsed`, means the exposition
    format drifted from what `parse_gufo_metrics` / `parse_vllm_metrics` expect."""
    body: dict[str, Any] = {
        "ok": True,
        "engine": engine,
        "upstream": upstream,
        "scraped_at": time.time(),
        "metrics": metrics,
    }
    if parsed is not None:
        body["series"] = len(parsed.series)
        body["unparsed"] = parsed.parse_errors
    return JSONResponse(body, headers=NO_STORE)


# --------------------------------------------------------------------------- #
# Routes
# --------------------------------------------------------------------------- #
@app.get("/", include_in_schema=False)
async def dashboard() -> FileResponse:
    return FileResponse(STATIC_DIR / "index.html", media_type="text/html", headers=NO_STORE)


@app.get("/api/config")
async def config() -> JSONResponse:
    """What the page is pointed at, and which engines exist at all. Never includes API keys.

    The `*_enabled` flags are what the page renders its layout from: an engine with no
    base URL gets no panel, no strip card and no polling. Derived from the base URL on
    purpose — there is no separate ENABLED flag that could disagree with it.
    """
    return JSONResponse(
        {
            "gufo_base_url": GUFO_BASE_URL,
            "vllm_base_url": VLLM_BASE_URL,
            "gufo_enabled": bool(GUFO_BASE_URL),
            "vllm_enabled": bool(VLLM_BASE_URL),
            "gufo_auth": bool(GUFO_API_KEY),
            "vllm_auth": bool(VLLM_API_KEY),
            "poll_ms": POLL_MS,
            "history_len": HISTORY_LEN,
            "upstream_timeout_seconds": UPSTREAM_TIMEOUT,
        },
        headers=NO_STORE,
    )


@app.get("/api/gufo/slots")
async def gufo_slots(client: httpx.AsyncClient = Depends(client_of)) -> Response:
    """Pass gufo's /slots JSON straight through (it is already JSON).

    /v1/slots is the documented alias, so a server that only exposes the alias
    still works. Queued/deferred requests never appear here — the deferred count
    comes from /metrics instead.
    """
    if not GUFO_BASE_URL:
        return not_configured("gufo", "GUFO_BASE_URL")
    headers = _auth_headers(GUFO_API_KEY)
    tried: list[str] = []
    last_error: UpstreamError | None = None
    for path in ("/slots", "/v1/slots"):
        url = GUFO_BASE_URL + path
        tried.append(path)
        try:
            resp = await fetch(client, url, headers)
        except UpstreamError as exc:
            if exc.upstream_status == 404:  # try the documented alias
                last_error = exc
                continue
            raise
        return Response(
            content=resp.content,
            media_type=resp.headers.get("content-type", "application/json"),
            headers=NO_STORE,
        )
    detail = f"{GUFO_BASE_URL} has no {' or '.join(tried)}"
    raise UpstreamError(detail if last_error is None else f"{detail} ({last_error.message})")


@app.get("/api/gufo/metrics")
async def gufo_metrics(client: httpx.AsyncClient = Depends(client_of)) -> JSONResponse:
    if not GUFO_BASE_URL:
        return not_configured("gufo", "GUFO_BASE_URL")
    url = GUFO_BASE_URL + "/metrics"
    text = await fetch_text(client, url, _auth_headers(GUFO_API_KEY))
    parsed = Prom(text)
    # RATES turns the cumulative token counter into an aggregate tok/s, which is the only
    # way to count every concurrent slot — gufo's speed gauge tracks one request.
    return ok_payload("gufo", url, parse_gufo_metrics(text, RATES), parsed)


@app.get("/api/vllm/metrics")
async def vllm_metrics(client: httpx.AsyncClient = Depends(client_of)) -> JSONResponse:
    if not VLLM_BASE_URL:
        return not_configured("vllm", "VLLM_BASE_URL")
    url = VLLM_BASE_URL + "/metrics"
    text = await fetch_text(client, url, _auth_headers(VLLM_API_KEY))
    parsed = Prom(text)
    # RATES carries the previous scrape's counters, so prompt/decode tok/s can be
    # measured as Δcounter/Δt (this vLLM build exports no rate gauge of its own).
    return ok_payload("vllm", url, parse_vllm_metrics(text, RATES), parsed)


@app.exception_handler(404)
async def not_found(request: Request, exc: Any) -> JSONResponse:
    return JSONResponse({"ok": False, "detail": f"no such route: {request.url.path}"}, status_code=404)


# Convenience for `python app.py` — equivalent to `uvicorn app:app --port 8090`.
if __name__ == "__main__":
    import uvicorn

    uvicorn.run(
        app,
        host=os.environ.get("DASHBOARD_HOST", "127.0.0.1"),
        port=int(os.environ.get("DASHBOARD_PORT", "8090")),
    )