kibble 0.1.0

use serde_json::Value;
use serde::Serialize;
use std::collections::HashSet;
use std::path::Path;
use crate::config::{load_config, EvalConfig};

pub struct Row {
    pub user: String,
    pub assistant: String,
}

pub struct SplitParse {
    pub rows: Vec<Row>,
    pub malformed: usize,
    pub total: usize,
}

pub fn norm_key(s: &str) -> String {
    s.split_whitespace().collect::<Vec<_>>().join(" ").to_lowercase()
}

/// Parse one split. A line is a valid SFT row iff `messages` is a non-empty array whose entries
/// all have a role in {system,user,assistant} with non-empty content, and it has >=1 user and
/// >=1 assistant turn. Missing file → empty.
pub fn parse_split(path: &Path) -> SplitParse {
    let text = match std::fs::read_to_string(path) {
        Ok(t) => t,
        Err(_) => return SplitParse { rows: Vec::new(), malformed: 0, total: 0 },
    };
    let (mut rows, mut malformed, mut total) = (Vec::new(), 0usize, 0usize);
    for line in text.lines() {
        let line = line.trim();
        if line.is_empty() { continue; }
        total += 1;
        let v: Value = match serde_json::from_str(line) { Ok(v) => v, Err(_) => { malformed += 1; continue; } };
        let msgs = match v.get("messages").and_then(|m| m.as_array()) {
            Some(m) if !m.is_empty() => m,
            _ => { malformed += 1; continue; }
        };
        let mut ok = true;
        let (mut last_user, mut last_asst) = (None, None);
        for m in msgs {
            let role = m.get("role").and_then(|r| r.as_str()).unwrap_or("");
            let content = m.get("content").and_then(|c| c.as_str()).unwrap_or("");
            if !matches!(role, "system" | "user" | "assistant") || content.trim().is_empty() {
                ok = false; break;
            }
            if role == "user" { last_user = Some(content.to_string()); }
            if role == "assistant" { last_asst = Some(content.to_string()); }
        }
        match (ok, last_user, last_asst) {
            (true, Some(u), Some(a)) => rows.push(Row { user: u, assistant: a }),
            _ => malformed += 1,
        }
    }
    SplitParse { rows, malformed, total }
}

pub fn duplicate_rate(rows: &[Row]) -> f64 {
    if rows.is_empty() { return 0.0; }
    let distinct: HashSet<String> = rows.iter().map(|r| norm_key(&r.assistant)).collect();
    (rows.len() - distinct.len()) as f64 / rows.len() as f64
}

pub fn leakage_rate(train: &[Row], held: &[Row]) -> f64 {
    if held.is_empty() { return 0.0; }
    let keys: HashSet<String> = train.iter().map(|r| norm_key(&r.assistant)).collect();
    let leaked = held.iter().filter(|r| keys.contains(&norm_key(&r.assistant))).count();
    leaked as f64 / held.len() as f64
}

pub struct LengthStats {
    pub min: usize,
    pub median: usize,
    pub p90: usize,
    pub max: usize,
    pub short_rate: f64,
    pub long_count: usize,
}

pub fn length_stats(rows: &[Row], short_char_min: usize, long_max: usize) -> LengthStats {
    if rows.is_empty() {
        return LengthStats { min: 0, median: 0, p90: 0, max: 0, short_rate: 0.0, long_count: 0 };
    }
    let mut lens: Vec<usize> = rows.iter().map(|r| r.assistant.chars().count()).collect();
    lens.sort_unstable();
    let n = lens.len();
    let pct = |p: f64| lens[((p * (n as f64 - 1.0)).round() as usize).min(n - 1)];
    let short = lens.iter().filter(|&&l| l < short_char_min).count();
    let long = lens.iter().filter(|&&l| l > long_max).count();
    LengthStats {
        min: lens[0], median: pct(0.5), p90: pct(0.9), max: lens[n - 1],
        short_rate: short as f64 / n as f64, long_count: long,
    }
}

