kibble 0.1.0

use crate::catalog::{classify, classify_with_fallback, load_overrides, load_taxonomy, CatalogEntry};
use crate::codebase::scan_codebase;
pub(crate) use crate::clean::clean_text;
use crate::config::{detect_type, load_config, SourceType};
use crate::corpus::{load_raw_docs, Document};
use crate::dataset::load_dataset_rows;
use crate::git::{cache_slug, clone_repo};
use crate::net::resolve_proxy;
use crate::text::{chunk_text, pick_template, split_bucket, topic_from_title};
use serde::Serialize;
use std::collections::{BTreeMap, HashSet};
use std::path::Path;

pub const SYSTEM_PROMPT: &str = "You are a knowledgeable assistant. Answer accurately and concisely, with clear structure and no filler.";

pub const USER_PROMPT_TEMPLATES: &[&str] = &[
    "Explain the following clearly: {topic}",
    "Write a well-structured overview of: {topic}",
    "Summarize the key points about: {topic}",
    "Give a detailed explanation of: {topic}",
];

pub const REFERENCE_USER_TEMPLATES: &[&str] = &[
    "Explain the following passage in context: {topic}",
    "Summarize this text accurately: {topic}",
];

pub const MAX_CHUNK_CHARS: usize = 3500;
pub const MIN_LONGFORM_CHARS: usize = 120;

fn collect_text_files(dir: &std::path::Path, out: &mut Vec<std::path::PathBuf>) -> std::io::Result<()> {
    for entry in std::fs::read_dir(dir)? {
        let path = entry?.path();
        if path.is_dir() {
            collect_text_files(&path, out)?;
        } else if matches!(path.extension().and_then(|e| e.to_str()), Some("txt") | Some("md")) {
            out.push(path);
        }
    }
    Ok(())
}

#[derive(Serialize, Debug, PartialEq)]
pub struct Msg {
    pub role: String,
    pub content: String,
}

#[derive(Serialize, Debug, PartialEq)]
pub struct Row {
    pub messages: Vec<Msg>,
}

fn make_row(user: String, assistant: String) -> Row {
    Row {
        messages: vec![
            Msg { role: "system".to_string(), content: SYSTEM_PROMPT.to_string() },
            Msg { role: "user".to_string(), content: user },
            Msg { role: "assistant".to_string(), content: assistant },
        ],
    }
}

pub fn doc_to_rows(doc: &Document) -> Vec<Row> {
    let templates: &[&str] = if doc.is_reference {
        REFERENCE_USER_TEMPLATES
    } else {
        USER_PROMPT_TEMPLATES
    };

    if doc.source == "twitter" {
        if doc.text.chars().count() < 20 {
            return vec![];
        }
        let topic: String = doc.text.chars().take(200).collect::<String>().replace('\n', " ");
        let user = pick_template(&doc.doc_id, 0, templates).replace("{topic}", &topic);
        return vec![make_row(user, doc.text.clone())];
    }

    let mut rows = Vec::new();
    for (i, chunk) in chunk_text(&doc.text, MAX_CHUNK_CHARS).into_iter().enumerate() {
        if chunk.chars().count() < MIN_LONGFORM_CHARS {
            continue;
        }
        let topic = topic_from_title(&doc.title);
        let user = pick_template(&doc.doc_id, i, templates).replace("{topic}", &topic);
        rows.push(make_row(user, chunk));
    }
    rows
}


#[derive(Serialize)]
pub struct BuildStats {
    pub train: usize,
    pub valid: usize,
    pub test: usize,
    pub total_documents: usize,
    pub sources: Vec<(String, usize, usize, usize)>,
    pub dropped_duplicates: usize,
    pub dropped_filtered: usize,
    pub dropped_near_duplicates: usize,
    pub dropped_semantic_duplicates: usize,
    pub dropped_topic_rebalanced: usize,
}

fn write_split(path: &Path, clean_rows: &[Row], clean_sources: &[String], raw_rows: &[Row], raw_sources: &[String]) -> std::io::Result<usize> {
    let mut out = String::new();
    let mut n = 0;
    for row in clean_rows {
        // Apply clean_text to corpus/dataset rows at write time.
        let cleaned = Row {
            messages: row
                .messages
                .iter()
                .map(|m| Msg { role: m.role.clone(), content: clean_text(&m.content) })
                .collect(),
        };
        let json = serde_json::to_string(&cleaned).expect("serialize row");
        out.push_str(&escape_non_ascii(&json));
        out.push('\n');
        n += 1;
    }
    for row in raw_rows {
        // Code rows: no clean_text — preserve $VAR, URLs, r/paths verbatim.
        let json = serde_json::to_string(row).expect("serialize row");
        out.push_str(&escape_non_ascii(&json));
        out.push('\n');
        n += 1;
    }
    std::fs::write(path, out)?;
    // Sidecar provenance manifest: one source per line, in write order (clean then raw).
    let mut manifest = String::new();
    for s in clean_sources.iter().chain(raw_sources.iter()) {
        manifest.push_str(s);
        manifest.push('\n');
    }
    std::fs::write(path.with_extension("sources.jsonl"), manifest)?;
    Ok(n)
}

/// Match Python json.dumps(ensure_ascii=True): escape any non-ASCII scalar as \uXXXX.
fn escape_non_ascii(s: &str) -> String {
    let mut out = String::with_capacity(s.len());
    for ch in s.chars() {
        if ch.is_ascii() {
            out.push(ch);
        } else {
            let mut buf = [0u16; 2];
            for unit in ch.encode_utf16(&mut buf) {
                out.push_str(&format!("\\u{:04x}", unit));
            }
        }
    }
    out
}

pub struct RowWithMeta {
    pub row: Row,
    pub doc_id: String,
    pub source: String,
    pub raw: bool,
}

#[derive(Default)]
pub struct SplitRows {
    pub clean: Vec<Row>,
    pub raw: Vec<Row>,
    pub clean_sources: Vec<String>,
    pub raw_sources: Vec<String>,
    pub clean_doc_ids: Vec<String>,
    pub raw_doc_ids: Vec<String>,
}

pub struct CuratedSplits {
    pub train: SplitRows,
    pub valid: SplitRows,
    pub test: SplitRows,
    pub sources: Vec<(String, usize, usize, usize)>,
    pub dropped_duplicates: usize,
    pub dropped_filtered: usize,
    pub dropped_near_duplicates: usize,
}

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

pub(crate) fn last_content<'a>(row: &'a Row, role: &str) -> &'a str {
    row.messages.iter().rev().find(|m| m.role == role).map(|m| m.content.as_str()).unwrap_or("")
}

pub fn is_valid_sft(row: &Row) -> bool {
    if row.messages.is_empty() { return false; }
    let (mut has_user, mut has_asst) = (false, false);
    for m in &row.messages {
        if !matches!(m.role.as_str(), "system" | "user" | "assistant") || m.content.trim().is_empty() {
            return false;
        }
        if m.role == "user" { has_user = true; }
        if m.role == "assistant" { has_asst = true; }
    }
    has_user && has_asst
}

pub fn is_degenerate(row: &Row, min_chars: usize, raw: bool) -> bool {
    let raw_a = last_content(row, "assistant");
    let a = if raw { raw_a.to_string() } else { clean_text(raw_a) };
    let at = a.trim();
    if at.is_empty() || at.chars().count() < min_chars {
        return true;
    }
    let raw_u = last_content(row, "user");
    let u = if raw { raw_u.to_string() } else { clean_text(raw_u) };
    norm_answer(&a) == norm_answer(&u)
}

/// The dedup/split key: last assistant content, cleaned for clean rows (to match what `write_split`
/// writes and `eval` reads) / verbatim for raw rows, then normalized.
pub fn answer_key(row: &Row, raw: bool) -> String {
    let a = row.messages.iter().rev()
        .find(|m| m.role == "assistant")
        .map(|m| m.content.clone())
        .unwrap_or_default();
    let content = if raw { a } else { clean_text(&a) };
    norm_answer(&content)
}

