inferencelayer 0.2.13

//! `scoreboard` — the hold-the-crown perf harness (Engine Leadership Program P0).
//!
//! Measures the engine's scoreboard rows on a checkpoint and gates them against committed pins:
//! any measured row `< pin × 0.97` exits non-zero, so a perf regression blocks its cluster the
//! same way a red test does. `--pin` advances the pins after an accepted win; pins carry load
//! provenance so a number captured under contention is recognizable forever.
//!
//! Local rows (one engine instance, `--model`):
//!   `mono`       — solo greedy decode tok/s (decode-only, median of runs)
//!   `prefill`    — cold prompt prefill tok/s (default 2048 tokens, median of runs)
//!   `aggregate`  — continuous-batching Scheduler throughput at `--streams` concurrent requests
//!
//! Fleet rows (35B duo/4-way mono, WAN P2P, serving-MTP) are measured by their own harnesses
//! (`lfm2-shard-run`, accept_bench) and ingested with `--record <row>=<value>`; the constrained
//! at-scale row is measured by the application-tier gate test and ingested the same way. This
//! ledger is the single source of truth for every crown.
//!
//! Preflight refuses a busy machine on BOTH contention axes, because neither implies the other:
//! ambient CPU load, and foreign compute processes resident on the GPUs — a CUDA job from another
//! user pins the SMs while `uptime` still reads near-idle, so load average alone cannot certify a
//! quiet box (learned the hard way on the shared V100 box, 2026-07-12). Contention distorts
//! absolutes; only ratios survive it. `OSFKB_SCOREBOARD_IGNORE_LOAD=1` downgrades the refusal to a
//! warning for exploratory runs, and `--pin` is then REFUSED — a contended number must never
//! become the reference every later run is judged against. Prints NVIDIA clock/throttle state when
//! `nvidia-smi` is present (lock clocks before pinning: `nvidia-smi -lgc <mhz>`).

use anyhow::{Context, Result, bail};
use inferencelayer::bench_guard::{
    ensure_quiet_machine, load_average_1m, nvidia_clock_note, read_pins, write_pins,
};
use inferencelayer::{GpuCtx, Lfm2Gpu, Scheduler, Weights};
use std::time::Instant;

/// Tolerated fraction of the pinned value: measured ≥ pin × HOLD_FRACTION passes.
const HOLD_FRACTION: f64 = 0.97;

fn main() {
    let args: Vec<String> = std::env::args().skip(1).collect();
    match run(&args) {
        Ok(true) => {}
        Ok(false) => std::process::exit(1),
        Err(e) => {
            eprintln!("scoreboard: {e:#}");
            std::process::exit(2);
        }
    }
}

struct Opts {
    model: String,
    rows: Vec<String>,
    pins_path: String,
    write_pins: bool,
    records: Vec<(String, f64)>,
    ngen: usize,
    streams: usize,
    prefill_len: usize,
    runs: usize,
}

fn parse_opts(args: &[String]) -> Result<Opts> {
    let mut o = Opts {
        model: std::env::var("OSFKB_QWEN_DIR").unwrap_or_else(|_| "data/qwen3-0.6b".into()),
        rows: vec![],
        pins_path: "bench/scoreboard_pins.json".into(),
        write_pins: false,
        records: vec![],
        ngen: 128,
        streams: 8,
        prefill_len: 2048,
        runs: 3,
    };
    let mut it = args.iter();
    while let Some(a) = it.next() {
        let mut val = |name: &str| {
            it.next()
                .cloned()
                .with_context(|| format!("{name} needs a value"))
        };
        match a.as_str() {
            "--model" => o.model = val("--model")?,
            "--rows" => o.rows = val("--rows")?.split(',').map(str::to_string).collect(),
            "--pins" => o.pins_path = val("--pins")?,
            "--pin" => o.write_pins = true,
            "--ngen" => o.ngen = val("--ngen")?.parse().context("--ngen")?,
            "--streams" => o.streams = val("--streams")?.parse().context("--streams")?,
            "--prefill-len" => {
                o.prefill_len = val("--prefill-len")?.parse().context("--prefill-len")?
            }
            "--runs" => o.runs = val("--runs")?.parse().context("--runs")?,
            "--record" => {
                let kv = val("--record")?;
                let (k, v) = kv
                    .split_once('=')
                    .with_context(|| format!("--record wants row=value, got {kv:?}"))?;
                o.records
                    .push((k.to_string(), v.parse().context("--record value")?));
            }
            other => bail!("unknown flag {other:?}"),
        }
    }
    // Default to the local engine rows only when this is not a record-only invocation
    // (fleet/constrained rows are ingested checkpoint-free via --record).
    if o.rows.is_empty() && o.records.is_empty() {
        o.rows = vec!["mono".into(), "prefill".into(), "aggregate".into()];
    }
    Ok(o)
}

