nornir 0.4.17

//! 📈 **Bench LIVE** — surface a benchmark run *as it runs* in the Bench pane
//! (plan item **P1.3**).
//!
//! Where the rest of the 📈 Bench tab charts **historical** metric series (one
//! bar per past run), this panel shows the **currently-/most-recently-running
//! bench's footprint** straight from the warehouse:
//!
//!   * the **bench unit NAME** (`znippy.compress`, `holger.ops`, …),
//!   * a **green/red status** (derived from the run's `tests` outcomes — a bench
//!     whose matching test flipped red is red; a clean bench is green; a bench
//!     with no paired test is neutral),
//!   * a **cores-busy / N** indicator (avg + peak from the `bench_telemetry`
//!     table P1.2 streams — `cores_busy_avg` / `cores_busy_max` over `n_cores`),
//!   * a tiny **CPU sparkline** (avg → max → peak-busy fraction), and
//!   * the peak-mem badge.
//!
//! **Source of truth = the warehouse** (LAW "persist-to-warehouse-stream-not-tmp").
//! Local mode opens the warehouse read-only and reads two tables:
//!   * [`bench_telemetry`](crate::warehouse::iceberg::BenchTelemetryRow) — one
//!     row per `(run_id, repo, bench)`, the CPU/mem footprint the in-bench
//!     sampler captured, AND
//!   * `bench_runs` — for the per-bench pass/fail `tests`.
//!
//! The pane re-reads on a timer while the tab is open, so rows that land mid-run
//! (P1.2 appends per-run) **appear as they arrive** — `draw` calls
//! `request_repaint_after` to keep the frame moving without user input.
//!
//! Remote (thin-client) mode shows a note — there is no `Bench.Telemetry` RPC
//! yet (same split as the 🧪 Test / 🚀 Release tabs).
//!
//! Everything rendered is exposed via `state_json()` (folded into the app's
//! `state_json["bench"]["live"]`) and the `$NORNIR_VIZ_TRACE` IN/OUT/END stream
//! (LAW #6: "see what the user sees" as data, not pixels).

use std::path::PathBuf;
use std::time::{Duration, Instant};

use eframe::egui::{self, Color32, RichText, ScrollArea};

use crate::warehouse::iceberg::{BenchTelemetryRow, IcebergWarehouse};
use crate::warehouse::{BenchFilter, Warehouse};

/// How often (while the tab is open) the live panel re-reads the warehouse so
/// telemetry rows that land mid-run show up. Cheap read-only scan.
const RELOAD_EVERY: Duration = Duration::from_millis(1500);

/// Where the live telemetry comes from (mirrors the other tabs' local/remote split).
enum Src {
    Local(PathBuf),
    /// No `Bench.Telemetry` RPC exists yet — remote mode shows a note.
    Remote { endpoint: String },
}

/// One live bench unit row the panel renders: name + status + cores-busy/N + the
/// CPU sparkline. Built by folding a `bench_telemetry` row with the matching
/// `bench_runs` test outcome.
#[derive(Debug, Clone, PartialEq)]
pub struct LiveBench {
    pub run_id: String,
    pub repo: String,
    pub bench: String,
    pub n_cores: u32,
    pub cores_busy_avg: f64,
    pub cores_busy_max: u32,
    pub cpu_pct_avg: f64,
    pub cpu_pct_max: f64,
    pub mem_peak_mb: f64,
    pub elapsed_ms: f64,
    /// `Some(true)` green, `Some(false)` red, `None` neutral (no paired test).
    pub ok: Option<bool>,
}

impl LiveBench {
    /// The CPU sparkline series the cell draws: a tiny avg→max→peak-busy-fraction
    /// trace (0..100), so an at-a-glance shape shows how hard the bench pushed the
    /// cores. Exposed verbatim in `state_json` so the headless matrix asserts it.
    pub fn cpu_spark(&self) -> Vec<f64> {
        let busy_frac = if self.n_cores > 0 {
            self.cores_busy_max as f64 / self.n_cores as f64 * 100.0
        } else {
            0.0
        };
        vec![self.cpu_pct_avg, self.cpu_pct_max, busy_frac]
    }