pub fn curate_split(rows: Vec<RowWithMeta>, cfg: &crate::config::CurateConfig) -> CuratedSplits {
    // Pass A: filter + exact dedup → survivors (with their answer keys).
    let mut seen: HashSet<String> = HashSet::new();
    let mut dropped_filtered = 0usize;
    let mut dropped_duplicates = 0usize;
    let mut survivors: Vec<(RowWithMeta, String)> = Vec::new();
    for rm in rows {
        if cfg.drop_malformed && !is_valid_sft(&rm.row) { dropped_filtered += 1; continue; }
        if cfg.drop_degenerate && is_degenerate(&rm.row, cfg.min_answer_chars, rm.raw) { dropped_filtered += 1; continue; }
        let key = answer_key(&rm.row, rm.raw);
        if cfg.dedup && !key.is_empty() && !seen.insert(key.clone()) { dropped_duplicates += 1; continue; }
        survivors.push((rm, key));
    }

    // Optional near-dedup over the survivors' answer keys.
    let mut dropped_near_duplicates = 0usize;
    if cfg.near_dedup {
        let keys: Vec<String> = survivors.iter().map(|(_, k)| k.clone()).collect();
        let drop = crate::minhash::near_dup_drop_indices(&keys, cfg.near_dedup_threshold, cfg.shingle_size);
        dropped_near_duplicates = drop.len();
        let mut idx = 0usize;
        survivors.retain(|_| { let keep = !drop.contains(&idx); idx += 1; keep });
    }

    // Pass B: split the survivors.
    let (mut train, mut valid, mut test) = (SplitRows::default(), SplitRows::default(), SplitRows::default());
    let mut counts: BTreeMap<String, (usize, usize, usize)> = BTreeMap::new();
    for (rm, key) in survivors {
        let bucket_input = if cfg.leakage_safe_split && !key.is_empty() { key.as_str() } else { rm.doc_id.as_str() };
        let b = split_bucket(bucket_input);
        let entry = counts.entry(rm.source.clone()).or_insert((0, 0, 0));
        let slot = if b < 78 { entry.0 += 1; &mut train }
                   else if b < 90 { entry.1 += 1; &mut valid }
                   else { entry.2 += 1; &mut test };
        let doc_id = rm.doc_id.clone();
        if rm.raw {
            slot.raw.push(rm.row);
            slot.raw_sources.push(rm.source);
            slot.raw_doc_ids.push(doc_id);
        } else {
            slot.clean.push(rm.row);
            slot.clean_sources.push(rm.source);
            slot.clean_doc_ids.push(doc_id);
        }
    }
    let sources = counts.into_iter().map(|(n, (t, v, te))| (n, t, v, te)).collect();
    CuratedSplits { train, valid, test, sources, dropped_duplicates, dropped_filtered, dropped_near_duplicates }
}

/// Remove the given combined-index drops from a train split, keeping rows and their
/// source manifest entries in lockstep. Indices `< n_clean` address `clean`; indices
/// `>= n_clean` address `raw` at `idx - n_clean` (matching `write_split`'s clean-then-raw order).
fn apply_train_drops(train: &mut SplitRows, drops: &[usize], n_clean: usize) {
    let drop_set: std::collections::HashSet<usize> = drops.iter().copied().collect();
    let mut i = 0usize;
    train.clean.retain(|_| { let keep = !drop_set.contains(&i); i += 1; keep });
    let mut ci = 0usize;
    train.clean_sources.retain(|_| { let keep = !drop_set.contains(&ci); ci += 1; keep });
    let mut j = n_clean;
    train.raw.retain(|_| { let keep = !drop_set.contains(&j); j += 1; keep });
    let mut cj = n_clean;
    train.raw_sources.retain(|_| { let keep = !drop_set.contains(&cj); cj += 1; keep });
    let mut di = 0usize;
    train.clean_doc_ids.retain(|_| { let keep = !drop_set.contains(&di); di += 1; keep });
    let mut dj = n_clean;
    train.raw_doc_ids.retain(|_| { let keep = !drop_set.contains(&dj); dj += 1; keep });
}

/// Embed `texts` and return indices to drop as semantic near-duplicates
/// (keeps the longest text per cluster).
pub async fn semantic_dedup_drop<E: crate::embed::Embedder>(
    embedder: &E,
    store_dir: &std::path::Path,
    model: &str,
    texts: &[String],
    threshold: f64,
    batch_size: usize,
) -> std::io::Result<std::collections::HashSet<usize>> {
    let vecs = crate::vectors::get_or_embed(embedder, store_dir, model, texts, batch_size).await?;
    Ok(crate::simhash::sim_dup_drop_indices(&vecs, texts, threshold))
}