pub fn degenerate_rate(rows: &[Row], short_char_min: usize) -> f64 {
    if rows.is_empty() { return 0.0; }
    let bad = rows.iter().filter(|r| {
        let a = r.assistant.trim();
        a.is_empty()
            || a.chars().count() < short_char_min
            || norm_key(&r.assistant) == norm_key(&r.user)
    }).count();
    bad as f64 / rows.len() as f64
}

pub fn distinct2(rows: &[Row]) -> f64 {
    use std::hash::{Hash, Hasher};
    let mut total = 0usize;
    // Hash each bigram to a u64 instead of storing (String, String) pairs — keeps memory/time
    // bounded on large corpora. DefaultHasher has fixed keys, so this is deterministic.
    let mut seen: HashSet<u64> = HashSet::new();
    for r in rows {
        let words: Vec<&str> = r.assistant.split_whitespace().collect();
        for w in words.windows(2) {
            total += 1;
            let mut h = std::collections::hash_map::DefaultHasher::new();
            w[0].to_lowercase().hash(&mut h);
            w[1].to_lowercase().hash(&mut h);
            seen.insert(h.finish());
        }
    }
    if total == 0 { return 1.0; }
    seen.len() as f64 / total as f64
}

pub fn max_source_share(stats_path: &Path) -> Option<f64> {
    let text = std::fs::read_to_string(stats_path).ok()?;
    let v: Value = serde_json::from_str(&text).ok()?;
    let sources = v.get("sources")?.as_object()?;
    if sources.is_empty() { return None; }
    let mut totals = Vec::new();
    for val in sources.values() {
        let t: u64 = ["train", "valid", "test"].iter()
            .filter_map(|k| val.get(*k).and_then(|x| x.as_u64()))
            .sum();
        totals.push(t);
    }
    let sum: u64 = totals.iter().sum();
    if sum == 0 { return None; }
    Some(*totals.iter().max().unwrap_or(&0) as f64 / sum as f64)
}

/// One topic's share of the assigned rows, for the report's diagnostic breakdown.
#[derive(Serialize)]
pub struct TopicShare {
    pub label: String,
    pub count: usize,
    pub share: f64,
}

const MAX_CHUNK_CHARS: usize = 3500;

#[derive(Serialize)]
pub struct MetricStatus { pub name: String, pub value: f64, pub threshold: f64, pub ok: bool }

#[derive(Serialize)]
pub struct QualityReport {
    pub method: String,
    pub train_rows: usize,
    pub valid_rows: usize,
    pub test_rows: usize,
    pub total_rows: usize,
    pub malformed_rate: f64,
    pub duplicate_rate: f64,
    pub leakage_rate: f64,
    pub short_rate: f64,
    pub degenerate_rate: f64,
    pub distinct2: f64,
    pub max_source_share: Option<f64>,
    pub length_min: usize,
    pub length_median: usize,
    pub length_p90: usize,
    pub length_max: usize,
    pub long_count: usize,
    pub metrics: Vec<MetricStatus>,
    pub quality_score: u32,
    pub overall: bool,
    pub near_duplicate_rate: Option<f64>,
    pub near_leakage_rate: Option<f64>,
    pub semantic_near_duplicate_rate: Option<f64>,
    pub semantic_leakage_rate: Option<f64>,
    pub max_topic_share: Option<f64>,
    pub topic_distribution: Option<Vec<TopicShare>>,
}

fn clone_rows(rows: &[Row]) -> Vec<Row> {
    rows.iter().map(|r| Row { user: r.user.clone(), assistant: r.assistant.clone() }).collect()
}

/// (semantic_near_duplicate_rate over train, semantic_leakage_rate of held vs train).
pub async fn semantic_rates<E: crate::embed::Embedder>(
    embedder: &E,
    store_dir: &std::path::Path,
    model: &str,
    train_keys: &[String],
    held_keys: &[String],
    threshold: f64,
    batch_size: usize,
) -> std::io::Result<(f64, f64)> {
    let train_vecs = crate::vectors::get_or_embed(embedder, store_dir, model, train_keys, batch_size).await?;
    let nd = crate::simhash::sim_dup_drop_indices(&train_vecs, train_keys, threshold).len() as f64
        / train_keys.len().max(1) as f64;
    let held_vecs = crate::vectors::get_or_embed(embedder, store_dir, model, held_keys, batch_size).await?;
    let nl = crate::simhash::sim_match_count(&train_vecs, &held_vecs, threshold) as f64
        / held_keys.len().max(1) as f64;
    Ok((nd, nl))
}

