app.py

"""Gradio app for the Video Benchmark Space.

One Python process, one language, no build step. ``gr.Blocks`` drives
every tab; the hero and About prose are rendered as ``gr.HTML`` for HF
brand fidelity, and the data tabs (Results / Leaderboards / Compare /
Submit) use native Gradio components fed by ``src.compute``.

Tab map:
    * Results — chip filters + column picker + ``gr.Dataframe`` with a
      pandas ``Styler`` rdylgn heatmap on metric columns.
    * Leaderboards — category / access-pattern / top_n controls driving
      a plotly radar chart (one polygon per top config, axes normalized
      so 1.0 = best-in-class on that metric).
    * Compare — three plotly panels (bar / scatter / stacked) fed by
      ``compute.compare_bar`` / ``compare_scatter`` / ``compare_stacked``.
    * Submit — param ``CheckboxGroup``s + repo ``Textbox`` + sliders.
      Submit commits one JSON file per sweep to the submissions dataset;
      the queue table is refreshed from a short-lived in-process cache.
    * Parameters / About — static cards and prose rendered from
      ``src.schema``.

Workers polling ``lerobot/video-benchmark-submissions`` consume the same
JSON schema as before, so this app is drop-in compatible with the
existing benchmark pipeline.
"""
from __future__ import annotations

import json
import logging
import math
import os
import re
import secrets
import threading
import time
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

import gradio as gr
import pandas as pd
import plotly.graph_objects as go
from huggingface_hub import HfApi
from huggingface_hub.utils import HfHubHTTPError
from packaging.version import InvalidVersion, Version
from pydantic import BaseModel, Field, ValidationError, field_validator, model_validator

from src import compute, schema

logger = logging.getLogger("video-benchmark")
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s")

ROOT = Path(__file__).parent
ASSETS_DIR = ROOT / "assets"

SUBMISSIONS_DATASET = os.environ.get("SUBMISSIONS_DATASET", schema.SUBMISSIONS_DATASET)
RESULTS_DATASET = os.environ.get("RESULTS_DATASET", schema.RESULTS_DATASET)
SUBMISSIONS_PREFIX = "submissions/"

import dataclasses as _dataclasses


def _with_env_overrides(bench: schema.BenchmarkConfig) -> schema.BenchmarkConfig:
    """Apply the ``RESULTS_DATASET`` / ``SUBMISSIONS_DATASET`` env overrides.

    All benchmarks share one submissions repo, so the ``SUBMISSIONS_DATASET``
    override applies to every benchmark. ``RESULTS_DATASET`` is RGB-specific
    (depth keeps its own results dataset), matching the historical behavior
    of pointing the RGB datasets at a scratch repo for local testing.
    """
    if bench.key == "rgb":
        return _dataclasses.replace(
            bench,
            results_dataset=RESULTS_DATASET,
            submissions_dataset=SUBMISSIONS_DATASET,
        )
    return _dataclasses.replace(bench, submissions_dataset=SUBMISSIONS_DATASET)


_BENCHMARKS: list[schema.BenchmarkConfig] = [_with_env_overrides(b) for b in schema.BENCHMARKS]
HF_TOKEN = os.environ.get("HF_TOKEN") or os.environ.get("HUGGING_FACE_HUB_TOKEN")

MAX_CONFIGS_PER_SUBMISSION = 5_000
MAX_ITEMS_PER_FIELD = 32
REPO_ID_RE = re.compile(r"^[A-Za-z0-9][\w\-.]*\/[A-Za-z0-9][\w\-.]*$")


# --------------------------------------------------------------------------- #
# Submission payload (unchanged shape — workers still read the same JSON)
# --------------------------------------------------------------------------- #
def _to_bool(v: Any) -> bool:
    return v is True or (isinstance(v, str) and v.strip().lower() == "true")


class SubmissionIn(BaseModel):
    """Validated payload from the Submit tab.

    Mirrors the JSON schema that workers polling the submissions dataset
    already consume — field names and types must stay stable across
    releases.
    """

    repos: list[str] = Field(min_length=1, max_length=MAX_ITEMS_PER_FIELD)
    vcodecs: list[str] = Field(min_length=1, max_length=MAX_ITEMS_PER_FIELD)
    pix_fmts: list[str] = Field(min_length=1, max_length=MAX_ITEMS_PER_FIELD)
    g: list[int] = Field(min_length=1, max_length=MAX_ITEMS_PER_FIELD)
    crf: list[int] = Field(min_length=1, max_length=MAX_ITEMS_PER_FIELD)
    timestamps_modes: list[str] = Field(min_length=1, max_length=MAX_ITEMS_PER_FIELD)
    backends: list[str] = Field(min_length=1, max_length=MAX_ITEMS_PER_FIELD)
    samples_per_config: int = Field(ge=1, le=1000)

    # Depth-only knobs (None for RGB submissions).
    lossless: list[str] | None = None
    use_log: list[str] | None = None
    depth_min: float | None = None
    depth_max: float | None = None
    shift: float | None = None

    @field_validator("repos", "vcodecs", "pix_fmts", "timestamps_modes", "backends")
    @classmethod
    def _strings_nonempty(cls, v: list[str]) -> list[str]:
        cleaned = [s.strip() for s in v if isinstance(s, str) and s.strip()]
        if not cleaned:
            raise ValueError("must contain at least one non-empty value")
        return cleaned

    @field_validator("repos")
    @classmethod
    def _repos_shape(cls, v: list[str]) -> list[str]:
        for s in v:
            if not REPO_ID_RE.match(s):
                raise ValueError(f"invalid repo_id: {s!r}")
        return v

    @model_validator(mode="after")
    def _depth_axes_nonempty(self) -> "SubmissionIn":
        # Depth submissions (lossless set) must sweep both lossless and
        # use_log; RGB (lossless is None) is unaffected.
        if self.lossless is not None and (not self.lossless or not self.use_log):
            raise ValueError("depth submissions require non-empty lossless and use_log values")
        return self

    def total_configs(self) -> int:
        """Cartesian product size across every sweep axis.

        Depth submissions additionally sweep ``lossless`` / ``use_log``;
        when present these multiply the count (matching the live
        ``_submission_summary`` preview), so the cap and queued count
        agree for both benchmarks. ``None`` axes (RGB) count as 1.
        """
        n = (
            len(self.repos) * len(self.vcodecs) * len(self.pix_fmts)
            * len(self.g) * len(self.crf) * len(self.timestamps_modes)
            * len(self.backends)
        )
        if self.lossless is not None:
            n *= max(len(self.lossless), 1)
        if self.use_log is not None:
            n *= max(len(self.use_log), 1)
        return n

    def to_worker_payload(self, submission_id: str, created_at: str, benchmark_type: str) -> dict[str, Any]:
        """Render the JSON the worker pipeline expects to find on the Hub.

        All benchmarks share one submissions repo, so ``benchmark_type``
        is always set and is what the worker routes on.
        """
        payload: dict[str, Any] = {
            "submission_id": submission_id,
            "status": "queued",
            "created_at": created_at,
            "started_at": None,
            "completed_at": None,
            "error": None,
            "benchmark_type": benchmark_type,
            "repo_ids": self.repos,
            "vcodecs": self.vcodecs,
            "pix_fmts": self.pix_fmts,
            "g_values": self.g,
            "crf_values": self.crf,
            "timestamps_modes": self.timestamps_modes,
            "backends": self.backends,
            "num_samples": self.samples_per_config,
            "progress": None,
        }
        if self.lossless is not None:
            payload["lossless_values"] = [_to_bool(v) for v in self.lossless]
            payload["use_log_values"] = [_to_bool(v) for v in (self.use_log or [])]
            if self.depth_min is not None:
                payload["depth_min"] = self.depth_min
            if self.depth_max is not None:
                payload["depth_max"] = self.depth_max
            if self.shift is not None:
                payload["shift"] = self.shift
        return payload


# --------------------------------------------------------------------------- #
# Hub plumbing
# --------------------------------------------------------------------------- #
_api = HfApi(token=HF_TOKEN) if HF_TOKEN else HfApi()

# Lazily create the submissions dataset on first write, so a fresh deployment
# with a valid HF_TOKEN can start accepting submissions without a human having
# to pre-create the repo. The flag + lock make the create_repo call happen at
# most once per process even under concurrent submits.
_submissions_repos_ready: set[str] = set()
_submissions_repo_lock = threading.Lock()


def _make_submission_id() -> str:
    return secrets.token_hex(6)


def _ensure_submissions_repo(bench: schema.BenchmarkConfig) -> None:
    """Create ``bench.submissions_dataset`` on the Hub if it doesn't exist yet.

    Idempotent (``exist_ok=True``) and thread-safe. Called lazily from
    :func:`_commit_submission` so the very first submission on a fresh
    deployment materializes the dataset rather than 404-ing against a
    missing repo. Requires ``HF_TOKEN`` with repo-create rights; any
    HTTP error is re-raised so the Submit handler can surface it.
    """
    dataset = bench.submissions_dataset
    if dataset in _submissions_repos_ready:
        return
    with _submissions_repo_lock:
        if dataset in _submissions_repos_ready:
            return
        try:
            _api.create_repo(
                repo_id=dataset,
                repo_type="dataset",
                exist_ok=True,
                token=HF_TOKEN,
            )
        except HfHubHTTPError as e:
            raise RuntimeError(
                f"Could not create or access submissions dataset "
                f"{dataset!r}. Check that HF_TOKEN has write "
                f"access to that namespace. Underlying error: {e}"
            ) from e
        _submissions_repos_ready.add(dataset)
        logger.info("submissions repo ready: %s", dataset)


def _commit_submission(bench: schema.BenchmarkConfig, submission_id: str, payload: dict[str, Any]) -> None:
    """Upload one submission JSON file to the Hub. Requires ``HF_TOKEN``."""
    if not HF_TOKEN:
        raise RuntimeError(
            "HF_TOKEN secret is not configured on this Space — cannot write to "
            f"{bench.submissions_dataset}."
        )
    _ensure_submissions_repo(bench)
    content = json.dumps(payload, indent=2, ensure_ascii=False).encode("utf-8")
    _api.upload_file(
        path_or_fileobj=content,
        path_in_repo=f"{SUBMISSIONS_PREFIX}{submission_id}.json",
        repo_id=bench.submissions_dataset,
        repo_type="dataset",
        token=HF_TOKEN,
        commit_message=f"Submit {submission_id}",
    )


# --------------------------------------------------------------------------- #
# Queue listing (briefly cached so tab opens don't hammer the Hub)
# --------------------------------------------------------------------------- #
_queue_caches: dict[str, dict[str, Any]] = {}
_QUEUE_TTL_SECONDS = 20
_QUEUE_LIMIT = 10


def _queue_cache_for(key: str) -> dict[str, Any]:
    c = _queue_caches.get(key)
    if c is None:
        c = {"at": 0.0, "items": []}
        _queue_caches[key] = c
    return c


def _fmt_relative(iso: str) -> str:
    """Render an ISO-8601 UTC timestamp as a coarse "N min ago" string."""
    if not iso:
        return ""
    try:
        t = datetime.fromisoformat(iso.replace("Z", "+00:00"))
    except ValueError:
        return iso
    now = datetime.now(timezone.utc)
    delta = (now - t).total_seconds()
    if delta < 60:
        return "just now"
    if delta < 3600:
        return f"{int(delta // 60)} min ago"
    if delta < 86400:
        return f"{int(delta // 3600)} h ago"
    return f"{int(delta // 86400)} d ago"


def _load_queue(bench: schema.BenchmarkConfig, limit: int = _QUEUE_LIMIT) -> list[dict[str, Any]]:
    """List the most recent submissions on the Hub, newest first.

    Reads at most ``limit`` JSON files from the submissions dataset. The
    result is cached for ``_QUEUE_TTL_SECONDS`` so opening the Submit tab
    or hitting *Refresh queue* doesn't hammer the Hub. Hub failures are
    logged and surfaced as an empty list — the UI falls back to "no
    submissions" rather than erroring.
    """
    dataset = bench.submissions_dataset
    queue_cache = _queue_cache_for(bench.key)
    now = time.time()
    if queue_cache["items"] and (now - queue_cache["at"] < _QUEUE_TTL_SECONDS):
        return queue_cache["items"]

    try:
        entries = list(_api.list_repo_tree(
            dataset,
            repo_type="dataset",
            path_in_repo=SUBMISSIONS_PREFIX.rstrip("/"),
            recursive=False,
            expand=True,
        ))
    except Exception as e:  # noqa: BLE001
        logger.warning("list_repo_tree failed: %s", e)
        return []

    files = [e for e in entries if getattr(e, "path", "").endswith(".json")]
    files.sort(
        key=lambda e: getattr(getattr(e, "last_commit", None), "date", None)
        or datetime.min.replace(tzinfo=timezone.utc),
        reverse=True,
    )
    items: list[dict[str, Any]] = []
    for entry in files:
        if len(items) >= limit:
            break
        path = entry.path
        try:
            local = _api.hf_hub_download(
                repo_id=dataset,
                repo_type="dataset",
                filename=path,
                token=HF_TOKEN,
            )
            payload = json.loads(Path(local).read_text(encoding="utf-8"))
        except Exception as e:
            logger.warning("could not read %s: %s", path, e)
            continue

        # Shared repo: legacy entries without a type are RGB submissions.
        if (payload.get("benchmark_type") or "rgb") != bench.key:
            continue

        repo_ids = payload.get("repo_ids") or payload.get("repos") or []
        if len(repo_ids) == 1:
            repo_label = repo_ids[0]
        elif repo_ids:
            repo_label = f"{len(repo_ids)} datasets"
        else:
            repo_label = "—"

        items.append({
            "id": payload.get("submission_id") or payload.get("id") or Path(path).stem,
            "status": (payload.get("status") or "queued").lower(),
            "repo": repo_label,
            "when": _fmt_relative(payload.get("created_at", "")),
        })

    queue_cache["items"] = items
    queue_cache["at"] = now
    return items