    /// `cores-busy avg/peak of N` label.
    pub fn cores_label(&self) -> String {
        format!(
            "cores-busy {:.1}/{} (peak {}/{})",
            self.cores_busy_avg, self.n_cores, self.cores_busy_max, self.n_cores
        )
    }
}

pub struct BenchLive {
    src: Src,
    /// Whether the first load has run (so `draw` lazy-loads once, then the timer
    /// re-reads). `false` after a `reload()` to force a re-read.
    loaded: bool,
    last_reload: Instant,
    error: Option<String>,
    /// The live bench units, newest run first, then by bench name.
    benches: Vec<LiveBench>,
    /// The run_id of the newest run the panel is showing (the "live" run).
    live_run_id: Option<String>,
}

impl BenchLive {
    pub fn local(root: PathBuf) -> Self {
        Self::with(Src::Local(root))
    }

    pub fn remote(endpoint: String) -> Self {
        Self::with(Src::Remote { endpoint })
    }

    fn with(src: Src) -> Self {
        Self {
            src,
            loaded: false,
            last_reload: Instant::now(),
            error: None,
            benches: Vec::new(),
            live_run_id: None,
        }
    }

    /// Test-only: inject `bench_telemetry` rows + `bench_runs` (for per-bench
    /// status) directly — no warehouse on disk — exactly as if both tables had
    /// been read. Mirrors the Test/Live tabs' `inject_for_test` so the
    /// inject-and-assert harness reads the rendered live benches back out of
    /// `state_json`.
    #[doc(hidden)]
    pub fn inject_for_test(
        &mut self,
        telemetry: Vec<BenchTelemetryRow>,
        runs: Vec<(String, crate::bench::BenchRun)>,
    ) {
        self.benches = fold_live(telemetry, &runs);
        self.live_run_id = self.benches.first().map(|b| b.run_id.clone());
        self.loaded = true;
        // Push the reload clock forward so the periodic warehouse re-read doesn't
        // immediately overwrite the injected rows (there is no warehouse on disk
        // in the inject-and-assert harness).
        self.last_reload = Instant::now() + Duration::from_secs(3600);
        self.error = None;
        self.emit_trace();
    }

    /// Re-scope (workspace switch / reload): force the next draw to re-read.
    pub fn reload(&mut self) {
        self.loaded = false;
        self.error = None;
    }

    fn source_label(&self) -> String {
        match &self.src {
            Src::Local(p) => format!("local {}", p.display()),
            Src::Remote { endpoint } => format!("remote {endpoint} (Bench.Telemetry RPC TODO)"),
        }
    }

    /// Read `bench_telemetry` + `bench_runs` from the warehouse (local only) and
    /// fold them into the live-bench rows. Lock-tolerant read-only open, like the
    /// Test/Release tabs, so it coexists with a live server / running bench
    /// holding the catalog lock. A read error is recorded (not fatal): an empty
    /// warehouse before the first bench is normal.
    fn load(&mut self) {
        let root = match &self.src {
            Src::Local(root) => root.clone(),
            Src::Remote { .. } => return,
        };
        super::trace::emit_in(
            "bench.live.load",
            &serde_json::json!({ "source": self.source_label() }),
        );
        let res = IcebergWarehouse::open_read_only(&root).and_then(|wh| {
            let telem = wh.query_bench_telemetry(None)?;
            let runs = wh.query_bench_runs(&BenchFilter::default())?;
            // Pair each run with a synthetic run_id key. The warehouse doesn't
            // re-surface the `run_id` on `BenchRun`, so we key the test-outcome
            // lookup on (repo, date+version) — the same compound the telemetry
            // run_id is derived from when appended. We index by repo+bench name.
            Ok::<_, anyhow::Error>((telem, runs))
        });
        match res {
            Ok((telem, runs)) => {
                // Build a (repo → BenchRun) keying so the per-bench test status
                // can be looked up. We don't have run_id on BenchRun, so fold by
                // repo + bench name against the newest run per repo.
                let keyed: Vec<(String, crate::bench::BenchRun)> =
                    runs.into_iter().map(|r| (String::new(), r)).collect();
                self.benches = fold_live(telem, &keyed);
                self.live_run_id = self.benches.first().map(|b| b.run_id.clone());
                self.error = None;
                self.emit_trace();
            }
            Err(e) => {
                self.error = Some(format!("{e:#}"));
                super::trace::emit_out("bench.live.load", &serde_json::json!({ "error": format!("{e:#}") }));
            }
        }
    }

