#!/usr/bin/env python3
# Copyright (c) 2026 Local AI Master (https://localaimaster.com/benchmarks)
# SPDX-License-Identifier: MIT
#
# Permission is hereby granted, free of charge, to any person obtaining a copy of this
# software and associated documentation files (the "Software"), to deal in the Software
# without restriction, including without limitation the rights to use, copy, modify, merge,
# publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons
# to whom the Software is furnished to do so, subject to the following conditions:
# The above copyright notice and this permission notice shall be included in all copies or
# substantial portions of the Software.
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED,
# INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR
# PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE
# FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR
# OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
# DEALINGS IN THE SOFTWARE.
#
# The script is MIT-licensed; the benchmark rows it writes, and the published dataset at
# localaimaster.com/benchmarks, are CC BY 4.0 (credit "Local AI Master").
"""Real, reproducible local-inference benchmarks (harness v2).

WHY THIS EXISTS
Most tokens-per-second figures for local LLMs are a screenshot or arithmetic.
This script produces numbers that are actually measured, on a machine you
actually have, with the environment recorded so they can be published and
re-checked later. It is the harness behind every row on
localaimaster.com/benchmarks. Python 3 standard library only; Ollama must be
running locally.

WHY THE HTTP API AND NOT `ollama run --verbose`
The CLI streams tokens with ANSI escape sequences embedded mid-line, so shell
parsing of its summary is unreliable — an early version of this harness produced
"1585 tok/s on an M3 Pro", which is impossible, because the escape codes corrupted
the parse. /api/generate returns clean JSON containing raw `eval_count` and
`eval_duration` (nanoseconds), so the rate is computed here from counts rather
than scraped from formatted text. v2 streams the request so it can also time the
first token; the counts still come from Ollama's final `done` message.

SANITY GATES (a benchmark that can emit nonsense is worse than none)
  * the first run of every (model, context) pair is a WARM-UP and is discarded
  * a run with fewer than MIN_TOKENS generated tokens is DISCARDED, not averaged
  * the MEDIAN of the valid runs is reported, not the mean, so one outlier cannot
    move the number; the slowest and fastest valid runs are reported beside it
  * a row needs at least MIN_VALID_RUNS valid runs
  * CEILING GATE: decode reads the whole weight file once per token, so a dense
    model cannot beat  memory bandwidth (GB/s) / file size (GB).  A median above
    that ceiling x 1.05 is a measurement error and the row is REJECTED
  * a model that is not fully in GPU memory (Ollama /api/ps size_vram < size)
    is rejected: that is a CPU-offload number, not a number for the card
  * mixture-of-experts models read only their active experts, so the file-size
    ceiling does not bound them; with no ceiling to validate against they are
    recorded in the raw file and REJECTED as rows
  * the device you name must match the device the harness detects, and the
    bandwidth must come from the built-in vendor table (or be passed explicitly
    for a device that is not in it) — a number measured on one GPU can never be
    filed under another
  * battery power, Low Power Mode, a busy machine or other loaded models are
    recorded as condition warnings in the raw file; rows from such a sweep are
    not marked verified

OUTPUT
  One raw JSON per sweep:  <out>/raw/<deviceId>/<YYYY-MM-DD>-<host>.json
  (an existing file is never overwritten: -2, -3 ... is appended). It holds every
  run, every gate decision and the host facts. Rows in the shared format
  (data/benchmarks/rows.json, built by scripts/bench-rows.py) are printed at the
  end and embedded in the raw file.

USAGE
  python3 scripts/benchmark-ollama.py                       # the Oct 2026 M3 Pro tag set
  python3 scripts/benchmark-ollama.py llama3.1:8b qwen3:8b --runs 7 --host owner-mac
  python3 scripts/benchmark-ollama.py --tags-file scripts/cloud-bench/tags.txt \\
      --device-id rtx-4090-24gb --host runpod:rtx-4090 --ctx 2048,8192 --pull
  python3 scripts/benchmark-ollama.py qwen3:8b --submit-json  # paste-ready block, no upload
  python3 scripts/benchmark-ollama.py --list-devices
  python3 scripts/benchmark-ollama.py --registry-info qwen3:8b   # tags.txt line from the registry

WHERE TO RUN
  Locally for Apple Silicon (docs/benchmarks/RUNBOOK.md). Rented NVIDIA cards via
  scripts/cloud-bench/ (README.md there). Never publish a number measured on one
  GPU as though it came from another.

PUBLISHING RULE
  Always publish the hardware, the Ollama version, the quantisation, the run count
  and the DATE next to the figure. A measurement without its environment is barely
  better than an invented one.
"""
from __future__ import annotations

import argparse
import hashlib
import json
import os
import pathlib
import platform
import re
import secrets
import socket
import statistics
import subprocess
import sys
import tempfile
import time
import urllib.error
import urllib.request
from datetime import datetime, timezone

HARNESS_VERSION = "2.0"
OLLAMA = os.environ.get("OLLAMA_URL", "http://127.0.0.1:11434").rstrip("/")
REGISTRY = "https://registry.ollama.ai/v2/library"
MIN_TOKENS = 30          # below this a rate is statistically meaningless
MIN_VALID_RUNS = 3       # a row needs at least this many post-warm-up runs
CEILING_TOLERANCE = 1.05  # median may exceed bandwidth/fileGB by at most 5%
UNSTABLE_SPREAD = 0.25    # (high - low) > 25% of the median -> flagged
FULL_GPU_SHARE = 0.99     # size_vram / size below this = partly on the CPU
DEFAULT_NUM_CTX = 2048
DEFAULT_NUM_PREDICT = 512
DEFAULT_SEED = 42
DEFAULT_PROMPT = (
    "Explain how attention works in a transformer model. "
    "Give a clear, technical answer of roughly 200 words."
)
# The nine tags measured on the M3 Pro in October 2026 (lib/measured-benchmarks.ts),
# so a bare run repeats that table.
DEFAULT_MODELS = [
    "llama3.2:3b", "qwen2.5-coder:3b-base", "phi4-mini:3.8b", "qwen3:4b", "mistral:7b",
    "qwen2.5-coder:7b", "llama3.1:8b", "qwen3:8b", "deepseek-r1:8b",
]
CREDIT = "Local AI Master (localaimaster.com/benchmarks)"
LICENSE = "CC-BY-4.0"