# --------------------------------------------------------------------------- #
# Results loader (Hub parquet shards via `datasets`, cached in-process)
# --------------------------------------------------------------------------- #
_RESULTS_TTL_SECONDS = 3600
_RESULTS_STALE_TTL_SECONDS = 24 * 3600
_results_caches: dict[str, dict[str, Any]] = {}
_results_fetch_lock = threading.Lock()


def _cache_for(dataset: str) -> dict[str, Any]:
    c = _results_caches.get(dataset)
    if c is None:
        c = {"at": 0.0, "rows": []}
        _results_caches[dataset] = c
    return c


def _fetch_results_rows(bench: schema.BenchmarkConfig) -> list[dict[str, Any]]:
    """Download ``bench``'s results dataset and project it to its keep-keys.

    The Hub dataset can carry extra columns from older runs; we only
    keep the ones the UI actually reads to keep the in-memory footprint
    small. Rows are then reduced to the latest LeRobot release and
    de-duplicated per config so both cache-population paths (TTL refresh
    and the manual "Refresh from Hub") stay duplicate-free.
    """
    from datasets import load_dataset

    dataset = bench.results_dataset
    ds = load_dataset(dataset, split="train", token=HF_TOKEN)
    keep = [c for c in bench.keep_keys if c in ds.column_names]
    if not keep:
        raise RuntimeError(
            f"{dataset} has none of the expected columns "
            f"(got {sorted(ds.column_names)!r})"
        )
    ds = ds.select_columns(keep)
    rows = ds.to_list()

    # Metrics are only comparable within a single LeRobot release, so keep
    # only the rows produced by the most recent version present.
    parsed: list[tuple[Version, dict[str, Any]]] = []
    for r in rows:
        try:
            parsed.append((Version(str(r.get("lerobot_version"))), r))
        except InvalidVersion:
            continue
    if parsed:
        latest = max(v for v, _ in parsed)
        rows = [r for v, r in parsed if v == latest]

    config_keys = [c["key"] for c in bench.columns if c["group"] == "Config"]
    return compute.dedupe_latest(rows, config_keys)


def _cached_rows(bench: schema.BenchmarkConfig) -> list[dict[str, Any]]:
    """Return cached result rows for ``bench``, refetching when stale.

    Single-flight: concurrent callers wait on one fetch instead of
    racing. If a refresh fails but a stale-but-usable cache exists
    (within ``_RESULTS_STALE_TTL_SECONDS``) it's returned with a
    warning, so a transient Hub outage doesn't blank the UI.
    """
    cache = _cache_for(bench.results_dataset)
    now = time.time()
    if cache["rows"] and now - cache["at"] < _RESULTS_TTL_SECONDS:
        return cache["rows"]
    with _results_fetch_lock:
        now = time.time()
        if cache["rows"] and now - cache["at"] < _RESULTS_TTL_SECONDS:
            return cache["rows"]
        try:
            rows = _fetch_results_rows(bench)
        except Exception as e:  # noqa: BLE001
            if cache["rows"] and now - cache["at"] < _RESULTS_STALE_TTL_SECONDS:
                logger.warning("Hub load failed (%s); serving stale rows", e)
                return cache["rows"]
            logger.exception("Hub load failed")
            raise
        cache.update({"at": now, "rows": rows})
        return rows


def _safe_cached_rows(bench: schema.BenchmarkConfig) -> list[dict[str, Any]]:
    """Like :func:`_cached_rows` but never raises."""
    try:
        return _cached_rows(bench)
    except Exception as e:  # noqa: BLE001
        logger.warning("rows unavailable: %s", e)
        return []


def _force_refresh_rows(bench: schema.BenchmarkConfig) -> tuple[bool, str, int]:
    """Bypass the TTL and re-download ``bench``'s results dataset now.

    Returns ``(ok, message, row_count)`` so the caller can pop a
    ``gr.Info`` / ``gr.Warning`` and refresh the row counter in the
    hero.
    """
    cache = _cache_for(bench.results_dataset)
    with _results_fetch_lock:
        try:
            rows = _fetch_results_rows(bench)
        except Exception as e:  # noqa: BLE001
            logger.exception("manual refresh failed")
            return False, f"Refresh failed: {e}", len(cache.get("rows") or [])
        cache.update({"at": time.time(), "rows": rows})
        return True, f"Loaded {len(rows):,} rows from {bench.results_dataset}.", len(rows)


# --------------------------------------------------------------------------- #
# Hero — built from native Gradio components. Using gr.Row/Column/Markdown/
# Image (instead of a hand-rolled <section> injected via gr.HTML) lets Gradio
# own the layout: widths come from the normal gr.Blocks flex container, not
# from brittle CSS that has to escape multiple nested wrappers.
# --------------------------------------------------------------------------- #
_MASCOT_PATH = ASSETS_DIR / "huggy-coding.png"


def _hero_stats_md(bench: schema.BenchmarkConfig, total_rows: int, repo_count: int) -> str:
    """Render the four-cell stats row at the bottom of the hero.

    Pulled out of :func:`_build_hero` so the "Refresh from Hub" footer
    button can recompute the same string and push it back into the hero
    Markdown without rebuilding the whole row.
    """
    rows_fmt = f"{total_rows:,}" if total_rows else "—"
    repos_fmt = str(repo_count) if repo_count else "—"
    return (
        f"| {rows_fmt} | {repos_fmt} | {bench.codec_count} | {bench.backend_count} |\n"
        "|---|---|---|---|\n"
        "| Configurations | Datasets | Codecs | Backends |"
    )


def _footer_note_md(bench: schema.BenchmarkConfig) -> str:
    """Footer note: LeRobot release of the results + a Submit-tab nudge."""
    hint = "Missing some values? Add the configuration from the **Submit** tab."
    vers = [str(r["lerobot_version"]) for r in _safe_cached_rows(bench) if r.get("lerobot_version")]
    if not vers:
        return hint
    return f"Results computed with LeRobot `v{max(vers, key=Version)}` · {hint}"


def _build_hero(bench: schema.BenchmarkConfig, total_rows: int, repo_count: int) -> gr.Markdown:
    """Render the top-of-page hero with live measurement / dataset counts.

    Returns the stats ``gr.Markdown`` so callers can wire it as an
    output of the "Refresh from Hub" footer button — that's the only
    hero element whose value depends on the cached rows.
    """
    with gr.Row(elem_classes="hero", equal_height=True):
        with gr.Column(scale=4, min_width=0):
            gr.Markdown(
                "**Open benchmark**\n\n"
                f"# {bench.title}\n\n"
                f"{bench.subtitle}\n\n"
                "_All measurements are currently CPU-only._",
                elem_classes="hero-copy",
            )
            stats = gr.Markdown(
                _hero_stats_md(bench, total_rows, repo_count),
                elem_classes="hero-stats",
            )
        with gr.Column(scale=1, min_width=0):
            gr.Image(
                value=str(_MASCOT_PATH),
                show_label=False,
                buttons=[],
                container=False,
                interactive=False,
                elem_classes="hero-mascot",
            )
    return stats


def _tab_intro(title: str, desc: str) -> gr.HTML:
    """Per-tab heading + subtitle block.

    Keeps every tab visually consistent with the same ``<h2>`` yellow-
    underline treatment as the leaderboard section heading.
    """
    import html as _html
    safe_title = _html.escape(title)
    safe_desc = _html.escape(desc)
    return gr.HTML(
        f'<div class="tab-intro">'
        f'<h2>{safe_title}</h2>'
        f'<p>{safe_desc}</p>'
        f'</div>',
        elem_classes="tab-intro-wrap",
    )


# --------------------------------------------------------------------------- #
# Results tab helpers — rdylgn heatmap via pandas Styler
# --------------------------------------------------------------------------- #
_RDYLGN_STOPS = [
    (0.00, (215, 48, 39)),    # red
    (0.25, (252, 141, 89)),   # orange
    (0.50, (255, 255, 191)),  # pale yellow
    (0.75, (166, 217, 106)),  # light green
    (1.00, (26, 152, 80)),    # green
]


def _rdylgn(t: float) -> str:
    """Linearly interpolate the rdylgn ramp at ``t`` in [0, 1].

    Returns a CSS ``rgb(...)`` string, or ``"transparent"`` for ``None``
    / NaN. Out-of-range ``t`` is clamped.
    """
    if t is None or t != t:  # NaN
        return "transparent"
    t = max(0.0, min(1.0, float(t)))
    for (t0, c0), (t1, c1) in zip(_RDYLGN_STOPS, _RDYLGN_STOPS[1:]):
        if t <= t1:
            span = (t1 - t0) or 1.0
            k = (t - t0) / span
            r = int(round(c0[0] + (c1[0] - c0[0]) * k))
            g = int(round(c0[1] + (c1[1] - c0[1]) * k))
            b = int(round(c0[2] + (c1[2] - c0[2]) * k))
            return f"rgb({r},{g},{b})"
    return "transparent"


def _cols_by_key(bench: schema.BenchmarkConfig) -> dict[str, dict[str, Any]]:
    return {c["key"]: c for c in bench.columns}


def _default_visible(bench: schema.BenchmarkConfig) -> list[str]:
    return [c["key"] for c in bench.columns
            if c["group"] == "Config" or c.get("metric")]


def _picker_default_visible(bench: schema.BenchmarkConfig) -> list[str]:
    """Columns shown by default: ``_default_visible`` minus the
    hidden-by-default "Quantized quality" group and any column flagged
    ``default_hidden``. Shared by the column picker and the Results-table
    empty-selection fallback so the two stay consistent."""
    return [c["key"] for c in bench.columns
            if (c["group"] == "Config" or c.get("metric"))
            and c["group"] != "Quantized quality"
            and not c.get("default_hidden")]


def _resolve_cols(bench: schema.BenchmarkConfig, keys: list[str]) -> list[dict[str, Any]]:
    cbk = _cols_by_key(bench)
    return [cbk[k] for k in keys if k in cbk]


def _nb_slash(text: str) -> str:
    """Wrap ``/`` with U+2060 WORD JOINER on both sides.

    The word joiner is an invisible zero-width character that suppresses
    line-break opportunities at its position. Without this, Chrome
    treats ``/`` inside a compound token (e.g. ``Video/Image``) as a
    soft break and will split the word onto two lines as soon as the
    column is narrow enough, pushing the trailing word past the 2-line
    ``-webkit-line-clamp`` cap on the Results-tab header and clipping
    it. CSS alone can't fix this in Chrome (``word-break: keep-all`` is
    a no-op for non-CJK text), so we patch the source string instead.
    """
    return text.replace("/", "\u2060/\u2060")


# Header label → ``desc`` mapping used to populate ``title=`` tooltips on
# the Results-tab dataframe headers (see ``_RESULTS_HEADER_TOOLTIP_JS``
# below). Keys mirror the ``short`` form ``_render_results_table`` puts on
# each column (whitespace collapsed so the JS's ``replace(/\s+/g, " ")``
# always matches, and U+2060 word joiners stripped — they're added on
# the rendered header to prevent slash-breaks but absent from these
# lookup keys, so the JS strips them from textContent before lookup).
# The value carries the *full* schema label as a header so the tooltip
# surfaces the unit + polarity that ``short`` drops.
def _results_header_text(c: dict[str, Any]) -> str:
    """Header label rendered in the Results-tab dataframe ``<th>``.

    Uses the schema's ``label`` (which carries the unit suffix, e.g.
    ``"PSNR (dB)"`` / ``"Decoding (ms)"``) rather than ``short`` so units
    stay visible in the table header. The trailing ↑ / ↓ polarity arrow
    is kept so the header signals lower-is-better / higher-is-better at a
    glance, and the embedded ``\\n`` (used to hint a wrap point in the
    schema) is collapsed so we control wrapping via CSS instead.
    """
    raw = (c.get("label") or c.get("short") or c["key"]).replace("\n", " ")
    return re.sub(r"\s+", " ", raw).strip()


def _header_descs(bench: schema.BenchmarkConfig) -> dict[str, str]:
    out: dict[str, str] = {}
    for c in bench.columns:
        desc = c.get("desc")
        if not desc:
            continue
        key = _results_header_text(c)
        full = re.sub(r"\s+", " ", (c.get("label") or c["key"]).replace("\n", " ")).strip()
        out[key] = f"{full} — {desc}" if full and full != key else desc
    return out


def _all_header_descs() -> dict[str, str]:
    merged: dict[str, str] = {}
    for b in schema.BENCHMARKS:
        merged.update(_header_descs(b))
    return merged