    fn emit_trace(&self) {
        super::trace::emit_out("bench.live.build", &self.benches_json());
        let (green, red, neutral) = self.tallies();
        super::trace::emit_end(
            "bench.live",
            &serde_json::json!({
                "live_run_id": self.live_run_id,
                "benches": self.benches.len(),
                "green": green, "red": red, "neutral": neutral,
            }),
        );
    }

    fn tallies(&self) -> (usize, usize, usize) {
        let mut green = 0;
        let mut red = 0;
        let mut neutral = 0;
        for b in &self.benches {
            match b.ok {
                Some(true) => green += 1,
                Some(false) => red += 1,
                None => neutral += 1,
            }
        }
        (green, red, neutral)
    }

    pub fn draw(&mut self, ui: &mut egui::Ui) {
        // Lazy first load, then a periodic re-read so mid-run rows appear live.
        if !self.loaded {
            self.loaded = true;
            self.load();
            self.last_reload = Instant::now();
        } else if matches!(self.src, Src::Local(_)) && self.last_reload.elapsed() >= RELOAD_EVERY {
            self.load();
            self.last_reload = Instant::now();
        }

        ui.heading("📡 Bench — live telemetry");

        if let Src::Remote { endpoint } = &self.src {
            ui.label(format!(
                "Live bench telemetry lives in the warehouse on `{endpoint}`. A \
                 streaming Bench.Telemetry RPC is not wired yet — point the viz at a \
                 local warehouse (NORNIR_WAREHOUSE=…) to watch a bench run live."
            ));
            ui.separator();
            return;
        }

        if let Some(err) = self.error.clone() {
            ui.colored_label(Color32::RED, format!("bench_telemetry read failed:\n{err}"));
            ui.separator();
            return;
        }

        ui.horizontal(|ui| {
            ui.label("live run:");
            ui.monospace(self.live_run_id.as_deref().unwrap_or("—"));
            ui.separator();
            let (green, red, neutral) = self.tallies();
            ui.colored_label(Color32::LIGHT_GREEN, format!("✓ {green}"));
            ui.colored_label(Color32::LIGHT_RED, format!("✗ {red}"));
            if neutral > 0 {
                ui.colored_label(Color32::GRAY, format!("· {neutral}"));
            }
            if ui.button("↻").on_hover_text("re-read bench_telemetry").clicked() {
                self.reload();
            }
        });

        if self.benches.is_empty() {
            ui.add_space(8.0);
            ui.weak("no bench telemetry recorded yet — run a bench to light this up:");
            ui.monospace("nornir bench <repo>");
            ui.separator();
            return;
        }

        ui.separator();
        ScrollArea::vertical().auto_shrink([false, false]).max_height(220.0).show(ui, |ui| {
            for b in &self.benches {
                draw_live_row(ui, b);
            }
        });
        ui.separator();

        // Keep the frame moving so live rows appear without user input.
        ui.ctx().request_repaint_after(Duration::from_millis(400));
    }

    /// LAW #6: the live benches the panel renders, as JSON folded into the app's
    /// `state_json["bench"]["live"]` — the proof the pane reads the warehouse
    /// telemetry, not pixels.
    pub fn state_json(&self) -> serde_json::Value {
        let (green, red, neutral) = self.tallies();
        serde_json::json!({
            "source": self.source_label(),
            "error": self.error,
            "live_run_id": self.live_run_id,
            "count": self.benches.len(),
            "green": green, "red": red, "neutral": neutral,
            "benches": self.benches_json_inner(),
        })
    }