# ---------------------------------------------------------------------------
# Vendor memory bandwidth for the target devices. Every figure is the vendor's
# own published number; the URL is where it was read. These are the same
# sources as the header of lib/vram-picks.ts (fetched 15 Oct 2026) and
# lib/measured-benchmarks.ts. scripts/cloud-bench/devices.txt repeats the NVIDIA
# rows for the shell bootstrap; the harness refuses to run if the two disagree.
#   vramGB = marketed memory size (binary GiB, as the vendor sells it).
# ---------------------------------------------------------------------------
DEVICES: dict[str, dict] = {
    "apple-m3-pro-18gb": {
        "name": "Apple M3 Pro (18 GB)", "vramGB": 18, "bandwidthGBs": 150,
        # https://support.apple.com/en-us/117736  (MacBook Pro 14-inch, M3 Pro/M3 Max, Nov 2023: "150GB/s memory bandwidth"; read 2 Nov 2026)
    },
    "rtx-3060-12gb": {
        "name": "NVIDIA GeForce RTX 3060 (12 GB)", "vramGB": 12, "bandwidthGBs": 360,
        # https://www.nvidia.com/en-us/geforce/news/game-on-you-asked-we-answered-qa/
    },
    "rtx-4060-ti-16gb": {
        "name": "NVIDIA GeForce RTX 4060 Ti (16 GB)", "vramGB": 16, "bandwidthGBs": 288,
        # https://www.nvidia.com/en-us/geforce/news/rtx-40-series-vram-video-memory-explained/
    },
    "rtx-4070-12gb": {
        "name": "NVIDIA GeForce RTX 4070 (12 GB)", "vramGB": 12, "bandwidthGBs": 504,
        # https://www.nvidia.com/en-us/geforce/news/rtx-50-series-graphics-cards-gpu-laptop-announcements/
    },
    "rtx-3090-24gb": {
        "name": "NVIDIA GeForce RTX 3090 (24 GB)", "vramGB": 24, "bandwidthGBs": 936,
        # https://www.nvidia.com/content/PDF/nvidia-ampere-ga-102-gpu-architecture-whitepaper-v2.pdf (Appendix A)
    },
    "rtx-4090-24gb": {
        "name": "NVIDIA GeForce RTX 4090 (24 GB)", "vramGB": 24, "bandwidthGBs": 1008,
        # https://images.nvidia.com/aem-dam/Solutions/geforce/ada/nvidia-ada-gpu-architecture.pdf (spec tables)
    },
    "rtx-5070-ti-16gb": {
        "name": "NVIDIA GeForce RTX 5070 Ti (16 GB)", "vramGB": 16, "bandwidthGBs": 896,
        # https://www.nvidia.com/en-us/geforce/graphics-cards/compare/
    },
    "rtx-5080-16gb": {
        "name": "NVIDIA GeForce RTX 5080 (16 GB)", "vramGB": 16, "bandwidthGBs": 960,
        # https://www.nvidia.com/en-us/geforce/graphics-cards/compare/
    },
    "rtx-5090-32gb": {
        "name": "NVIDIA GeForce RTX 5090 (32 GB)", "vramGB": 32, "bandwidthGBs": 1792,
        # https://www.nvidia.com/en-us/geforce/graphics-cards/compare/
    },
    # Cloud datacenter cards (Google Cloud G2 / N1 GPU VMs). Real hardware, but not consumer
    # cards: rows are published under their own device id and never on a GeForce page.
    "nvidia-l4-24gb": {
        "name": "NVIDIA L4 (24 GB)", "vramGB": 24, "bandwidthGBs": 300,
        # https://www.nvidia.com/en-us/data-center/l4/  (spec table: "GPU memory 24GB", "GPU memory bandwidth 300GB/s"; read 3 Nov 2026)
    },
    "nvidia-t4-16gb": {
        "name": "NVIDIA T4 (16 GB)", "vramGB": 16, "bandwidthGBs": 320,
        # https://www.nvidia.com/en-us/data-center/tesla-t4/  (spec table: "16 GB GDDR6", "320+ GB/s"; read 3 Nov 2026)
    },
}

DEVICE_ID_RE = re.compile(r"^[a-z0-9]+(-[a-z0-9]+)*-\d+gb$")
DIGEST_RE = re.compile(r"^sha256:[0-9a-f]{64}$")
# Host labels in the shared format. A rented host is never verified by one sweep.
HOST_RE = re.compile(r"^(owner-mac|local|(runpod|vast|cloud):[a-z0-9][a-z0-9-]*)$")
RENTED_PREFIXES = ("runpod:", "vast:", "cloud:")


class HarnessError(Exception):
    """A condition that makes the whole sweep untrustworthy — stop, don't guess."""


# ---------------------------------------------------------------------------
# small helpers
# ---------------------------------------------------------------------------
def sh(cmd: list[str], timeout: int = 30) -> str:
    try:
        return subprocess.run(cmd, capture_output=True, text=True, timeout=timeout).stdout.strip()
    except Exception:
        return ""


def utc_now() -> str:
    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")


def api(path: str, body: dict | None = None, timeout: int = 60) -> dict:
    data = json.dumps(body).encode() if body is not None else None
    req = urllib.request.Request(f"{OLLAMA}{path}", data=data,
                                 headers={"Content-Type": "application/json"})
    with urllib.request.urlopen(req, timeout=timeout) as r:
        return json.loads(r.read())


def normalise_tag(tag: str) -> str:
    tag = tag.strip()
    tag = re.sub(r"^ollama\s+(pull|run)\s+", "", tag)
    return tag if ":" in tag else f"{tag}:latest"


def slug(s: str) -> str:
    return re.sub(r"[^a-z0-9]+", "-", s.lower()).strip("-") or "host"


def median(xs: list[float]) -> float:
    return float(statistics.median(xs))


# ---------------------------------------------------------------------------
# device table helpers
# ---------------------------------------------------------------------------
def devices_file() -> pathlib.Path:
    return pathlib.Path(__file__).resolve().parent / "cloud-bench" / "devices.txt"


def check_devices_file() -> None:
    """scripts/cloud-bench/devices.txt must agree with DEVICES (one source of truth)."""
    f = devices_file()
    if not f.exists():
        return
    for n, line in enumerate(f.read_text().splitlines(), 1):
        line = line.split("#", 1)[0].strip()
        if not line:
            continue
        cols = line.split()
        if len(cols) < 3:
            raise HarnessError(f"{f}:{n}: expected '<deviceId> <vramGB> <bandwidthGBs> ...'")
        dev_id, vram, bw = cols[0], cols[1], cols[2]
        known = DEVICES.get(dev_id)
        if not known:
            raise HarnessError(f"{f}:{n}: {dev_id} is not in the harness DEVICES table")
        if float(vram) != known["vramGB"] or float(bw) != known["bandwidthGBs"]:
            raise HarnessError(
                f"{f}:{n}: {dev_id} says {vram} GB / {bw} GB/s but the harness table says "
                f"{known['vramGB']} GB / {known['bandwidthGBs']} GB/s — fix one of them")