pub async fn run_build(repo_root: &Path) -> std::io::Result<BuildStats> {
    let cfg = load_config(&repo_root.join(crate::config::CONFIG_FILE));
    let data_root = repo_root.join(&cfg.paths.data_root);
    let dataset_dir = repo_root.join(&cfg.paths.dataset_dir);
    // Local ingest pile. Prefer the neutral "raw/local"; fall back to the legacy "raw/me".
    let raw_root = {
        let local = data_root.join("raw/local");
        if local.is_dir() { local } else { data_root.join("raw/me") }
    };
    let taxonomy = load_taxonomy(&data_root.join("catalog/taxonomy.yaml"));
    let overrides = load_overrides(&data_root.join("catalog/overrides.json"));

    let docs = load_raw_docs(&raw_root)?;

    let mut all_rows: Vec<RowWithMeta> = Vec::new();
    let mut entries: Vec<CatalogEntry> = Vec::new();
    for doc in &docs {
        let entry = classify(doc, &taxonomy, &overrides);
        for row in doc_to_rows(doc) {
            all_rows.push(RowWithMeta { row, doc_id: doc.doc_id.clone(), source: doc.source.clone(), raw: false });
        }
        entries.push(entry);
    }

    // Fold configured sources.
    let proxy = resolve_proxy(cfg.network.proxy.as_deref(), |k| std::env::var(k).ok());
    let client = crate::net::build_client(proxy.as_deref())
        .map_err(|e| std::io::Error::other(format!("http client: {e}")))?;
    let mut source_meta: Vec<(String, String, usize)> = Vec::new(); // (name, resolved_path, skipped)
    for source in &cfg.source {
        match detect_type(&source.path, source.r#type.as_deref()) {
            SourceType::Dataset => {
                // Resolve relative paths against repo_root so kibble.toml paths work.
                let abs_path = {
                    let p = std::path::Path::new(&source.path);
                    if p.is_absolute() { source.path.clone() } else { repo_root.join(p).to_string_lossy().into_owned() }
                };
                let mut resolved_source = source.clone();
                resolved_source.path = abs_path;
                let load = load_dataset_rows(&resolved_source)
                    .map_err(|e| std::io::Error::new(e.kind(), format!("source '{}': {e}", source.source_name())))?;
                source_meta.push((source.source_name(), load.resolved_path.display().to_string(), load.skipped));
                entries.extend(synth_entries_from_rows(&load.rows, &source.source_name(), None, &taxonomy, &overrides));
                let raw = source.preserve_code.unwrap_or(false);
                for (id, row) in load.rows {
                    all_rows.push(RowWithMeta { row, doc_id: id, source: source.source_name(), raw });
                }
            }
            SourceType::Codebase => {
                let lower = source.path.to_lowercase();
                let scan_dir = if lower.starts_with("http://") || lower.starts_with("https://") || lower.starts_with("file://") {
                    let dest = repo_root.join(".kibble-cache").join(cache_slug(&source.path));
                    let cached = dest.exists();
                    clone_repo(&source.path, &dest, proxy.as_deref())
                        .map_err(|e| std::io::Error::new(e.kind(), format!("source '{}': {e}", source.source_name())))?;
                    // refresh = true updates an already-cached clone via git pull.
                    if cached && source.refresh.unwrap_or(false) {
                        crate::git::pull_repo(&dest, proxy.as_deref())
                            .map_err(|e| std::io::Error::new(e.kind(), format!("source '{}': {e}", source.source_name())))?;
                    }
                    dest
                } else {
                    let p = std::path::Path::new(&source.path);
                    if p.is_absolute() { p.to_path_buf() } else { repo_root.join(p) }
                };
                let load = scan_codebase(source, &scan_dir)
                    .map_err(|e| std::io::Error::new(e.kind(), format!("source '{}': {e}", source.source_name())))?;
                source_meta.push((source.source_name(), scan_dir.display().to_string(), 0));
                entries.extend(synth_entries_from_rows(&load.rows, &source.source_name(), Some("code"), &taxonomy, &overrides));
                for (id, row) in load.rows {
                    all_rows.push(RowWithMeta { row, doc_id: id, source: source.source_name(), raw: true });
                }
            }
            SourceType::Web => {
                let html = crate::web::fetch_url(&client, &source.path)
                    .await
                    .map_err(|e| std::io::Error::new(e.kind(), format!("source '{}': {e}", source.source_name())))?;
                let text = crate::web::extract_main_text(&html);
                let name = source.source_name();
                let doc = crate::corpus::Document {
                    doc_id: format!("{name}:{}", crate::git::cache_slug(&source.path)),
                    text,
                    source: name.clone(),
                    title: name.clone(),
                    is_reference: false,
                };
                for row in doc_to_rows(&doc) {
                    all_rows.push(RowWithMeta { row, doc_id: doc.doc_id.clone(), source: name.clone(), raw: false });
                }
                entries.push(classify(&doc, &taxonomy, &overrides));
                source_meta.push((name, source.path.clone(), 0));
            }
            SourceType::Files => {
                let dir = {
                    let p = std::path::Path::new(&source.path);
                    if p.is_absolute() { p.to_path_buf() } else { repo_root.join(p) }
                };
                if !dir.is_dir() {
                    return Err(std::io::Error::new(
                        std::io::ErrorKind::NotFound,
                        format!("source '{}': not a directory: {}", source.source_name(), dir.display()),
                    ));
                }
                let name = source.source_name();
                let mut paths = Vec::new();
                collect_text_files(&dir, &mut paths)?;
                for path in paths {
                    let rel = path.strip_prefix(&dir).unwrap_or(&path).to_string_lossy().replace('\\', "/");
                    let text = match std::fs::read_to_string(&path) {
                        Ok(s) => s,
                        Err(e) => { eprintln!("kibble: skipping {}: {e}", path.display()); continue; }
                    };
                    if text.trim().is_empty() { continue; }
                    let doc = crate::corpus::Document {
                        doc_id: format!("{name}:{rel}"),
                        text,
                        source: name.clone(),
                        title: rel.clone(),
                        is_reference: false,
                    };
                    for row in doc_to_rows(&doc) {
                        all_rows.push(RowWithMeta { row, doc_id: doc.doc_id.clone(), source: name.clone(), raw: false });
                    }
                    entries.push(classify(&doc, &taxonomy, &overrides));
                }
                source_meta.push((name, dir.display().to_string(), 0));
            }
            SourceType::Blog => {
                let file = {
                    let p = std::path::Path::new(&source.path);
                    if p.is_absolute() { p.to_path_buf() } else { repo_root.join(p) }
                };
                let html = std::fs::read_to_string(&file).map_err(|e| {
                    std::io::Error::new(e.kind(), format!("source '{}': {e}", source.source_name()))
                })?;
                let text = crate::web::extract_main_text(&html);
                let name = source.source_name();
                let doc = crate::corpus::Document {
                    doc_id: format!("{name}:{}", crate::git::cache_slug(&source.path)),
                    text,
                    source: name.clone(),
                    title: name.clone(),
                    is_reference: false,
                };
                for row in doc_to_rows(&doc) {
                    all_rows.push(RowWithMeta { row, doc_id: doc.doc_id.clone(), source: name.clone(), raw: false });
                }
                entries.push(classify(&doc, &taxonomy, &overrides));
                source_meta.push((name, file.display().to_string(), 0));
            }
        }
    }

    let mut dropped_semantic_duplicates = 0usize;
    if cfg.curate.semantic_dedup && !cfg.understand.embed.base_url.is_empty() {
        let texts: Vec<String> = all_rows.iter().map(|rm| answer_key(&rm.row, rm.raw)).collect();
        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_dedup_drop(
                &embedder,
                &store_dir,
                &cfg.understand.embed.model,
                &texts,
                cfg.curate.semantic_threshold,
                cfg.understand.embed.batch_size,
            )
            .await
            {
                Ok(drop) => {
                    dropped_semantic_duplicates = drop.len();
                    let mut kept = Vec::with_capacity(all_rows.len() - drop.len());
                    for (i, rm) in all_rows.into_iter().enumerate() {
                        if !drop.contains(&i) {
                            kept.push(rm);
                        }
                    }
                    all_rows = kept;
                }
                Err(e) => eprintln!("kibble: semantic dedup skipped (embed failed): {e}"),
            },
            Err(e) => eprintln!("kibble: semantic dedup skipped (embed init failed): {e}"),
        }
    }

    let total_documents = all_rows.iter().map(|r| r.doc_id.as_str()).collect::<HashSet<_>>().len();
    let mut by_source_docs: BTreeMap<String, HashSet<&str>> = BTreeMap::new();
    for r in &all_rows {
        by_source_docs.entry(r.source.clone()).or_default().insert(r.doc_id.as_str());
    }
    let by_source: BTreeMap<String, usize> = by_source_docs.into_iter().map(|(k, v)| (k, v.len())).collect();

    let mut curated = curate_split(all_rows, &cfg.curate);

    let mut dropped_topic_rebalanced = 0usize;
    let mut inline_clusters: Option<crate::cluster::ClusterResult> = None;
    if cfg.cluster.rebalance {
        let n_clean = curated.train.clean.len();
        let mut train_answers: Vec<String> = curated.train.clean.iter()
            .map(|r| clean_text(last_content(r, "assistant"))).collect();
        train_answers.extend(curated.train.raw.iter().map(|r| last_content(r, "assistant").to_string()));
        match crate::rebalance::rebalance_inline(repo_root, &train_answers).await {
            Ok(Some(outcome)) => {
                dropped_topic_rebalanced = outcome.drops.len();
                apply_train_drops(&mut curated.train, &outcome.drops, n_clean);
                inline_clusters = Some(outcome.clusters);
            }
            Ok(None) => {}
            Err(e) => eprintln!("kibble: rebalance skipped: {e}"),
        }
    }

    let ds_dir = dataset_dir.clone();
    std::fs::create_dir_all(&ds_dir)?;
    let n_train = write_split(&ds_dir.join("train.jsonl"), &curated.train.clean, &curated.train.clean_sources, &curated.train.raw, &curated.train.raw_sources)?;
    let n_valid = write_split(&ds_dir.join("valid.jsonl"), &curated.valid.clean, &curated.valid.clean_sources, &curated.valid.raw, &curated.valid.raw_sources)?;
    let n_test = write_split(&ds_dir.join("test.jsonl"), &curated.test.clean, &curated.test.clean_sources, &curated.test.raw, &curated.test.raw_sources)?;

    // Mirror to data_root/.
    for name in ["train.jsonl", "valid.jsonl", "test.jsonl"] {
        std::fs::copy(ds_dir.join(name), data_root.join(name))?;
    }

    // Catalog documents.jsonl + summary.json.
    let cat_dir = data_root.join("catalog");
    std::fs::create_dir_all(&cat_dir)?;
    if cfg.classify.enabled {
        crate::classify::apply_auto_topics(&cfg, repo_root, &cat_dir, &curated, inline_clusters.as_ref(), &mut entries).await?;
    }
    write_catalog(&cat_dir, &entries)?;

    // stats.json.
    let meta_by_name: BTreeMap<String, (String, usize)> = source_meta
        .into_iter().map(|(n, p, s)| (n, (p, s))).collect();
    let sources_json: serde_json::Map<String, serde_json::Value> = curated.sources
        .iter()
        .map(|(name, t, v, te)| {
            let (path, skipped) = meta_by_name.get(name).cloned().unwrap_or_default();
            (name.clone(), serde_json::json!({ "train": t, "valid": v, "test": te, "resolved_path": path, "skipped": skipped }))
        })
        .collect();
    let stats = serde_json::json!({
        "train_examples": n_train,
        "valid_examples": n_valid,
        "test_examples": n_test,
        "total_documents": total_documents,
        "documents_by_source": by_source,
        "dropped_duplicates": curated.dropped_duplicates,
        "dropped_filtered": curated.dropped_filtered,
        "dropped_near_duplicates": curated.dropped_near_duplicates,
        "dropped_semantic_duplicates": dropped_semantic_duplicates,
        "dropped_topic_rebalanced": dropped_topic_rebalanced,
        "sources": sources_json,
    });
    std::fs::write(ds_dir.join("stats.json"), serde_json::to_string_pretty(&stats)?)?;

    if let Some(c) = inline_clusters {
        crate::cluster::write_clusters(repo_root, &cfg.cluster.out, &c)?;
        println!("Clustered {} rows into {} topics -> {} (inline, rebalanced)", c.sizes.iter().sum::<usize>(), c.k, cfg.cluster.out);
    } else if cfg.cluster.enabled && !cfg.understand.embed.base_url.is_empty() {
        match crate::cluster::run_cluster(repo_root, None).await {
            Ok(Some(r)) => println!("Clustered {} rows into {} topics -> {}", r.sizes.iter().sum::<usize>(), r.k, cfg.cluster.out),
            Ok(None) => {}
            Err(e) => eprintln!("kibble: clustering skipped: {e}"),
        }
    }

    Ok(BuildStats { train: n_train, valid: n_valid, test: n_test, total_documents,
                    sources: curated.sources, dropped_duplicates: curated.dropped_duplicates,
                    dropped_filtered: curated.dropped_filtered,
                    dropped_near_duplicates: curated.dropped_near_duplicates,
                    dropped_semantic_duplicates,
                    dropped_topic_rebalanced })
}

fn write_catalog(cat_dir: &Path, entries: &[CatalogEntry]) -> std::io::Result<()> {
    let mut docs_jsonl = String::new();
    let mut by_role: BTreeMap<String, usize> = BTreeMap::new();
    let mut by_bucket: BTreeMap<String, usize> = BTreeMap::new();
    let mut by_topic: BTreeMap<String, usize> = BTreeMap::new();
    for e in entries {
        *by_role.entry(e.role.clone()).or_insert(0) += 1;
        *by_bucket.entry(e.lora_bucket.clone()).or_insert(0) += 1;
        for t in &e.topics {
            *by_topic.entry(t.clone()).or_insert(0) += 1;
        }
        let mut row = serde_json::json!({
            "doc_id": e.doc_id, "source": e.source, "title": e.title,
            "role": e.role, "topics": e.topics, "lora_bucket": e.lora_bucket,
            "rag": e.rag, "is_reference": e.is_reference, "chars": e.chars,
        });
        if let Some(t) = &e.auto_topic {
            row["auto_topic"] = serde_json::json!(t);
        }
        if let Some(c) = e.topic_confidence {
            row["topic_confidence"] = serde_json::json!(c);
        }
        docs_jsonl.push_str(&serde_json::to_string(&row).expect("serialize catalog row"));
        docs_jsonl.push('\n');
    }
    std::fs::write(cat_dir.join("documents.jsonl"), docs_jsonl)?;
    let summary = serde_json::json!({
        "documents": entries.len(), "by_role": by_role,
        "by_lora_bucket": by_bucket, "by_topic": by_topic,
    });
    std::fs::write(cat_dir.join("summary.json"), serde_json::to_string_pretty(&summary)?)?;
    Ok(())
}