    fn benches_json(&self) -> serde_json::Value {
        serde_json::json!({ "benches": self.benches_json_inner() })
    }

    fn benches_json_inner(&self) -> Vec<serde_json::Value> {
        self.benches
            .iter()
            .map(|b| {
                serde_json::json!({
                    "run_id": b.run_id,
                    "repo": b.repo,
                    "bench": b.bench,
                    "status": match b.ok {
                        Some(true) => "green",
                        Some(false) => "red",
                        None => "neutral",
                    },
                    "ok": b.ok,
                    "n_cores": b.n_cores,
                    "cores_busy_avg": round2(b.cores_busy_avg),
                    "cores_busy_max": b.cores_busy_max,
                    "cores_label": b.cores_label(),
                    "cpu_pct_avg": round2(b.cpu_pct_avg),
                    "cpu_pct_max": round2(b.cpu_pct_max),
                    "mem_peak_mb": round2(b.mem_peak_mb),
                    "elapsed_ms": round2(b.elapsed_ms),
                    "cpu_spark": b.cpu_spark().iter().map(|v| round2(*v)).collect::<Vec<_>>(),
                })
            })
            .collect()
    }
}

/// Fold `bench_telemetry` rows with `bench_runs` (for per-bench pass/fail) into
/// the rendered [`LiveBench`] list, newest run first then by bench name.
///
/// The per-bench **status** is derived from the run's `tests`: a bench is red if
/// a `TestOutcome` whose `name` matches the bench (exact, or the bench's last
/// `.`-segment) failed; green if a matching test passed; neutral if no test
/// pairs with it. `runs` is `(key, BenchRun)` — the key is unused by the
/// warehouse path (we match on repo+bench name) but lets a test inject explicit
/// run_ids if desired.
fn fold_live(
    mut telemetry: Vec<BenchTelemetryRow>,
    runs: &[(String, crate::bench::BenchRun)],
) -> Vec<LiveBench> {
    // Newest run first: telemetry rows don't carry a timestamp, so we keep the
    // order the warehouse returns (append order ≈ chronological) but surface the
    // run with the most rows / latest run_id first. We sort by run_id descending
    // as a stable proxy (run_ids are time-prefixed UUIDs/labels), then bench name.
    telemetry.sort_by(|a, b| {
        b.run_id.cmp(&a.run_id).then(a.bench.cmp(&b.bench))
    });

    telemetry
        .into_iter()
        .map(|t| {
            let ok = bench_status(&t.repo, &t.bench, runs);
            LiveBench {
                run_id: t.run_id,
                repo: t.repo,
                bench: t.bench,
                n_cores: t.n_cores,
                cores_busy_avg: t.cores_busy_avg,
                cores_busy_max: t.cores_busy_max,
                cpu_pct_avg: t.cpu_pct_avg,
                cpu_pct_max: t.cpu_pct_max,
                mem_peak_mb: t.mem_peak_mb,
                elapsed_ms: t.elapsed_ms,
                ok,
            }
        })
        .collect()
}

/// Derive a bench's green/red status from any `bench_runs` row of the same repo
/// whose `tests` pairs with the bench name. Returns `None` (neutral) if nothing
/// pairs. Matches a test name exactly OR against the bench's last `.`-segment
/// (`znippy.compress` ↔ test `compress`).
fn bench_status(repo: &str, bench: &str, runs: &[(String, crate::bench::BenchRun)]) -> Option<bool> {
    let leaf = bench.rsplit('.').next().unwrap_or(bench);
    for (_key, run) in runs {
        // The repo isn't carried on BenchRun; we match on bench/test name only,
        // but prefer runs whose results name this bench (same repo's run).
        let _ = repo;
        let names_this_bench = run.results.iter().any(|r| r.name == bench || r.name == leaf);
        if !names_this_bench && !run.tests.iter().any(|t| t.name == bench || t.name == leaf) {
            continue;
        }
        for t in &run.tests {
            if t.name == bench || t.name == leaf {
                return Some(t.passed);
            }
        }
    }
    None
}