# Inline JS that wires native ``title=`` tooltips onto the Results-tab
# dataframe headers. Kept as a single string so it can be injected via
# ``gr.Blocks(head=...)``. The MutationObserver is necessary because
# Gradio re-renders the <th> cells when the visible columns change.
_RESULTS_HEADER_TOOLTIP_JS: str = r"""
(function () {
  var DESCS = window.__RESULTS_HEADER_DESCS__ || {};
  function apply(root) {
    var ths = root.querySelectorAll(".results-table thead th");
    for (var i = 0; i < ths.length; i++) {
      var th = ths[i];
      var label = (th.textContent || "")
        .replace(/[\u2060\u200B-\u200D\uFEFF]/g, "")
        .trim()
        .replace(/\s+/g, " ");
      var desc = DESCS[label];
      if (desc && th.getAttribute("title") !== desc) {
        th.setAttribute("title", desc);
        th.setAttribute("aria-label", label + " — " + desc);
      }
    }
  }
  function start() {
    apply(document);
    var obs = new MutationObserver(function () { apply(document); });
    obs.observe(document.body, { childList: true, subtree: true });
  }
  if (document.readyState === "loading") {
    document.addEventListener("DOMContentLoaded", start);
  } else {
    start();
  }
})();
"""

# Bundle the header→desc JSON and the observer JS into a single ``<head>``
# payload. Passed to ``demo.launch(head=...)`` (Gradio 6+ moved the kwarg
# off ``Blocks``); HF Spaces invokes ``launch`` under the hood, so this
# keeps tooltips working in both local runs and on the Space.
_RESULTS_HEADER_TOOLTIP_HEAD: str = (
    "<script>window.__RESULTS_HEADER_DESCS__ = "
    + json.dumps(_all_header_descs())
    + ";</script>"
    + "<script>" + _RESULTS_HEADER_TOOLTIP_JS + "</script>"
)


def _fmt_cell(spec: str | None):
    """Build a printf-style cell formatter, or ``None`` for non-numeric columns."""
    if not spec:
        return None

    def fmt(v: Any) -> str:
        if v is None or (isinstance(v, float) and v != v):
            return ""
        try:
            return spec % v
        except (TypeError, ValueError):
            return str(v)

    return fmt


# Columns whose raw values are numeric even though they live in the
# ``"Config"`` group. These need ``datatype="number"`` so clicking the
# header sorts by numeric value (``"10" < "2"`` the other way).
_NUMERIC_CONFIG_KEYS: frozenset[str] = frozenset({"g", "crf"})


def _col_datatype(c: dict[str, Any]) -> str:
    """Gradio Dataframe datatype for one column.

    ``"number"`` for metric columns and numeric-valued config columns
    (``g`` / ``crf``), ``"str"`` everywhere else. Drives both the cell
    value coercion on the frontend (``cast_value_to_type``) and — more
    importantly — the comparator TanStack uses when the user clicks a
    column header to sort. String columns sort lexicographically;
    number columns sort by numeric value.
    """
    if c.get("metric") or c["key"] in _NUMERIC_CONFIG_KEYS:
        return "number"
    return "str"


def _chip_style(color: str) -> str:
    """Inline CSS custom-property declaration that paints the heatmap chip.

    ``metadata.styling`` is emitted by Gradio as an inline ``style=`` on
    the *outer* ``<div class="body-cell">`` wrapper — see
    ``DataCell.svelte`` (``style="{col_style} {cell_style}"``), not on
    the inner ``<span>`` that holds the value. The body-cell is a fixed
    column track with ``padding: 0; overflow: hidden``, so any
    ``background`` set here floods the whole cell and ``display:
    inline-block`` / ``width: fit-content`` are no-ops at that level.
    Setting a CSS variable instead lets the matching rule in
    ``styles.css`` (``.results-table .body-cell[style*="--chip-bg"]
    .cell-wrap > span``) pick the color up and paint just the inner
    span as a content-sized pill — same visual as the old
    ``<span class="heatmap-chip">`` HTML, but without wrapping the
    value in an HTML string (which forced ``datatype="html"`` and broke
    numeric sorting, since TanStack would compare the ``<span style=
    "background:rgb(...)">`` prefix instead of the underlying value).
    """
    return f"--chip-bg:{color};"


def _render_results_table(
    bench: schema.BenchmarkConfig,
    selections: dict[str, list[str]],
    visible_keys: list[str],
) -> tuple[dict[str, Any], str, list[str]]:
    """Build the Results-tab table from the current chip / column selections.

    Filters → composite-ranks → projects down to the visible columns →
    emits a Gradio Dataframe payload where each cell carries three things:

      * the raw (numeric) value — used by TanStack for header-click
        sorting; this is why metric columns stay numbers instead of
        pre-rendered HTML chips, so sorting compares values rather
        than ``<span style="background:rgb(...)">`` strings;
      * a formatted display value, including the ``± std`` suffix on
        metrics that carry one;
      * an inline ``style=`` that paints the rdylgn heatmap chip on
        metric cells (applied to the innermost span by
        ``EditableCell.svelte``).

    Returns ``(value, count_md, datatypes)`` where ``value`` is the
    ``{headers, data, metadata}`` dict ``gr.Dataframe`` accepts directly
    (see ``Dataframe.postprocess`` / ``get_cell_data`` / ``get_metadata``),
    ``count_md`` is the row-count markdown shown under the table, and
    ``datatypes`` is the per-column list driving numeric vs. string sort.
    """
    rows = _safe_cached_rows(bench)
    filters = {k: selections.get(k, []) for k in bench.filter_order}
    filtered = compute.filter_rows(rows, filters)
    cols = _resolve_cols(bench, visible_keys) or _resolve_cols(bench, _picker_default_visible(bench))
    metric_cols = [c for c in cols if c.get("metric")]
    ranked = compute.composite_rank(filtered, metric_cols)
    scales = compute.metric_scales(filtered, metric_cols)

    # Headers use the schema's full ``label`` (e.g. "Decoding (ms) ↓",
    # "PSNR (dB) ↑") including the trailing ↑ / ↓ polarity arrow so the
    # header states lower-is-better / higher-is-better at a glance, and
    # keeping the unit suffix visible makes the table self-explanatory
    # without forcing the user into the tooltip. The full description
    # (and any extra context) still surfaces via the native ``title=``
    # tooltip wired by ``_RESULTS_HEADER_TOOLTIP_JS``.
    headers = [_results_header_text(c) for c in cols]
    datatypes = [_col_datatype(c) for c in cols]

    total_matches = len(ranked)
    count_md = f"**{total_matches:,}** / {len(rows):,} rows"

    if not ranked:
        return (
            {"headers": headers, "data": [], "metadata": None},
            count_md,
            datatypes,
        )

    formatters = {c["key"]: _fmt_cell(c.get("fmt_spec")) for c in cols}

    def _fmt_text(c: dict[str, Any], v: Any) -> str:
        fmt = formatters.get(c["key"])
        if fmt is not None:
            return fmt(v)
        if v is None or (isinstance(v, float) and v != v):
            return ""
        return str(v)

    data: list[list[Any]] = []
    display_values: list[list[str]] = []
    styling: list[list[str]] = []

    for row in ranked:
        row_data: list[Any] = []
        row_disp: list[str] = []
        row_style: list[str] = []
        for c in cols:
            key = c["key"]
            v = row.get(key)
            is_metric = bool(c.get("metric"))
            is_numeric = is_metric or key in _NUMERIC_CONFIG_KEYS

            # Underlying cell value — kept as a real number on metric /
            # numeric-config columns so TanStack sorts correctly. None
            # flows through as null and lands at the bottom of a sort.
            if is_numeric:
                row_data.append(v if compute._is_num(v) else None)
            else:
                row_data.append("" if v is None else str(v))

            text = _fmt_text(c, v)
            std_key = c.get("std_key")
            if std_key and is_metric and text:
                std = row.get(std_key)
                # Skip zero / missing stds — an empty std column or a
                # single measurement per config both come back as 0, and
                # rendering "± 0.00" is noise, not information.
                if compute._is_num(std) and std > 0:
                    fmt = formatters.get(key)
                    std_text = fmt(std) if fmt is not None else str(std)
                    text = f"{text} ± {std_text}"
            row_disp.append(text)

            if is_metric and compute._is_num(v) and text:
                sc = scales.get(key)
                t = compute._normalize_value(v, sc) if sc else None
                color = _rdylgn(t) if t is not None else "transparent"
                row_style.append(_chip_style(color))
            else:
                row_style.append("")

        data.append(row_data)
        display_values.append(row_disp)
        styling.append(row_style)

    value = {
        "headers": headers,
        "data": data,
        "metadata": {"display_value": display_values, "styling": styling},
    }
    return value, count_md, datatypes


# --------------------------------------------------------------------------- #
# Leaderboards tab — plotly radar chart
# --------------------------------------------------------------------------- #
# Plotly renders on top of both the light and the dark Gradio themes,
# and the server can't read the current theme — so every baked-in
# color here is a mid-slate (``#8B93A7``) that has readable contrast
# on *both* ``#FFFFFF`` and the dark ``#1E2233`` page. This matches
# the tick / grid palette the leaderboard radar already uses so the
# Compare and Leaderboards tabs share a visual language.
_PLOTLY_FG = "#8B93A7"
_PLOTLY_GRID = "rgba(139, 147, 167, 0.28)"
_PLOTLY_LAYOUT = dict(
    paper_bgcolor="rgba(0,0,0,0)",
    plot_bgcolor="rgba(0,0,0,0)",
    font=dict(family="Source Sans 3, ui-sans-serif, system-ui, sans-serif",
              size=13, color=_PLOTLY_FG),
    margin=dict(l=40, r=20, t=40, b=40),
    hoverlabel=dict(bgcolor="#1B1B1D", font_color="#FFFFFF",
                    font_family="IBM Plex Mono, monospace"),
)


def _plotly_axis(title: str = "") -> dict[str, Any]:
    """Axis dict with theme-neutral colors for compare/scatter/stacked plots."""
    return dict(
        title=dict(text=title, font=dict(color=_PLOTLY_FG)),
        color=_PLOTLY_FG,
        gridcolor=_PLOTLY_GRID,
        linecolor=_PLOTLY_GRID,
        zerolinecolor=_PLOTLY_GRID,
        tickfont=dict(color=_PLOTLY_FG),
    )


def _plotly_title(text: str) -> dict[str, Any]:
    """Plot-title dict with a themed font color."""
    return dict(text=text, font=dict(size=13, color=_PLOTLY_FG),
                x=0.01, xanchor="left")


def _plotly_legend(title: str | None = None) -> dict[str, Any]:
    """Legend dict — themed font color, transparent bg, subtle border."""
    return dict(
        font=dict(size=12, color=_PLOTLY_FG),
        title=dict(text=title or "", font=dict(size=12, color=_PLOTLY_FG)),
        bgcolor="rgba(0,0,0,0)",
        bordercolor=_PLOTLY_GRID,
    )


# The radar/cards render these group keys in a bespoke RGB layout; any
# additional leaderboard group keys (e.g. depth's lossless / use_log) are
# appended generically so distinct configs don't collapse visually.
_LB_BASE_GROUP_KEYS: tuple[str, ...] = ("vcodec", "pix_fmt", "g", "crf", "backend")

# Friendly display labels for the submit-form codec picker; the checkbox
# *values* stay the raw ffmpeg codec names the worker expects.
_VCODEC_SUBMIT_LABELS: dict[str, str] = {
    "libsvtav1": "AV1 (libsvtav1)",
    "libaom-av1": "AV1 (libaom-av1)",
}


def _lb_extra_group_keys(bench: schema.BenchmarkConfig) -> list[str]:
    return [k for k in bench.leaderboard_group_keys if k not in _LB_BASE_GROUP_KEYS]


def _label_for_item(bench: schema.BenchmarkConfig, row: dict[str, Any]) -> str:
    """One-line config label used in legends and hover tooltips."""
    label = (
        f"{row.get('vcodec','?')} · {row.get('pix_fmt','?')} · "
        f"g={row.get('g','?')} crf={row.get('crf','?')} · {row.get('backend','?')}"
    )
    cbk = _cols_by_key(bench)
    for k in _lb_extra_group_keys(bench):
        col = cbk.get(k)
        short = (col.get("short") or col.get("label") or k).replace("\n", " ") if col else k
        label += f" · {short}={row.get(k, '?')}"
    return label


def _hex_to_rgba(hex_color: str, alpha: float) -> str:
    """Convert ``#RRGGBB`` to ``rgba(r,g,b,alpha)`` for translucent polygon fills."""
    h = hex_color.lstrip("#")
    r, g, b = int(h[0:2], 16), int(h[2:4], 16), int(h[4:6], 16)
    return f"rgba({r},{g},{b},{alpha})"


def _leaderboard_data(bench: schema.BenchmarkConfig, ts: str, cat: str) -> dict[str, Any]:
    """Compute the leaderboard payload once, catching Hub failures."""
    rows = _safe_cached_rows(bench)
    try:
        return compute.leaderboard(
            rows, ts=ts, cat=cat,
            axes=bench.leaderboard_axes,
            cats=bench.leaderboard_cats,
            group_keys=bench.leaderboard_group_keys,
            columns=bench.columns,
        )
    except Exception as e:  # noqa: BLE001
        logger.warning("leaderboard failed: %s", e)
        return {"axes": [], "items": []}