pub async fn run_eval(repo_root: &Path, _strict: bool) -> std::io::Result<QualityReport> {
    let cfg = load_config(&repo_root.join(crate::config::CONFIG_FILE));
    let eval: EvalConfig = cfg.eval.unwrap_or_default();
    if eval.method != "sft" {
        return Err(std::io::Error::new(std::io::ErrorKind::InvalidInput,
            format!("unsupported eval method '{}': only 'sft' is supported", eval.method)));
    }
    let ds = repo_root.join(eval.dataset_dir.clone().unwrap_or_else(|| cfg.paths.dataset_dir.clone()));
    let t = &eval.thresholds;

    let train = parse_split(&ds.join("train.jsonl"));
    let valid = parse_split(&ds.join("valid.jsonl"));
    let test = parse_split(&ds.join("test.jsonl"));
    let total = train.total + valid.total + test.total;
    let malformed = train.malformed + valid.malformed + test.malformed;
    let malformed_rate = if total == 0 { 0.0 } else { malformed as f64 / total as f64 };

    let mut held = clone_rows(&valid.rows);
    held.extend(clone_rows(&test.rows));
    let mut all = clone_rows(&train.rows);
    all.extend(clone_rows(&valid.rows));
    all.extend(clone_rows(&test.rows));

    let dup = duplicate_rate(&train.rows);
    let leak = leakage_rate(&train.rows, &held);
    let ls = length_stats(&all, t.short_char_min, MAX_CHUNK_CHARS);
    let degen = degenerate_rate(&all, t.short_char_min);
    let div = distinct2(&all);
    let share = max_source_share(&ds.join("stats.json"));

    let mut metrics = vec![
        MetricStatus { name: "min_rows".into(), value: total as f64, threshold: t.min_rows as f64, ok: total >= t.min_rows },
        MetricStatus { name: "malformed_rate".into(), value: malformed_rate, threshold: t.max_malformed_rate, ok: malformed_rate <= t.max_malformed_rate },
        MetricStatus { name: "duplicate_rate".into(), value: dup, threshold: t.max_duplicate_rate, ok: dup <= t.max_duplicate_rate },
        MetricStatus { name: "leakage_rate".into(), value: leak, threshold: t.max_leakage_rate, ok: leak <= t.max_leakage_rate },
        MetricStatus { name: "short_rate".into(), value: ls.short_rate, threshold: t.max_short_rate, ok: ls.short_rate <= t.max_short_rate },
        MetricStatus { name: "degenerate_rate".into(), value: degen, threshold: t.max_degenerate_rate, ok: degen <= t.max_degenerate_rate },
        MetricStatus { name: "distinct2".into(), value: div, threshold: t.min_distinct2, ok: div >= t.min_distinct2 },
    ];
    if let Some(s) = share {
        metrics.push(MetricStatus { name: "max_source_share".into(), value: s, threshold: t.max_source_share, ok: s <= t.max_source_share });
    }
    let (train_keys, held_keys): (Vec<String>, Vec<String>) = if eval.near_dup || eval.semantic {
        (
            train.rows.iter().map(|r| norm_key(&r.assistant)).collect(),
            held.iter().map(|r| norm_key(&r.assistant)).collect(),
        )
    } else {
        (Vec::new(), Vec::new())
    };
    let (near_duplicate_rate, near_leakage_rate) = if eval.near_dup {
        let nd = if train_keys.is_empty() { 0.0 } else {
            crate::minhash::near_dup_drop_indices(&train_keys, t.near_dup_threshold, t.near_shingle_size).len() as f64 / train_keys.len() as f64
        };
        let nl = if held_keys.is_empty() { 0.0 } else {
            crate::minhash::near_match_count(&train_keys, &held_keys, t.near_dup_threshold, t.near_shingle_size) as f64 / held_keys.len() as f64
        };
        metrics.push(MetricStatus { name: "near_duplicate_rate".into(), value: nd, threshold: t.max_near_duplicate_rate, ok: nd <= t.max_near_duplicate_rate });
        metrics.push(MetricStatus { name: "near_leakage_rate".into(), value: nl, threshold: t.max_near_leakage_rate, ok: nl <= t.max_near_leakage_rate });
        (Some(nd), Some(nl))
    } else {
        (None, None)
    };
    let (semantic_near_duplicate_rate, semantic_leakage_rate) = if eval.semantic
        && !cfg.understand.embed.base_url.is_empty()
    {
        let store_dir = repo_root.join(&cfg.understand.embed.store);
        let proxy = cfg.network.proxy.as_deref();
        match crate::embed::EndpointEmbedder::new(&cfg.understand.embed, proxy) {
            Ok(embedder) => match semantic_rates(
                &embedder,
                &store_dir,
                &cfg.understand.embed.model,
                &train_keys,
                &held_keys,
                t.semantic_threshold,
                cfg.understand.embed.batch_size,
            )
            .await
            {
                Ok((nd, nl)) => {
                    metrics.push(MetricStatus { name: "semantic_near_duplicate_rate".into(), value: nd, threshold: t.max_semantic_near_dup_rate, ok: nd <= t.max_semantic_near_dup_rate });
                    metrics.push(MetricStatus { name: "semantic_leakage_rate".into(), value: nl, threshold: t.max_semantic_leakage_rate, ok: nl <= t.max_semantic_leakage_rate });
                    (Some(nd), Some(nl))
                }
                Err(e) => {
                    eprintln!("kibble: semantic eval skipped (embed failed): {e}");
                    (None, None)
                }
            },
            Err(e) => {
                eprintln!("kibble: semantic eval skipped (embed init failed): {e}");
                (None, None)
            }
        }
    } else {
        (None, None)
    };
    let (max_topic_share, topic_distribution): (Option<f64>, Option<Vec<TopicShare>>) = {
        let clusters_path = repo_root.join(&cfg.cluster.out);
        if clusters_path.is_file()
            && !cfg.understand.embed.base_url.is_empty()
            && !train.rows.is_empty()
        {
            match crate::cluster::load_clusters(&clusters_path, &cfg.understand.embed.model) {
                Some(cr) => {
                    let store_dir = repo_root.join(&cfg.understand.embed.store);
                    let raw_answers: Vec<String> = train.rows.iter().map(|r| r.assistant.clone()).collect();
                    match crate::embed::EndpointEmbedder::new(&cfg.understand.embed, cfg.network.proxy.as_deref()) {
                        Ok(embedder) => match crate::vectors::get_or_embed(
                            &embedder, &store_dir, &cfg.understand.embed.model, &raw_answers, cfg.understand.embed.batch_size,
                        ).await {
                            Ok(vecs) => {
                                if vecs.is_empty() || cr.centroids[0].len() != vecs[0].len() {
                                    if !vecs.is_empty() {
                                        eprintln!("kibble: topic-balance skipped — centroid dim {} != embedding dim {}", cr.centroids[0].len(), vecs[0].len());
                                    }
                                    (None, None)
                                } else {
                                    let assign = crate::cluster::assign_topics(&cr.centroids, &vecs);
                                    let mut counts = vec![0usize; cr.centroids.len()];
                                    for &tpc in &assign { counts[tpc] += 1; }
                                    let n = assign.len() as f64;
                                    let mut dist: Vec<TopicShare> = counts.iter().enumerate().map(|(i, &ct)| TopicShare {
                                        label: cr.labels.get(i).cloned().unwrap_or_else(|| format!("topic-{i}")),
                                        count: ct,
                                        share: ct as f64 / n,
                                    }).collect();
                                    dist.sort_by(|a, b| b.count.cmp(&a.count).then(a.label.cmp(&b.label)));
                                    let mts = dist.first().map(|d| d.share).unwrap_or(0.0);
                                    metrics.push(MetricStatus {
                                        name: "max_topic_share".into(), value: mts,
                                        threshold: t.max_topic_share, ok: mts <= t.max_topic_share,
                                    });
                                    (Some(mts), Some(dist))
                                }
                            }
                            Err(e) => { eprintln!("kibble: topic-balance skipped (embed failed): {e}"); (None, None) }
                        },
                        Err(e) => { eprintln!("kibble: topic-balance skipped (embed init failed): {e}"); (None, None) }
                    }
                }
                None => (None, None),
            }
        } else {
            (None, None)
        }
    };
    let ok_count = metrics.iter().filter(|m| m.ok).count();
    let overall = ok_count == metrics.len();
    let quality_score = (100.0 * ok_count as f64 / metrics.len() as f64).round() as u32;

    let report = QualityReport {
        method: eval.method, train_rows: train.rows.len(), valid_rows: valid.rows.len(),
        test_rows: test.rows.len(), total_rows: total, malformed_rate, duplicate_rate: dup,
        leakage_rate: leak, short_rate: ls.short_rate, degenerate_rate: degen, distinct2: div,
        max_source_share: share, length_min: ls.min, length_median: ls.median, length_p90: ls.p90,
        length_max: ls.max, long_count: ls.long_count, metrics, quality_score, overall,
        near_duplicate_rate, near_leakage_rate,
        semantic_near_duplicate_rate, semantic_leakage_rate,
        max_topic_share,
        topic_distribution,
    };

    let out_dir = repo_root.join(&eval.out);
    std::fs::create_dir_all(&out_dir)?;
    std::fs::write(out_dir.join("report.json"), serde_json::to_string_pretty(&report)
        .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?)?;
    std::fs::write(out_dir.join("report.md"), render_md(&report))?;
    Ok(report)
}