fn draw_live_row(ui: &mut egui::Ui, b: &LiveBench) {
    ui.horizontal(|ui| {
        // status dot.
        let (mark, col) = match b.ok {
            Some(true) => ("✓", Color32::LIGHT_GREEN),
            Some(false) => ("✗", Color32::LIGHT_RED),
            None => ("·", Color32::GRAY),
        };
        ui.colored_label(col, mark);
        // bench name.
        ui.add_sized(
            [180.0, 16.0],
            egui::Label::new(RichText::new(&b.bench).monospace().size(12.0)),
        );
        // cores-busy / N indicator.
        let busy_col = busy_color(b.cores_busy_max, b.n_cores);
        ui.colored_label(
            busy_col,
            RichText::new(format!("{}/{}", b.cores_busy_max, b.n_cores)).monospace().strong(),
        );
        ui.weak("cores");
        // CPU sparkline.
        draw_cpu_spark(ui, &b.cpu_spark());
        // CPU% + mem.
        ui.label(RichText::new(format!("{:.0}% cpu", b.cpu_pct_max)).monospace().size(11.0));
        if b.mem_peak_mb > 0.0 {
            ui.weak(format!("{:.0} MB", b.mem_peak_mb));
        }
    });
}

/// Colour the cores-busy/N indicator on a load ramp: green few, amber mid, red
/// pegged.
fn busy_color(busy: u32, n: u32) -> Color32 {
    if n == 0 {
        return Color32::GRAY;
    }
    let frac = busy as f32 / n as f32;
    if frac >= 0.75 {
        Color32::from_rgb(230, 120, 80)
    } else if frac >= 0.4 {
        Color32::from_rgb(230, 200, 90)
    } else {
        Color32::LIGHT_GREEN
    }
}

/// A tiny CPU sparkline (values 0..100).
fn draw_cpu_spark(ui: &mut egui::Ui, series: &[f64]) {
    let (rect, _) = ui.allocate_exact_size(egui::Vec2::new(46.0, 14.0), egui::Sense::hover());
    if series.len() < 2 {
        return;
    }
    let painter = ui.painter_at(rect);
    let n = series.len();
    let pts: Vec<egui::Pos2> = series
        .iter()
        .enumerate()
        .map(|(i, &v)| {
            let t = i as f32 / (n - 1) as f32;
            let x = rect.min.x + t * rect.width();
            let norm = (v.clamp(0.0, 100.0) / 100.0) as f32;
            let y = rect.max.y - norm * (rect.height() - 2.0) - 1.0;
            egui::Pos2::new(x, y)
        })
        .collect();
    for w in pts.windows(2) {
        painter.line_segment([w[0], w[1]], egui::Stroke::new(1.4, Color32::from_rgb(80, 170, 230)));
    }
    if let Some(last) = pts.last() {
        painter.circle_filled(*last, 1.8, Color32::from_rgb(80, 170, 230));
    }
}