/// Group `(doc_id, Row)` pairs by doc_id (first-seen order) and classify one synthetic
/// Document per doc_id (title = doc_id; text = the rows' non-`system` message contents,
/// joined by `\n`). `fallback_role` is the role used when the taxonomy has no source_default
/// for `source` (see `catalog::classify_with_fallback`).
fn synth_entries_from_rows(
    rows: &[(String, Row)],
    source: &str,
    fallback_role: Option<&str>,
    taxonomy: &crate::catalog::Taxonomy,
    overrides: &crate::catalog::Overrides,
) -> Vec<CatalogEntry> {
    let mut order: Vec<String> = Vec::new();
    let mut texts: std::collections::HashMap<String, String> = std::collections::HashMap::new();
    for (doc_id, row) in rows {
        let text = texts.entry(doc_id.clone()).or_insert_with(|| {
            order.push(doc_id.clone());
            String::new()
        });
        for m in &row.messages {
            if m.role == "system" { continue; }
            if !text.is_empty() { text.push('\n'); }
            text.push_str(&m.content);
        }
    }
    order
        .into_iter()
        .map(|doc_id| {
            let text = texts.remove(&doc_id).unwrap_or_default();
            let doc = crate::corpus::Document {
                doc_id: doc_id.clone(),
                text,
                source: source.to_string(),
                title: doc_id,
                is_reference: false,
            };
            classify_with_fallback(&doc, taxonomy, overrides, fallback_role)
        })
        .collect()
}

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

    fn mkrow(assistant: &str) -> Row {
        Row { messages: vec![
            Msg { role: "user".into(), content: "q".into() },
            Msg { role: "assistant".into(), content: assistant.into() },
        ]}
    }
    fn rwm(assistant: &str, doc_id: &str, source: &str, raw: bool) -> RowWithMeta {
        RowWithMeta { row: mkrow(assistant), doc_id: doc_id.into(), source: source.into(), raw }
    }

    #[test]
    fn answer_key_normalizes_and_cleans() {
        // raw=true → verbatim (only normalized); raw=false → clean_text applied
        assert_eq!(answer_key(&mkrow("  Hello   World "), true), "hello world");
    }

    #[test]
    fn curate_dedups_keeps_first() {
        let cfg = crate::config::CurateConfig { dedup: true, leakage_safe_split: true, drop_malformed: false, drop_degenerate: false, ..Default::default() };
        let rows = vec![
            rwm("Same Answer", "d1", "s", false),
            rwm("same   answer", "d2", "s", false), // dup by normalized key
            rwm("different", "d3", "s", false),
        ];
        let c = curate_split(rows, &cfg);
        let total = c.train.clean.len()+c.valid.clean.len()+c.test.clean.len();
        assert_eq!(total, 2);
        assert_eq!(c.dropped_duplicates, 1);
    }

    #[test]
    fn curate_leakage_safe_colocates_same_answer() {
        // dedup off so both survive; same answer + different doc_id must land in the SAME split
        let cfg = crate::config::CurateConfig { dedup: false, leakage_safe_split: true, drop_malformed: false, drop_degenerate: false, ..Default::default() };
        let rows = vec![ rwm("identical answer text", "docA", "s", false),
                         rwm("identical answer text", "docB", "s", false) ];
        let c = curate_split(rows, &cfg);
        let in_train = c.train.clean.len();
        // both in one split → that split has 2, others 0
        assert!(in_train == 2 || c.valid.clean.len() == 2 || c.test.clean.len() == 2);
        assert_eq!(in_train + c.valid.clean.len() + c.test.clean.len(), 2);
    }

    #[test]
    fn curate_raw_partition_and_sources() {
        let cfg = crate::config::CurateConfig { dedup: true, leakage_safe_split: true, drop_malformed: false, drop_degenerate: false, ..Default::default() };
        let rows = vec![ rwm("alpha beta gamma", "d1", "src1", false),
                         rwm("delta epsilon zeta", "d2", "src2", true) ];
        let c = curate_split(rows, &cfg);
        let raw_total = c.train.raw.len()+c.valid.raw.len()+c.test.raw.len();
        let clean_total = c.train.clean.len()+c.valid.clean.len()+c.test.clean.len();
        assert_eq!(raw_total, 1);
        assert_eq!(clean_total, 1);
        let names: Vec<&str> = c.sources.iter().map(|(n,_,_,_)| n.as_str()).collect();
        assert!(names.contains(&"src1") && names.contains(&"src2"));
    }

    fn doc(source: &str, text: &str) -> Document {
        Document {
            doc_id: format!("{source}:x"),
            text: text.to_string(),
            source: source.to_string(),
            title: "my_post".to_string(),
            is_reference: source == "textfile",
        }
    }

    #[test]
    fn twitter_makes_one_row_with_system_user_assistant() {
        let rows = doc_to_rows(&doc("twitter", "this is a long enough tweet body"));
        assert_eq!(rows.len(), 1);
        let m = &rows[0].messages;
        assert_eq!(m.len(), 3);
        assert_eq!(m[0].role, "system");
        assert_eq!(m[0].content, SYSTEM_PROMPT);
        assert_eq!(m[1].role, "user");
        assert_eq!(m[2].role, "assistant");
        assert_eq!(m[2].content, "this is a long enough tweet body");
    }

    #[test]
    fn short_twitter_is_skipped() {
        let rows = doc_to_rows(&doc("twitter", "too short"));
        assert!(rows.is_empty());
    }

    #[test]
    fn reference_uses_reference_templates() {
        let long = "word ".repeat(40); // > 120 chars
        let rows = doc_to_rows(&doc("textfile", &long));
        assert_eq!(rows.len(), 1);
        // user content must come from a REFERENCE template
        let u = &rows[0].messages[1].content;
        assert!(u.contains("passage") || u.contains("Summarize this text"));
    }

    #[tokio::test]
    async fn run_build_folds_configured_dataset() {
        use super::run_build;
        use std::fs;
        let root = std::env::temp_dir().join(format!("kibble_build_ds_{}", std::process::id()));
        let _ = fs::remove_dir_all(&root);
        fs::create_dir_all(root.join("data/raw/me/twitter")).unwrap();
        fs::create_dir_all(root.join("data/catalog")).unwrap();
        fs::write(
            root.join("data/raw/me/twitter/tweet_1.txt"),
            "a sufficiently long tweet body here",
        )
        .unwrap();
        // external dataset + kibble.toml pointing at it
        fs::write(
            root.join("ext.jsonl"),
            r#"{"id":"x","messages":[{"role":"user","content":"hi"},{"role":"assistant","content":"there and back again over the hill"}]}"#,
        )
        .unwrap();
        fs::write(
            root.join(crate::config::CONFIG_FILE),
            "[[source]]\npath = \"ext.jsonl\"\nname = \"ext\"\nsystem_prompt = \"SYS\"\n",
        )
        .unwrap();

        let stats = run_build(&root).await.unwrap();
        // 1 tweet doc + 1 dataset row across the splits
        let total: usize = stats.train + stats.valid + stats.test;
        assert_eq!(total, 2);
        assert!(stats.sources.iter().any(|(n, _, _, _)| n == "ext"));

        let combined = ["data/datasets/unsloth/train.jsonl", "data/datasets/unsloth/valid.jsonl", "data/datasets/unsloth/test.jsonl"]
            .iter()
            .map(|p| fs::read_to_string(root.join(p)).unwrap_or_default())
            .collect::<String>();
        assert!(combined.contains("there"));
        assert!(combined.contains("SYS"));
        // stats.json carries the sources map
        let stats_txt = fs::read_to_string(root.join("data/datasets/unsloth/stats.json")).unwrap();
        assert!(stats_txt.contains("\"sources\""));
        assert!(stats_txt.contains("\"ext\""));
    }

    #[tokio::test]
    async fn run_build_counts_source_documents() {
        use super::run_build;
        use std::fs;
        let root = std::env::temp_dir().join(format!("kibble_build_docs_{}", std::process::id()));
        let _ = fs::remove_dir_all(&root);
        fs::create_dir_all(root.join("data/catalog")).unwrap();
        // Two dataset rows from one [[source]] jsonl, no data/raw pile at all.
        fs::write(
            root.join("ext.jsonl"),
            "{\"id\":\"a\",\"messages\":[{\"role\":\"user\",\"content\":\"q1\"},{\"role\":\"assistant\",\"content\":\"a distinct first answer here\"}]}\n{\"id\":\"b\",\"messages\":[{\"role\":\"user\",\"content\":\"q2\"},{\"role\":\"assistant\",\"content\":\"a distinct second answer here\"}]}\n",
        ).unwrap();
        fs::write(
            root.join(crate::config::CONFIG_FILE),
            "[[source]]\npath = \"ext.jsonl\"\nname = \"ext\"\nsystem_prompt = \"SYS\"\n",
        ).unwrap();

        let stats = run_build(&root).await.unwrap();
        assert!(stats.total_documents >= 2, "counts the [[source]] docs, not just data/raw (got {})", stats.total_documents);
        let s: serde_json::Value = serde_json::from_str(
            &fs::read_to_string(root.join("data/datasets/unsloth/stats.json")).unwrap()).unwrap();
        assert!(s["documents_by_source"]["ext"].as_u64().unwrap() >= 2, "by_source counts the source's docs");

        fs::remove_dir_all(&root).ok();
    }

    #[tokio::test]
    async fn run_build_writes_source_manifest() {
        use super::run_build;
        use std::fs;
        let root = std::env::temp_dir().join(format!("kibble_build_src_{}", std::process::id()));
        let _ = fs::remove_dir_all(&root);
        fs::create_dir_all(root.join("data/catalog")).unwrap();
        fs::write(
            root.join("ext.jsonl"),
            r#"{"id":"x","messages":[{"role":"user","content":"hi"},{"role":"assistant","content":"there and back again over the hill"}]}"#,
        ).unwrap();
        fs::write(
            root.join(crate::config::CONFIG_FILE),
            "[[source]]\npath = \"ext.jsonl\"\nname = \"ext\"\nsystem_prompt = \"SYS\"\n",
        ).unwrap();

        let stats = run_build(&root).await.unwrap();
        let manifest = fs::read_to_string(root.join("data/datasets/unsloth/train.sources.jsonl")).unwrap();
        let lines: Vec<&str> = manifest.lines().collect();
        assert_eq!(lines.len(), stats.train, "one manifest line per train row");
        assert!(lines.iter().all(|s| *s == "ext"), "every source is the single configured source");

        fs::remove_dir_all(&root).ok();
    }

    #[tokio::test]
    async fn run_build_rebalance_failsoft_without_clusters() {
        use super::run_build;
        use std::fs;
        let root = std::env::temp_dir().join(format!("kibble_build_rb_{}", std::process::id()));
        let _ = fs::remove_dir_all(&root);
        fs::create_dir_all(root.join("data/catalog")).unwrap();
        fs::write(
            root.join("ext.jsonl"),
            r#"{"id":"x","messages":[{"role":"user","content":"hi"},{"role":"assistant","content":"there and back again over the hill"}]}"#,
        ).unwrap();
        // rebalance enabled, but no clusters.json / no embed backend → fail-soft (0 drops)
        fs::write(
            root.join(crate::config::CONFIG_FILE),
            "[[source]]\npath = \"ext.jsonl\"\nname = \"ext\"\nsystem_prompt = \"SYS\"\n[cluster]\nrebalance = true\n",
        ).unwrap();

        let stats = run_build(&root).await.unwrap();
        assert_eq!(stats.dropped_topic_rebalanced, 0, "no clusters.json → rebalance is a no-op");

        let s: serde_json::Value = serde_json::from_str(
            &fs::read_to_string(root.join("data/datasets/unsloth/stats.json")).unwrap()).unwrap();
        assert_eq!(s["dropped_topic_rebalanced"], 0);

        fs::remove_dir_all(&root).ok();
    }

    #[tokio::test]
    async fn run_build_preserves_code_in_dataset_when_flagged() {
        use super::run_build;
        use std::fs;
        let root = std::env::temp_dir().join(format!("kibble_build_dscode_{}", std::process::id()));
        let _ = fs::remove_dir_all(&root);
        fs::create_dir_all(root.join("data/raw/me")).unwrap();
        fs::create_dir_all(root.join("data/catalog")).unwrap();
        fs::write(
            root.join("ds.jsonl"),
            r#"{"id":"c","messages":[{"role":"user","content":"show me a path"},{"role":"assistant","content":"use $PATH and r/rust and https://example.com/x"}]}"#,
        )
        .unwrap();
        fs::write(
            root.join(crate::config::CONFIG_FILE),
            "[[source]]\npath = \"ds.jsonl\"\nname = \"code_ds\"\npreserve_code = true\n",
        )
        .unwrap();

        let stats = run_build(&root).await.unwrap();
        assert!(stats.sources.iter().any(|(n, _, _, _)| n == "code_ds"));
        let combined = ["data/datasets/unsloth/train.jsonl", "data/datasets/unsloth/valid.jsonl", "data/datasets/unsloth/test.jsonl"]
            .iter()
            .map(|p| fs::read_to_string(root.join(p)).unwrap_or_default())
            .collect::<String>();
        // preserved verbatim — the prose cleaner would otherwise strip these
        assert!(combined.contains("$PATH"));
        assert!(combined.contains("r/rust"));
        assert!(combined.contains("https://example.com/x"));
    }

    #[tokio::test]
    async fn run_build_errors_on_missing_blog_file() {
        use super::run_build;
        use std::fs;
        let root = std::env::temp_dir().join(format!("kibble_build_missblog_{}", std::process::id()));
        let _ = fs::remove_dir_all(&root);
        fs::create_dir_all(root.join("data/raw/me")).unwrap();
        fs::write(root.join(crate::config::CONFIG_FILE), "[[source]]\npath = \"nope.html\"\n").unwrap();
        assert!(run_build(&root).await.is_err());
    }

    #[tokio::test]
    async fn run_build_folds_blog_html_file() {
        use super::run_build;
        use std::fs;
        let root = std::env::temp_dir().join(format!("kibble_build_blog_{}", std::process::id()));
        let _ = fs::remove_dir_all(&root);
        fs::create_dir_all(root.join("data/raw/me")).unwrap();
        fs::create_dir_all(root.join("data/catalog")).unwrap();
        let html = "<html><body><article><h1>Post</h1><p>A saved blog paragraph long enough to comfortably clear the longform minimum and make a training row here today right now.</p></article></body></html>";
        fs::write(root.join("post.html"), html).unwrap();
        fs::write(
            root.join(crate::config::CONFIG_FILE),
            "[[source]]\npath = \"post.html\"\nname = \"saved\"\n",
        )
        .unwrap();

        let stats = run_build(&root).await.unwrap();
        assert!(stats.sources.iter().any(|(n, _, _, _)| n == "saved"));
        let combined = ["data/datasets/unsloth/train.jsonl", "data/datasets/unsloth/valid.jsonl", "data/datasets/unsloth/test.jsonl"]
            .iter()
            .map(|p| fs::read_to_string(root.join(p)).unwrap_or_default())
            .collect::<String>();
        assert!(combined.contains("saved blog paragraph"));
    }

    #[tokio::test]
    async fn run_build_writes_outputs_and_cleans_content() {
        use super::run_build;
        use std::fs;
        let root = std::env::temp_dir().join(format!("kibble_build_it_{}", std::process::id()));
        let _ = fs::remove_dir_all(&root);
        fs::create_dir_all(root.join("data/raw/me/twitter")).unwrap();
        fs::create_dir_all(root.join("data/catalog")).unwrap();
        // a tweet with an @mention that clean_text must strip at write time
        fs::write(
            root.join("data/raw/me/twitter/tweet_1.txt"),
            "hello @bob this is a sufficiently long tweet body",
        )
        .unwrap();

        let stats = run_build(&root).await.unwrap();
        assert_eq!(stats.total_documents, 1);
        assert_eq!(stats.train + stats.valid + stats.test, 1);

        // The three JSONL exist and the mention was stripped in the written content.
        let combined = ["data/datasets/unsloth/train.jsonl", "data/datasets/unsloth/valid.jsonl", "data/datasets/unsloth/test.jsonl"]
            .iter()
            .map(|p| fs::read_to_string(root.join(p)).unwrap_or_default())
            .collect::<String>();
        assert!(combined.contains("hello this is a sufficiently long tweet body"));
        assert!(!combined.contains("@bob"));
        // stats.json + catalog written
        assert!(root.join("data/datasets/unsloth/stats.json").exists());
        assert!(root.join("data/catalog/documents.jsonl").exists());
        // mirrored copy
        assert!(root.join("data/train.jsonl").exists());
    }

    #[tokio::test]
    async fn run_build_preserves_code_tokens_verbatim() {
        use super::run_build;
        use std::fs;
        let root = std::env::temp_dir().join(format!("kibble_build_code_tok_{}", std::process::id()));
        let _ = fs::remove_dir_all(&root);
        fs::create_dir_all(root.join("data/raw/me")).unwrap();
        fs::create_dir_all(root.join("data/catalog")).unwrap();
        fs::create_dir_all(root.join("repo/src")).unwrap();
        // Content with tokens that clean_text would strip: $PATH, https://, r/rust
        let body = "fn main() {\n    let p = \"$PATH\";\n    let u = \"https://example.com/x\";\n    let r = \"r/rust\";\n    println!(\"{p}{u}{r}\");\n}\n".repeat(2);
        fs::write(root.join("repo/src/conf.rs"), &body).unwrap();
        fs::write(
            root.join(crate::config::CONFIG_FILE),
            "[[source]]\npath = \"repo\"\ntype = \"codebase\"\nname = \"repo\"\n",
        )
        .unwrap();

        let _stats = run_build(&root).await.unwrap();

        let combined = ["data/datasets/unsloth/train.jsonl", "data/datasets/unsloth/valid.jsonl", "data/datasets/unsloth/test.jsonl"]
            .iter()
            .map(|p| fs::read_to_string(root.join(p)).unwrap_or_default())
            .collect::<String>();

        assert!(combined.contains("$PATH"), "clean_text stripped $PATH from code row");
        assert!(combined.contains("https://example.com/x"), "clean_text stripped URL from code row");
        assert!(combined.contains("r/rust"), "clean_text stripped r/rust from code row");
    }

    #[tokio::test]
    async fn run_build_folds_codebase_source() {
        use super::run_build;
        use std::fs;
        let root = std::env::temp_dir().join(format!("kibble_build_cb_{}", std::process::id()));
        let _ = fs::remove_dir_all(&root);
        fs::create_dir_all(root.join("data/raw/me")).unwrap();
        fs::create_dir_all(root.join("data/catalog")).unwrap();
        fs::create_dir_all(root.join("repo/src")).unwrap();
        let body = "fn main() {
    println!(\"a real source file here\");
}
".repeat(2);
        fs::write(root.join("repo/src/main.rs"), &body).unwrap();
        fs::write(
            root.join(crate::config::CONFIG_FILE),
            "[[source]]
path = \"repo\"
type = \"codebase\"
name = \"repo\"
",
        )
        .unwrap();

        let stats = run_build(&root).await.unwrap();
        assert!(stats.sources.iter().any(|(n, _, _, _)| n == "repo"));
        let combined = ["data/datasets/unsloth/train.jsonl", "data/datasets/unsloth/valid.jsonl", "data/datasets/unsloth/test.jsonl"]
            .iter()
            .map(|p| fs::read_to_string(root.join(p)).unwrap_or_default())
            .collect::<String>();
        assert!(combined.contains("```rust"));
    }

    #[tokio::test]
    async fn run_build_clones_and_folds_git_url_codebase() {
        use super::run_build;
        use std::fs;
        use std::process::Command;

        let root = std::env::temp_dir().join(format!("kibble_build_giturl_{}", std::process::id()));
        let _ = fs::remove_dir_all(&root);
        fs::create_dir_all(root.join("data/raw/me")).unwrap();
        fs::create_dir_all(root.join("data/catalog")).unwrap();

        // a real local git repo to act as the "remote"
        let remote = root.join("remote_repo");
        fs::create_dir_all(remote.join("src")).unwrap();
        let run = |args: &[&str], cwd: &std::path::Path| {
            let o = Command::new("git").args(args).current_dir(cwd).output().unwrap();
            assert!(o.status.success(), "git {:?}: {}", args, String::from_utf8_lossy(&o.stderr));
        };
        run(&["init", "-q"], &remote);
        run(&["config", "user.email", "t@t"], &remote);
        run(&["config", "user.name", "t"], &remote);
        let body = "fn main() {\n    println!(\"cloned source file body here\");\n}\n".repeat(2);
        fs::write(remote.join("src/main.rs"), &body).unwrap();
        run(&["add", "."], &remote);
        run(&["commit", "-q", "-m", "init"], &remote);

        // kibble.toml points the codebase source at the repo via a file:// URL
        let url = format!("file://{}", remote.display());
        fs::write(
            root.join(crate::config::CONFIG_FILE),
            format!("[[source]]\npath = \"{url}\"\ntype = \"codebase\"\nname = \"remote\"\n"),
        )
        .unwrap();

        let stats = run_build(&root).await.unwrap();
        assert!(stats.sources.iter().any(|(n, _, _, _)| n == "remote"));
        // cache dir created
        assert!(root.join(".kibble-cache").is_dir());
        let combined = ["data/datasets/unsloth/train.jsonl", "data/datasets/unsloth/valid.jsonl", "data/datasets/unsloth/test.jsonl"]
            .iter()
            .map(|p| fs::read_to_string(root.join(p)).unwrap_or_default())
            .collect::<String>();
        assert!(combined.contains("```rust"));
    }

    #[tokio::test]
    async fn run_build_folds_files_source() {
        use super::run_build;
        use std::fs;
        let root = std::env::temp_dir().join(format!("kibble_build_files_{}", std::process::id()));
        let _ = fs::remove_dir_all(&root);
        fs::create_dir_all(root.join("data/raw/me")).unwrap();
        fs::create_dir_all(root.join("data/catalog")).unwrap();
        fs::create_dir_all(root.join("notes/sub")).unwrap();
        let body = "This is a longform note with plenty of words so it comfortably exceeds the longform minimum of one hundred and twenty characters and reliably yields at least one training row.";
        fs::write(root.join("notes/a.md"), body).unwrap();
        fs::write(root.join("notes/sub/b.txt"), body).unwrap();
        fs::write(
            root.join(crate::config::CONFIG_FILE),
            "[[source]]\npath = \"notes\"\ntype = \"files\"\nname = \"notes\"\n",
        )
        .unwrap();

        let stats = run_build(&root).await.unwrap();
        assert!(stats.sources.iter().any(|(n, _, _, _)| n == "notes"));
        let combined = ["data/datasets/unsloth/train.jsonl", "data/datasets/unsloth/valid.jsonl", "data/datasets/unsloth/test.jsonl"]
            .iter()
            .map(|p| fs::read_to_string(root.join(p)).unwrap_or_default())
            .collect::<String>();
        assert!(combined.contains("longform note"));
    }

    #[tokio::test]
    async fn run_build_catalogs_files_source() {
        use super::run_build;
        let root = std::env::temp_dir().join(format!("kibble_cat_files_{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&root);
        std::fs::create_dir_all(root.join("data/catalog")).unwrap();
        std::fs::create_dir_all(root.join("notes")).unwrap();
        std::fs::write(root.join("notes/alpha.md"),
            "# Alpha networking notes\n\nA reasonably long paragraph about TCP sockets, congestion \
             control, and how packets traverse a network so the row survives curation. More text \
             here to be safe about length thresholds during the build's curation stage.\n").unwrap();
        std::fs::write(root.join(crate::config::CONFIG_FILE),
            "[[source]]\npath = \"notes\"\ntype = \"files\"\nname = \"notes\"\n").unwrap();

        run_build(&root).await.unwrap();

        let cat = std::fs::read_to_string(root.join("data/catalog/documents.jsonl")).unwrap();
        assert!(cat.lines().any(|l| l.contains("\"source\":\"notes\"") && l.contains("alpha.md")),
            "the files-source doc must appear in the catalog: {cat}");
        // Reporting-only: building the catalog must not create a retrieval index.
        assert!(!root.join("data/index/chunks.jsonl").exists(),
            "build must not write a retrieval index");
        std::fs::remove_dir_all(&root).ok();
    }

    #[tokio::test]
    async fn run_build_fetches_and_folds_web_source() {
        use super::run_build;
        use std::fs;
        use std::io::{Read, Write};
        use std::net::TcpListener;

        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
        let port = listener.local_addr().unwrap().port();
        let server = std::thread::spawn(move || {
            if let Ok((mut stream, _)) = listener.accept() {
                let mut buf = [0u8; 1024];
                let _ = stream.read(&mut buf);
                let body = "<html><body><article><h1>Post</h1><p>A real blog paragraph with enough words to pass the longform minimum so it produces a training row here for sure yes.</p></article></body></html>";
                let resp = format!("HTTP/1.1 200 OK\r\nContent-Length: {}\r\nContent-Type: text/html\r\n\r\n{}", body.len(), body);
                let _ = stream.write_all(resp.as_bytes());
                let _ = stream.flush();
            }
        });

        let root = std::env::temp_dir().join(format!("kibble_build_web_{}", std::process::id()));
        let _ = fs::remove_dir_all(&root);
        fs::create_dir_all(root.join("data/raw/me")).unwrap();
        fs::create_dir_all(root.join("data/catalog")).unwrap();
        fs::write(
            root.join(crate::config::CONFIG_FILE),
            format!("[[source]]\npath = \"http://127.0.0.1:{port}/\"\ntype = \"web\"\nname = \"blog\"\n"),
        )
        .unwrap();

        let stats = run_build(&root).await.unwrap();
        server.join().unwrap();
        assert!(stats.sources.iter().any(|(n, _, _, _)| n == "blog"));
        let combined = ["data/datasets/unsloth/train.jsonl", "data/datasets/unsloth/valid.jsonl", "data/datasets/unsloth/test.jsonl"]
            .iter()
            .map(|p| fs::read_to_string(root.join(p)).unwrap_or_default())
            .collect::<String>();
        assert!(combined.contains("A real blog paragraph"));
    }

    #[test]
    fn curate_near_dedup_collapses_paraphrases() {
        let base = crate::config::CurateConfig { dedup: true, leakage_safe_split: true, drop_malformed: true, drop_degenerate: true, min_answer_chars: 5, near_dedup: true, near_dedup_threshold: 0.6, shingle_size: 5, semantic_dedup: false, semantic_threshold: 0.90 };
        let rows = vec![
            rwm("the quick brown fox jumps over the lazy dog in the meadow at dawn each morning", "d1", "s", false),
            rwm("the quick brown fox jumps over the lazy dog in the meadow at dusk each morning", "d2", "s", false),
            rwm("a completely different sentence about astronomy and the expanding universe entirely now", "d3", "s", false),
        ];
        let c = curate_split(rows, &base);
        let kept = c.train.clean.len()+c.valid.clean.len()+c.test.clean.len();
        assert_eq!(kept, 2, "the two paraphrases collapse to one");
        assert_eq!(c.dropped_near_duplicates, 1);

        // near_dedup off → all three kept (no near drops)
        let off = crate::config::CurateConfig { near_dedup: false, ..base };
        let rows2 = vec![
            rwm("the quick brown fox jumps over the lazy dog in the meadow at dawn each morning", "d1", "s", false),
            rwm("the quick brown fox jumps over the lazy dog in the meadow at dusk each morning", "d2", "s", false),
        ];
        let c2 = curate_split(rows2, &off);
        assert_eq!(c2.dropped_near_duplicates, 0);
        assert_eq!(c2.train.clean.len()+c2.valid.clean.len()+c2.test.clean.len(), 2);
    }

    #[tokio::test]
    async fn semantic_dedup_drop_collapses_same_topic() {
        use crate::embed::Embedder;
        // Vector depends only on the first word -> same first word => identical vector.
        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_semdedup_{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&dir);
        let texts = vec![
            "apple pie recipe".to_string(),       // idx 0  (first word "apple")
            "apple orchard tour today".to_string(), // idx 1  (first word "apple", longer)
            "banana bread".to_string(),           // idx 2  (first word "banana")
        ];
        let drop = semantic_dedup_drop(&FirstWord, &dir, "m", &texts, 0.99, 64).await.unwrap();
        assert!(drop.contains(&0));      // collapsed with idx 1
        assert!(!drop.contains(&1));     // longest "apple" text kept
        assert!(!drop.contains(&2));     // distinct topic survives
        std::fs::remove_dir_all(&dir).ok();
    }

    fn mkrow2(user: &str, assistant: &str) -> Row {
        Row { messages: vec![
            Msg { role: "user".into(), content: user.into() },
            Msg { role: "assistant".into(), content: assistant.into() },
        ]}
    }

    #[test]
    fn is_valid_sft_rules() {
        assert!(is_valid_sft(&mkrow2("q", "a clear answer")));
        // no assistant
        assert!(!is_valid_sft(&Row { messages: vec![Msg{role:"user".into(),content:"q".into()}] }));
        // empty content
        assert!(!is_valid_sft(&mkrow2("q", "   ")));
        // bad role
        assert!(!is_valid_sft(&Row { messages: vec![
            Msg{role:"system".into(),content:"s".into()},
            Msg{role:"bot".into(),content:"x".into()}]}));
        // empty messages
        assert!(!is_valid_sft(&Row { messages: vec![] }));
    }

    #[test]
    fn is_degenerate_rules() {
        assert!(!is_degenerate(&mkrow2("question here", "a sufficiently long real answer"), 20, false));
        assert!(is_degenerate(&mkrow2("q", ""), 20, false));                       // empty
        assert!(is_degenerate(&mkrow2("q", "too short"), 20, false));              // < 20 chars
        assert!(is_degenerate(&mkrow2("Echo This", "echo   this"), 20, false));    // echo (normalized)

        // ≥20 raw chars but all-URL → cleaned to empty for non-raw → degenerate; raw keeps it
        let urly = mkrow2("question", "https://example.com/a/very/long/path/that/exceeds/twenty");
        assert!(is_degenerate(&urly, 20, false));   // cleaned → empty → degenerate
        assert!(!is_degenerate(&urly, 20, true));    // raw → survives
    }

    #[test]
    fn curate_filters_before_dedup() {
        let cfg = crate::config::CurateConfig { dedup: true, leakage_safe_split: true, drop_malformed: true, drop_degenerate: true, min_answer_chars: 20, ..Default::default() };
        let rows = vec![
            rwm("a perfectly good and sufficiently long answer", "d1", "s", false),
            rwm("", "d2", "s", false),                                  // degenerate empty
            rwm("short", "d3", "s", false),                             // degenerate short
            RowWithMeta { row: Row { messages: vec![Msg{role:"user".into(),content:"q".into()}] }, doc_id: "d4".into(), source: "s".into(), raw: false }, // malformed
        ];
        let c = curate_split(rows, &cfg);
        let kept = c.train.clean.len()+c.valid.clean.len()+c.test.clean.len();
        assert_eq!(kept, 1);
        assert_eq!(c.dropped_filtered, 3);
        assert_eq!(c.dropped_duplicates, 0);

        // toggles off → nothing filtered (4 kept, minus any dedup; all distinct here)
        let cfg2 = crate::config::CurateConfig { drop_malformed: false, drop_degenerate: false, ..cfg };
        let rows2 = vec![ rwm("good long answer that is fine here", "d1", "s", false), rwm("", "d2", "s", false) ];
        let c2 = curate_split(rows2, &cfg2);
        assert_eq!(c2.dropped_filtered, 0);
    }

    #[tokio::test]
    async fn run_build_drops_degenerate_rows() {
        let root = std::env::temp_dir().join(format!("kibble_fdrop_{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&root);
        std::fs::create_dir_all(root.join("data")).unwrap();
        let ds = root.join("data/ext.jsonl");
        let lines = [
            r#"{"messages":[{"role":"user","content":"q1"},{"role":"assistant","content":"a clearly long and perfectly valid answer with real substance here"}]}"#,
            r#"{"messages":[{"role":"user","content":"q2"},{"role":"assistant","content":"   "}]}"#,
            r#"{"messages":[{"role":"user","content":"q3"},{"role":"assistant","content":"short"}]}"#,
        ].join("\n");
        std::fs::write(&ds, lines).unwrap();
        std::fs::write(root.join(crate::config::CONFIG_FILE), format!("[paths]\ndata_root=\"data\"\ndataset_dir=\"data/ds\"\n[[source]]\npath=\"{}\"\nname=\"ext\"\n", ds.display())).unwrap();
        let stats = run_build(&root).await.unwrap();
        assert!(stats.dropped_filtered >= 2, "empty + short rows must be filtered (got {})", stats.dropped_filtered);
        let s: serde_json::Value = serde_json::from_str(&std::fs::read_to_string(root.join("data/ds/stats.json")).unwrap()).unwrap();
        assert!(s.get("dropped_filtered").is_some());
    }

    #[tokio::test]
    async fn run_build_near_dedup_drops_paraphrases() {
        let root = std::env::temp_dir().join(format!("kibble_nd_{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&root);
        std::fs::create_dir_all(root.join("data")).unwrap();
        let ds = root.join("data/ext.jsonl");
        // two near-paraphrase answers + one distinct
        let lines = [
            r#"{"messages":[{"role":"user","content":"q1"},{"role":"assistant","content":"the quick brown fox jumps over the lazy dog in the meadow at dawn each and every morning"}]}"#,
            r#"{"messages":[{"role":"user","content":"q2"},{"role":"assistant","content":"the quick brown fox jumps over the lazy dog in the meadow at dusk each and every morning"}]}"#,
            r#"{"messages":[{"role":"user","content":"q3"},{"role":"assistant","content":"an entirely separate passage discussing astronomy galaxies and the expanding cosmos in detail"}]}"#,
        ].join("\n");
        std::fs::write(&ds, lines).unwrap();
        std::fs::write(root.join(crate::config::CONFIG_FILE), format!("[paths]\ndata_root=\"data\"\ndataset_dir=\"data/ds\"\n[curate]\nnear_dedup=true\nnear_dedup_threshold=0.6\n[[source]]\npath=\"{}\"\nname=\"ext\"\n", ds.display())).unwrap();
        let stats = run_build(&root).await.unwrap();
        assert!(stats.dropped_near_duplicates >= 1, "paraphrase should be near-deduped (got {})", stats.dropped_near_duplicates);
        let s: serde_json::Value = serde_json::from_str(&std::fs::read_to_string(root.join("data/ds/stats.json")).unwrap()).unwrap();
        assert!(s.get("dropped_near_duplicates").is_some());
    }

    #[tokio::test]
    async fn run_build_dedups_and_writes_stats() {
        let root = std::env::temp_dir().join(format!("kibble_curbld_{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&root);
        std::fs::create_dir_all(root.join("data/raw/local/docs")).unwrap();
        // two identical-content files (different doc_id) + one distinct → 1 dedup drop
        std::fs::write(root.join("data/raw/local/docs/a.txt"), "the quick brown fox jumps over the lazy dog repeatedly and clearly yes this sentence is definitely long enough to pass the minimum longform character threshold").unwrap();
        std::fs::write(root.join("data/raw/local/docs/b.txt"), "the quick brown fox jumps over the lazy dog repeatedly and clearly yes this sentence is definitely long enough to pass the minimum longform character threshold").unwrap();
        std::fs::write(root.join("data/raw/local/docs/c.txt"), "an entirely different sentence about something else altogether here and this one is also comfortably above the minimum longform character threshold too").unwrap();
        std::fs::write(root.join(crate::config::CONFIG_FILE), "[paths]\ndata_root=\"data\"\ndataset_dir=\"data/ds\"\n").unwrap();
        let stats = run_build(&root).await.unwrap();
        assert!(stats.dropped_duplicates >= 1, "identical docs should dedup");
        let s: serde_json::Value = serde_json::from_str(&std::fs::read_to_string(root.join("data/ds/stats.json")).unwrap()).unwrap();
        assert!(s.get("dropped_duplicates").is_some());
    }

    #[test]
    fn apply_train_drops_drops_rows_and_sources_in_lockstep() {
        fn row(c: &str) -> Row { Row { messages: vec![Msg { role: "user".into(), content: c.into() }] } }
        let mut t = SplitRows {
            clean: vec![row("c0"), row("c1"), row("c2")],
            clean_sources: vec!["a".into(), "b".into(), "c".into()],
            clean_doc_ids: vec!["d0".into(), "d1".into(), "d2".into()],
            raw: vec![row("r0"), row("r1")],
            raw_sources: vec!["d".into(), "e".into()],
            raw_doc_ids: vec!["d3".into(), "d4".into()],
        };
        // combined indices: clean 0,1,2 ; raw 3(=raw[0]),4(=raw[1]). Drop clean idx1 + raw idx4.
        apply_train_drops(&mut t, &[1, 4], 3);
        assert_eq!(t.clean_sources, vec!["a".to_string(), "c".to_string()]);
        assert_eq!(t.clean.len(), 2);
        assert_eq!(t.raw_sources, vec!["d".to_string()]);
        assert_eq!(t.raw.len(), 1);
        assert_eq!(t.clean_doc_ids, vec!["d0".to_string(), "d2".to_string()]);
        assert_eq!(t.raw_doc_ids, vec!["d3".to_string()]);
    }

    #[test]
    fn synth_entries_group_and_classify() {
        use super::{synth_entries_from_rows, Msg, Row};
        let rows = vec![
            ("d1".to_string(), Row { messages: vec![
                Msg { role: "system".into(), content: "SYS".into() },
                Msg { role: "user".into(), content: "q1".into() },
                Msg { role: "assistant".into(), content: "a1".into() }] }),
            ("d1".to_string(), Row { messages: vec![
                Msg { role: "assistant".into(), content: "a2".into() }] }),
            ("d2".to_string(), Row { messages: vec![
                Msg { role: "assistant".into(), content: "b1".into() }] }),
        ];
        let tax = crate::catalog::load_taxonomy(std::path::Path::new("/nonexistent-kibble-tax"));
        let ov = crate::catalog::Overrides::default();
        let entries = synth_entries_from_rows(&rows, "ds", None, &tax, &ov);
        assert_eq!(entries.len(), 2, "one entry per distinct doc_id");
        assert_eq!(entries[0].doc_id, "d1");   // first-seen order
        assert_eq!(entries[0].source, "ds");
        assert_eq!(entries[0].chars, "q1\na1\na2".chars().count(), "system excluded, contents joined");
        // fallback role applies when the taxonomy has no source_default for this source
        let code = synth_entries_from_rows(&rows, "cb", Some("code"), &tax, &ov);
        assert_eq!(code[0].role, "code");
    }

    #[tokio::test]
    async fn run_build_catalogs_dataset_source() {
        use super::run_build;
        let root = std::env::temp_dir().join(format!("kibble_cat_ds_{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&root);
        std::fs::create_dir_all(root.join("data/catalog")).unwrap();
        std::fs::write(root.join("ext.jsonl"),
            "{\"id\":\"a\",\"messages\":[{\"role\":\"user\",\"content\":\"q1\"},{\"role\":\"assistant\",\"content\":\"a distinct first answer with enough words to survive curation here\"}]}\n\
             {\"id\":\"b\",\"messages\":[{\"role\":\"user\",\"content\":\"q2\"},{\"role\":\"assistant\",\"content\":\"a distinct second answer with enough words to survive curation here\"}]}\n").unwrap();
        std::fs::write(root.join(crate::config::CONFIG_FILE),
            "[[source]]\npath = \"ext.jsonl\"\nname = \"ext\"\nsystem_prompt = \"SYS\"\n").unwrap();

        run_build(&root).await.unwrap();

        let cat = std::fs::read_to_string(root.join("data/catalog/documents.jsonl")).unwrap();
        let n = cat.lines().filter(|l| l.contains("\"source\":\"ext\"")).count();
        assert!(n >= 2, "each dataset doc_id is cataloged (got {n}): {cat}");
        std::fs::remove_dir_all(&root).ok();
    }

    #[tokio::test]
    async fn run_build_codebase_role_default_and_override() {
        use super::run_build;
        let root = std::env::temp_dir().join(format!("kibble_cat_code_{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&root);
        std::fs::create_dir_all(root.join("data/catalog")).unwrap();
        std::fs::create_dir_all(root.join("srcpile")).unwrap();
        std::fs::write(root.join("srcpile/lib.rs"),
            "pub fn add(a: i32, b: i32) -> i32 { a + b } // a small but real code file with enough text\n").unwrap();
        std::fs::write(root.join(crate::config::CONFIG_FILE),
            "[[source]]\npath = \"srcpile\"\ntype = \"codebase\"\nname = \"mycode\"\n").unwrap();

        run_build(&root).await.unwrap();
        let cat = std::fs::read_to_string(root.join("data/catalog/documents.jsonl")).unwrap();
        assert!(cat.lines().any(|l| l.contains("\"source\":\"mycode\"") && l.contains("\"role\":\"code\"")),
            "codebase defaults to role=code when unconfigured: {cat}");

        // Configured source_defaults must win over the code fallback.
        std::fs::write(root.join("data/catalog/taxonomy.yaml"),
            "topics: {}\nsource_defaults:\n  mycode:\n    role: knowledge\n    topics: []\n").unwrap();
        run_build(&root).await.unwrap();
        let cat2 = std::fs::read_to_string(root.join("data/catalog/documents.jsonl")).unwrap();
        assert!(cat2.lines().any(|l| l.contains("\"source\":\"mycode\"") && l.contains("\"role\":\"knowledge\"")),
            "source_defaults role wins over the code fallback: {cat2}");
        assert!(!cat2.lines().any(|l| l.contains("\"source\":\"mycode\"") && l.contains("\"role\":\"code\"")));
        std::fs::remove_dir_all(&root).ok();
    }

    #[tokio::test]
    async fn catalog_count_matches_total_in_controlled_build() {
        use super::run_build;
        let root = std::env::temp_dir().join(format!("kibble_cat_count_{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&root);
        std::fs::create_dir_all(root.join("data/catalog")).unwrap();
        // One files source, distinct docs, each with enough text to yield >=1 row and no id collision.
        std::fs::create_dir_all(root.join("notes")).unwrap();
        std::fs::write(root.join("notes/one.md"),
            "First doc about CSS grid layout, covering fr units, named template areas, and gap sizing, \
             with sufficient descriptive text to survive curation cleanly.\n").unwrap();
        std::fs::write(root.join("notes/two.md"),
            "Second doc about TCP networking sockets, covering the three-way handshake, congestion \
             windows, and retransmission timers, with sufficient descriptive text to survive curation cleanly.\n").unwrap();
        std::fs::write(root.join(crate::config::CONFIG_FILE), "[[source]]\npath = \"notes\"\ntype = \"files\"\nname = \"notes\"\n").unwrap();

        let stats = run_build(&root).await.unwrap();
        let cat = std::fs::read_to_string(root.join("data/catalog/documents.jsonl")).unwrap();
        let lines = cat.lines().filter(|l| !l.trim().is_empty()).count();
        // Sanity check for THIS fixture (every doc yields >=1 row, no id collisions) — not a contract.
        assert_eq!(lines, stats.total_documents, "catalog covers exactly the counted documents here");
        std::fs::remove_dir_all(&root).ok();
    }

    #[tokio::test]
    async fn classify_disabled_leaves_catalog_byte_identical() {
        use super::run_build;
        let root = std::env::temp_dir().join(format!("kibble_classify_off_{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&root);
        std::fs::create_dir_all(root.join("data/catalog")).unwrap();
        std::fs::create_dir_all(root.join("notes")).unwrap();
        std::fs::write(root.join("notes/a.md"),
            "A paragraph long enough to survive curation about networking, sockets, and packets on a busy link with plenty of words here.\n").unwrap();
        std::fs::write(root.join(crate::config::CONFIG_FILE),
            "[[source]]\npath = \"notes\"\ntype = \"files\"\nname = \"notes\"\n").unwrap();
        run_build(&root).await.unwrap();
        let cat = std::fs::read_to_string(root.join("data/catalog/documents.jsonl")).unwrap();
        assert!(!cat.contains("auto_topic"), "classify off → no auto_topic keys");
        assert!(!root.join("data/catalog/topics.json").exists(), "classify off → no topics.json");
        std::fs::remove_dir_all(&root).ok();
    }

    #[tokio::test]
    async fn classify_failsoft_without_embed_backend() {
        use super::run_build;
        let root = std::env::temp_dir().join(format!("kibble_classify_failsoft_{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&root);
        std::fs::create_dir_all(root.join("data/catalog")).unwrap();
        std::fs::create_dir_all(root.join("notes")).unwrap();
        std::fs::write(root.join("notes/a.md"),
            "A paragraph long enough to survive curation about networking, sockets, and packets on a busy link with plenty of words here.\n").unwrap();
        // classify enabled but no embed backend ([understand.embed].base_url empty) → fail-soft.
        std::fs::write(root.join(crate::config::CONFIG_FILE),
            "[classify]\nenabled = true\n[[source]]\npath = \"notes\"\ntype = \"files\"\nname = \"notes\"\n").unwrap();
        run_build(&root).await.unwrap();   // must not panic / error
        let cat = std::fs::read_to_string(root.join("data/catalog/documents.jsonl")).unwrap();
        assert!(!cat.contains("auto_topic"), "no embed backend → fail-soft, no auto_topic");
        std::fs::remove_dir_all(&root).ok();
    }
}