fn render_md(r: &QualityReport) -> String {
    let mut s = format!("# Dataset quality — {} / 100 — {}\n\n", r.quality_score, if r.overall { "PASS" } else { "FAIL" });
    s.push_str(&format!("method `{}` · rows: train {} / valid {} / test {} (total {})\n\n",
        r.method, r.train_rows, r.valid_rows, r.test_rows, r.total_rows));
    s.push_str("| metric | value | threshold | status |\n|---|---|---|---|\n");
    for m in &r.metrics {
        s.push_str(&format!("| {} | {:.4} | {:.4} | {} |\n", m.name, m.value, m.threshold, if m.ok { "ok" } else { "FAIL" }));
    }
    let issues: Vec<&str> = r.metrics.iter().filter(|m| !m.ok).map(|m| m.name.as_str()).collect();
    if issues.is_empty() {
        s.push_str("\nNo issues — dataset is training-ready.\n");
    } else {
        s.push_str(&format!("\n## Top issues\n{}\n", issues.iter().map(|i| format!("- {i}")).collect::<Vec<_>>().join("\n")));
    }
    s.push_str(&format!("\nlength chars: min {} · median {} · p90 {} · max {} · over-{} {}\n",
        r.length_min, r.length_median, r.length_p90, r.length_max, MAX_CHUNK_CHARS, r.long_count));
    if let (Some(nd), Some(nl)) = (r.near_duplicate_rate, r.near_leakage_rate) {
        s.push_str(&format!("\nnear-dup: duplicate {nd:.4} · leakage {nl:.4}\n"));
    }
    if let (Some(nd), Some(nl)) = (r.semantic_near_duplicate_rate, r.semantic_leakage_rate) {
        s.push_str(&format!("\nsemantic: duplicate {nd:.4} · leakage {nl:.4}\n"));
    }
    if let Some(dist) = &r.topic_distribution {
        s.push_str("\n## Topics\n");
        for d in dist {
            s.push_str(&format!("- {} — {:.1}% ({} rows)\n", d.label, d.share * 100.0, d.count));
        }
    }
    if let Some(mts) = r.max_topic_share {
        s.push_str(&format!("\nmax topic share: {mts:.4}\n"));
    }
    s
}