def device_id_for(accel: str, mem_gb: int | None) -> str | None:
    """Map a detected accelerator to a table id (None if it is not a target device)."""
    a = accel.lower()
    if "laptop" in a or "mobile" in a or mem_gb is None:
        return None  # laptop parts have different bandwidth from the desktop card
    m = re.search(r"apple (m\d+)(?: (pro|max|ultra))?", a)
    if m:
        return "-".join(x for x in ["apple", m.group(1), m.group(2) or "", f"{mem_gb}gb"] if x)
    # Datacenter cards report less than their nominal memory to nvidia-smi (ECC / reserved
    # memory: an L4 shows ~22.5 GiB), so they map by name to the vendor's nominal capacity.
    if re.search(r"\bl4\b", a) and (mem_gb or 0) >= 20:
        return "nvidia-l4-24gb"
    if re.search(r"\bt4\b", a) and (mem_gb or 0) >= 14:
        return "nvidia-t4-16gb"
    # Variant words stay in the id, so an RTX 4070 SUPER or an RTX 4090 D can never be
    # filed as the plain RTX 4070 / RTX 4090 of the table.
    m = re.search(r"rtx\s*(\d{4})((?:\s*(?:ti|super|d)\b)*)", a)
    if m:
        return "-".join(["rtx", m.group(1), *m.group(2).split(), f"{mem_gb}gb"])
    return None


# ---------------------------------------------------------------------------
# host facts
# ---------------------------------------------------------------------------
def host_facts() -> dict:
    facts: dict = {
        "platform": f"{platform.system()} {platform.release()} {platform.machine()}",
        "python": platform.python_version(),
        "cpuCount": os.cpu_count(),
        "gpuCount": 0,
        "accelerator": "unknown",
        "memoryGB": None,
        "driver": None,
        "powerLimitW": None,
        "os": f"{platform.system()} {platform.release()} ({platform.machine()})",
        "nvidia": [],
        "warnings": [],
    }
    try:
        facts["loadAvgStart"] = [round(x, 2) for x in os.getloadavg()]
    except OSError:
        facts["loadAvgStart"] = None
    gpu_uuid = ""
    nv = sh(["nvidia-smi", "--query-gpu=name,driver_version,power.limit,memory.total,uuid",
             "--format=csv,noheader,nounits"])
    if nv:
        gpus = []
        for line in nv.splitlines():
            p = [x.strip() for x in line.split(",")]
            if len(p) < 5:
                continue
            try:
                power = float(p[2])
            except ValueError:
                power = None
            try:
                mem_mib = float(p[3])
            except ValueError:
                mem_mib = None
            gpus.append({"name": p[0], "driver": p[1], "powerLimitW": power, "memoryMiB": mem_mib})
            gpu_uuid = gpu_uuid or p[4]
        facts["nvidia"] = gpus
        facts["gpuCount"] = len(gpus)
        if gpus:
            g = gpus[0]
            facts["accelerator"] = g["name"]
            facts["memoryGB"] = round(g["memoryMiB"] / 1024) if g["memoryMiB"] else None
            facts["driver"] = g["driver"]
            facts["powerLimitW"] = g["powerLimitW"]
        osr = {}
        try:
            for line in pathlib.Path("/etc/os-release").read_text().splitlines():
                if "=" in line:
                    k, v = line.split("=", 1)
                    osr[k] = v.strip('"')
        except OSError:
            pass
        facts["os"] = f"{osr.get('PRETTY_NAME', platform.system())} (kernel {platform.release()}, {platform.machine()}), CUDA"
    elif platform.system() == "Darwin":
        cpu = sh(["sysctl", "-n", "machdep.cpu.brand_string"]) or "Apple Silicon"
        mem = sh(["sysctl", "-n", "hw.memsize"])
        facts["accelerator"] = cpu
        facts["memoryGB"] = int(mem) // 1073741824 if mem.isdigit() else None
        facts["gpuCount"] = 1
        try:
            disp = json.loads(sh(["system_profiler", "SPDisplaysDataType", "-json"], timeout=60) or "{}")
            items = disp.get("SPDisplaysDataType", [])
            if items:
                facts["gpuCores"] = items[0].get("sppci_cores")
                facts["gpuModel"] = items[0].get("sppci_model")
                facts["metal"] = items[0].get("spdisplays_mtlgpufamilysupport")
        except (ValueError, AttributeError):
            pass
        ver = sh(["sw_vers", "-productVersion"])
        facts["os"] = f"macOS {ver} (Darwin {platform.release()}, {platform.machine()}), Metal"
        batt = sh(["pmset", "-g", "batt"])
        facts["powerSource"] = ("battery" if "Battery Power" in batt else
                                "ac" if "AC Power" in batt else "unknown")
        if facts["powerSource"] == "battery":
            facts["warnings"].append("running on battery — plug in power")
        if re.search(r"lowpowermode\s+1", sh(["pmset", "-g"])):
            facts["lowPowerMode"] = True
            facts["warnings"].append("Low Power Mode is on — turn it off")
        m = re.search(r'"IOPlatformUUID"\s*=\s*"([^"]+)"',
                      sh(["ioreg", "-rd1", "-c", "IOPlatformExpertDevice"]))
        gpu_uuid = m.group(1) if m else ""
    # A stable, non-identifying id for "is this the same physical host?" (two-host
    # agreement on rented cards). It hashes the PHYSICAL id only (NVIDIA GPU UUID,
    # Mac platform UUID): a container's hostname changes with every rental, so
    # re-renting the same card must not count as a second host. The hostname is
    # used only when no hardware id is readable, and is never stored itself.
    facts["hostInstance"] = hashlib.sha256(
        (gpu_uuid or "hostname|" + socket.gethostname()).encode()).hexdigest()[:16]
    la = facts.get("loadAvgStart")
    if la and facts["cpuCount"] and la[0] > 0.25 * facts["cpuCount"]:
        facts["warnings"].append(
            f"machine is busy (1-min load {la[0]} on {facts['cpuCount']} cores) — quit other apps")
    return facts