def _wrap_axis_label(s: str, max_chars: int = 14) -> str:
    """Insert a single ``<br>`` to wrap a long radar tick label.

    Plotly tick labels don't word-wrap; once a metric short-name gets
    past ~14 characters it overlaps neighbouring spokes on the radar.
    We split on the first space so e.g. "Video/Image size ratio"
    becomes two lines without us having to widen the figure margins.
    Short labels and single-token labels fall through unchanged.
    """
    if len(s) <= max_chars or " " not in s:
        return s
    return s.replace(" ", "<br>", 1)


def _leaderboard_plot(bench: schema.BenchmarkConfig, data: dict[str, Any], cat: str) -> go.Figure:
    """Radar plot: one polygon per top-``top_n`` config, no legend.

    Each axis is a leaderboard metric, normalized so 1.0 = best-in-class
    on that axis (``compute.leaderboard`` already orients polarity).
    Colors match the rank badges in the companion "Top configurations"
    card list (see :func:`_leaderboard_cards_html`) so users can pair
    each polygon with its card by color.

    ``cat`` is used to bold the angular tick labels of axes whose weight
    is ≥ 1 in the active category, so users can see at a glance which
    metrics drive the current ranking.
    """
    items = data.get("items", [])
    axes = data.get("axes", [])
    axis_keys = data.get("axis_keys", [])
    weights = bench.leaderboard_cats.get(cat, {}).get("weights", {})
    cbk = _cols_by_key(bench)
    # Plotly renders across light and dark Gradio themes. We can't read the
    # current theme from the server, so axes / ticks / fonts use mid-gray
    # tones that have sufficient contrast on both #FFFFFF and #1B1B1D.
    fg = "#8B93A7"
    grid = "rgba(139, 147, 167, 0.28)"
    if not items or not axes:
        fig = go.Figure()
        fig.update_layout(
            annotations=[dict(text="No rows match this category.",
                              xref="paper", yref="paper", x=0.5, y=0.5,
                              showarrow=False, font=dict(size=14, color=fg))],
            xaxis=dict(visible=False), yaxis=dict(visible=False),
            **{**_PLOTLY_LAYOUT, "height": 520},
        )
        return fig

    # Close each polygon by repeating the first axis at the end so the
    # line returns to its starting vertex (plotly doesn't auto-close
    # scatterpolar traces).
    theta = list(axes) + [axes[0]]

    fig = go.Figure()
    for i, it in enumerate(items):
        label = _label_for_item(bench, it["row"])
        color = schema.PODIUM_COLORS[i % len(schema.PODIUM_COLORS)]
        values = list(it["values"]) + [it["values"][0]]
        score_pct = 100 * (1 - it["score"])
        rank = i + 1
        # Every polygon stays as a stroked outline so users can compare
        # shapes. The #1 config gets a subtle 15% fill to anchor the
        # chart on the winner.
        fill_alpha = 0.18 if rank == 1 else 0.0
        line_w = 2.6 if rank == 1 else 1.9
        marker_size = 7 if rank == 1 else 5
        fig.add_trace(go.Scatterpolar(
            r=values,
            theta=theta,
            name=f"#{rank} · {label}",
            mode="lines+markers",
            fill="toself" if fill_alpha > 0 else "none",
            fillcolor=_hex_to_rgba(color, fill_alpha) if fill_alpha > 0 else None,
            line=dict(color=color, width=line_w),
            marker=dict(color=color, size=marker_size,
                        line=dict(width=1, color="rgba(255,255,255,0.8)")),
            opacity=1.0,
            hovertemplate=(
                f"<b>#{rank} · {label}</b><br>"
                "%{theta}: %{r:.2f}<br>"
                f"score {score_pct:.1f}"
                "<extra></extra>"
            ),
        ))

    fig.update_layout(
        polar=dict(
            bgcolor="rgba(0,0,0,0)",
            radialaxis=dict(
                visible=True,
                range=[0, 1],
                tickvals=[0.25, 0.5, 0.75, 1.0],
                ticktext=[".25", ".50", ".75", "1.0"],
                tickfont=dict(family="IBM Plex Mono, monospace",
                              size=10, color=fg),
                gridcolor=grid,
                linecolor=grid,
                angle=90,
                tickangle=0,
            ),
            angularaxis=dict(
                gridcolor=grid,
                linecolor=grid,
                tickfont=dict(size=12, color=fg),
                rotation=90,
                direction="clockwise",
                tickmode="array",
                tickvals=list(axes),
                ticktext=[
                    (
                        f'<b><span style="color:#FFFFFF">'
                        f'{_wrap_axis_label(a)}{_polarity_arrow(cbk.get(k))}</span></b>'
                        if weights.get(k, 0) >= 1
                        else _wrap_axis_label(a) + _polarity_arrow(cbk.get(k))
                    )
                    for a, k in zip(axes, axis_keys)
                ],
            ),
        ),
        showlegend=False,
        height=520,
        **{**_PLOTLY_LAYOUT,
           "font": dict(family="Source Sans 3, ui-sans-serif, system-ui, sans-serif",
                        size=13, color=fg),
           "margin": dict(l=60, r=60, t=40, b=40)},
    )
    return fig


def _codec_color(codec: Any) -> str:
    """Look up a codec's accent color, falling back to a neutral slate."""
    return schema.CODEC_COLORS.get(
        str(codec).lower(), schema.CODEC_FALLBACK_COLOR,
    )


def _codec_chip_html(codec: Any) -> str:
    """Render a codec name as a plain pill. Reused by cards and table."""
    c = "" if codec is None else str(codec)
    return f'<span class="lb-codec">{c}</span>'


def _rank_chip_html(rank: int, color: str) -> str:
    """Render a rank index as the same numbered badge used on the cards."""
    return (
        f'<span class="lb-badge lb-badge-inline" '
        f'style="background:{color}">{rank}</span>'
    )


def _leaderboard_cards_html(bench: schema.BenchmarkConfig, data: dict[str, Any]) -> str:
    """Right-column HTML: one podium card per ranked config.

    Card colors (rank badge, left accent) track ``schema.PODIUM_COLORS``
    in the same order as the radar polygons, so the polygon and the
    card for the same config share a color. Raw metric values are
    surfaced as labeled chips using each column's ``fmt_spec``.
    """
    items = data.get("items", [])
    if not items:
        return (
            '<div class="lb-cards">'
            '<div class="lb-empty">No rows match this category.</div>'
            '</div>'
        )

    cbk = _cols_by_key(bench)
    axis_cols = [cbk[k] for k in bench.leaderboard_axes if k in cbk]

    def _fmt(col: dict[str, Any], v: Any) -> str:
        fmt = _fmt_cell(col.get("fmt_spec"))
        if fmt is None:
            return "—" if v is None else str(v)
        return fmt(v) or "—"

    parts: list[str] = ['<div class="lb-cards">']
    for i, it in enumerate(items):
        row = it["row"]
        color = schema.PODIUM_COLORS[i % len(schema.PODIUM_COLORS)]
        rank = i + 1
        codec = row.get("vcodec", "?")
        config_bits = (
            f'<code>{row.get("pix_fmt", "?")}</code>'
            f' · <code>g={row.get("g", "?")}</code>'
            f' · <code>crf={row.get("crf", "?")}</code>'
            f' · <code>{row.get("backend", "?")}</code>'
        )
        for k in _lb_extra_group_keys(bench):
            col = cbk.get(k)
            short = (col.get("short") or col.get("label") or k).replace("\n", " ") if col else k
            config_bits += f' · <code>{short}={row.get(k, "?")}</code>'
        metric_chips = "".join(
            '<span class="lb-metric">'
            f'<span class="lb-metric-k">{(c.get("short") or c["label"]).replace(chr(10), " ")}{_polarity_arrow(c)}</span>'
            f'<span class="lb-metric-v">{_fmt(c, row.get(c["key"]))}</span>'
            '</span>'
            for c in axis_cols
        )
        parts.append(
            '<div class="lb-card" '
            f'style="--rank-color: {color}">'
            f'  <div class="lb-badge">{rank}</div>'
            '  <div class="lb-card-body">'
            '    <div class="lb-card-head">'
            f'      {_codec_chip_html(codec)}'
            f'      <span class="lb-config">{config_bits}</span>'
            '    </div>'
            f'    <div class="lb-metrics">{metric_chips}</div>'
            '  </div>'
            '</div>'
        )
    parts.append("</div>")
    return "".join(parts)


def _leaderboard_table(bench: schema.BenchmarkConfig, data: dict[str, Any]) -> pd.DataFrame:
    """Tabular view of the same top-N configs shown in the cards / radar.

    Columns are ordered like the cards read them (rank → config →
    metrics → score). The Rank cell carries the same colored badge
    used on the cards (via ``schema.PODIUM_COLORS``) so the radar
    polygon, card, and table row share one visual identity per config.
    Codec is rendered as plain text in the table — the colored pill is
    kept on the cards where it doubles as the heading, but adding it to
    a tabular row just adds visual noise next to the rank badge. Numeric
    columns are pre-formatted with their ``fmt_spec`` so chip text and
    table text match to the last decimal. Score is shown on a 0–100
    scale (higher = better).
    """
    items = data.get("items", [])
    cbk = _cols_by_key(bench)
    axis_cols = [cbk[k] for k in bench.leaderboard_axes if k in cbk]
    # Use the schema's full label (e.g. "Decoding (ms)", "PSNR (dB)") via
    # ``_results_header_text`` so unit suffixes survive into the header,
    # matching the Results table's treatment. ``_nb_slash`` keeps tokens
    # like ``Video/Image`` unbreakable so they don't wrap mid-word.
    metric_labels = [_nb_slash(_results_header_text(c)) for c in axis_cols]
    # Config columns come from the benchmark's group keys so depth's
    # lossless / use_log surface as their own columns. For RGB the group
    # keys (vcodec, pix_fmt, g, crf, backend) reproduce the previous
    # fixed Codec / Pixel format / GOP / CRF / Backend headers exactly.
    cfg_cols = [cbk[k] for k in bench.leaderboard_group_keys if k in cbk]
    cfg_labels = [(c.get("short") or c["label"]).replace("\n", " ") for c in cfg_cols]
    score_label = "Score (0\u2013100)"
    cols = ["Rank", *cfg_labels, *metric_labels, score_label]

    def _fmt(col: dict[str, Any], v: Any) -> str:
        fmt = _fmt_cell(col.get("fmt_spec"))
        if fmt is None:
            return "" if v is None else str(v)
        return fmt(v)

    if not items:
        return pd.DataFrame(columns=cols)

    rows_out: list[dict[str, Any]] = []
    for i, it in enumerate(items):
        row = it["row"]
        rank = i + 1
        rank_color = schema.PODIUM_COLORS[i % len(schema.PODIUM_COLORS)]
        score_pct = 100 * (1 - float(it.get("score", 1.0)))
        record: dict[str, Any] = {"Rank": _rank_chip_html(rank, rank_color)}
        for c, label in zip(cfg_cols, cfg_labels):
            record[label] = str(row.get(c["key"], ""))
        record[score_label] = f"{score_pct:.1f}"
        for c, label in zip(axis_cols, metric_labels):
            record[label] = _fmt(c, row.get(c["key"]))
        rows_out.append(record)
    return pd.DataFrame(rows_out, columns=cols)


def _render_leaderboard(
    bench: schema.BenchmarkConfig, ts: str, cat: str, top_n: int,
) -> tuple[go.Figure, str, pd.DataFrame]:
    """Bundle radar + cards + table so one handler updates them.

    ``compute.leaderboard`` runs over every aggregated configuration;
    ``top_n`` is a *display* cap applied here so the radar / cards /
    table only surface the requested number of best-scoring rows.
    """
    data = _leaderboard_data(bench, ts, cat)
    n = max(0, int(top_n))
    items = data.get("items", [])
    if n and len(items) > n:
        data = {**data, "items": items[:n]}
    return (
        _leaderboard_plot(bench, data, cat),
        _leaderboard_cards_html(bench, data),
        _leaderboard_table(bench, data),
    )


# --------------------------------------------------------------------------- #
# Compare tab — plotly figures rendered via gr.Plot
#
# Gradio's native BarPlot / ScatterPlot (Altair under the hood) shipped a
# rendering regression in 6.12 / 6.13 where two or more nativeplot
# components on the same page race each other and only the first one
# renders (see gradio#12515). The Leaderboards radar chart already uses
# gr.Plot + plotly successfully, so the Compare tab does the same and
# side-steps the buggy component entirely. Each handler returns a
# ``go.Figure`` fed into a ``gr.Plot`` output.
# --------------------------------------------------------------------------- #
def _polarity_hint(col: dict[str, Any] | None) -> str:
    """Return ``"lower is better"`` / ``"higher is better"`` for a metric column.

    Returns the empty string for non-metric / unknown columns so callers
    can append the hint unconditionally.
    """
    if not col:
        return ""
    if col.get("lower"):
        return "lower is better"
    if col.get("higher"):
        return "higher is better"
    return ""