fn run(args: &[String]) -> Result<bool> {
    let o = parse_opts(args)?;
    let mut measured: Vec<(String, f64)> = o.records.clone();
    let wants_local = o
        .rows
        .iter()
        .any(|r| ["mono", "prefill", "aggregate"].contains(&r.as_str()));

    // The quiet-machine guard protects LOCAL measurement. A record-only invocation measures nothing
    // here — it ingests a value another harness produced elsewhere (the fleet rows on the V100 box,
    // the constrained row from the app-tier gate). Gating that on THIS machine's load is a false
    // positive, and it refused a legitimate fleet pin from a laptop mid-build.
    //
    // The flip side, stated rather than hidden: a recorded row carries NO local quiet guarantee,
    // and it never could. So such rows are tagged `source: "record"` in the pins, and their
    // `load_1m` is omitted — stamping the INGESTING machine's load on a number measured on another
    // box is provenance that reads as meaningful and is not.
    let quiet = if wants_local {
        let q = ensure_quiet_machine()?;
        if o.write_pins && !q {
            bail!(
                "--pin refused: the quiet-machine guard was overridden, so these numbers carry \
                 contention and must never become the reference every later run is judged against. \
                 Re-measure on a quiet box to pin."
            )
        }
        nvidia_clock_note();
        q
    } else {
        eprintln!(
            "record-only: {} row(s) ingested from another harness; the local quiet-machine guard \
             does not apply and is skipped",
            o.records.len()
        );
        true
    };
    let _ = quiet;

    if wants_local {
        let dir = std::path::Path::new(&o.model);
        if !dir.join("model.safetensors").exists()
            && !dir.join("model.safetensors.index.json").exists()
        {
            bail!(
                "no checkpoint at {} (pass --model or set OSFKB_QWEN_DIR)",
                o.model
            );
        }
        let ctx = GpuCtx::new().context("gpu adapter")?;
        let w = Weights::load(&ctx, &o.model).context("load weights")?;
        let vocab = w.cfg.vocab;
        let gpu = Lfm2Gpu::new(&ctx, w);
        let model_key = dir
            .file_name()
            .map(|s| s.to_string_lossy().into_owned())
            .unwrap_or_else(|| o.model.clone());

        // Cold-GPU warmup: the first run after pipeline compilation reads ~3× slow and would
        // invert every comparison (documented engine gotcha) — one throwaway decode first.
        let _ = gpu.decode(&ctx, &synth_prompt(vocab, 16, 0), 16)?;

        for row in &o.rows {
            let (key, value) = match row.as_str() {
                "mono" => {
                    let v = row_mono(&ctx, &gpu, vocab, o.ngen, o.runs)?;
                    (format!("mono@{model_key}"), v)
                }
                "prefill" => {
                    let v = row_prefill(&ctx, &gpu, vocab, o.prefill_len, o.runs)?;
                    (format!("prefill{}@{model_key}", o.prefill_len), v)
                }
                "aggregate" => {
                    let v = row_aggregate(&ctx, &gpu, vocab, o.streams, o.ngen)?;
                    (format!("aggregate{}@{model_key}", o.streams), v)
                }
                other => bail!("unknown local row {other:?} (fleet rows go through --record)"),
            };
            eprintln!("  {key}: {value:.1} tok/s");
            measured.push((key, value));
        }
    }
    if measured.is_empty() {
        bail!("nothing to do: no local rows selected and no --record given");
    }

    let mut pins = read_pins(&o.pins_path)?;
    let rows_obj = pins
        .as_object_mut()
        .and_then(|p| p.get_mut("rows"))
        .and_then(|r| r.as_object_mut())
        .context("pins file malformed: expected {\"rows\": {...}}")?;

    let load = load_average_1m();
    let mut all_hold = true;
    for (key, value) in &measured {
        match rows_obj
            .get(key)
            .and_then(|e| e.get("value"))
            .and_then(|v| v.as_f64())
        {
            Some(pin) if *value < pin * HOLD_FRACTION => {
                all_hold = false;
                eprintln!(
                    "REGRESSION {key}: {value:.1} < pin {pin:.1} × {HOLD_FRACTION} = {:.1}",
                    pin * HOLD_FRACTION
                );
            }
            Some(pin) => {
                eprintln!("hold {key}: {value:.1} vs pin {pin:.1} ✓");
            }
            None => {
                eprintln!(
                    "new  {key}: {value:.1} (no pin yet{})",
                    if o.write_pins {
                        "; pinning"
                    } else {
                        " — run with --pin to record"
                    }
                );
            }
        }
        if o.write_pins {
            // A row measured HERE carries this machine's load as provenance. A row ingested via
            // --record was measured on another box entirely, so this machine's load says nothing
            // about it — recording it anyway would manufacture provenance that reads as real.
            let recorded = o.records.iter().any(|(k, _)| k == key);
            let mut entry = serde_json::json!({
                "value": value,
                "captured_epoch_s": std::time::SystemTime::now()
                    .duration_since(std::time::UNIX_EPOCH)
                    .map(|d| d.as_secs())
                    .unwrap_or(0),
            });
            let obj = entry.as_object_mut().expect("json object");
            if recorded {
                obj.insert("source".into(), serde_json::json!("record"));
            } else {
                obj.insert("load_1m".into(), serde_json::json!(load));
            }
            rows_obj.insert(key.clone(), entry);
        }
    }
    if o.write_pins {
        write_pins(&o.pins_path, &pins)?;
        eprintln!("pins written to {}", o.pins_path);
    }
    Ok(all_hold)
}