def resolve_device(args, facts: dict) -> dict:
    detected = device_id_for(facts["accelerator"], facts["memoryGB"])
    dev_id = args.device_id or detected
    if not dev_id:
        raise HarnessError(
            f"could not map '{facts['accelerator']}' ({facts['memoryGB']} GB) to a device id. "
            "Pass --device-id <vendor-model-vramgb> and --bandwidth-gbs <vendor figure>.")
    if not DEVICE_ID_RE.match(dev_id):
        raise HarnessError(f"--device-id '{dev_id}' must look like vendor-model-<N>gb, lowercase")
    if detected and args.device_id and detected != args.device_id:
        raise HarnessError(
            f"--device-id {args.device_id} but this machine is {detected} "
            f"('{facts['accelerator']}', {facts['memoryGB']} GB). Refusing to file one "
            "device's numbers under another.")
    if facts.get("gpuCount", 0) > 1:
        raise HarnessError(
            f"{facts['gpuCount']} GPUs visible to nvidia-smi — rent a single-GPU host "
            "(CUDA_VISIBLE_DEVICES does not hide cards from nvidia-smi, so the recorded "
            "host facts could not be tied to one card).")
    known = DEVICES.get(dev_id)
    if known:
        if not detected:
            raise HarnessError(
                f"{dev_id} is a table device but the harness could not confirm it on this host "
                f"(detected '{facts['accelerator']}'). Is nvidia-smi on PATH?")
        if args.bandwidth_gbs is not None and args.bandwidth_gbs != known["bandwidthGBs"]:
            raise HarnessError(
                f"--bandwidth-gbs {args.bandwidth_gbs} contradicts the vendor figure for {dev_id} "
                f"({known['bandwidthGBs']} GB/s). Fix the table with a source, don't override it.")
        return {"deviceId": dev_id, "deviceName": known["name"], "bandwidthGBs": known["bandwidthGBs"],
                "bandwidthSource": "harness table (vendor spec)"}
    if args.bandwidth_gbs is None:
        raise HarnessError(f"{dev_id} is not in the built-in table — pass --bandwidth-gbs with the "
                           "vendor's published figure (and keep the URL with the result).")
    name = args.device_name or f"{facts['accelerator']} ({facts['memoryGB']} GB)"
    return {"deviceId": dev_id, "deviceName": name, "bandwidthGBs": args.bandwidth_gbs,
            "bandwidthSource": "--bandwidth-gbs (supplied by the person running the sweep)"}


# ---------------------------------------------------------------------------
# model facts
# ---------------------------------------------------------------------------
def installed_models() -> dict[str, dict]:
    return {m["name"]: m for m in api("/api/tags", timeout=30).get("models", [])}


def pull(tag: str) -> None:
    print(f"   pulling {tag} ...", flush=True)
    body = json.dumps({"model": tag, "stream": True}).encode()
    req = urllib.request.Request(f"{OLLAMA}/api/pull", data=body,
                                 headers={"Content-Type": "application/json"})
    last = ""
    with urllib.request.urlopen(req, timeout=7200) as r:
        for line in r:
            if not line.strip():
                continue
            d = json.loads(line)
            if d.get("error"):
                raise HarnessError(f"pull {tag}: {d['error']}")
            st = d.get("status", "")
            if st != last and not st.startswith("pulling ") or st == "success":
                print(f"     {st}", flush=True)
            last = st


def manifest_model_bytes(manifest: dict) -> int | None:
    sizes = [l.get("size", 0) for l in manifest.get("layers", [])
             if str(l.get("mediaType", "")).endswith("image.model")]
    return sum(sizes) if sizes else None


def file_size_gb(tag: str, local_digest: str | None, tags_size: int, offline: bool) -> tuple[float, str, bool | None]:
    """Weights file size in GB (1e9). Prefer the exact local manifest, then the
    registry manifest IF its digest matches what is installed, then the /api/tags
    total (which also counts projector/template layers, so the ceiling is lower =
    the gate is stricter, never looser)."""
    name, _, version = tag.partition(":")
    ns = name if "/" in name else f"library/{name}"
    roots = [os.environ.get("OLLAMA_MODELS", ""), str(pathlib.Path.home() / ".ollama" / "models"),
             "/usr/share/ollama/.ollama/models", "/root/.ollama/models"]
    for root in filter(None, roots):
        p = pathlib.Path(root) / "manifests" / "registry.ollama.ai" / ns / version
        try:
            raw = p.read_bytes()
        except OSError:
            continue
        if local_digest and hashlib.sha256(raw).hexdigest() != local_digest:
            continue
        b = manifest_model_bytes(json.loads(raw))
        if b:
            return round(b / 1e9, 2), "local manifest (model layer)", None
    registry_match: bool | None = None
    if not offline and "/" not in name:
        try:
            req = urllib.request.Request(
                f"{REGISTRY}/{name}/manifests/{version}",
                headers={"Accept": "application/vnd.docker.distribution.manifest.v2+json"})
            with urllib.request.urlopen(req, timeout=20) as r:
                raw = r.read()
            registry_match = bool(local_digest) and hashlib.sha256(raw).hexdigest() == local_digest
            if registry_match:
                b = manifest_model_bytes(json.loads(raw))
                if b:
                    return round(b / 1e9, 2), "registry manifest (model layer, digest matches)", True
        except (urllib.error.URLError, OSError, ValueError):
            registry_match = None
    return round(tags_size / 1e9, 2), "/api/tags total size (all layers)", registry_match


def model_facts(tag: str, entry: dict, offline: bool) -> dict:
    show = api("/api/show", {"model": tag}, timeout=60)
    det = show.get("details", {}) or {}
    info = show.get("model_info", {}) or {}
    experts = 0
    ctx_len = None
    for k, v in info.items():
        if k.endswith(".expert_count") and isinstance(v, (int, float)):
            experts = max(experts, int(v))
        if k.endswith(".context_length") and isinstance(v, (int, float)):
            ctx_len = int(v)
    digest_hex = str(entry.get("digest", "")).replace("sha256:", "")
    digest = f"sha256:{digest_hex}" if re.fullmatch(r"[0-9a-f]{64}", digest_hex) else None
    file_gb, file_src, reg_match = file_size_gb(tag, digest_hex or None, int(entry.get("size", 0)), offline)
    return {
        "digest": digest,
        "params": det.get("parameter_size") or "unknown",
        "quant": det.get("quantization_level") or "unknown",
        "family": det.get("family"),
        "format": det.get("format"),
        "fileGB": file_gb,
        "fileGBSource": file_src,
        "registryDigestMatch": reg_match,
        "totalSizeGB": round(int(entry.get("size", 0)) / 1e9, 2),
        "expertCount": experts,
        "moe": experts > 1,
        "contextLength": ctx_len,
    }