def _polarity_arrow(col: dict[str, Any] | None) -> str:
    """Return ``" \u2193"`` / ``" \u2191"`` for a metric column's polarity, else ``""``.

    Mirrors the trailing arrows baked into the schema ``label`` so the
    Leaderboards radar / cards (which use the arrow-free ``short`` names)
    signal lower-is-better / higher-is-better the same way the tables do.
    """
    if not col:
        return ""
    if col.get("lower"):
        return " \u2193"
    if col.get("higher"):
        return " \u2191"
    return ""


def _metric_axis_label(col: dict[str, Any] | None, fallback: str) -> str:
    """Y-axis label for a metric column: full ``label`` (with unit) + polarity.

    Strips the trailing ↑ / ↓ from the schema label since the axis label
    spells the polarity out in words instead. When the label already ends
    in a parenthesised unit (e.g. ``"PSNR (dB)"``), the polarity hint is
    folded into the same parens (``"PSNR (dB, higher is better)"``)
    rather than tacked on as a second pair, so the axis title reads as
    one annotation instead of two.
    """
    if not col:
        return fallback
    raw = (col.get("label") or col.get("short") or fallback).replace("\n", " ")
    base = re.sub(r"\s*[↑↓]\s*$", "", raw).strip()
    hint = _polarity_hint(col)
    if not hint:
        return base
    m = re.match(r"^(.*)\(([^()]+)\)\s*$", base)
    if m:
        head, inner = m.group(1).rstrip(), m.group(2).strip()
        return f"{head} ({inner}, {hint})"
    return f"{base} ({hint})"


def _empty_figure(msg: str, height: int = 380) -> go.Figure:
    fig = go.Figure()
    fig.update_layout(
        annotations=[dict(text=msg, xref="paper", yref="paper", x=0.5, y=0.5,
                          showarrow=False, font=dict(size=14, color="#8B93A7"))],
        xaxis=dict(visible=False), yaxis=dict(visible=False),
        **{**_PLOTLY_LAYOUT, "height": height},
    )
    return fig


def _render_compare_bar(bench: schema.BenchmarkConfig, metric: str, group_by: str) -> go.Figure:
    """Mean of ``metric`` grouped by ``group_by`` as a plotly bar figure."""
    rows = _safe_cached_rows(bench)
    data = compute.compare_bar(rows, metric=metric, group_by=group_by)
    if not data:
        return _empty_figure("No data for this selection.", height=380)

    xs = [d["k"] for d in data]
    ys = [d["v"] for d in data]
    # Color the bars by their group, matching the codec chip palette when
    # grouping by codec, and falling back to the stack palette otherwise.
    if group_by == "vcodec":
        colors = [_codec_color(k) for k in xs]
    else:
        colors = [schema.STACK_COLORS[i % len(schema.STACK_COLORS)]
                  for i in range(len(xs))]

    cbk = _cols_by_key(bench)
    metric_col = cbk.get(metric)
    group_col = cbk.get(group_by)
    y_label = _metric_axis_label(metric_col, fallback=metric)
    x_label = (
        (group_col.get("short") or group_col.get("label") or group_by).replace("\n", " ")
        if group_col else group_by
    )

    fig = go.Figure(go.Bar(
        x=xs, y=ys, marker_color=colors,
        hovertemplate="<b>%{x}</b><br>%{y:.3f}<extra></extra>",
    ))
    fig.update_layout(
        xaxis=_plotly_axis(x_label),
        yaxis=_plotly_axis(y_label),
        height=380,
        **_PLOTLY_LAYOUT,
    )
    return fig


def _render_compare_scatter(bench: schema.BenchmarkConfig) -> go.Figure:
    """Quality-vs-compression scatter, colored by codec."""
    rows = _safe_cached_rows(bench)
    pts = compute.compare_scatter(rows, x=bench.scatter_x, y=bench.scatter_y)
    if not pts:
        return _empty_figure("No data yet.", height=420)

    ycol = _cols_by_key(bench).get(bench.scatter_y)
    yshort = (ycol.get("short") or ycol.get("label") or bench.scatter_y) if ycol else bench.scatter_y
    fig = go.Figure()
    # Group points by codec so each gets its own legend entry + color.
    by_codec: dict[str, list[dict[str, Any]]] = {}
    for p in pts:
        by_codec.setdefault(p.get("c") or "unknown", []).append(p)

    for codec, group in sorted(by_codec.items()):
        color = _codec_color(codec)
        fig.add_trace(go.Scatter(
            x=[p["x"] for p in group],
            y=[p["y"] for p in group],
            mode="markers",
            name=codec,
            marker=dict(color=color, size=7, opacity=0.7,
                        line=dict(width=0.5, color="rgba(0,0,0,0.25)")),
            text=[p["label"] for p in group],
            hovertemplate=(
                f"<b>{codec}</b><br>"
                f"%{{text}}<br>size=%{{x:.3f}} · {yshort}=%{{y:.3f}}"
                "<extra></extra>"
            ),
        ))
    # Pin a vertical codec legend on the right side of the plot so the
    # color → codec mapping is always visible without hovering.
    legend = _plotly_legend("Codec")
    legend.update(
        orientation="v",
        yanchor="top", y=1.0,
        xanchor="left", x=1.02,
    )
    fig.update_layout(
        xaxis=_plotly_axis("Video / image size ratio (lower is smaller)"),
        yaxis=_plotly_axis(_metric_axis_label(ycol, "Quality")),
        showlegend=True,
        legend=legend,
        height=420,
        **_PLOTLY_LAYOUT,
    )
    return fig


def _render_compare_stacked(bench: schema.BenchmarkConfig) -> go.Figure:
    """Stacked decoding-latency bars: one stack per codec, segments per access pattern."""
    rows = _safe_cached_rows(bench)
    data = compute.compare_stacked(rows)
    if not data["data"]:
        return _empty_figure("No data yet.", height=420)

    codecs = [entry["k"] for entry in data["data"]]
    modes = data["modes"]
    # Build one Bar trace per access pattern so plotly stacks them.
    fig = go.Figure()
    for i, mode in enumerate(modes):
        color = schema.STACK_COLORS[i % len(schema.STACK_COLORS)]
        ys = []
        for entry in data["data"]:
            seg = next((s for s in entry["segments"] if s["key"] == mode), None)
            ys.append(seg["v"] if seg else 0.0)
        fig.add_trace(go.Bar(
            x=codecs, y=ys, name=mode,
            marker_color=color,
            hovertemplate=(
                f"<b>%{{x}}</b> · {mode}<br>"
                "%{y:.1f} ms<extra></extra>"
            ),
        ))
    fig.update_layout(
        barmode="stack",
        xaxis=_plotly_axis("Codec"),
        yaxis=_plotly_axis("Mean decoding time (ms, lower is better)"),
        legend=_plotly_legend("Access pattern"),
        height=420,
        **_PLOTLY_LAYOUT,
    )
    return fig


# --------------------------------------------------------------------------- #
# Submit tab
# --------------------------------------------------------------------------- #
def _split_values(text: str) -> list[str]:
    """Split whitespace/comma-separated free-text into a trimmed list."""
    return [s.strip() for s in re.split(r"[\s,]+", text or "") if s.strip()]


def _merge_values(checked: list[str] | None, custom_text: str) -> list[str]:
    """Merge a CheckboxGroup selection with its "custom values" textbox.

    Dedupes while preserving order: checked presets first, then any
    additional tokens typed into the custom field. Empty strings are
    dropped so the submission validator doesn't choke on stray commas.
    """
    seen: set[str] = set()
    out: list[str] = []
    for s in list(checked or []) + _split_values(custom_text):
        s = str(s).strip()
        if s and s not in seen:
            seen.add(s)
            out.append(s)
    return out


def _parse_int_list(values: list[str]) -> list[int]:
    out: list[int] = []
    for v in values or []:
        try:
            out.append(int(v))
        except (TypeError, ValueError):
            pass
    return out


def _submission_summary(
    bench: schema.BenchmarkConfig,
    repos: list[str], repos_custom: str,
    vcodecs: list[str], vcodecs_custom: str,
    pix_fmts: list[str], pix_fmts_custom: str,
    g: list[str], g_custom: str,
    crf: list[str], crf_custom: str,
    ts_modes: list[str],
    backends: list[str],
    samples: int,
    lossless: list[str] | None = None,
    use_log: list[str] | None = None,
) -> str:
    """Live preview shown next to the Submit button.

    Returns the total config count as markdown so the Submit-tab side
    panel can render it as-is. For knobs with a custom-value textbox,
    the final list is the union of the checkbox selection and any
    comma/whitespace-separated custom values.
    """
    repos = _merge_values(repos, repos_custom)
    vcodecs = _merge_values(vcodecs, vcodecs_custom)
    pix_fmts = _merge_values(pix_fmts, pix_fmts_custom)
    g = _merge_values(g, g_custom)
    crf = _merge_values(crf, crf_custom)
    n = max(len(repos), 1) * max(len(vcodecs), 1) * max(len(pix_fmts), 1) \
        * max(len(g), 1) * max(len(crf), 1) * max(len(ts_modes), 1) \
        * max(len(backends), 1)
    if lossless is not None:
        n *= max(len(lossless), 1)
    if use_log is not None:
        n *= max(len(use_log), 1)
    return f"**{n:,}** total configs"


def _queue_dataframe(bench: schema.BenchmarkConfig) -> pd.DataFrame:
    """Recent submissions as a DataFrame for the Submit-tab queue table."""
    items = _load_queue(bench)
    if not items:
        return pd.DataFrame(columns=["ID", "Status", "Dataset", "When"])
    return pd.DataFrame([
        {"ID": it["id"], "Status": it["status"], "Dataset": it["repo"], "When": it["when"]}
        for it in items
    ])


def _submit(
    bench: schema.BenchmarkConfig,
    repos: list[str], repos_custom: str,
    vcodecs: list[str], vcodecs_custom: str,
    pix_fmts: list[str], pix_fmts_custom: str,
    g: list[str], g_custom: str,
    crf: list[str], crf_custom: str,
    ts_modes: list[str],
    backends: list[str],
    samples: int,
    lossless: list[str] | None = None,
    use_log: list[str] | None = None,
    depth_min: str | float | None = None,
    depth_max: str | float | None = None,
    shift: str | float | None = None,
) -> tuple[str, pd.DataFrame]:
    """Validate the form, commit the submission JSON, and refresh the queue.

    Raises ``gr.Error`` on validation failure, missing token, or Hub
    write errors so Gradio surfaces a toast instead of a stack trace.
    Returns ``(status_md, queue_df)`` for the two output components.
    For knobs with a custom-value textbox, the final list is the union
    of the checkbox selection and any comma/whitespace-separated custom
    values from the sibling textbox.
    """
    repos = _merge_values(repos, repos_custom)
    vcodecs = _merge_values(vcodecs, vcodecs_custom)
    pix_fmts = _merge_values(pix_fmts, pix_fmts_custom)
    g_all = _merge_values(g, g_custom)
    crf_all = _merge_values(crf, crf_custom)
    extra: dict[str, Any] = {}
    if bench.key == "depth" and lossless is not None:
        def _opt_float(x: str | float | None) -> float | None:
            s = str(x).strip() if x is not None else ""
            if not s:
                return None
            try:
                return float(s)
            except ValueError as e:
                raise gr.Error(f"Invalid number: {s!r}") from e
        extra = {
            "lossless": list(lossless or []),
            "use_log": list(use_log or []),
            "depth_min": _opt_float(depth_min),
            "depth_max": _opt_float(depth_max),
            "shift": _opt_float(shift),
        }
    try:
        payload = SubmissionIn(
            repos=repos,
            vcodecs=vcodecs,
            pix_fmts=pix_fmts,
            g=_parse_int_list(g_all),
            crf=_parse_int_list(crf_all),
            timestamps_modes=ts_modes,
            backends=backends,
            samples_per_config=int(samples or 1),
            **extra,
        )
    except ValidationError as e:
        msgs = "; ".join(
            f"{'.'.join(str(p) for p in err['loc'])}: {err['msg']}"
            for err in e.errors()
        )
        raise gr.Error(f"Invalid submission: {msgs}") from e

    return _commit_and_refresh(bench, payload)


def _commit_and_refresh(
    bench: schema.BenchmarkConfig,
    payload: SubmissionIn,
    *,
    enforce_cap: bool = True,
    label: str = "submission",
) -> tuple[str, pd.DataFrame]:
    """Validate size, commit ``payload`` to the Hub, and refresh the queue.

    ``enforce_cap=False`` is used by the maintainer-only full-sweep path,
    which intentionally overshoots ``MAX_CONFIGS_PER_SUBMISSION``.
    """
    n = payload.total_configs()
    if enforce_cap and n > MAX_CONFIGS_PER_SUBMISSION:
        raise gr.Error(
            f"Too many configurations ({n:,}). Max {MAX_CONFIGS_PER_SUBMISSION:,} per submission."
        )

    submission_id = _make_submission_id()
    created_at = datetime.now(timezone.utc).isoformat(timespec="seconds")
    body = payload.to_worker_payload(submission_id, created_at, bench.key)

    try:
        _commit_submission(bench, submission_id, body)
    except RuntimeError as e:
        raise gr.Error(str(e)) from e
    except HfHubHTTPError as e:
        logger.exception("Hub commit failed")
        raise gr.Error(f"Hub commit failed: {e}") from e

    queue_cache = _queue_cache_for(bench.key)
    queue_cache["items"] = []
    queue_cache["at"] = 0.0

    gr.Info(f"Queued {label} {submission_id} ({n:,} configs).")
    return f"Queued `{submission_id}` ({n:,} configs).", _queue_dataframe(bench)