#[cfg(test)]
mod tests {
    use super::*;

    fn qrow(u: &str, a: &str) -> Row { Row { user: u.into(), assistant: a.into() } }

    #[test]
    fn q_norm_key_normalizes() {
        assert_eq!(norm_key("  Hello   World \n"), "hello world");
        assert_eq!(norm_key("A\tB"), "a b");
    }

    #[test]
    fn q_parse_split_counts_valid_and_malformed() {
        let dir = std::env::temp_dir().join(format!("kibble_ps_{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&dir); std::fs::create_dir_all(&dir).unwrap();
        let p = dir.join("train.jsonl");
        let lines = [
            r#"{"messages":[{"role":"system","content":"s"},{"role":"user","content":"q"},{"role":"assistant","content":"a"}]}"#,
            r#"{"messages":[{"role":"user","content":"q2"},{"role":"assistant","content":"a2"}]}"#,
            r#"{"messages":[{"role":"user","content":"only user"}]}"#,
            r#"{"messages":[{"role":"user","content":"q"},{"role":"assistant","content":""}]}"#,
            "not json",
        ].join("\n");
        std::fs::write(&p, lines).unwrap();
        let sp = parse_split(&p);
        assert_eq!(sp.total, 5);
        assert_eq!(sp.rows.len(), 2);
        assert_eq!(sp.malformed, 3);
        assert_eq!(sp.rows[0].assistant, "a");
        assert_eq!(parse_split(&dir.join("nope.jsonl")).total, 0);
    }

    #[test]
    fn q_duplicate_and_leakage() {
        let rows = vec![qrow("q","Same Answer"), qrow("q","same   answer"), qrow("q","other")];
        assert!((duplicate_rate(&rows) - (1.0/3.0)).abs() < 1e-9);
        assert_eq!(duplicate_rate(&[]), 0.0);
        let train = vec![qrow("q","leaked answer"), qrow("q","train only")];
        let held = vec![qrow("q","Leaked   Answer"), qrow("q","held only")];
        assert!((leakage_rate(&train, &held) - 0.5).abs() < 1e-9);
        assert_eq!(leakage_rate(&train, &[]), 0.0);
    }

    #[test]
    fn q_length_degen_diversity() {
        let rows = vec![qrow("q","tiny"), qrow("q",&"x".repeat(100)), qrow("q",&"y".repeat(50))];
        let ls = length_stats(&rows, 20, 3500);
        assert_eq!(ls.min, 4);
        assert_eq!(ls.max, 100);
        assert!((ls.short_rate - (1.0/3.0)).abs() < 1e-9);
        assert_eq!(ls.long_count, 0);
        let degs = vec![
            qrow("a normal user turn that is fairly long","a perfectly fine sufficiently long assistant answer"),
            qrow("echo me back exactly as i wrote it","echo me back exactly as i wrote it"),
            qrow("q","ok"),
        ];
        assert!((degenerate_rate(&degs, 20) - (2.0/3.0)).abs() < 1e-9);
        let templated = vec![qrow("q","same words over and over"), qrow("q","same words over and over")];
        let varied = vec![qrow("q","alpha beta gamma delta epsilon"), qrow("q","one two three four five")];
        assert!(distinct2(&varied) > distinct2(&templated));
    }

    #[test]
    fn distinct2_exact_and_deterministic() {
        // "a b a b" → bigrams (a,b),(b,a),(a,b): 3 total, 2 distinct → 2/3; stable across runs
        let rows = vec![qrow("q", "a b a b")];
        let v = distinct2(&rows);
        assert!((v - (2.0 / 3.0)).abs() < 1e-9, "got {v}");
        assert_eq!(distinct2(&rows), v);
        assert_eq!(distinct2(&[]), 1.0);
    }

    #[test]
    fn q_max_source_share() {
        let dir = std::env::temp_dir().join(format!("kibble_bal_{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&dir); std::fs::create_dir_all(&dir).unwrap();
        let sp = dir.join("stats.json");
        std::fs::write(&sp, r#"{"sources":{"big":{"train":60,"valid":10,"test":10},"small":{"train":10,"valid":5,"test":5}}}"#).unwrap();
        assert!((max_source_share(&sp).unwrap() - 0.8).abs() < 1e-9);
        assert!(max_source_share(&dir.join("nope.json")).is_none());
        std::fs::write(&sp, r#"{"total_documents":5}"#).unwrap();
        assert!(max_source_share(&sp).is_none());
    }

    fn write_split(dir: &std::path::Path, name: &str, rows: &[(&str,&str)]) {
        let body: String = rows.iter().map(|(u,a)| format!(
            r#"{{"messages":[{{"role":"user","content":{}}},{{"role":"assistant","content":{}}}]}}"#,
            serde_json::to_string(u).unwrap(), serde_json::to_string(a).unwrap())).collect::<Vec<_>>().join("\n");
        std::fs::write(dir.join(name), body).unwrap();
    }

    #[tokio::test]
    async fn semantic_rates_flags_leakage() {
        use crate::embed::Embedder;
        struct FirstWord;
        impl Embedder for FirstWord {
            async fn embed_batch(&self, texts: &[String]) -> std::io::Result<Vec<Vec<f32>>> {
                Ok(texts.iter().map(|t| {
                    let w = t.split_whitespace().next().unwrap_or("");
                    let mut h: u64 = 0xcbf29ce484222325;
                    for b in w.bytes() { h ^= b as u64; h = h.wrapping_mul(0x100000001b3); }
                    (0..8).map(|i| ((h >> (i * 8)) & 0xff) as f32).collect()
                }).collect())
            }
        }
        let dir = std::env::temp_dir().join(format!("kibble_semeval_{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&dir);
        let train = vec!["apple pie".to_string(), "banana bread".to_string()];
        let held = vec!["apple orchard".to_string(), "cherry cake".to_string()]; // "apple" leaks
        let (nd, nl) = semantic_rates(&FirstWord, &dir, "m", &train, &held, 0.99, 64).await.unwrap();
        assert_eq!(nd, 0.0);       // train has no internal dup
        assert_eq!(nl, 0.5);       // 1 of 2 held rows matches a train row
        std::fs::remove_dir_all(&dir).ok();
    }

    #[tokio::test]
    async fn run_eval_near_dup_flags_leakage() {
        let base = std::env::temp_dir().join(format!("kibble_neval_{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&base);
        let ds = base.join("data/datasets/unsloth");
        std::fs::create_dir_all(&ds).unwrap();
        // 50 distinct long train rows + one extra train row that a valid row paraphrases
        let mut train: Vec<(String,String)> = (0..50).map(|i| (format!("q{i}"), format!("distinct training answer number {i} alpha beta gamma delta epsilon zeta eta theta iota number {i}"))).collect();
        train.push(("qx".into(), "the quick brown fox jumps over the lazy dog in the meadow at dawn each and every morning".into()));
        let tr: Vec<(&str,&str)> = train.iter().map(|(u,a)| (u.as_str(), a.as_str())).collect();
        write_split(&ds, "train.jsonl", &tr);
        write_split(&ds, "valid.jsonl", &[("qy","the quick brown fox jumps over the lazy dog in the meadow at dusk each and every morning")]); // near-paraphrase of qx
        write_split(&ds, "test.jsonl", &[]);
        std::fs::write(base.join(crate::config::CONFIG_FILE),
            "[paths]\ndataset_dir=\"data/datasets/unsloth\"\n[eval]\nnear_dup=true\n[eval.thresholds]\nnear_dup_threshold=0.6\n").unwrap();

        let rep = run_eval(&base, false).await.unwrap();
        assert!(rep.near_leakage_rate.unwrap() > 0.0, "valid paraphrase should near-leak");
        assert!(!rep.overall, "near-leakage (max 0.0) should FAIL");

        // near_dup off → fields absent, and this dataset (exact-clean) PASSes near-dup-wise
        std::fs::write(base.join(crate::config::CONFIG_FILE), "[paths]\ndataset_dir=\"data/datasets/unsloth\"\n").unwrap();
        let rep2 = run_eval(&base, false).await.unwrap();
        assert!(rep2.near_leakage_rate.is_none());
    }

    #[tokio::test]
    async fn q_run_eval_passes_clean_fails_leaky() {
        let base = std::env::temp_dir().join(format!("kibble_re_{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&base);
        let ds = base.join("data/datasets/unsloth");
        std::fs::create_dir_all(&ds).unwrap();
        // Build answers that are fully unique so bigrams don't repeat much → high distinct2.
        // Use a unique number pair per row so every bigram is unique.
        let train: Vec<(String,String)> = (0..60usize).map(|i| {
            let u = format!("what is item number {i}");
            // Each answer is entirely unique text: row index appears in every adjacent pair,
            // making all bigrams globally distinct across rows.
            let a = format!("row{i}start row{i}word1 row{i}word2 row{i}word3 row{i}word4 row{i}word5 row{i}word6 row{i}end");
            (u, a)
        }).collect();
        let tr: Vec<(&str,&str)> = train.iter().map(|(u,a)| (u.as_str(), a.as_str())).collect();
        write_split(&ds, "train.jsonl", &tr);
        write_split(&ds, "valid.jsonl", &[("a fresh held-out question here","a fresh held out answer alpha beta gamma delta epsilon")]);
        write_split(&ds, "test.jsonl", &[]);
        std::fs::write(base.join(crate::config::CONFIG_FILE), "[paths]\ndataset_dir=\"data/datasets/unsloth\"\n").unwrap();

        let rep = run_eval(&base, false).await.unwrap();
        assert!(rep.overall, "clean dataset should PASS (score {})", rep.quality_score);
        assert!(base.join("eval/report/report.json").is_file());
        assert!(base.join("eval/report/report.md").is_file());

        write_split(&ds, "valid.jsonl", &[(tr[0].0, tr[0].1)]); // leak a train answer into valid
        let rep2 = run_eval(&base, false).await.unwrap();
        assert!(!rep2.overall, "leakage should FAIL");
        assert!(rep2.leakage_rate > 0.0);
    }

    #[tokio::test]
    async fn run_eval_semantic_failsoft_on_unreachable_endpoint() {
        let base = std::env::temp_dir().join(format!("kibble_semfs_{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&base);
        let ds = base.join("data/datasets/unsloth");
        std::fs::create_dir_all(&ds).unwrap();
        write_split(&ds, "train.jsonl", &[("q0", "a small distinct training answer"), ("q1", "another distinct training answer")]);
        write_split(&ds, "valid.jsonl", &[("q2", "a fresh held-out answer")]);
        write_split(&ds, "test.jsonl", &[]);
        std::fs::write(
            base.join(crate::config::CONFIG_FILE),
            "[paths]\ndataset_dir=\"data/datasets/unsloth\"\n[eval]\nsemantic=true\n[understand.embed]\nbase_url=\"http://127.0.0.1:1/v1\"\n",
        )
        .unwrap();

        let rep = run_eval(&base, false).await.unwrap();
        assert!(rep.semantic_near_duplicate_rate.is_none(), "unreachable embed endpoint should fail soft, not error");
        assert!(rep.semantic_leakage_rate.is_none());
        assert!(rep.total_rows > 0, "eval should still complete and produce a report");

        std::fs::remove_dir_all(&base).ok();
    }

    #[tokio::test]
    async fn run_eval_topic_balance_absent_without_clusters_json() {
        let base = std::env::temp_dir().join(format!("kibble_tbfs_{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&base);
        let ds = base.join("data/datasets/unsloth");
        std::fs::create_dir_all(&ds).unwrap();
        write_split(&ds, "train.jsonl", &[("q0", "a small distinct training answer"), ("q1", "another distinct training answer")]);
        write_split(&ds, "valid.jsonl", &[("q2", "a fresh held-out answer")]);
        write_split(&ds, "test.jsonl", &[]);
        // base_url is set (non-empty) but there is NO data/clusters.json → the
        // clusters-file guard short-circuits before any network call.
        std::fs::write(
            base.join(crate::config::CONFIG_FILE),
            "[paths]\ndataset_dir=\"data/datasets/unsloth\"\n[understand.embed]\nbase_url=\"http://127.0.0.1:1/v1\"\n",
        ).unwrap();

        let rep = run_eval(&base, false).await.unwrap();
        assert!(rep.max_topic_share.is_none(), "no clusters.json → metric absent");
        assert!(rep.topic_distribution.is_none());
        assert!(!rep.metrics.iter().any(|m| m.name == "max_topic_share"), "metric must not be pushed");
        assert!(rep.total_rows > 0, "eval still completes");

        std::fs::remove_dir_all(&base).ok();
    }
}