# ---------------------------------------------------------------------------
# one generation (streamed, so time-to-first-token can be measured)
# ---------------------------------------------------------------------------
def generate(tag: str, prompt: str, num_ctx: int, num_predict: int, seed: int) -> dict:
    body = json.dumps({
        "model": tag, "prompt": prompt, "stream": True, "keep_alive": "10m",
        "options": {"num_ctx": num_ctx, "num_predict": num_predict, "seed": seed},
    }).encode()
    req = urllib.request.Request(f"{OLLAMA}/api/generate", data=body,
                                 headers={"Content-Type": "application/json"})
    t0 = time.perf_counter()
    first = None
    final: dict = {}
    with urllib.request.urlopen(req, timeout=1800) as r:
        for line in r:
            if not line.strip():
                continue
            d = json.loads(line)
            if d.get("error"):
                raise RuntimeError(d["error"])
            if first is None and (d.get("response") or d.get("thinking")):
                first = time.perf_counter()
            if d.get("done"):
                final = d
                break
    wall = time.perf_counter() - t0
    ec, ed = int(final.get("eval_count", 0)), int(final.get("eval_duration", 0))
    pc, pd = int(final.get("prompt_eval_count", 0)), int(final.get("prompt_eval_duration", 0))
    return {
        "evalCount": ec, "evalDurationNs": ed,
        "promptEvalCount": pc, "promptEvalDurationNs": pd,
        "loadMs": round(int(final.get("load_duration", 0)) / 1e6, 1),
        "totalMs": round(int(final.get("total_duration", 0)) / 1e6, 1),
        "wallMs": round(wall * 1000, 1),
        "ttftMs": round((first - t0) * 1000, 1) if first is not None else None,
        "doneReason": final.get("done_reason"),
        "tokS": round(ec / (ed / 1e9), 2) if ed else None,
        "promptTokS": round(pc / (pd / 1e9), 2) if pd else None,
    }


def loaded_models() -> list[dict]:
    try:
        return api("/api/ps", timeout=15).get("models", [])
    except (urllib.error.URLError, OSError, ValueError):
        return []


def unload(tag: str) -> None:
    try:
        api("/api/generate", {"model": tag, "keep_alive": 0}, timeout=120)
    except (urllib.error.URLError, OSError, ValueError):
        pass


# ---------------------------------------------------------------------------
# gates -> row (also imported by scripts/bench-rows.py, so the rules live here)
# ---------------------------------------------------------------------------
def summarise(result: dict, sweep: dict) -> dict:
    """Apply every gate to one (tag, ctx) result and build its shared-format row.
    Returns {"status": "valid"|"rejected", "reasons": [...], "row": {...} | None,
    "unstable": bool}. Pure function of the recorded runs."""
    reasons: list[str] = []
    runs = result.get("runs", [])
    valid = [r for r in runs if r.get("status") == "ok"]
    rates = [float(r["tokS"]) for r in valid]
    bw = float(sweep["bandwidthGBs"])
    file_gb = float(result["fileGB"]) if result.get("fileGB") else 0.0
    ceiling = round(bw / file_gb, 1) if file_gb > 0 else None
    if result.get("moe"):
        reasons.append("mixture-of-experts: the file-size ceiling does not bound it, so the row "
                       "cannot be validated")
    if result.get("skipReason"):
        reasons.append(result["skipReason"])
    if not ceiling:
        reasons.append("no file size, so no ceiling")
    if len(valid) < MIN_VALID_RUNS:
        reasons.append(f"only {len(valid)} valid runs after the warm-up (need {MIN_VALID_RUNS})")
    off = result.get("offload")
    if off and off.get("sizeGB") and off.get("vramGB") is not None \
            and off["vramGB"] < off["sizeGB"] * FULL_GPU_SHARE:
        reasons.append(f"not fully in GPU memory ({off['vramGB']} of {off['sizeGB']} GB) — CPU offload")
    tok = round(median(rates), 1) if rates else None
    if tok is not None and ceiling and tok > ceiling * CEILING_TOLERANCE:
        reasons.append(f"median {tok} tok/s is above the bandwidth ceiling {ceiling} x {CEILING_TOLERANCE} "
                       "— measurement error")
    unstable = bool(rates) and len(rates) > 1 and (max(rates) - min(rates)) > (tok or 0) * UNSTABLE_SPREAD
    if reasons or tok is None:
        return {"status": "rejected", "reasons": reasons or ["no valid runs"], "row": None,
                "unstable": unstable, "medianTokS": tok, "ceilingTokS": ceiling}
    ttfts = [float(r["ttftMs"]) for r in valid if r.get("ttftMs") is not None]
    measured = str(result.get("finishedUtc") or sweep["startedUtc"])[:10]
    num_ctx = int(result["numCtx"])
    row = {
        "id": f"{sweep['deviceId']}__{result['tag']}__{measured}__ctx{num_ctx}",
        "deviceId": sweep["deviceId"],
        "deviceName": sweep["deviceName"],
        "bandwidthGBs": sweep["bandwidthGBs"],
        "host": sweep["host"],
        "tag": result["tag"],
        "digest": result.get("digest"),
        "params": result.get("params"),
        "quant": result.get("quant"),
        "fileGB": result["fileGB"],
        "ollama": sweep["ollama"],
        "driver": sweep.get("driver"),
        "os": sweep["os"],
        "powerLimitW": sweep.get("powerLimitW"),
        "numCtx": num_ctx,
        "runs": len(runs),
        "validRuns": len(valid),
        "tokS": tok,
        "low": round(min(rates), 1),
        "high": round(max(rates), 1),
        "ttftMs": round(median(ttfts)) if ttfts else None,
        "vramGB": (off or {}).get("vramGB"),
        "ceilingTokS": ceiling,
        "measured": measured,
        "rawFile": sweep.get("rawFile"),
        "source": sweep.get("source", "localaimaster"),
        "verified": bool(sweep.get("source", "localaimaster") == "localaimaster"
                         and not sweep.get("warnings") and not unstable
                         and not str(sweep["host"]).startswith(RENTED_PREFIXES)),
    }
    # Rented hosts are only verified once a second host agrees (scripts/bench-rows.py).
    return {"status": "valid", "reasons": [], "row": row, "unstable": unstable,
            "medianTokS": tok, "ceilingTokS": ceiling}