/// Deterministic synthetic prompt: ids spread over the vocab interior (never id 0 / specials
/// tail), index-seeded so every run measures identical inputs without a tokenizer.
fn synth_prompt(vocab: usize, len: usize, seed: usize) -> Vec<u32> {
    let span = vocab.saturating_sub(20).max(1);
    (0..len)
        .map(|i| (10 + (seed * 7919 + i * 6151) % span) as u32)
        .collect()
}

/// Solo greedy decode tok/s: median of `runs` decode-only rates over distinct prompts.
fn row_mono(ctx: &GpuCtx, gpu: &Lfm2Gpu, vocab: usize, ngen: usize, runs: usize) -> Result<f64> {
    let mut rates = Vec::with_capacity(runs);
    for r in 0..runs {
        let prompt = synth_prompt(vocab, 32, r + 1);
        let (_, decode_tok_s, _) = gpu.decode(ctx, &prompt, ngen)?;
        rates.push(decode_tok_s);
    }
    Ok(median(rates))
}

/// Cold prefill tok/s over a `plen`-token prompt, measured as prompt-tokens / TTFT through the
/// SCHEDULER — i.e. the real prefill path (batched prefill columns through the batch plan).
///
/// It must NOT go through `Lfm2Gpu::decode`: that is the M=1 solo loop, which ingests the prompt
/// one token at a time and so measures sequential ingestion, not prefill. (It reported 16 tok/s
/// here — slower than decode itself, which is the tell: prefill is batched and is always far
/// faster per token.)
fn row_prefill(ctx: &GpuCtx, gpu: &Lfm2Gpu, vocab: usize, plen: usize, runs: usize) -> Result<f64> {
    let mut rates = Vec::with_capacity(runs);
    for r in 0..runs {
        let prompt = synth_prompt(vocab, plen, 100 + r);
        let mut sched = Scheduler::new(gpu, ctx, 8, vec![]);
        let id = sched.submit(prompt, 1)?;
        let t0 = Instant::now();
        // Step until this request emits its first token: that wall time IS the prefill.
        loop {
            let ems = sched.step(gpu, ctx)?;
            if ems.iter().any(|e| e.id == id) {
                break;
            }
        }
        let ttft = t0.elapsed().as_secs_f64();
        rates.push(plen as f64 / ttft.max(1e-9));
    }
    Ok(median(rates))
}