def _submit_full_sweep(bench: schema.BenchmarkConfig, confirmed: bool) -> tuple[str, pd.DataFrame]:
    """Maintainer-only: queue the canonical full Cartesian sweep.

    Reads ``bench.full_sweep`` so the parameter set lives next to every
    other knob vocabulary. Bypasses ``MAX_CONFIGS_PER_SUBMISSION`` because
    the curated full sweep is deliberately huge (tens of thousands of
    configs) and is gated behind the "I understand" checkbox in the
    collapsed Submit-tab accordion.
    """
    if not confirmed:
        raise gr.Error("Tick the confirm checkbox before triggering the full sweep.")

    fs = bench.full_sweep
    extra: dict[str, Any] = {}
    if bench.key == "depth":
        extra = {
            "lossless": list(fs.get("lossless", [])),
            "use_log": list(fs.get("use_log", [])),
            "depth_min": fs.get("depth_min"),
            "depth_max": fs.get("depth_max"),
            "shift": fs.get("shift"),
        }
    try:
        payload = SubmissionIn(
            repos=list(fs["repos"]),
            vcodecs=list(fs["vcodecs"]),
            pix_fmts=list(fs["pix_fmts"]),
            g=_parse_int_list(list(fs["g"])),
            crf=_parse_int_list(list(fs["crf"])),
            timestamps_modes=list(fs["timestamps_modes"]),
            backends=list(fs["backends"]),
            samples_per_config=int(fs["samples_per_config"]),
            **extra,
        )
    except ValidationError as e:
        msgs = "; ".join(
            f"{'.'.join(str(p) for p in err['loc'])}: {err['msg']}"
            for err in e.errors()
        )
        raise gr.Error(f"Invalid full-sweep config: {msgs}") from e

    return _commit_and_refresh(bench, payload, enforce_cap=False, label="full sweep")


# --------------------------------------------------------------------------- #
# Parameters tab — render cards from bench.param_groups
# --------------------------------------------------------------------------- #
def _params_html(bench: schema.BenchmarkConfig) -> str:
    """Render the Parameters tab as grouped cards driven by ``bench.param_groups``."""
    cbk = _cols_by_key(bench)
    parts: list[str] = ['<div class="params-page">']
    for grp in bench.param_groups:
        keys = grp["keys"]
        cards: list[str] = []
        for k in keys:
            c = cbk.get(k) or bench.param_notes.get(k)
            if not c:
                continue
            cards.append(
                f'<div class="param-card">'
                f'  <div class="param-card-head">'
                f'    <span class="param-key">{k}</span>'
                f'    <span class="param-label">{(c.get("short") or c["label"]).replace(chr(10), " ")}</span>'
                f'  </div>'
                f'  <p>{c.get("desc", "")}</p>'
                f'</div>'
            )
        parts.append(
            f'<section class="param-group">'
            f'  <div class="param-group-head"><h3>{grp["t"]}</h3><p>{grp["desc"]}</p></div>'
            f'  <div class="param-cards">{"".join(cards)}</div>'
            f'</section>'
        )
    parts.append("</div>")
    return "".join(parts)


# --------------------------------------------------------------------------- #
# Theme + Blocks layout
# --------------------------------------------------------------------------- #
def _hf_theme() -> gr.themes.Base:
    """Gradio theme tuned to the Hugging Face brand palette and typography."""
    return gr.themes.Default(
        primary_hue=gr.themes.colors.yellow,
        secondary_hue=gr.themes.colors.blue,
        neutral_hue=gr.themes.colors.gray,
        font=(gr.themes.GoogleFont("Source Sans 3"), "ui-sans-serif", "system-ui", "sans-serif"),
        font_mono=(gr.themes.GoogleFont("IBM Plex Mono"), "ui-monospace", "monospace"),
    ).set(
        body_background_fill="*neutral_50",
        body_background_fill_dark="#16192A",
        body_text_color="#1B1B1D",
        body_text_color_dark="#F5F6F8",
        block_background_fill="#FFFFFF",
        block_background_fill_dark="#1E2233",
        block_border_color="#EEEFF2",
        block_border_color_dark="rgba(255, 255, 255, 0.08)",
        block_border_width="1px",
        block_radius="12px",
        button_primary_background_fill="#1B1B1D",
        button_primary_background_fill_hover="#32343D",
        button_primary_text_color="#FFFFFF",
        button_secondary_background_fill="#FFFFFF",
        button_secondary_background_fill_dark="#1E2233",
        button_secondary_border_color="#EEEFF2",
        button_secondary_border_color_dark="rgba(255, 255, 255, 0.18)",
        checkbox_background_color_selected="#496EF0",
        checkbox_border_color_selected="#496EF0",
        slider_color="#496EF0",
        input_background_fill="#FFFFFF",
        input_background_fill_dark="#1E2233",
        input_border_color="#EEEFF2",
        input_border_color_dark="rgba(255, 255, 255, 0.18)",
    )


def _load_css() -> str:
    return (ROOT / "styles.css").read_text(encoding="utf-8")


def _initial_counts(bench: schema.BenchmarkConfig) -> tuple[int, int]:
    rows = _safe_cached_rows(bench)
    repos = {r.get("repo_id") for r in rows if r.get("repo_id")}
    return len(rows), len(repos)