fn round2(f: f64) -> f64 {
    (f * 100.0).round() / 100.0
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::bench::{BenchResult, BenchRun, TestOutcome};

    fn telem(run: &str, repo: &str, bench: &str, n: u32, busy_avg: f64, busy_max: u32) -> BenchTelemetryRow {
        BenchTelemetryRow {
            run_id: run.into(),
            repo: repo.into(),
            bench: bench.into(),
            n_cores: n,
            cpu_pct_avg: 50.0,
            cpu_pct_max: 90.0,
            cores_busy_avg: busy_avg,
            cores_busy_max: busy_max,
            mem_peak_mb: 256.0,
            mem_pct_max: 30.0,
            elapsed_ms: 1234.0,
        }
    }

    fn run_with_tests(tests: Vec<(&str, bool)>, results: Vec<&str>) -> BenchRun {
        BenchRun {
            date: "2026-06-13".into(),
            timestamp: None,
            version: "1.0".into(),
            machine: "host".into(),
            cores: 8,
            results: results
                .into_iter()
                .map(|n| BenchResult { name: n.into(), metrics: Default::default() })
                .collect(),
            tests: tests
                .into_iter()
                .map(|(n, ok)| TestOutcome { name: n.into(), passed: ok, duration_ms: None, message: None })
                .collect(),
        }
    }

    #[test]
    fn fold_pairs_status_and_cores() {
        // INJECT: one telemetry row + a run whose test for that bench is red.
        let telem_rows = vec![telem("run-2", "znippy", "znippy.compress", 8, 6.5, 8)];
        let runs = vec![(
            String::new(),
            run_with_tests(vec![("compress", false)], vec!["znippy.compress"]),
        )];
        let live = fold_live(telem_rows, &runs);
        assert_eq!(live.len(), 1);
        let b = &live[0];
        // ASSERT real values, not "didn't panic".
        assert_eq!(b.bench, "znippy.compress");
        assert_eq!(b.ok, Some(false), "red test → red bench");
        assert_eq!(b.n_cores, 8);
        assert_eq!(b.cores_busy_max, 8);
        assert!((b.cores_busy_avg - 6.5).abs() < 1e-9);
        assert_eq!(b.cores_label(), "cores-busy 6.5/8 (peak 8/8)");
        // CPU sparkline: avg, max, peak-busy-fraction (8/8 → 100).
        assert_eq!(b.cpu_spark(), vec![50.0, 90.0, 100.0]);
    }

    #[test]
    fn newest_run_first_then_bench_name() {
        let telem_rows = vec![
            telem("run-1", "z", "z.b", 4, 2.0, 3),
            telem("run-2", "z", "z.a", 4, 1.0, 2),
            telem("run-2", "z", "z.c", 4, 4.0, 4),
        ];
        let live = fold_live(telem_rows, &[]);
        // run-2 (newest) first, sorted by bench name within the run.
        assert_eq!(live[0].run_id, "run-2");
        assert_eq!(live[0].bench, "z.a");
        assert_eq!(live[1].bench, "z.c");
        assert_eq!(live[2].run_id, "run-1");
        // No paired tests → neutral.
        assert_eq!(live[0].ok, None);
    }

    #[test]
    fn inject_renders_into_state_json() {
        // INJECT-AND-ASSERT (the P1.3 LAW): seed bench_telemetry + bench_runs,
        // "render" via state_json, assert the bench NAME + STATUS + cores-busy
        // are real values read back out — not merely no-crash.
        let mut live = BenchLive::local(PathBuf::new());
        live.inject_for_test(
            vec![
                telem("run-9", "znippy", "znippy.compress", 8, 5.0, 7),
                telem("run-9", "holger", "holger.ops", 8, 3.0, 4),
            ],
            vec![
                (String::new(), run_with_tests(vec![("compress", true)], vec!["znippy.compress"])),
                (String::new(), run_with_tests(vec![("ops", false)], vec!["holger.ops"])),
            ],
        );
        let v = live.state_json();
        assert_eq!(v["live_run_id"], "run-9");
        assert_eq!(v["count"], 2);
        assert_eq!(v["green"], 1);
        assert_eq!(v["red"], 1);
        let benches = v["benches"].as_array().unwrap();
        // Find the compress bench and assert name/status/cores read back.
        let compress = benches.iter().find(|b| b["bench"] == "znippy.compress").unwrap();
        assert_eq!(compress["status"], "green");
        assert_eq!(compress["n_cores"], 8);
        assert_eq!(compress["cores_busy_max"], 7);
        assert_eq!(compress["cores_label"], "cores-busy 5.0/8 (peak 7/8)");
        assert_eq!(compress["cpu_spark"].as_array().unwrap().len(), 3);
        let ops = benches.iter().find(|b| b["bench"] == "holger.ops").unwrap();
        assert_eq!(ops["status"], "red", "failed test → red");
        assert_eq!(ops["cores_busy_max"], 4);
    }
}