/// Continuous-batching aggregate tok/s: `streams` concurrent requests of `ngen` tokens each
/// through the Scheduler (no EOS — every request runs to max_tokens, so the workload is exact).
fn row_aggregate(
    ctx: &GpuCtx,
    gpu: &Lfm2Gpu,
    vocab: usize,
    streams: usize,
    ngen: usize,
) -> Result<f64> {
    let mut sched = Scheduler::new(
        gpu,
        ctx,
        streams.min(inferencelayer::forward::MAX_SLOTS),
        vec![],
    );
    for s in 0..streams {
        sched.submit(synth_prompt(vocab, 32, 1000 + s), ngen)?;
    }
    let t0 = Instant::now();
    let results = sched.run_to_completion(gpu, ctx)?;
    let secs = t0.elapsed().as_secs_f64();
    let total: usize = results.values().map(Vec::len).sum();
    Ok(total as f64 / secs)
}

fn median(mut v: Vec<f64>) -> f64 {
    v.sort_by(|a, b| a.partial_cmp(b).expect("finite"));
    v[v.len() / 2]
}

#[cfg(test)]
mod tests {
    use inferencelayer::bench_guard::parse_compute_apps;

    /// Verbatim capture from the shared V100 box (2026-07-12) at the moment a foreign workload
    /// landed mid-session: three root-owned CUDA jobs holding 9-18 GiB while the 1-min load
    /// average read 2.05 against a 30.0 limit (60 cores). CPU load is structurally blind to this,
    /// which is why the guard has to ask the GPU directly.
    const V100_CONTENDED: &str = "2110607, 9462 MiB, python\n\
                                  2110983, 8896 MiB, python\n\
                                  2111120, 18566 MiB, python\n";

    #[test]
    fn should_report_foreign_compute_procs_occupying_the_gpus() {
        let procs = parse_compute_apps(V100_CONTENDED, 999);
        assert_eq!(procs.len(), 3, "all three foreign jobs are contention");
        assert!(
            procs[2].contains("18566 MiB"),
            "row kept verbatim for the operator"
        );
    }

    /// The scoreboard itself appears in `--query-compute-apps` once it opens a GPU context; it is
    /// not contention with itself.
    #[test]
    fn should_not_count_our_own_process_as_contention() {
        let procs = parse_compute_apps("4242, 512 MiB, scoreboard\n", 4242);
        assert!(
            procs.is_empty(),
            "our own pid is not a foreign job: {procs:?}"
        );
    }

    /// An idle GPU emits no rows — that is the only state in which absolutes may be pinned.
    #[test]
    fn should_report_a_clear_gpu_as_having_no_foreign_procs() {
        assert!(parse_compute_apps("", 1).is_empty());
        assert!(
            parse_compute_apps("\n  \n", 1).is_empty(),
            "blank rows are not jobs"
        );
    }
}