# ---------------------------------------------------------------------------
# sweep
# ---------------------------------------------------------------------------
def bench_one(tag: str, facts: dict, num_ctx: int, args) -> dict:
    res: dict = {"tag": tag, "numCtx": num_ctx, **facts, "runs": [], "offload": None,
                 "startedUtc": utc_now()}
    if facts.get("contextLength") and num_ctx > facts["contextLength"]:
        res["skipReason"] = f"num_ctx {num_ctx} exceeds the model's context length {facts['contextLength']}"
        res["finishedUtc"] = utc_now()
        return res
    for i in range(args.runs):
        warm = i == 0
        label = "warm-up" if warm else f"run {i}/{args.runs - 1}"
        try:
            r = generate(tag, DEFAULT_PROMPT, num_ctx, args.num_predict, args.seed)
        except Exception as e:  # noqa: BLE001 — any failure is recorded, never averaged
            res["runs"].append({"i": i, "status": "failed", "error": str(e)[:200]})
            print(f"   {label:<10} FAILED: {str(e)[:80]}", flush=True)
            continue
        if warm:
            r["status"] = "warmup"
            ps = [m for m in loaded_models() if m.get("name") == tag or m.get("model") == tag]
            if ps:
                res["offload"] = {"sizeGB": round(int(ps[0].get("size", 0)) / 1e9, 2),
                                  "vramGB": round(int(ps[0].get("size_vram", 0)) / 1e9, 2)}
            note = ""
            if res["offload"] and res["offload"]["vramGB"] < res["offload"]["sizeGB"] * FULL_GPU_SHARE:
                note = f"  ! only {res['offload']['vramGB']} of {res['offload']['sizeGB']} GB on the GPU"
            print(f"   {label:<10} {r['tokS']} tok/s ({r['evalCount']} tokens, load {r['loadMs']} ms) "
                  f"— discarded{note}", flush=True)
            res["runs"].append({"i": i, **r})
            if note:
                print("   stopping this model: a CPU-offload number is not a number for the card", flush=True)
                break
            continue
        if r["evalCount"] < MIN_TOKENS or not r["tokS"]:
            r["status"] = "short"
            print(f"   {label:<10} discarded — only {r['evalCount']} tokens generated", flush=True)
        else:
            r["status"] = "ok"
            print(f"   {label:<10} {r['tokS']} tok/s ({r['evalCount']} tokens, TTFT {r['ttftMs']} ms)", flush=True)
        res["runs"].append({"i": i, **r})
    res["finishedUtc"] = utc_now()
    return res


def reserve_raw_path(out: pathlib.Path, dev_id: str, host: str, date: str) -> pathlib.Path:
    d = out / "raw" / dev_id
    d.mkdir(parents=True, exist_ok=True)
    base = f"{date}-{slug(host)}"
    n = 1
    while True:
        p = d / (f"{base}.json" if n == 1 else f"{base}-{n}.json")
        try:
            fd = os.open(p, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o644)
            os.close(fd)
            return p
        except FileExistsError:
            n += 1


def write_atomic(p: pathlib.Path, obj: dict) -> None:
    fd, tmp = tempfile.mkstemp(dir=p.parent, prefix=".tmp-", suffix=".json")
    with os.fdopen(fd, "w") as f:
        json.dump(obj, f, indent=1)
        f.write("\n")
    os.replace(tmp, p)


def read_tags_file(path: str) -> list[str]:
    tags = []
    for line in pathlib.Path(path).read_text().splitlines():
        line = line.split("#", 1)[0].strip()
        if line:
            tags.append(line.split()[0])
    return tags


def parse_args(argv: list[str]):
    ap = argparse.ArgumentParser(
        description="Measure Ollama generation speed with honesty gates; writes one raw JSON per sweep.")
    ap.add_argument("models", nargs="*", help="Ollama tags (default: the Oct 2026 M3 Pro set)")
    ap.add_argument("--runs", type=int, default=7,
                    help="runs per model and context, INCLUDING the discarded warm-up (default 7)")
    ap.add_argument("--ctx", default=str(DEFAULT_NUM_CTX),
                    help="comma list of num_ctx values to sweep, e.g. 2048,8192,32768 (default 2048)")
    ap.add_argument("--tags-file", help="file with one tag per line (first column; # comments)")
    ap.add_argument("--device-id", help="vendor-model-vramgb id, e.g. rtx-4090-24gb (default: detected)")
    ap.add_argument("--device-name", help="display name for a device that is not in the table")
    ap.add_argument("--bandwidth-gbs", type=float,
                    help="vendor memory bandwidth, only for devices not in the built-in table")
    ap.add_argument("--host", default="local",
                    help="host label: owner-mac, runpod:<gpu-type>, vast:<gpu-type> (default: local)")
    ap.add_argument("--out", default=str(pathlib.Path(__file__).resolve().parent.parent / "data" / "benchmarks"),
                    help="output root; raw files go to <out>/raw/<deviceId>/ (default data/benchmarks)")
    ap.add_argument("--num-predict", type=int, default=DEFAULT_NUM_PREDICT,
                    help=f"max tokens generated per run (default {DEFAULT_NUM_PREDICT})")
    ap.add_argument("--seed", type=int, default=DEFAULT_SEED)
    ap.add_argument("--pull", action="store_true", help="pull tags that are not installed")
    ap.add_argument("--unload-others", action="store_true",
                    help="unload models other processes left loaded before starting (bench hosts only)")
    ap.add_argument("--offline", action="store_true", help="do not contact registry.ollama.ai")
    ap.add_argument("--submit-json", action="store_true",
                    help="end with one JSON block to paste into the site's submission form (no upload)")
    ap.add_argument("--list-devices", action="store_true", help="print the built-in device table and exit")
    ap.add_argument("--registry-info", metavar="TAG", action="append",
                    help="print a tag's registry status, layer sizes and quant as a tags.txt line, then exit "
                         "(repeatable; needs network, not Ollama)")
    a = ap.parse_args(argv)
    if a.runs < MIN_VALID_RUNS + 1:
        print(f"note: --runs {a.runs} cannot produce a valid row (warm-up + {MIN_VALID_RUNS} valid "
              "runs needed); results will be recorded and rejected.", file=sys.stderr)
    if a.runs < 2:
        ap.error("--runs must be at least 2 (one warm-up + one measured run)")
    if not HOST_RE.match(a.host):
        ap.error("--host must be owner-mac, local, or runpod:/vast:/cloud:<gpu-slug> (lowercase), "
                 f"not '{a.host}'")
    if a.num_predict < MIN_TOKENS:
        ap.error(f"--num-predict must be at least {MIN_TOKENS}, or every run is discarded as too short")
    try:
        a.ctx_list = sorted({int(x) for x in a.ctx.split(",") if x.strip()})
    except ValueError:
        ap.error("--ctx must be a comma list of integers")
    if not a.ctx_list or min(a.ctx_list) < 256:
        ap.error("--ctx values must be >= 256")
    return a