def _build_benchmark_tabs(bench: schema.BenchmarkConfig, *, maybe_tab) -> None:
        """Build one benchmark's hero + sub-tabs + refresh footer.

        Called once per benchmark inside a top-level ``gr.Tab`` so the
        same machinery renders both the RGB and Depth benchmarks; every
        handler below closes over ``bench``.
        """
        total_rows, repo_count = _initial_counts(bench)

        filter_keys = list(bench.filter_order)
        n_filters = len(filter_keys)
        cbk = _cols_by_key(bench)
        _FILTER_LABELS = {
            "vcodec": "Codec", "pix_fmt": "Pixel format", "backend": "Backend",
            "g": "GOP", "crf": "CRF", "lossless": "Lossless", "use_log": "Log encode",
        }

        def _filter_label(key: str) -> str:
            if key in _FILTER_LABELS:
                return _FILTER_LABELS[key]
            c = cbk.get(key)
            return (c.get("short") or c.get("label") or key).replace("\n", " ") if c else key

        # Submit defaults / options come from the benchmark config.
        sd = bench.submit_defaults
        is_depth = bench.key == "depth"

        if not os.environ.get("BISECT_NO_HERO"):
            hero_stats = _build_hero(bench, total_rows, repo_count)
        else:
            # Hidden sink so the "Refresh from Hub" wiring below can
            # always include ``hero_stats`` in its outputs list without
            # branching on the bisect flag.
            hero_stats = gr.Markdown(
                _hero_stats_md(bench, total_rows, repo_count), visible=False,
            )

        with gr.Tabs():
            # -------------------- Results --------------------
            with maybe_tab("Results"):
                # Filters default to *empty* so all chips start un-tinted
                # (yellow = user selection). `compute.filter_rows` treats an
                # empty selection as "no filter applied", so the table still
                # shows every row until the user clicks a chip to narrow it.
                # The chips are built dynamically from ``bench.filter_order``
                # so depth picks up its extra ``lossless`` / ``use_log`` knobs.
                filter_comps: dict[str, gr.CheckboxGroup] = {}
                with gr.Accordion("Filters", open=True,
                                  elem_classes="results-accordion filters-accordion"):
                    with gr.Row(equal_height=False, elem_classes="filters-row"):
                        for _fkey in filter_keys:
                            filter_comps[_fkey] = gr.CheckboxGroup(
                                bench.filter_options[_fkey], label=_filter_label(_fkey),
                                value=[],
                                elem_classes=["filter-chips", "filter-cat"],
                            )

                # Column-picker groups derive from the benchmark's declared
                # column order (RGB: Config/Compression/Speed/Quality; Depth
                # adds "Quantized quality", hidden by default).
                groups_in_order = list(dict.fromkeys(c["group"] for c in bench.columns))
                col_pairs_by_group = {
                    grp: [((c.get("short") or c["label"]).replace("\n", " "), c["key"])
                          for c in bench.columns if c["group"] == grp]
                    for grp in groups_in_order
                }
                _dv = set(_picker_default_visible(bench))
                default_by_group = {
                    grp: [k for _, k in pairs if k in _dv]
                    for grp, pairs in col_pairs_by_group.items()
                }

                def _all_keys(group: str) -> list[str]:
                    return [k for _, k in col_pairs_by_group[group]]

                col_masters: dict[str, gr.Checkbox] = {}
                col_subgroups: dict[str, gr.CheckboxGroup] = {}
                with gr.Accordion("Visible columns", open=False,
                                  elem_classes="results-accordion"):
                    with gr.Row():
                        for grp in groups_in_order:
                            _slug = grp.lower().replace(" ", "-")
                            with gr.Column(min_width=0):
                                col_masters[grp] = gr.Checkbox(
                                    label=grp,
                                    value=default_by_group[grp] == _all_keys(grp),
                                    elem_classes=f"col-cat-master col-cat-{_slug}",
                                )
                                col_subgroups[grp] = gr.CheckboxGroup(
                                    choices=col_pairs_by_group[grp],
                                    value=default_by_group[grp],
                                    show_label=False,
                                    elem_classes=f"col-cat col-cat-{_slug}",
                                )

                for _group_name in groups_in_order:
                    _master = col_masters[_group_name]
                    _sub = col_subgroups[_group_name]
                    _all = _all_keys(_group_name)
                    if os.environ.get("BISECT_NO_MASTER"):
                        continue
                    # master → group: toggle all on / all off when the user
                    # clicks the category master checkbox.
                    #
                    # We deliberately do NOT also wire a reverse "group →
                    # master" handler. In Gradio 6 the ``.input`` event
                    # fires on programmatic value updates as well as user
                    # clicks, so a bidirectional binding causes Svelte's
                    # reactivity to enter an infinite update loop
                    # (``effect_update_depth_exceeded``) that pins the
                    # browser's main thread for several seconds — long
                    # enough to trigger the "page slows down browser"
                    # warning on every tab click. Losing the auto-toggle
                    # of the master when the user hand-edits individual
                    # columns is a small UX cost compared to the page
                    # being responsive.
                    _master.input(
                        lambda checked, _all=_all: _all if checked else [],
                        _master, _sub,
                    )

                # Standard ``gr.Dataframe``. Per-column ``datatype`` keeps
                # numeric columns sortable as numbers; ``_render_results_table``
                # provides the formatted text and the heatmap chip color
                # via ``metadata.display_value`` / ``metadata.styling``,
                # so the cell value stays a real number for sort while
                # the rendered cell still shows "42.13 ± 0.45" inside a
                # tinted pill (painted via the ``--chip-bg`` CSS var
                # picked up by ``.results-table .body-cell[style*=
                # "--chip-bg"] .cell-wrap > span`` in styles.css).
                #
                # The initial table/count are pre-rendered server-side and
                # passed as ``value=`` on the output components instead of
                # being wired through ``demo.load(_render_results_table,
                # results_inputs, ...)``. In Gradio 6.13 a ``demo.load``
                # whose input list contains a component living inside a
                # ``gr.Tab`` triggers a Svelte ``effect_update_depth_exceeded``
                # infinite loop on every tab click — so the whole page
                # freezes as soon as the user switches tabs. Baking the
                # initial render into ``value=`` side-steps the buggy
                # code path; user-driven updates still flow through the
                # per-input ``.change`` handlers below.
                _visible_default = [k for grp in groups_in_order
                                    for k in default_by_group[grp]]
                _initial_table, _initial_count, _initial_datatypes = _render_results_table(
                    bench, {k: [] for k in filter_keys}, _visible_default,
                )

                def _column_widths_for(value: dict) -> list[str]:
                    """Column widths sized to the data, not the headers.

                    Uses the median display-value length per column so
                    short numeric columns ("0.22") get a narrow track
                    and long string columns ("lerobot/pusht_imageh264")
                    get a wider one, capped at MAX_PX.  The median is
                    used instead of the max so a single outlier row
                    doesn't blow out the whole column.
                    """
                    CH_PX  = 9    # px per char, IBM Plex Mono 13 px
                    PAD    = 24   # cell padding (left + right)
                    MIN_PX = 48   # floor: enough for 2-3 char values
                    MAX_PX = 260  # soft ceiling; long values get room
                    headers  = value.get("headers") or []
                    meta     = value.get("metadata") or {}
                    rows     = meta.get("display_value") or []
                    n        = len(headers)
                    widths: list[str] = []
                    for col in range(n):
                        lengths = [
                            len(str(row[col] or ""))
                            for row in rows
                            if col < len(row)
                        ]
                        non_empty = [x for x in lengths if x > 0]
                        max_len = max(non_empty) if non_empty else 0
                        data_px = max(MIN_PX, max_len * CH_PX + PAD)
                        # Special case: if the header label is long,
                        # ensure the column is at least wide enough to
                        # show it in 2 lines (ceil(chars/2) per line). The
                        # header only wraps at whitespace, so a single long
                        # word (e.g. "Decoder") must still fit on one line
                        # or it gets clipped — floor the estimate at the
                        # longest word's width.
                        header_text = headers[col] if col < len(headers) else ""
                        header_chars = len(header_text)
                        longest_word = max((len(w) for w in header_text.split()), default=0)
                        header_px = max(math.ceil(header_chars / 2), longest_word) * CH_PX + PAD
                        px = min(MAX_PX, max(data_px, header_px))
                        widths.append(f"{px}px")
                    return widths

                _initial_widths = _column_widths_for(_initial_table)
                table = gr.Dataframe(
                    value=_initial_table,
                    interactive=False,
                    wrap=False,
                    datatype=_initial_datatypes,
                    show_search="none",
                    column_widths=_initial_widths,
                    elem_classes="results-table",
                )
                count_md = gr.Markdown(value=_initial_count, elem_classes=["results-count"])

                results_inputs = [
                    *filter_comps.values(),
                    *(col_subgroups[grp] for grp in groups_in_order),
                ]
                results_outputs = [table, count_md]

                def _render_results_with_widths(*args):
                    fvals = args[:n_filters]
                    cvals = args[n_filters:]
                    selections = {k: (v or []) for k, v in zip(filter_keys, fvals)}
                    visible_keys = [k for group in cvals for k in (group or [])]
                    value, count, datatypes = _render_results_table(
                        bench, selections, visible_keys,
                    )
                    widths = _column_widths_for(value)
                    return (
                        gr.update(
                            value=value,
                            datatype=datatypes,
                            column_widths=widths,
                        ),
                        count,
                    )

                if not os.environ.get("BISECT_NO_RESULTS_HANDLERS"):
                    for inp in results_inputs:
                        inp.change(
                            _render_results_with_widths,
                            results_inputs, results_outputs,
                        )

            # -------------------- Leaderboards --------------------
            # Initial header / radar / cards / table are pre-computed
            # server-side and baked into the output components' ``value=``
            # kwargs. We used to populate them via
            # ``tab_lb.select(_lb_cb, lb_inputs, lb_outputs)``, but in
            # Gradio 6.12 / 6.13 a tab's ``.select`` handler whose input
            # list references components living inside the *same* tab
            # triggers a Svelte recursion ("Maximum call stack size
            # exceeded" on 6.12, ``effect_update_depth_exceeded`` on
            # 6.13) the moment the user clicks the tab — the page
            # visibly freezes. Rendering eagerly sidesteps the buggy
            # code path; the ``.change`` handlers below still keep the
            # panel in sync when the user tweaks Ranking / Access
            # pattern / Top N.
            _lb_initial_plot, _lb_initial_cards, _lb_initial_table = (
                _render_leaderboard(bench, "1_frame", "Overall", 6)
            )
            tab_lb = maybe_tab("Leaderboards")
            with tab_lb:
                _axis_names = ", ".join(
                    (_cols_by_key(bench).get(k, {}).get("short") or k)
                    for k in bench.leaderboard_axes
                )
                _tab_intro(
                    "Leaderboards",
                    f"The top configurations on {len(bench.leaderboard_axes)} "
                    f"normalized axes — {_axis_names}. For each configuration, "
                    "every axis is averaged across all datasets the config has "
                    "results for at the selected access pattern, then rescaled "
                    "so 1.0 = best-in-class. Bigger polygon means a better "
                    "all-rounder.",
                )

                with gr.Row(elem_classes="lb-controls", equal_height=True):
                    with gr.Column(scale=1, min_width=0):
                        lb_ts = gr.Dropdown(
                            choices=schema.TS_MODES, value="1_frame",
                            label="Access pattern",
                            elem_classes="lb-topn",
                        )
                    with gr.Column(scale=1, min_width=0, elem_classes="lb-topn-col"):
                        lb_top = gr.Slider(1, 12, value=6, step=1,
                                           label="Top N configs",
                                           elem_classes="lb-topn")
                    with gr.Column(scale=2, min_width=0, elem_classes="lb-cats-col"):
                        # Single-select pill group; styling in
                        # .ranking-pills CSS turns Gradio's default radio
                        # list into a row of toggleable pill buttons
                        # aligned to the right so the ranking selection
                        # stays visible at a glance.
                        lb_cat = gr.Radio(
                            choices=list(bench.leaderboard_cats.keys()),
                            value="Overall", show_label=False,
                            elem_classes="ranking-pills",
                        )

                with gr.Row(elem_classes="lb-main", equal_height=False):
                    with gr.Column(scale=1, min_width=0):
                        lb_plot = gr.Plot(value=_lb_initial_plot, elem_classes="lb-radar")
                    with gr.Column(scale=1, min_width=0):
                        lb_cards = gr.HTML(value=_lb_initial_cards, elem_classes="lb-cards-wrap")

                # Tabular mirror of the podium — same rows, same order,
                # but surfaced as a dataframe for users who want to
                # sort / search / copy values instead of scanning chips.
                gr.Markdown("### Selected configurations",
                            elem_classes="lb-table-head")
                lb_table = gr.Dataframe(
                    value=_lb_initial_table,
                    interactive=False,
                    wrap=False,
                    show_search="none",
                    datatype="html",
                    elem_classes="lb-table",
                )

                def _lb_cb(cat, ts, n):
                    return _render_leaderboard(bench, ts, cat, int(n))

                lb_inputs = [lb_cat, lb_ts, lb_top]
                lb_outputs = [lb_plot, lb_cards, lb_table]
                lb_cat.change(_lb_cb, lb_inputs, lb_outputs)
                lb_ts.change(_lb_cb, lb_inputs, lb_outputs)
                lb_top.change(_lb_cb, lb_inputs, lb_outputs)

            # -------------------- Compare --------------------
            # Initial figures are pre-computed server-side and baked into
            # each ``gr.Plot``'s ``value=`` kwarg — same reasoning as
            # the Leaderboards tab above. A ``tab_cmp.select`` handler
            # whose ``inputs`` list references components living inside
            # the same tab (``cmp_metric`` / ``cmp_group``) matches the
            # open upstream bug gradio-app/gradio#13198 and can freeze
            # the page on tab click with either
            # ``effect_update_depth_exceeded`` (Gradio 6.11 / 6.13) or
            # ``Maximum call stack size exceeded`` (Gradio 6.12). Empirically
            # the two ``Dropdown`` inputs here don't currently blow up,
            # but the shape is identical to the Leaderboards one that
            # did — eager initial render avoids the latent freeze.
            _cmp_default_metric = bench.scatter_y
            _cmp_default_group = "vcodec"
            _cmp_initial_bar = _render_compare_bar(bench, _cmp_default_metric, _cmp_default_group)
            _cmp_initial_scatter = _render_compare_scatter(bench)
            _cmp_initial_stacked = _render_compare_stacked(bench)
            tab_cmp = maybe_tab("Compare")
            with tab_cmp:
                _tab_intro(
                    "Compare parameters",
                    "Pick any metric and any dimension to see how the field "
                    "behaves.",
                )
                metric_cols = [c for c in bench.columns if c.get("metric")]
                group_cols = [c for c in bench.columns if c["group"] == "Config"]
                metric_choices = [((c.get("short") or c["label"]).replace("\n", " "), c["key"])
                                  for c in metric_cols]
                group_choices = [((c.get("short") or c["label"]).replace("\n", " "), c["key"])
                                 for c in group_cols]

                with gr.Group(elem_classes=["compare-panel"]):
                    with gr.Column(elem_classes=["compare-controls"]):
                        gr.Markdown(
                            "### Metric averages by configuration axis\n\n"
                            "Pick a metric and the configuration axis to "
                            "group it by. Each bar is the mean of the chosen "
                            "metric across every run sharing that group "
                            "value.",
                            elem_classes=["compare-caption"],
                        )
                        with gr.Row():
                            cmp_metric = gr.Dropdown(choices=metric_choices, value=_cmp_default_metric,
                                                     label="Metric",
                                                     elem_classes="cmp-control")
                            cmp_group = gr.Dropdown(choices=group_choices, value=_cmp_default_group,
                                                    label="Group by",
                                                    elem_classes="cmp-control")
                    cmp_bar = gr.Plot(value=_cmp_initial_bar, elem_classes=["compare-plot"])
                _ylabel = (_cols_by_key(bench).get(bench.scatter_y, {}).get("label")
                           or bench.scatter_y).replace("\n", " ")
                with gr.Group(elem_classes=["compare-panel"]):
                    with gr.Column(elem_classes=["compare-controls"]):
                        gr.Markdown(
                            "### Quality vs. compression\n\n"
                            "Each point is one configuration. The x-axis is "
                            "the video / image size ratio (lower is smaller, "
                            f"so better compression); the y-axis is {_ylabel} "
                            "(higher is better). Colors identify the codec — "
                            "the top-left corner is the sweet spot.",
                            elem_classes=["compare-caption"],
                        )
                    cmp_scatter = gr.Plot(value=_cmp_initial_scatter, elem_classes=["compare-plot"])
                with gr.Group(elem_classes=["compare-panel"]):
                    with gr.Column(elem_classes=["compare-controls"]):
                        gr.Markdown(
                            "### Decoding latency by codec and access pattern\n\n"
                            "Each bar is one codec; segments stack the mean "
                            "decoding time (ms, lower is better) for each access "
                            "pattern. Tall bars flag codecs that are slow "
                            "overall; tall single segments flag codecs whose "
                            "cost scales poorly with random or multi-frame "
                            "access.",
                            elem_classes=["compare-caption"],
                        )
                    cmp_stacked = gr.Plot(value=_cmp_initial_stacked, elem_classes=["compare-plot"])

                def _cmp_bar_cb(metric, group_by):
                    return _render_compare_bar(bench, metric, group_by)

                cmp_bar_inputs = [cmp_metric, cmp_group]
                cmp_metric.change(_cmp_bar_cb, cmp_bar_inputs, cmp_bar)
                cmp_group.change(_cmp_bar_cb, cmp_bar_inputs, cmp_bar)

            # -------------------- Submit --------------------
            with maybe_tab("Submit"):
                _tab_intro(
                    "Submit a benchmark",
                    "Queue a configuration sweep; results are pushed to the "
                    "Hub and appear here automatically.",
                )
                # Every parameter is a preset CheckboxGroup paired with a
                # free-text "Custom" box. On submit we union the two (see
                # _merge_values) so users can either stay within the
                # curated vocabulary or punch through it with arbitrary
                # codec / pix_fmt / dataset / GOP / CRF / etc. values.
                #
                # Layout mirrors the reference HF Space: each parameter
                # group is its own bordered card on the left (`.submit-section`)
                # so users can scan the form as a list of distinct knobs,
                # while the right column stays card-less — total / estimate
                # / Submit CTA / recent-submissions queue stack directly on
                # the page background as a quiet action rail.
                with gr.Row(equal_height=False):
                    with gr.Column(scale=2):
                        with gr.Group(elem_classes=["submit-section"]):
                            s_repos = gr.CheckboxGroup(
                                bench.submit_options["repos"],
                                value=sd["repos"], label="Datasets",
                                elem_classes="submit-checks",
                            )
                            s_repos_custom = gr.Textbox(
                                value="", label="Custom datasets",
                                placeholder="org/name — one per line or comma-separated",
                                lines=2,
                            )

                        with gr.Row(equal_height=False):
                            with gr.Column(min_width=0):
                                with gr.Group(elem_classes=["submit-section"]):
                                    s_vcodecs = gr.CheckboxGroup(
                                        [(_VCODEC_SUBMIT_LABELS.get(v, v), v)
                                         for v in bench.submit_options["vcodecs"]],
                                        value=sd["vcodecs"], label="Codecs",
                                        elem_classes="submit-checks",
                                    )
                                    s_vcodecs_custom = gr.Textbox(
                                        value="", label="Custom codecs",
                                        placeholder="e.g. libvpx-vp9, libaom-av1",
                                    )
                            with gr.Column(min_width=0):
                                with gr.Group(elem_classes=["submit-section"]):
                                    s_pix = gr.CheckboxGroup(
                                        bench.submit_options["pix_fmts"],
                                        value=sd["pix_fmts"], label="Pixel formats",
                                        elem_classes="submit-checks",
                                    )
                                    s_pix_custom = gr.Textbox(
                                        value="", label="Custom pixel formats",
                                        placeholder="e.g. yuv422p, yuv420p10le",
                                    )
                        with gr.Row(equal_height=False):
                            with gr.Column(min_width=0):
                                with gr.Group(elem_classes=["submit-section"]):
                                    s_g = gr.CheckboxGroup(
                                        bench.submit_options["g"],
                                        value=sd["g"], label="GOP values",
                                        elem_classes="submit-checks",
                                    )
                                    s_g_custom = gr.Textbox(
                                        value="", label="Custom GOP",
                                        placeholder="e.g. 8, 50, 250",
                                    )
                            with gr.Column(min_width=0):
                                with gr.Group(elem_classes=["submit-section"]):
                                    s_crf = gr.CheckboxGroup(
                                        bench.submit_options["crf"],
                                        value=sd["crf"], label="CRF values",
                                        elem_classes="submit-checks",
                                    )
                                    s_crf_custom = gr.Textbox(
                                        value="", label="Custom CRF",
                                        placeholder="e.g. 12, 38",
                                    )

                        with gr.Row(equal_height=False):
                            with gr.Column(min_width=0):
                                with gr.Group(elem_classes=["submit-section"]):
                                    s_ts = gr.CheckboxGroup(
                                        bench.submit_options["timestamps_modes"],
                                        value=sd["timestamps_modes"],
                                        label="Access patterns",
                                        elem_classes="submit-checks",
                                    )
                            with gr.Column(min_width=0):
                                with gr.Group(elem_classes=["submit-section"]):
                                    s_back = gr.CheckboxGroup(
                                        bench.submit_options["backends"],
                                        value=sd["backends"], label="Backends",
                                        elem_classes="submit-checks",
                                    )

                        # Depth-only knobs, kept in two distinct cards:
                        # `lossless` is a codec encoding knob, while
                        # `use_log` + depth-range mapping define how depth is
                        # quantized into the encoder's pixel range.
                        s_lossless = s_use_log = None
                        s_depth_min = s_depth_max = s_shift = None
                        if is_depth:
                            with gr.Group(elem_classes=["submit-section"]):
                                s_lossless = gr.CheckboxGroup(
                                    bench.submit_options["lossless"],
                                    value=sd["lossless"], label="Lossless",
                                    elem_classes="submit-checks",
                                )
                            with gr.Group(elem_classes=["submit-section"]):
                                s_use_log = gr.CheckboxGroup(
                                    bench.submit_options["use_log"],
                                    value=sd["use_log"], label="Log encode",
                                    elem_classes="submit-checks",
                                )
                                # Textboxes (not gr.Number) so the fields start
                                # empty: gr.Number coerces an unset value to 0,
                                # and its placeholder/callable workarounds crash
                                # Gradio 6.13 for Tab-nested components. Parsed
                                # to float|None in ``_submit``.
                                with gr.Row():
                                    s_depth_min = gr.Textbox(
                                        value="", placeholder="auto",
                                        label="Depth min (m) — optional",
                                        info="leave empty to derive from dataset stats",
                                    )
                                    s_depth_max = gr.Textbox(
                                        value="", placeholder="auto",
                                        label="Depth max (m) — optional",
                                        info="leave empty to derive from dataset stats",
                                    )
                                    s_shift = gr.Textbox(
                                        value="", placeholder="auto",
                                        label="Shift — optional",
                                        info="leave empty to derive from dataset stats",
                                    )

                        with gr.Group(elem_classes=["submit-section"]):
                            with gr.Row():
                                s_samples = gr.Slider(
                                    1, 200, value=sd["samples_per_config"],
                                    step=1, label="Samples per config",
                                )

                    # Initial summary counts are pre-computed server-side
                    # and baked into the two markdowns as ``value=``. This
                    # used to flow through ``demo.load(_submission_summary,
                    # summary_inputs, ...)``, but Gradio 6.13 enters a
                    # Svelte ``effect_update_depth_exceeded`` infinite loop
                    # on every tab click whenever a ``demo.load`` reads a
                    # component that lives inside a ``gr.Tab`` — so the
                    # whole page freezes the moment the user switches tab.
                    # ``.change`` handlers below still keep the summary in
                    # sync once the user edits anything.
                    _initial_total = _submission_summary(
                        bench,
                        sd["repos"], "",
                        sd["vcodecs"], "",
                        sd["pix_fmts"], "",
                        sd["g"], "",
                        sd["crf"], "",
                        sd["timestamps_modes"],
                        sd["backends"],
                        sd["samples_per_config"],
                        *((sd["lossless"], sd["use_log"]) if is_depth else ()),
                    )
                    with gr.Column(scale=1, elem_classes=["submit-right"]):
                        total_md = gr.Markdown(
                            value=_initial_total,
                            elem_classes=["submit-summary"],
                        )
                        submit_btn = gr.Button(
                            "Submit sweep", variant="primary",
                            elem_classes=["submit-cta"],
                        )
                        submit_status = gr.Markdown()
                        gr.HTML(
                            '<div class="submit-queue-label">Recent submissions</div>'
                        )
                        queue_tbl = gr.Dataframe(
                            value=_queue_dataframe(bench),
                            interactive=False,
                            wrap=True,
                            elem_classes=["submit-queue"],
                        )
                        refresh_btn = gr.Button("Refresh queue", variant="secondary",
                                                size="sm")

                base_inputs = [
                    s_repos, s_repos_custom,
                    s_vcodecs, s_vcodecs_custom,
                    s_pix, s_pix_custom,
                    s_g, s_g_custom,
                    s_crf, s_crf_custom,
                    s_ts,
                    s_back,
                    s_samples,
                ]
                # Order matches the trailing optional params of
                # ``_submission_summary`` / ``_submit`` so the flat Gradio
                # args map positionally.
                summary_inputs = base_inputs + ([s_lossless, s_use_log] if is_depth else [])
                submit_inputs = base_inputs + (
                    [s_lossless, s_use_log, s_depth_min, s_depth_max, s_shift]
                    if is_depth else []
                )

                def _summary_cb(*args):
                    return _submission_summary(bench, *args)

                def _submit_cb(*args):
                    return _submit(bench, *args)

                for inp in summary_inputs:
                    inp.change(_summary_cb, summary_inputs, total_md)

                submit_btn.click(
                    _submit_cb, submit_inputs,
                    [submit_status, queue_tbl],
                )
                refresh_btn.click(lambda: _queue_dataframe(bench), None, queue_tbl)

                # Maintainer-only "full sweep" trigger. Hidden behind a
                # collapsed accordion so it doesn't compete with the
                # primary submit flow; gated by an "I understand" checkbox
                # because the resulting submission deliberately exceeds
                # ``MAX_CONFIGS_PER_SUBMISSION`` and saturates worker
                # capacity for hours. The parameter set itself lives in
                # ``bench.full_sweep`` so this stays a one-knob trigger.
                _fs = bench.full_sweep
                _fs_total = (
                    len(_fs["repos"]) * len(_fs["vcodecs"]) * len(_fs["pix_fmts"])
                    * len(_fs["g"]) * len(_fs["crf"])
                    * len(_fs["timestamps_modes"]) * len(_fs["backends"])
                )
                with gr.Accordion(
                    "Maintainer tools", open=False,
                    elem_classes=["submit-maintainer"],
                ):
                    gr.Markdown(
                        "### Full sweep\n\n"
                        "Queues the canonical Cartesian product across every "
                        "curated knob — used to re-baseline the results table "
                        "after a worker, codec, or schema change. This "
                        "**bypasses** the per-submission config cap and will "
                        "saturate the worker pool for hours.\n\n"
                        f"- Datasets: `{', '.join(_fs['repos'])}`\n"
                        f"- Codecs: `{', '.join(_fs['vcodecs'])}`\n"
                        f"- Pixel formats: `{', '.join(_fs['pix_fmts'])}`\n"
                        f"- GOP: `{', '.join(_fs['g'])}`\n"
                        f"- CRF: `{', '.join(_fs['crf'])}`\n"
                        f"- Access patterns: `{', '.join(_fs['timestamps_modes'])}`\n"
                        f"- Backends: `{', '.join(_fs['backends'])}`\n"
                        f"- Samples per config: **{_fs['samples_per_config']}**\n\n"
                        f"Total: **{_fs_total:,} configs**."
                    )
                    full_sweep_confirm = gr.Checkbox(
                        value=False,
                        label="I understand this triggers a multi-hour worker run.",
                    )
                    full_sweep_btn = gr.Button(
                        "Trigger full sweep", variant="stop",
                    )
                    def _full_sweep_cb(confirmed):
                        return _submit_full_sweep(bench, confirmed)

                    full_sweep_btn.click(
                        _full_sweep_cb, [full_sweep_confirm],
                        [submit_status, queue_tbl],
                    )

            # -------------------- About --------------------
            # The old standalone "Parameters" tab is folded into the bottom
            # of About so users get methodology and the full parameter
            # reference in one scroll without a context switch. The
            # ``.params-page`` grid below stretches to the tab width
            # (see ``.param-cards`` in styles.css) so the two tabs' worth
            # of content fits comfortably side-by-side.
            with maybe_tab("About"):
                _tab_intro(
                    "About this benchmark",
                    "Methodology, metrics, and how to reproduce — followed by "
                    "a full parameter reference.",
                )
                gr.HTML(f'<div class="prose">{bench.about_html}</div>')
                gr.HTML(_params_html(bench))

        # -------------------- Footer: Refresh from Hub --------------------
        # Manual escape hatch for the 1-hour ``_RESULTS_TTL_SECONDS`` cache
        # in ``_cached_rows``. By default the Space only re-pulls
        # ``RESULTS_DATASET`` once an hour, so a row pushed to the Hub
        # right now stays invisible until the TTL elapses *and* a handler
        # happens to ask for rows. The button bypasses both conditions:
        # ``_force_refresh_rows`` re-downloads under ``_results_fetch_lock``
        # and ``_refresh_from_hub`` then re-renders every panel that reads
        # cached rows (Results table + count, Leaderboards radar / cards
        # / table, Compare bar / scatter / stacked) plus the hero
        # measurement / dataset counters.
        with gr.Row(elem_classes="refresh-footer"):
            version_md = gr.Markdown(
                value=_footer_note_md(bench), elem_classes="footer-version",
            )
            refresh_status = gr.Markdown(
                value="", elem_classes="refresh-status",
            )
            refresh_hub_btn = gr.Button(
                "Refresh from Hub", variant="secondary", size="sm",
            )

        _n_cols = len(groups_in_order)

        def _refresh_from_hub(*args):
            fvals = args[:n_filters]
            cvals = args[n_filters:n_filters + _n_cols]
            lb_cat_v, lb_ts_v, lb_top_v, cmp_metric_v, cmp_group_v = args[n_filters + _n_cols:]

            ok, msg, _row_count = _force_refresh_rows(bench)
            if ok:
                gr.Info(msg)
            else:
                gr.Warning(msg)

            selections = {k: (v or []) for k, v in zip(filter_keys, fvals)}
            visible_keys = [k for group in cvals for k in (group or [])]
            value, count, datatypes = _render_results_table(
                bench, selections, visible_keys,
            )
            widths = _column_widths_for(value)
            table_update = gr.update(
                value=value, datatype=datatypes, column_widths=widths,
            )

            lb_plot_v, lb_cards_v, lb_table_v = _render_leaderboard(
                bench, lb_ts_v, lb_cat_v, int(lb_top_v),
            )
            cmp_bar_v = _render_compare_bar(bench, cmp_metric_v, cmp_group_v)
            cmp_scatter_v = _render_compare_scatter(bench)
            cmp_stacked_v = _render_compare_stacked(bench)

            rows = _safe_cached_rows(bench)
            repos = {r.get("repo_id") for r in rows if r.get("repo_id")}
            hero_md_v = _hero_stats_md(bench, len(rows), len(repos))

            stamp = datetime.now(timezone.utc).strftime("%H:%M:%S UTC")
            status_md = (
                f"Last refresh **{stamp}** — {msg}"
                if ok else f"Last refresh **{stamp}** failed — {msg}"
            )

            return (
                table_update, count,
                lb_plot_v, lb_cards_v, lb_table_v,
                cmp_bar_v, cmp_scatter_v, cmp_stacked_v,
                hero_md_v,
                status_md,
                _footer_note_md(bench),
            )

        refresh_hub_btn.click(
            _refresh_from_hub,
            [
                *filter_comps.values(),
                *(col_subgroups[grp] for grp in groups_in_order),
                lb_cat, lb_ts, lb_top,
                cmp_metric, cmp_group,
            ],
            [
                table, count_md,
                lb_plot, lb_cards, lb_table,
                cmp_bar, cmp_scatter, cmp_stacked,
                hero_stats,
                refresh_status,
                version_md,
            ],
        )