# Keys of the shared row format that the site's submission endpoint accepts
# (lib/community-benchmarks.ts SubmissionRowSchema is strict: anything else is rejected).
SUBMISSION_ROW_KEYS = (
    "id", "deviceId", "deviceName", "bandwidthGBs", "host", "tag", "digest", "params", "quant",
    "fileGB", "ollama", "driver", "os", "powerLimitW", "numCtx", "runs", "validRuns", "tokS",
    "low", "high", "ttftMs", "vramGB", "ceilingTokS", "measured", "rawFile", "source", "verified",
)


def machine_id() -> str:
    """A random id generated once per machine and kept in a local file, so moderators can
    tell two submitting machines apart. It is never derived from hardware or the network,
    and the site stores only a salted hash of it."""
    path = pathlib.Path(os.environ.get("LAM_BENCH_ID_FILE")
                        or pathlib.Path.home() / ".config" / "localaimaster" / "bench-machine-id")
    try:
        value = path.read_text().strip()
        if re.fullmatch(r"[A-Za-z0-9_-]{16,128}", value):
            return value
    except OSError:
        pass
    value = secrets.token_hex(16)
    try:
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(value + "\n")
    except OSError:
        pass  # still usable for this one submission
    return value


def submission_block(sweep: dict, results: list[dict]) -> dict:
    """The paste-ready block for localaimaster.com/benchmarks/submit, in exactly the
    envelope the endpoint validates: {schemaVersion, format, harness, machineId, license,
    rows}. Rows are shared-format rows marked community / unverified."""
    rows = []
    sub_sweep = {**sweep, "source": "community", "host": "community", "rawFile": None}
    for r in results:
        s = summarise(r, sub_sweep)
        if s["row"]:
            row = {**s["row"], "verified": False}
            rows.append({k: row[k] for k in SUBMISSION_ROW_KEYS if k in row})
    return {
        "schemaVersion": 1,
        "format": "localaimaster-bench-submission",
        "harness": HARNESS_VERSION,
        "machineId": machine_id(),
        "license": LICENSE,
        "rows": rows[:25],
    }


def registry_info(tag: str) -> str:
    """One tags.txt line for a tag, straight from registry.ollama.ai (release-day helper)."""
    name, _, version = normalise_tag(tag).partition(":")
    url = f"{REGISTRY}/{name}/manifests/{version}"
    req = urllib.request.Request(url, headers={"Accept": "application/vnd.docker.distribution.manifest.v2+json"})
    try:
        with urllib.request.urlopen(req, timeout=20) as r:
            man = json.loads(r.read())
    except urllib.error.HTTPError as e:
        return f"# {name}:{version}  HTTP {e.code} from {url} — this tag does not exist; do not write it anywhere"
    except (urllib.error.URLError, OSError, ValueError) as e:
        return f"# {name}:{version}  registry unreachable ({e})"
    model = manifest_model_bytes(man) or 0
    proj = sum(l.get("size", 0) for l in man.get("layers", []) if "projector" in str(l.get("mediaType", "")))
    cfg: dict = {}
    try:
        with urllib.request.urlopen(f"{REGISTRY}/{name}/blobs/{man['config']['digest']}", timeout=20) as r:
            cfg = json.loads(r.read())
    except (urllib.error.URLError, OSError, ValueError, KeyError):
        pass
    need = model / 1e9 + proj / 1e9 + 1.5
    tier = next((t for t in (12, 16, 24, 32) if need <= t * 1.0737), None)
    mac = "yes" if need <= 12 * 1.0737 else "no"
    return (f"{name + ':' + version:<23} {model / 1e9:6.2f}  {proj / 1e9:5.2f}   {str(tier or '>32'):<9}  {mac:<5}  "
            f"# HTTP 200; {cfg.get('model_type', '?')} {cfg.get('file_type', '?')} {cfg.get('model_family', '')}; "
            "MoE is detected by the harness at run time")