def build_app() -> gr.Blocks:
    """Wire the full ``gr.Blocks`` layout: top-level RGB | Depth tabs."""
    # DEBUG: skip tabs by setting BISECT_SKIP env var to a comma-separated
    # list of tab labels (e.g. BISECT_SKIP=Submit,Compare,Leaderboards).
    # The skipped tab's components are still created (so wiring code that
    # references them keeps working) but inside an off-screen ``gr.Group``,
    # which means clicking the tab in the UI is impossible.
    _bisect_skip = {s.strip() for s in os.environ.get("BISECT_SKIP", "").split(",") if s.strip()}

    class _SkippedTab:
        """Drop-in for ``gr.Tab`` that hides its children from the UI."""
        def __init__(self):
            self._group = gr.Group(visible=False)
        def __enter__(self):
            self._group.__enter__()
            return self
        def __exit__(self, *a):
            return self._group.__exit__(*a)
        def select(self, *a, **k):
            return None

    def _maybe_tab(label):
        if label in _bisect_skip:
            print(f"[bisect] hiding tab: {label}")
            return _SkippedTab()
        return gr.Tab(label)

    with gr.Blocks(title="Video Benchmark", analytics_enabled=False) as demo:
        with gr.Tabs():
            for bench in _BENCHMARKS:
                with gr.Tab(bench.label):
                    _build_benchmark_tabs(bench, maybe_tab=_maybe_tab)
    return demo


demo = build_app()


if __name__ == "__main__":
    demo.launch(
        server_name="0.0.0.0",
        server_port=7860,
        theme=_hf_theme(),
        css=_load_css(),
        head=_RESULTS_HEADER_TOOLTIP_HEAD,
        ssr_mode=False,
    )