def main(argv: list[str]) -> int:
    args = parse_args(argv)
    if args.registry_info:
        print("# tag                   fileGB  projGB  minVramGB  mac18")
        for t in args.registry_info:
            print(registry_info(t))
        return 0
    if args.list_devices:
        for k, v in DEVICES.items():
            print(f"{k:<20} {v['vramGB']:>3} GB  {v['bandwidthGBs']:>5} GB/s  {v['name']}")
        return 0
    check_devices_file()
    try:
        ollama_version = api("/api/version", timeout=5).get("version", "unknown")
    except Exception:
        print(f"Ollama is not reachable at {OLLAMA} — start it with `ollama serve`.")
        return 1

    tags = list(args.models)
    if args.tags_file:
        tags += read_tags_file(args.tags_file)
    tags = list(dict.fromkeys(normalise_tag(t) for t in (tags or DEFAULT_MODELS)))

    facts = host_facts()
    dev = resolve_device(args, facts)

    others = [m.get("name") for m in loaded_models()]
    if others:
        if args.unload_others:
            for m in others:
                unload(m)
            print(f"unloaded models left in memory: {', '.join(others)}")
        else:
            facts["warnings"].append(f"other models were loaded at start: {', '.join(others)}")

    started = utc_now()
    raw_path = reserve_raw_path(pathlib.Path(args.out), dev["deviceId"], args.host, started[:10])
    rel_raw = f"raw/{dev['deviceId']}/{raw_path.name}"
    sweep: dict = {
        "kind": "localaimaster-ollama-sweep",
        "schemaVersion": 1,
        "harness": f"scripts/benchmark-ollama.py {HARNESS_VERSION}",
        **dev,
        "host": args.host,
        "hostInstance": facts["hostInstance"],
        "ollama": ollama_version,
        "driver": facts["driver"],
        "os": facts["os"],
        "powerLimitW": facts["powerLimitW"],
        "hostFacts": facts,
        "warnings": facts["warnings"],
        "startedUtc": started,
        "finishedUtc": None,
        "complete": False,
        "rawFile": rel_raw,
        "source": "localaimaster",
        "license": LICENSE,
        "credit": CREDIT,
        "method": {"prompt": DEFAULT_PROMPT, "runsPerPair": args.runs, "warmupDiscarded": 1,
                   "minTokens": MIN_TOKENS, "minValidRuns": MIN_VALID_RUNS,
                   "ceilingTolerance": CEILING_TOLERANCE, "numPredict": args.num_predict,
                   "seed": args.seed, "ctx": args.ctx_list, "stream": True,
                   "ceiling": "bandwidthGBs / fileGB (GB = 1e9 bytes)"},
        "results": [],
    }

    print("=" * 66)
    for k in ("deviceId", "deviceName", "bandwidthGBs", "host", "ollama", "driver", "os", "powerLimitW"):
        print(f" {k:<13}: {sweep[k]}")
    print(f" {'runs':<13}: {args.runs} per model and context (first = warm-up, discarded)")
    print(f" {'contexts':<13}: {', '.join(map(str, args.ctx_list))}")
    print(f" {'raw file':<13}: {raw_path}")
    for w in facts["warnings"]:
        print(f" WARNING      : {w}")
    print("=" * 66, flush=True)

    installed = installed_models()
    for tag in tags:
        print(f"\n── {tag} " + "─" * max(4, 56 - len(tag)), flush=True)
        if tag not in installed:
            if not args.pull:
                print("   not installed — `ollama pull` it or pass --pull; skipped (not recorded as zero)")
                sweep["results"].append({"tag": tag, "skipReason": "not installed", "runs": []})
                write_atomic(raw_path, sweep)
                continue
            try:
                pull(tag)
            except Exception as e:  # noqa: BLE001
                print(f"   pull failed: {str(e)[:120]}")
                sweep["results"].append({"tag": tag, "skipReason": f"pull failed: {str(e)[:120]}", "runs": []})
                write_atomic(raw_path, sweep)
                continue
            installed = installed_models()
            if tag not in installed:
                sweep["results"].append({"tag": tag, "skipReason": "not listed after pull", "runs": []})
                write_atomic(raw_path, sweep)
                continue
        try:
            mf = model_facts(tag, installed[tag], args.offline)
        except Exception as e:  # noqa: BLE001
            print(f"   /api/show failed: {str(e)[:120]}")
            sweep["results"].append({"tag": tag, "skipReason": f"/api/show failed: {str(e)[:120]}", "runs": []})
            write_atomic(raw_path, sweep)
            continue
        print(f"   {mf['params']} {mf['quant']}, file {mf['fileGB']} GB ({mf['fileGBSource']})"
              + ("  [mixture-of-experts: no ceiling gate -> rows rejected]" if mf["moe"] else ""))
        for ctx in args.ctx_list:
            if len(args.ctx_list) > 1:
                print(f"   · num_ctx {ctx}", flush=True)
            res = bench_one(tag, mf, ctx, args)
            s = summarise(res, sweep)
            res["gate"] = {k: s[k] for k in ("status", "reasons", "unstable", "medianTokS", "ceilingTokS")}
            res["row"] = s["row"]
            sweep["results"].append(res)
            write_atomic(raw_path, sweep)
            if s["status"] == "valid":
                share = round(100 * s["medianTokS"] / s["ceilingTokS"])
                flag = "  ! unstable spread" if s["unstable"] else ""
                print(f"   -> {s['medianTokS']} tok/s median of {s['row']['validRuns']} "
                      f"(ceiling {s['ceilingTokS']}, {share}%){flag}", flush=True)
            else:
                print(f"   -> REJECTED: {'; '.join(s['reasons'])}", flush=True)
        unload(tag)

    try:
        sweep["hostFacts"]["loadAvgEnd"] = [round(x, 2) for x in os.getloadavg()]
    except OSError:
        pass
    sweep["finishedUtc"] = utc_now()
    sweep["complete"] = True
    write_atomic(raw_path, sweep)

    rows = [r["row"] for r in sweep["results"] if r.get("row")]
    print("\n" + "=" * 66)
    print("| Tag | Ctx | Params | Quant | File GB | tok/s (low–high) | Ceiling | TTFT ms | VRAM GB | Status |")
    print("|---|---|---|---|---|---|---|---|---|---|")
    for r in sweep["results"]:
        if "gate" not in r:
            print(f"| {r['tag']} | – | – | – | – | – | – | – | – | skipped: {r.get('skipReason')} |")
            continue
        row = r.get("row") or {}
        rate = f"{row['tokS']} ({row['low']}–{row['high']})" if row else str(r["gate"]["medianTokS"])
        status = "valid" if row else "rejected: " + "; ".join(r["gate"]["reasons"])
        print(f"| {r['tag']} | {r['numCtx']} | {r.get('params')} | {r.get('quant')} | {r.get('fileGB')} | {rate} | "
              f"{r['gate']['ceilingTokS']} | {row.get('ttftMs', '–')} | {(r.get('offload') or {}).get('vramGB', '–')} "
              f"| {status} |")
    print(f"\n*{dev['deviceName']}, Ollama {ollama_version}, {facts['os']}, {started[:10]} UTC. Median of "
          f"post-warm-up runs; runs under {MIN_TOKENS} tokens discarded; ceiling = {dev['bandwidthGBs']} GB/s "
          "÷ file size.*")
    print(f"\nraw sweep: {raw_path}")
    print(f"{len(rows)} valid row(s). Merge with: python3 scripts/bench-rows.py {raw_path}")
    if facts["warnings"]:
        print("conditions were not clean, so no row is marked verified:\n  - " + "\n  - ".join(facts["warnings"]))
    if args.submit_json:
        print("\n----- paste everything below this line into the submission form -----")
        print(json.dumps(submission_block(sweep, [r for r in sweep["results"] if "gate" in r]), indent=1))
    else:
        print("\nrows (shared format):")
        print(json.dumps(rows, indent=1))
    return 0


if __name__ == "__main__":
    try:
        sys.exit(main(sys.argv[1:]))
    except HarnessError as e:
        print(f"harness refused to run: {e}", file=sys.stderr)
        sys.exit(2)
    except KeyboardInterrupt:
        print("\ninterrupted — the raw file keeps every finished model (complete: false)", file=sys.stderr)
        sys.exit(130)