//! Does the layer give the same answer twice? The queue-stability meter.
//!
//! [`crate::layer_eval`] asks whether an answer is right. This asks whether it is
//! the same answer next week. For a layer whose model is the user's own library
//! those are independent properties, and only the second one is measured here: a
//! consistently wrong answer is perfectly stable, so nothing below can be inferred
//! from the 95.5% / 97.4% family-resolution accuracy figures, and nothing below
//! substitutes for them.
//!
//! # Why this is an attention gate and not a safety gate
//!
//! Exemplars from outside the user's own labelling are **suggest-only** (decided
//! 2026-08-07 for the bundled layer, and inherited by imported layers when that
//! was retired): they never auto-apply, they populate a review queue. That kills
//! the failure this
//! measurement was originally filed against. The apply path is monotonic
//! (`apply_policy` skips tags the sample already has, `apply_tag_sourced` is
//! INSERT OR IGNORE, only `remove_tags_by_source` removes), so a changed answer
//! used to mean the sample kept its old tag *and* gained the new one, permanently.
//! Nothing is written unasked now, so that cannot happen.
//!
//! What survives is cheaper and still real: a queue whose contents reshuffle
//! between runs spends the user's attention twice. Someone who read 340 low
//! suggestions last month should not be handed a substantially different 340 this
//! month for the same unchanged samples. So the unit of measurement is the
//! **queue entry**, not the written tag: a sample's answer is its top-scoring tag
//! if that tag clears the review threshold, and `silent` otherwise.
//!
//! # The bar, written before the first run
//!
//! Same discipline as the ship gate, and for the same reason: a number chosen
//! after reading one is not a bar. See [`BAR`].
//!
//! # What varies, and why those four
//!
//! Each measurement holds the probe set fixed and varies one property of the
//! index, because a flip rate is meaningless without saying what moved.
//!
//! 1. [`composition`] — same size, different class mix. The user whose library is
//!    mostly kicks and the user whose library is mostly cymbals are running the
//!    same code against different models.
//! 2. [`size`] — same mix, growing index. A library grows by accretion, so this is
//!    the shape of every real user's second month.
//! 3. [`deployment_shape`] — the mixed index the app actually builds: the user's
//!    own labels at 1.0 beside an imported layer at `DEFAULT_IMPORT_WEIGHT` 0.5.
//!    Never measured before this module, and the most decision-relevant of the
//!    four. Every earlier number came off a uniform-weight index, where the weight
//!    is a constant multiplier inside a sum divided by its own total and therefore
//!    cancels. It does not cancel here, and neither does the second-order effect:
//!    `build_index` fits the standardization params over local and imported
//!    exemplars together, so a user's labels move the space distances are measured
//!    in. This is the difference between a queue that helps and one that repeats
//!    what the user already knows.
//! 4. [`feedback`] — the queue changes its own inputs. An accepted suggestion
//!    becomes a user label at weight 1.0 and re-enters the index, so working
//!    through the queue rewrites the rest of it. Converges or oscillates is a
//!    question no other measurement here can answer.
//!
//! Usage: `cargo run --release -p audiofiles-bench -- layer-stability`
//! Env: `AF_BENCH_CORPUS`, `AF_BENCH_VAULT`, `AF_BENCH_EVAL_K` (first entry wins,
//! default `DEFAULT_K`), `AF_BENCH_EVAL_LABELS`, `AF_BENCH_STABILITY_PROBE`
//! (probe fraction denominator, default 5), `AF_BENCH_JSON`.

use std::collections::{BTreeMap, BTreeSet, HashMap};
use std::path::Path;

use audiofiles_core::analysis::config::AnalysisConfig;
use audiofiles_core::analysis::exemplar::{self, DEFAULT_K, DEFAULT_REVIEW_THRESHOLD};
use audiofiles_core::analysis::features::FEATURE_VERSION;
use audiofiles_core::rules::{RuleContext, RuleField};
use audiofiles_core::{rules, starter_rules};

use crate::families::{self, LabelSpace};
use crate::labelled;
use crate::report::Report;
use crate::rows::{self, Row};

/// The bar, stated before the first run.
///
/// Per-run family-label flip rate on a fixed held-out probe set, measured at the
/// deployment weight, counting only flips between two answers that both cleared
/// the review threshold. A sample moving between suggested and silent costs the
/// user nothing (the queue is shorter or longer, not wrong), so it is reported
/// separately as churn rather than counted here.
///
/// 2% is proposed rather than settled: on a 170-sample probe set it is three
/// samples, which is the resolution this corpus supports and not a claim that 2%
/// is where a user stops noticing. Max accepts or moves it.
const BAR: f64 = 0.02;

/// A sample either has a queue entry or it does not.
///
/// The threshold is [`DEFAULT_REVIEW_THRESHOLD`] and not the auto threshold on
/// purpose: under suggest-only nothing auto-applies, so the review threshold is
/// the only line that decides whether the user ever sees the answer.
#[derive(Clone, PartialEq, Eq)]
enum Answer {
    Silent,
    Tag(String),
}

impl Answer {
    fn of(index: &exemplar::ExemplarIndex, vector: &[f64], k: usize) -> Self {
        index
            .score(vector, k, None)
            .first()
            .filter(|s| s.score >= DEFAULT_REVIEW_THRESHOLD)
            .map_or(Self::Silent, |s| Self::Tag(s.tag.clone()))
    }

    fn tag(&self) -> Option<&str> {
        match self {
            Self::Silent => None,
            Self::Tag(t) => Some(t),
        }
    }
}

/// One index variant's answer for every probe sample, in probe order.
type Answers = Vec<Answer>;

/// How two runs' queues differ.
struct Churn {
    /// Probe samples suggested in both runs. The denominator of [`Self::flip_rate`].
    both: usize,
    /// Suggested in both, and the tag changed. What the bar is set against.
    flips: usize,
    /// Silent then suggested: the queue grew.
    appeared: usize,
    /// Suggested then silent: the queue shrank.
    vanished: usize,
}

impl Churn {
    fn between(before: &Answers, after: &Answers) -> Self {
        let mut churn = Self {
            both: 0,
            flips: 0,
            appeared: 0,
            vanished: 0,
        };
        for (x, y) in before.iter().zip(after) {
            match (x.tag(), y.tag()) {
                (Some(was), Some(now)) => {
                    churn.both += 1;
                    if was != now {
                        churn.flips += 1;
                    }
                }
                (None, Some(_)) => churn.appeared += 1,
                (Some(_), None) => churn.vanished += 1,
                (None, None) => {}
            }
        }
        churn
    }

    /// `None` when nothing was suggested in both runs, which is not a flip rate of
    /// zero: it is a pair of runs with no overlapping queue to compare.
    fn flip_rate(&self) -> Option<f64> {
        (self.both > 0).then(|| self.flips as f64 / self.both as f64)
    }

    /// The flip rate, but only when enough samples stood behind it to mean
    /// anything. See [`MIN_COMPARABLE`].
    fn comparable_flip_rate(&self) -> Option<f64> {
        self.flip_rate().filter(|_| self.both >= MIN_COMPARABLE)
    }

    fn passes(&self) -> bool {
        self.flip_rate().is_some_and(|r| r <= BAR)
    }
}

/// Fewest overlapping queue entries a pair of runs needs before its flip rate
/// counts toward a verdict.
///
/// Same job as `layer_eval`'s `min_support` and learned the same way. The first
/// run of [`feedback`] ended with seven samples still open and two of them
/// flipping, which is 28.6% and became the worst figure in the whole report. It
/// is one sample either way. Rates over thin tails are still printed — a tail
/// that thrashes is worth seeing — but they do not decide a pass.
const MIN_COMPARABLE: usize = 30;

fn pct(v: Option<f64>) -> String {
    v.map_or_else(|| "-".to_string(), |x| format!("{:.1}%", x * 100.0))
}

fn round4(v: f64) -> f64 {
    (v * 10_000.0).round() / 10_000.0
}

/// Score every probe row against one index.
fn answers(index: &exemplar::ExemplarIndex, probe: &[&Row], k: usize) -> Answers {
    probe
        .iter()
        .map(|r| Answer::of(index, &r.vector, k))
        .collect()
}

/// Build an index from `local` (weight 1.0) plus `imported` (the layer weight) and
/// score the probe against it.
fn run_variant(local: &[&Row], imported: &[&Row], probe: &[&Row], k: usize, what: &str) -> Answers {
    let db = match rows::mixed_db(local, imported) {
        Ok(db) => db,
        Err(e) => {
            eprintln!("{what}: {e}");
            std::process::exit(1);
        }
    };
    let index = match exemplar::build_index(&db) {
        Ok(i) => i,
        Err(e) => {
            eprintln!("{what}: build_index: {e}");
            std::process::exit(1);
        }
    };
    answers(&index, probe, k)
}

/// Take `n` rows of class `class` from `pool`, in pool order.
///
/// Prefix rather than a sample: pool order is content-hash order, which is already
/// arbitrary with respect to the pack a file came from, so this needs no RNG and
/// two runs produce the same variants. It also makes the size sweep nested by
/// construction, which is the property [`size`] wants: a growing library accretes,
/// it does not resample.
fn take_class<'a>(pool: &[&'a Row], class: &str, n: usize) -> Vec<&'a Row> {
    pool.iter()
        .filter(|r| r.truth.as_deref() == Some(class))
        .take(n)
        .copied()
        .collect()
}

/// Split a pool into two by stratified round-robin, so both halves carry the same
/// class mix and neither is a prefix of the other.
fn halve<'a>(pool: &[&'a Row]) -> (Vec<&'a Row>, Vec<&'a Row>) {
    let mut seen: HashMap<&str, usize> = HashMap::new();
    let mut a = Vec::new();
    let mut b = Vec::new();
    for r in pool {
        let key = r.truth.as_deref().unwrap_or("");
        let n = seen.entry(key).or_default();
        if (*n).is_multiple_of(2) {
            a.push(*r);
        } else {
            b.push(*r);
        }
        *n += 1;
    }
    (a, b)
}

pub(crate) fn run(
    corpus: &Path,
    vault: &Path,
    config: &AnalysisConfig,
    k: usize,
    probe_denominator: usize,
    space: LabelSpace,
) {
    println!("━━━ CLASSIFIER LAYER STABILITY ━━━");
    println!();
    println!("  corpus   {}", corpus.display());
    println!("  scratch  {}", vault.display());
    println!("  features v{FEATURE_VERSION}");
    println!("  k        {k}");
    println!("  labels   {}", space.describe());
    println!(
        "  answer   top-scoring tag at score >= {DEFAULT_REVIEW_THRESHOLD} (the review\n           \
         threshold), else silent. Under suggest-only that line is what decides\n           \
         whether the user ever sees the answer."
    );
    println!();
    println!(
        "  Bar: per-run family-label flip rate <= {:.0}% on the fixed probe set, at",
        BAR * 100.0
    );
    println!("  the deployment weight, counting only flips between two suggested answers.");
    println!("  Silent <-> suggested is reported as churn and is not a flip: the queue got");
    println!("  longer or shorter, it did not contradict itself.");
    println!();
    println!("  Flip rate and error rate are independent. A consistently wrong answer is");
    println!("  perfectly stable, so nothing here can be read off the accuracy figures and");
    println!("  nothing here substitutes for them.");
    println!();

    let built = match labelled::build_vault(corpus, vault, config) {
        Ok(v) => v,
        Err(e) => {
            eprintln!("corpus: {e}");
            std::process::exit(1);
        }
    };
    let (all, dropped) = match rows::load_rows(&built.db, space) {
        Ok(r) => r,
        Err(e) => {
            eprintln!("reading the vault back: {e}");
            std::process::exit(1);
        }
    };
    if dropped.unprojectable > 0 {
        println!(
            "  {} row(s) carry no label in this space ({}) and are excluded from",
            dropped.unprojectable,
            dropped
                .origins
                .iter()
                .cloned()
                .collect::<Vec<_>>()
                .join(", ")
        );
        println!("  every index and every probe:");
        println!("{}", families::DROPPED_NOTE);
        println!();
    }

    // Probe set: one stratified slice, held out of every index in every
    // measurement below. Fixed on purpose. A probe set that moved between
    // variants would mix "the layer changed its mind" with "we asked about
    // different samples", which is the whole thing this module exists to separate.
    let fold_of = rows::assign_folds(&all, probe_denominator);
    let probe: Vec<&Row> = all
        .iter()
        .zip(&fold_of)
        .filter(|(_, f)| **f == Some(0))
        .map(|(r, _)| r)
        .collect();
    let pool: Vec<&Row> = all
        .iter()
        .zip(&fold_of)
        .filter(|(_, f)| **f != Some(0))
        .map(|(r, _)| r)
        .collect();

    let classes: Vec<String> = all
        .iter()
        .filter_map(|r| r.truth.clone())
        .collect::<BTreeSet<_>>()
        .into_iter()
        .collect();

    println!(
        "  {} row(s) scoreable: {} held out as the fixed probe, {} in the pool every",
        all.len(),
        probe.len(),
        pool.len()
    );
    println!("  index is drawn from. No probe sample is ever an exemplar.");
    for c in &classes {
        let in_probe = probe
            .iter()
            .filter(|r| r.truth.as_deref() == Some(c.as_str()))
            .count();
        let in_pool = pool
            .iter()
            .filter(|r| r.truth.as_deref() == Some(c.as_str()))
            .count();
        println!(
            "    {:<14} probe {:>4}   pool {:>4}",
            families::label_for(space, c),
            in_probe,
            in_pool
        );
    }
    println!();

    if probe.is_empty() || pool.is_empty() {
        eprintln!("nothing to measure: probe or pool is empty");
        std::process::exit(1);
    }

    let mut report = Report::new("layer-stability");
    report.set("label_space", format!("{space:?}"));
    report.set("k", k);
    report.set("feat_version", FEATURE_VERSION);
    report.set("bar_flip_rate", BAR);
    report.set("review_threshold", DEFAULT_REVIEW_THRESHOLD);
    report.set("import_weight", rows::IMPORT_WEIGHT);
    report.set("probe", probe.len());
    report.set("pool", pool.len());

    review_threshold_note(&pool, &probe, &classes, k, space, &mut report);
    let miss = value_add_population(&probe, &mut report);
    let m1 = composition(&pool, &probe, &classes, k, space, &mut report);
    let m2 = size(&pool, &probe, &classes, k, &mut report);
    let m3 = deployment_shape(
        &pool,
        &probe,
        k,
        space,
        "3. DEPLOYMENT SHAPE",
        "deploy",
        &mut report,
    );
    let m3b = value_add_deployment(&pool, &miss, k, space, &mut report);
    let m4 = feedback(&pool, k, space, &mut report);

    let mut outcomes = vec![m1, m2, m3];
    outcomes.extend(m3b);
    outcomes.push(m4);
    verdict(&outcomes, space, &classes, &mut report);
    report.write();
}

/// Whether the review threshold is doing any work at this label resolution.
///
/// It is not, on a two-class corpus, and that has to be said before any number
/// below is read. A sample's score for a tag is that tag's share of the
/// neighbourhood's kernel weight, so when every exemplar carries exactly one of
/// two tags the two scores sum to 1 and the higher one is at or above 0.5 by
/// arithmetic. `silent` is unreachable, every probe sample is always a queue
/// entry, and the `appeared` / `vanished` columns are structurally zero.
///
/// Two consequences a reader must not miss:
///
/// - The queue-length half of "does the queue hold still" is untested here. Only
///   the contents were measured, because the length cannot move.
/// - Any metric defined as "the layer answers where the user's own labels do not"
///   is identically zero for the same reason, whatever the layer is worth. That
///   is why [`deployment_shape`] measures the layer's contribution as a change in
///   the answer and its correctness rather than as an answer appearing.
///
/// This resolves at three classes or more and is not a property of the layer, so
/// it is a scope note on the corpus rather than a finding about the code.
fn review_threshold_note(
    pool: &[&Row],
    probe: &[&Row],
    classes: &[String],
    k: usize,
    space: LabelSpace,
    report: &mut Report,
) {
    println!("━━━ IS THE REVIEW THRESHOLD BINDING? ━━━");
    println!();

    let full = run_variant(pool, &[], probe, k, "full pool");
    let silent = full.iter().filter(|a| a.tag().is_none()).count();
    println!(
        "  Against the full pool, {} of {} probe samples fall below the {} review",
        silent,
        probe.len(),
        DEFAULT_REVIEW_THRESHOLD
    );
    println!("  threshold and stay out of the queue.");
    println!();

    report.set("silent_at_full_pool", silent);
    report.set("classes", classes.len());

    if silent == 0 && classes.len() == 2 {
        println!("  Zero, and it is arithmetic rather than luck. A score is a tag's share of");
        println!("  the neighbourhood's kernel weight; with every exemplar carrying exactly one");
        println!("  of two tags the two shares sum to 1, so the larger is always at or above");
        println!("  0.5. On this corpus `silent` is unreachable and the threshold decides");
        println!("  nothing.");
        println!();
        println!("  What that costs the measurements below, stated plainly:");
        println!();
        println!("  - The queue can only change its CONTENTS, never its LENGTH. The churn");
        println!("    columns are structurally zero and prove nothing.");
        println!("  - 'The layer answers where the user's labels do not' is identically zero");
        println!("    for every layer, good or useless. Measurement 3 therefore reads the");
        println!("    layer's contribution off the answer and its correctness instead.");
        println!();
        println!("  Both resolve at three classes or more, so this is a limit of the drums-");
        println!("  only corpus and not a property of the layer. Phase C is where it lifts.");
        report.set("review_threshold_binding", false);
    } else {
        report.set("review_threshold_binding", true);
    }
    println!();
    let _ = space;
}

/// A named measurement's answer to "did it pass".
struct Outcome {
    name: &'static str,
    /// `None` when the measurement could not be run at all, which is not a pass.
    worst: Option<f64>,
    note: String,
}

// The value-add population

/// How much of the probe set the filename rules already answer.
///
/// The layer's stated job is libraries whose filenames say nothing.
/// `starter_rules` labels 97.7% of this corpus correctly off the name alone, and
/// every measurement of this layer so far — accuracy and now stability — runs on
/// exactly that population. So the number that matters is measured on the
/// complement, and this reports whether the complement is big enough to measure
/// on at all.
///
/// Returns the complement — the probe samples no filename rule answers — so
/// [`deployment_shape`] can be re-run on it. Empty when it is too thin to carry a
/// rate, which is a real possibility on a corpus whose folder labels were derived
/// from these same filenames. Phase C is where it stops being marginal: NSynth
/// names are `bass_synthetic_033-052-100`-shaped and carry no instrument keyword
/// the starter pack knows.
fn value_add_population<'a>(probe: &[&'a Row], report: &mut Report) -> Vec<&'a Row> {
    println!("━━━ THE VALUE-ADD POPULATION ━━━");
    println!();

    let Some(name_rules) = filename_rules() else {
        println!("  could not seed the starter rules; skipped");
        println!();
        return Vec::new();
    };

    let mut hit = 0usize;
    let mut miss: Vec<&Row> = Vec::new();
    for r in probe {
        let ctx = RuleContext {
            name: r.name.clone(),
            ..RuleContext::default()
        };
        if name_rules
            .iter()
            .any(|rule| rules::rule_matches(rule, &ctx))
        {
            hit += 1;
        } else {
            miss.push(r);
        }
    }

    let rate = hit as f64 / probe.len() as f64;
    println!(
        "  {} of {} probe samples ({}) already carry a filename a starter rule fires on.",
        hit,
        probe.len(),
        pct(Some(rate))
    );
    println!(
        "  The layer's job is the other {}: libraries whose filenames say nothing.",
        miss.len()
    );
    println!("  This counts only rules reading the filename, and only whether one FIRES —");
    println!("  not whether it fires correctly. It is a different quantity from the 97.7%");
    println!("  in `af-browse-axes`, which is an accuracy over the whole corpus, so the two");
    println!("  are not comparable and the gap between them is not a regression.");
    println!();
    if miss.len() < MIN_COMPARABLE {
        println!("  Too thin to carry a flip rate, and thin by construction: this corpus's");
        println!("  folder labels were derived from these same filenames, so a rule-miss here");
        println!("  is close to a corpus artifact. Every number below is therefore measured on");
        println!("  the population the rules already answer, which is the easy half.");
        println!("  Phase C (24cd7747) is where this lifts.");
        println!();
        report.set("rule_hit", hit);
        report.set("rule_miss", miss.len());
        report.set("rule_hit_rate", round4(rate));
        return Vec::new();
    }
    println!("  Large enough to carry a rate. Measurement 3 is re-run on it below, because");
    println!("  a layer that is stable and useful only where the filename already said the");
    println!("  answer is stable and useful nowhere that matters.");
    println!();

    report.set("rule_hit", hit);
    report.set("rule_miss", miss.len());
    report.set("rule_hit_rate", round4(rate));
    miss
}

/// The starter pack's filename rules, enabled, as the app would evaluate them.
///
/// Seeded into a throwaway in-memory vault rather than reconstructed from
/// `starter_rules::rules()` by hand: the seeding path is what assigns ids and
/// priorities, and guard rules depend on that order. Rules touching any field but
/// the name are dropped — this asks what the *filename* already answers, and a
/// rule reading spectral centroid is the classifier's own input wearing a
/// different hat.
fn filename_rules() -> Option<Vec<rules::Rule>> {
    let db = audiofiles_core::db::Database::open_in_memory().ok()?;
    starter_rules::seed(&db).ok()?;
    let all = rules::list_rules(&db).ok()?;
    Some(
        all.into_iter()
            .filter(|r| {
                !r.conditions.is_empty() && r.conditions.iter().all(|c| c.field == RuleField::Name)
            })
            .collect(),
    )
}

// 1. Composition

/// Same index size, different class mix.
///
/// Two users with libraries of the same size but different contents are running
/// the same code against different models, and this is the spread between them.
/// The variants are held to one common size so composition is not confounded with
/// [`size`]: an index that is both bigger and differently shaped explains nothing.
///
/// On this corpus the label space is two families, so the mix is varied both
/// between them (`low`-heavy against `bright`-heavy) and *within* `low` by corpus
/// origin (all-kick against all-tom). The second is the sharper test: `low` is
/// kick plus tom and `af-coarse-families` leaves open whether those behave as one
/// family, so a user whose low end is all toms and one whose low end is all kicks
/// are the realistic worst case for a family-level answer.
fn composition(
    pool: &[&Row],
    probe: &[&Row],
    classes: &[String],
    k: usize,
    space: LabelSpace,
    report: &mut Report,
) -> Outcome {
    println!("━━━ 1. COMPOSITION: same size, different mix ━━━");
    println!();

    let per_class: Vec<usize> = classes
        .iter()
        .map(|c| {
            pool.iter()
                .filter(|r| r.truth.as_deref() == Some(c.as_str()))
                .count()
        })
        .collect();
    let smallest = per_class.iter().copied().min().unwrap_or(0);
    if smallest < 20 || classes.len() < 2 {
        println!("  the pool cannot supply two differently-shaped indexes of a common size");
        println!();
        return Outcome {
            name: "composition",
            worst: None,
            note: "not runnable on this corpus".into(),
        };
    }

    // Common size: what the most lopsided mix can afford. 80/20 over the
    // smallest class is the binding constraint.
    let minor = smallest / 4;
    let major = smallest;
    let total = major + minor;

    let mut variants: Vec<(String, Vec<&Row>)> = Vec::new();
    let balanced_each = total / classes.len();
    variants.push((
        "balanced".into(),
        classes
            .iter()
            .flat_map(|c| take_class(pool, c, balanced_each))
            .collect(),
    ));
    for heavy in classes {
        let mut v = Vec::new();
        for c in classes {
            let n = if c == heavy {
                major
            } else {
                minor / (classes.len() - 1).max(1)
            };
            v.extend(take_class(pool, c, n));
        }
        variants.push((format!("{}-heavy", families::label_for(space, heavy)), v));
    }
    // Within-family origin variants, where the corpus has more than one folder
    // behind a class.
    for (label, origin) in origin_variants(pool, classes) {
        let mut v = Vec::new();
        for c in classes {
            let want = if Some(c.as_str()) == origin.class.as_deref() {
                pool.iter()
                    .filter(|r| r.truth.as_deref() == Some(c.as_str()) && r.origin == origin.folder)
                    .take(balanced_each)
                    .copied()
                    .collect()
            } else {
                take_class(pool, c, balanced_each)
            };
            v.extend(want);
        }
        variants.push((label, v));
    }

    println!("  Each index holds one class mix at a common size; the probe set never moves.");
    println!();
    println!("  {:<22} {:>8}   class mix", "variant", "n");
    println!("  {}", "─".repeat(72));
    let mut computed: Vec<(String, Answers)> = Vec::new();
    for (name, exemplars) in &variants {
        let mix: Vec<String> = classes
            .iter()
            .map(|c| {
                let n = exemplars
                    .iter()
                    .filter(|r| r.truth.as_deref() == Some(c.as_str()))
                    .count();
                format!("{} {}", families::label_for(space, c), n)
            })
            .collect();
        println!("  {:<22} {:>8}   {}", name, exemplars.len(), mix.join(", "));
        let a = run_variant(exemplars, &[], probe, k, name);
        computed.push((name.clone(), a));
    }
    println!();

    let worst = print_pairwise(&computed, report, "composition");
    Outcome {
        name: "composition",
        worst,
        note: format!("{} variants", computed.len()),
    }
}

/// A within-class origin split worth building a variant for.
struct OriginVariant {
    class: Option<String>,
    folder: String,
}

/// Classes the corpus fills from more than one folder, and the folders behind
/// them. `low` is kick plus tom; an all-kick and an all-tom index are the same
/// class holding two different things.
fn origin_variants(pool: &[&Row], classes: &[String]) -> Vec<(String, OriginVariant)> {
    let mut out = Vec::new();
    for c in classes {
        let mut folders: BTreeMap<&str, usize> = BTreeMap::new();
        for r in pool
            .iter()
            .filter(|r| r.truth.as_deref() == Some(c.as_str()))
        {
            *folders.entry(r.origin.as_str()).or_default() += 1;
        }
        if folders.len() < 2 {
            continue;
        }
        for (folder, n) in folders {
            if n < 20 {
                continue;
            }
            out.push((
                format!("{folder}-only"),
                OriginVariant {
                    class: Some(c.clone()),
                    folder: folder.to_string(),
                },
            ));
        }
    }
    out
}

/// Every pair of variants' flip rate, and the worst of them.
fn print_pairwise(computed: &[(String, Answers)], report: &mut Report, key: &str) -> Option<f64> {
    println!("  Pairwise flip rate (both suggested, tag changed):");
    println!();
    println!(
        "  {:<22} {:<22} {:>7} {:>8} {:>9} {:>9}",
        "a", "b", "both", "flips", "rate", "churn"
    );
    println!("  {}", "─".repeat(84));

    let mut worst: Option<f64> = None;
    for i in 0..computed.len() {
        for j in (i + 1)..computed.len() {
            let c = Churn::between(&computed[i].1, &computed[j].1);
            let rate = c.flip_rate();
            if let Some(r) = c.comparable_flip_rate() {
                worst = Some(worst.map_or(r, |w: f64| w.max(r)));
            }
            println!(
                "  {:<22} {:<22} {:>7} {:>8} {:>9} {:>9}",
                computed[i].0,
                computed[j].0,
                c.both,
                c.flips,
                pct(rate),
                format!("+{} -{}", c.appeared, c.vanished)
            );
        }
    }
    println!();
    match worst {
        Some(w) => {
            println!(
                "  worst pairwise flip rate {}  ({} the {:.0}% bar)",
                pct(Some(w)),
                if w <= BAR { "within" } else { "OVER" },
                BAR * 100.0
            );
            report.set(&format!("{key}_worst_flip_rate"), round4(w));
        }
        None => println!("  no pair had a suggested answer in common; nothing to compare"),
    }
    println!();
    worst
}

// 2. Size

/// Same mix, growing index: what a user's second month looks like.
///
/// Nested by construction — each index is a prefix of the next, so growth is
/// accretion rather than resampling and a flip is the layer changing its mind
/// about a sample rather than an artifact of two unrelated draws. The number to
/// read is the *consecutive* rate: whether it decays toward the bar says whether
/// the answer stabilises, and where it crosses says at what library size.
fn size(
    pool: &[&Row],
    probe: &[&Row],
    classes: &[String],
    k: usize,
    report: &mut Report,
) -> Outcome {
    println!("━━━ 2. SIZE: same mix, growing index ━━━");
    println!();
    println!("  Each index is a prefix of the next: a library accretes, it does not");
    println!("  resample. Read the consecutive column — the pairwise-with-full column is");
    println!("  the same information seen from the end state.");
    println!();

    const FRACTIONS: &[f64] = &[0.1, 0.2, 0.4, 0.6, 0.8, 1.0];
    let mut computed: Vec<(String, Answers, usize)> = Vec::new();
    for f in FRACTIONS {
        let exemplars: Vec<&Row> = classes
            .iter()
            .flat_map(|c| {
                let have = pool
                    .iter()
                    .filter(|r| r.truth.as_deref() == Some(c.as_str()))
                    .count();
                #[allow(clippy::cast_possible_truncation, clippy::cast_sign_loss)]
                take_class(pool, c, (have as f64 * f).round() as usize)
            })
            .collect();
        if exemplars.is_empty() {
            continue;
        }
        let label = format!("{:.0}%", f * 100.0);
        let a = run_variant(&exemplars, &[], probe, k, &label);
        computed.push((label, a, exemplars.len()));
    }
    if computed.len() < 2 {
        println!("  the pool is too small to sweep");
        println!();
        return Outcome {
            name: "size",
            worst: None,
            note: "not runnable on this corpus".into(),
        };
    }

    println!(
        "  {:<8} {:>8} {:>8} {:>8} {:>10} {:>12}",
        "index", "n", "queued", "flips", "vs prev", "vs full"
    );
    println!("  {}", "─".repeat(60));

    let full = &computed[computed.len() - 1].1;
    let mut worst_consecutive: Option<f64> = None;
    for (i, (label, a, n)) in computed.iter().enumerate() {
        let queued = a.iter().filter(|x| x.tag().is_some()).count();
        let prev = (i > 0).then(|| Churn::between(&computed[i - 1].1, a));
        let vs_full = Churn::between(a, full);
        if let Some(r) = prev.as_ref().and_then(Churn::comparable_flip_rate) {
            worst_consecutive = Some(worst_consecutive.map_or(r, |w: f64| w.max(r)));
        }
        println!(
            "  {:<8} {:>8} {:>8} {:>8} {:>10} {:>12}",
            label,
            n,
            queued,
            prev.as_ref().map_or(0, |c| c.flips),
            prev.as_ref()
                .map_or_else(|| "-".to_string(), |c| pct(c.flip_rate())),
            pct(vs_full.flip_rate()),
        );
        report.set(
            &format!("size_{}_queued", label.trim_end_matches('%')),
            queued,
        );
    }
    println!();

    // Where it settles: the first index size from which every later step stays
    // within the bar. "Stabilises" has to mean it stays stable, not that one step
    // happened to be quiet.
    let settles = (0..computed.len() - 1).find(|&i| {
        (i..computed.len() - 1).all(|j| Churn::between(&computed[j].1, &computed[j + 1].1).passes())
    });
    let settle_note = settles.map_or_else(
        || "never settles".to_string(),
        |i| format!("settles at {} exemplars", computed[i].2),
    );
    match settles {
        Some(i) => {
            println!(
                "  Settles at {} ({} exemplars): every step from there stays within the bar.",
                computed[i].0, computed[i].2
            );
            report.set("size_settles_at", computed[i].2);
        }
        None => {
            println!("  Never settles: no index size after which every step stays within the");
            println!("  bar. On this pool that is a statement about the pool as much as the");
            println!("  layer — the largest index here is still small.");
        }
    }
    println!();
    if let Some(w) = worst_consecutive {
        report.set("size_worst_consecutive_flip_rate", round4(w));
    }

    Outcome {
        name: "size",
        worst: worst_consecutive,
        note: format!("{} steps, {settle_note}", computed.len()),
    }
}

// 3. Deployment shape

/// The mixed index the app actually builds, and the one nothing had measured.
///
/// An imported layer sits at [`rows::IMPORT_WEIGHT`] beneath the user's own labels
/// at 1.0. Every earlier number came off a uniform index where that weight
/// cancels; here it does not, and neither does `build_index` fitting the
/// standardization params over both populations together.
///
/// Two questions, and the second is the one the ship decision turns on:
///
/// - **Stability.** As the user's own labels accumulate beside a fixed layer, does
///   the queue for the samples they have *not* labelled hold still? That is the
///   consecutive flip rate, measured at the deployment weight, and it is what
///   [`BAR`] is set against.
/// - **Value add.** At each user-library size, what does the layer contribute over
///   the user's own labels alone? Measured as the probe samples the mixed index
///   answers and the user-only index does not, and how often that answer is right.
///   A layer whose contribution goes to zero by the time a user has a few hundred
///   labels is a first-week feature, which is a fine thing to be and a different
///   thing from what it is currently described as.
fn deployment_shape(
    pool: &[&Row],
    probe: &[&Row],
    k: usize,
    space: LabelSpace,
    population: &str,
    key_prefix: &str,
    report: &mut Report,
) -> Outcome {
    println!(
        "━━━ {population}: user labels at 1.0, layer at {} ━━━",
        rows::IMPORT_WEIGHT
    );
    println!();

    let (layer_rows, user_pool) = halve(pool);
    println!(
        "  The pool splits stratified into a simulated imported layer ({} exemplars,\n  \
         imported at {}) and a user pool ({} labels, weight 1.0) the user's own\n  \
         library is drawn from. Same class mix on both sides.",
        layer_rows.len(),
        rows::IMPORT_WEIGHT,
        user_pool.len()
    );
    println!();

    const USER_FRACTIONS: &[f64] = &[0.0, 0.05, 0.1, 0.25, 0.5, 1.0];
    let classes: Vec<String> = user_pool
        .iter()
        .filter_map(|r| r.truth.clone())
        .collect::<BTreeSet<_>>()
        .into_iter()
        .collect();

    println!(
        "  {:<8} {:>7} {:>9} {:>9} {:>8} {:>9} {:>9} {:>7}",
        "user", "labels", "vs prev", "vs layer", "differs", "mixed ok", "user ok", "delta"
    );
    println!("  {}", "─".repeat(74));

    let mut computed: Vec<(String, Answers)> = Vec::new();
    let mut worst_consecutive: Option<f64> = None;
    let mut last_value_add: Option<(usize, f64)> = None;

    for f in USER_FRACTIONS {
        let user: Vec<&Row> = classes
            .iter()
            .flat_map(|c| {
                let refs: Vec<&Row> = user_pool.clone();
                let have = refs
                    .iter()
                    .filter(|r| r.truth.as_deref() == Some(c.as_str()))
                    .count();
                #[allow(clippy::cast_possible_truncation, clippy::cast_sign_loss)]
                let n = (have as f64 * f).round() as usize;
                take_class(&refs, c, n)
            })
            .collect();

        let label = format!("{:.0}%", f * 100.0);
        let mixed = run_variant(&user, &layer_rows, probe, k, &label);
        // The counterfactual: the same user labels with no layer at all. This is
        // what the layer has to beat to be worth shipping.
        let user_only = run_variant(&user, &[], probe, k, &format!("{label} user-only"));

        let queued = mixed.iter().filter(|a| a.tag().is_some()).count();
        let prev = computed.last().map(|(_, a)| Churn::between(a, &mixed));
        if let Some(r) = prev.as_ref().and_then(Churn::comparable_flip_rate) {
            worst_consecutive = Some(worst_consecutive.map_or(r, |w: f64| w.max(r)));
        }
        let vs_layer_only = computed
            .first()
            .map(|(_, a)| Churn::between(a, &mixed))
            .and_then(|c| c.flip_rate());

        // What the layer contributes, read off the answer rather than off an
        // answer appearing. "The layer answers where the user's labels do not"
        // is identically zero at two classes whatever the layer is worth (see
        // `review_threshold_note`), so the contribution is measured as: how often
        // the two disagree, and whether the layer's presence leaves the probe
        // more often right. The delta is the number the ship decision turns on —
        // a layer that changes answers without improving them is spending the
        // user's attention for nothing.
        let differs = (0..probe.len())
            .filter(|&i| mixed[i].tag() != user_only[i].tag())
            .count();
        let accuracy = |a: &Answers| {
            a.iter()
                .zip(probe)
                .filter(|(x, r)| x.tag().is_some() && x.tag() == r.truth.as_deref())
                .count() as f64
                / probe.len() as f64
        };
        let mixed_ok = accuracy(&mixed);
        let user_ok = accuracy(&user_only);
        last_value_add = Some((differs, mixed_ok - user_ok));

        println!(
            "  {:<8} {:>7} {:>9} {:>9} {:>8} {:>9} {:>9} {:>+7.1}",
            label,
            user.len(),
            prev.as_ref()
                .map_or_else(|| "-".to_string(), |c| pct(c.flip_rate())),
            pct(vs_layer_only),
            differs,
            pct(Some(mixed_ok)),
            pct(Some(user_ok)),
            (mixed_ok - user_ok) * 100.0,
        );

        let key = label.trim_end_matches('%');
        report.set(&format!("{key_prefix}_user{key}_queued"), queued);
        report.set(&format!("{key_prefix}_user{key}_differs"), differs);
        report.set(
            &format!("{key_prefix}_user{key}_mixed_accuracy"),
            round4(mixed_ok),
        );
        report.set(
            &format!("{key_prefix}_user{key}_user_accuracy"),
            round4(user_ok),
        );

        computed.push((label, mixed));
    }
    println!();
    println!("  'differs' is probe samples where the mixed index and the same user labels");
    println!("  alone give different answers; 'delta' is what the layer's presence does to");
    println!("  accuracy, in points. Together they are the layer's contribution surviving");
    println!("  contact with a user's own data.");
    println!();
    println!("  The 0% row is the layer alone, which is the cold-start case and the only row");
    println!("  where 'user ok' is not a real alternative: a user with no labels has nothing");
    println!("  to fall back on.");
    println!();

    if let Some((differs, delta)) = last_value_add {
        println!(
            "  At a full user library the layer changes {differs} answer(s) and moves accuracy\n  \
             by {:+.1} points.",
            delta * 100.0
        );
        if differs == 0 {
            println!();
            println!("  Zero is the strong version of the cold-start reading: once the user has");
            println!("  their own labels the layer is not merely adding little, it is changing");
            println!("  nothing at all. Worth shipping for the first week, and the documentation");
            println!("  has to say that rather than describe an ongoing contribution.");
        } else if delta.abs() < 0.005 {
            println!();
            println!("  Answers move and accuracy does not. That is the worst shape available:");
            println!("  the user re-reads a queue for no gain.");
        }
        println!();
    }

    if let Some(w) = worst_consecutive {
        report.set(
            &format!("{key_prefix}_worst_consecutive_flip_rate"),
            round4(w),
        );
        println!(
            "  worst consecutive flip rate at the deployment weight {}  ({} the {:.0}% bar)",
            pct(Some(w)),
            if w <= BAR { "within" } else { "OVER" },
            BAR * 100.0
        );
        println!();
    }

    let _ = space;
    Outcome {
        name: "deployment shape",
        worst: worst_consecutive,
        note: format!("layer {} + user pool {}", layer_rows.len(), user_pool.len()),
    }
}

/// [`deployment_shape`] again, restricted to the probe samples no filename rule
/// answers.
///
/// The layer's stated job is libraries whose filenames say nothing, and every
/// number in this file up to here is measured on the population the starter rules
/// already handle. This is the same measurement on the population that motivates
/// the feature. It is the one to read first when the two disagree.
fn value_add_deployment(
    pool: &[&Row],
    miss: &[&Row],
    k: usize,
    space: LabelSpace,
    report: &mut Report,
) -> Option<Outcome> {
    if miss.is_empty() {
        return None;
    }
    println!(
        "  Probe restricted to the {} samples no filename rule answers.",
        miss.len()
    );
    println!();
    let mut o = deployment_shape(
        pool,
        miss,
        k,
        space,
        "3b. DEPLOYMENT SHAPE, VALUE-ADD POPULATION ONLY",
        "valueadd",
        report,
    );
    o.name = "value-add deployment";
    Some(o)
}

// 4. Accepted-tag feedback

/// The queue changes its own inputs.
///
/// An accepted suggestion becomes a user label at weight 1.0 and re-enters the
/// index, so working through the queue rewrites the rest of it. This is the one
/// property no held-out measurement can see, because the thing that moves is the
/// index and the thing being measured is the same set of samples.
///
/// Simulated as a user actually working: build the mixed index, queue the
/// unlabelled library, accept the highest-scoring batch, fold those in as weight
/// 1.0 labels, rebuild, repeat. Two acceptance policies, because they bracket
/// real behaviour and they fail differently:
///
/// - **accept-all** — the user trusts the queue. Wrong suggestions become wrong
///   exemplars at full weight, so this is where a feedback loop would compound.
/// - **accept-correct** — an oracle user who accepts only right answers. If this
///   one oscillates, the instability is in the mechanism and not in the user's
///   mistakes.
///
/// What is measured per round, over the samples still unlabelled and suggested
/// before and after: the flip rate (does it decay), and per sample whether its
/// answer has changed more than once across the whole run (oscillation, which a
/// decaying average can hide).
fn feedback(pool: &[&Row], k: usize, space: LabelSpace, report: &mut Report) -> Outcome {
    println!("━━━ 4. ACCEPTED-TAG FEEDBACK: the queue rewrites itself ━━━");
    println!();

    let (layer_rows, user_pool) = halve(pool);
    // A small seed, because the interesting case is a user who has barely
    // labelled anything and is leaning on the queue to bootstrap.
    let seed_n = (user_pool.len() / 20).max(4);
    let seed: Vec<&Row> = user_pool.iter().take(seed_n).copied().collect();
    let library: Vec<&Row> = user_pool.iter().skip(seed_n).copied().collect();
    let batch = (library.len() / 8).max(1);

    println!(
        "  Layer {} at {}, user seed {} at 1.0, library {} unlabelled. Each round",
        layer_rows.len(),
        rows::IMPORT_WEIGHT,
        seed.len(),
        library.len()
    );
    println!("  accepts the {batch} highest-scoring suggestions and rebuilds the index.");
    println!();

    if library.len() < 20 {
        println!("  library too small to work through");
        println!();
        return Outcome {
            name: "feedback",
            worst: None,
            note: "not runnable on this corpus".into(),
        };
    }

    let mut worst: Option<f64> = None;
    for accept_all in [true, false] {
        let policy = if accept_all {
            "accept-all"
        } else {
            "accept-correct"
        };
        println!("  {policy}:");
        println!();
        println!(
            "    {:<7} {:>8} {:>8} {:>8} {:>9} {:>9}",
            "round", "labels", "open", "queued", "flips", "rate"
        );
        println!("    {}", "─".repeat(56));

        // `accepted` are library rows folded in as user labels; `open` are the
        // rest, and only they are measured — a sample the user has labelled is no
        // longer a queue entry, so its answer moving is not churn.
        let mut accepted: Vec<&Row> = Vec::new();
        let mut open: Vec<&Row> = library.clone();
        let mut previous: Option<HashMap<&str, Answer>> = None;
        let mut changes: HashMap<&str, usize> = HashMap::new();
        let mut round = 0usize;

        while !open.is_empty() {
            let local: Vec<&Row> = seed.iter().chain(&accepted).copied().collect();
            let db = match rows::mixed_db(&local, &layer_rows) {
                Ok(db) => db,
                Err(e) => {
                    eprintln!("{policy} round {round}: {e}");
                    std::process::exit(1);
                }
            };
            let index = match exemplar::build_index(&db) {
                Ok(i) => i,
                Err(e) => {
                    eprintln!("{policy} round {round}: build_index: {e}");
                    std::process::exit(1);
                }
            };

            // Score every open sample. `exclude_hash` is unnecessary: an open
            // sample is not in the index, which is the point of holding it out.
            let scored: Vec<(&Row, Option<exemplar::TagScore>)> = open
                .iter()
                .map(|r| {
                    let top = index
                        .score(&r.vector, k, None)
                        .into_iter()
                        .next()
                        .filter(|s| s.score >= DEFAULT_REVIEW_THRESHOLD);
                    (*r, top)
                })
                .collect();

            let now: HashMap<&str, Answer> = scored
                .iter()
                .map(|(r, top)| {
                    (
                        r.hash.as_str(),
                        top.as_ref()
                            .map_or(Answer::Silent, |s| Answer::Tag(s.tag.clone())),
                    )
                })
                .collect();
            let queued = now.values().filter(|a| a.tag().is_some()).count();

            // Flips are counted only over samples open in both rounds: a sample
            // that left the library because it was accepted did not change its
            // mind, it stopped being a question.
            let (both, flips) = previous.as_ref().map_or((0, 0), |before| {
                let mut b = 0;
                let mut f = 0;
                for (hash, after) in &now {
                    let Some(prev) = before.get(hash) else {
                        continue;
                    };
                    if let (Some(p), Some(q)) = (prev.tag(), after.tag()) {
                        b += 1;
                        if p != q {
                            f += 1;
                            *changes.entry(*hash).or_default() += 1;
                        }
                    }
                }
                (b, f)
            });
            let rate = (both > 0).then(|| flips as f64 / both as f64);
            // A tail of a handful of stubborn samples produces enormous rates off
            // one flip. Printed, marked, and kept out of the verdict.
            let thin = both > 0 && both < MIN_COMPARABLE;
            if let Some(r) = rate.filter(|_| round > 0 && !thin) {
                worst = Some(worst.map_or(r, |w: f64| w.max(r)));
            }

            println!(
                "    {:<7} {:>8} {:>8} {:>8} {:>9} {:>9}{}",
                round,
                local.len(),
                open.len(),
                queued,
                flips,
                rate.map_or_else(|| "-".to_string(), |r| pct(Some(r))),
                if thin { "  (thin)" } else { "" }
            );

            // Accept the batch: highest-scoring suggestions first, which is the
            // order the review screen presents and therefore the order a user
            // works in.
            let mut candidates: Vec<(&Row, &exemplar::TagScore)> = scored
                .iter()
                .filter_map(|(r, top)| top.as_ref().map(|s| (*r, s)))
                .filter(|(r, s)| accept_all || Some(s.tag.as_str()) == r.truth.as_deref())
                .collect();
            candidates.sort_by(|a, b| b.1.score.total_cmp(&a.1.score));
            let taking: Vec<&str> = candidates
                .iter()
                .take(batch)
                .map(|(r, _)| r.hash.as_str())
                .collect();
            if taking.is_empty() {
                println!("    nothing left above the review threshold; the queue is dry");
                break;
            }
            // Accepting the layer's answer means the row enters the index with
            // the tag the layer proposed, right or wrong. Using the row's own
            // corpus tags instead would quietly make every accepted label
            // correct, which is the accept-all failure mode being simulated.
            let taken: BTreeSet<&str> = taking.iter().copied().collect();
            for (r, s) in candidates
                .iter()
                .filter(|(r, _)| taken.contains(r.hash.as_str()))
            {
                accepted.push(accepted_row(r, &s.tag));
            }
            open.retain(|r| !taken.contains(r.hash.as_str()));
            previous = Some(now);
            round += 1;
            if round > 20 {
                break;
            }
        }

        let oscillating = changes.values().filter(|n| **n > 1).count();
        println!();
        println!(
            "    (thin) marks a round with fewer than {MIN_COMPARABLE} entries still open. Those rates"
        );
        println!("    are one sample either way and do not decide the verdict.");
        println!("    {oscillating} sample(s) changed answer more than once across the run.");
        if oscillating == 0 {
            println!("    No oscillation: a sample that moved, moved once and stayed.");
        } else {
            println!("    Oscillation is the failure a decaying average hides: the mean flip");
            println!("    rate can fall while individual entries keep swapping back.");
        }
        println!();
        report.set(
            &format!("feedback_{}_oscillating", policy.replace('-', "_")),
            oscillating,
        );
    }

    if let Some(w) = worst {
        report.set("feedback_worst_flip_rate", round4(w));
    }
    let _ = space;
    Outcome {
        name: "feedback",
        worst,
        note: format!("batch {batch}"),
    }
}

/// A row as it enters the index after the user accepts `tag` for it.
///
/// Leaks a `Row` deliberately: the accepted set is built round by round and has to
/// outlive the loop iteration that created it, while every other row in the run is
/// borrowed from the corpus. The run is a bounded number of rounds over a bounded
/// library, so the leak is bounded by the library size and the process exits
/// immediately after.
fn accepted_row(r: &Row, tag: &str) -> &'static Row {
    Box::leak(Box::new(Row {
        hash: r.hash.clone(),
        vector: r.vector.clone(),
        tags: vec![tag.to_string()],
        truth: r.truth.clone(),
        origin: r.origin.clone(),
        name: r.name.clone(),
    }))
}

// Verdict

fn verdict(outcomes: &[Outcome], space: LabelSpace, classes: &[String], report: &mut Report) {
    println!("━━━ VERDICT ━━━");
    println!();
    println!("  {:<20} {:>12}", "measurement", "worst flip");
    println!("  {}", "─".repeat(64));

    let mut failed = Vec::new();
    let mut unrun = Vec::new();
    for o in outcomes {
        let state = match o.worst {
            Some(w) if w <= BAR => "within the bar",
            Some(_) => {
                failed.push(o.name);
                "OVER THE BAR"
            }
            None => {
                unrun.push(o.name);
                "not measured"
            }
        };
        println!(
            "  {:<20} {:>12}   {state} ({})",
            o.name,
            o.worst.map_or_else(|| "-".to_string(), |w| pct(Some(w))),
            o.note
        );
    }
    println!();

    report.set("failed_measurements", failed.len());
    report.set("unrun_measurements", unrun.len());

    if failed.is_empty() && unrun.is_empty() {
        println!("  PASS on every measurement that ran.");
    } else if failed.is_empty() {
        println!(
            "  PASS on what ran; {} not measurable on this corpus: {}.",
            unrun.len(),
            unrun.join(", ")
        );
    } else {
        println!("  FAIL: {}.", failed.join(", "));
    }
    println!();

    // Scope, restated at the bottom where a verdict gets quoted from.
    let (covered, uncovered) = families::covered_families(classes);
    if space != LabelSpace::Instrument && !uncovered.is_empty() {
        println!(
            "  SCOPE: this corpus reaches {} of 7 families ({}). It says nothing about",
            covered.len(),
            covered.join(", ")
        );
        println!(
            "  {}, and a stability number measured over two",
            uncovered.join(", ")
        );
        println!("  classes is an easier question than the shipped layer will face: fewer");
        println!("  classes means fewer things an answer can flip to. Read every figure above");
        println!("  as a floor on the flip rate, not an estimate of it.");
        println!();
    }
    println!("  And the standing caveat: this measures whether the answer is the SAME, not");
    println!("  whether it is RIGHT. The two are independent. layer-eval is the other half.");
    println!();
}

/// Probe fraction denominator from the env: 5 means one in five rows is held out.
pub(crate) fn probe_denominator_from_env() -> usize {
    std::env::var("AF_BENCH_STABILITY_PROBE")
        .ok()
        .and_then(|v| v.parse().ok())
        .filter(|n| *n >= 2)
        .unwrap_or(5)
}

/// The single `k` this mode runs at.
///
/// Stability is measured at the `k` that ships, not swept: `k` is a global
/// constant at runtime, so a user never experiences two of them, and sweeping it
/// here would answer a question about the mechanism when the bar is about the
/// product. `AF_BENCH_EVAL_K`'s first entry overrides it so the two modes can be
/// pointed at the same non-default `k` for a comparison.
pub(crate) fn k_from_env() -> usize {
    std::env::var("AF_BENCH_EVAL_K")
        .ok()
        .and_then(|v| v.split(',').next()?.trim().parse().ok())
        .filter(|k| *k > 0)
        .unwrap_or(DEFAULT_K)
}

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

    fn ans(tags: &[Option<&str>]) -> Answers {
        tags.iter()
            .map(|t| t.map_or(Answer::Silent, |x| Answer::Tag(x.into())))
            .collect()
    }

    #[test]
    fn churn_separates_a_flip_from_a_queue_getting_longer() {
        // The distinction the whole module turns on: a sample that gains an
        // answer costs the user a read, a sample that changes its answer costs
        // them a read they already did.
        let a = ans(&[Some("low"), Some("low"), None, Some("low")]);
        let b = ans(&[Some("low"), Some("bright"), Some("low"), None]);
        let c = Churn::between(&a, &b);
        assert_eq!(c.both, 2);
        assert_eq!(c.flips, 1);
        assert_eq!(c.appeared, 1);
        assert_eq!(c.vanished, 1);
        assert_eq!(c.flip_rate(), Some(0.5));
    }

    #[test]
    fn no_overlapping_queue_is_not_a_flip_rate_of_zero() {
        // Two runs that never both suggested anything have no measured
        // agreement. Reporting 0% would read as perfect stability.
        let a = ans(&[Some("low"), None]);
        let b = ans(&[None, Some("low")]);
        let c = Churn::between(&a, &b);
        assert_eq!(c.flip_rate(), None);
        assert!(!c.passes(), "an unmeasurable pair must not pass");
    }

    #[test]
    fn the_review_threshold_is_what_gates_an_answer() {
        // Not the auto threshold: under suggest-only nothing auto-applies, so a
        // score between review and auto is a queue entry and must count. If the
        // two ever collapse, every `Answer` here silently becomes an auto-apply
        // decision and the module is measuring the wrong line.
        const {
            assert!(DEFAULT_REVIEW_THRESHOLD < exemplar::DEFAULT_AUTO_THRESHOLD);
        }
    }

    #[test]
    fn halving_keeps_the_class_mix_on_both_sides() {
        let rows: Vec<Row> = (0..10)
            .map(|i| Row {
                hash: format!("h{i}"),
                vector: Vec::new(),
                tags: Vec::new(),
                truth: Some(if i < 6 { "low" } else { "bright" }.to_string()),
                origin: String::new(),
                name: String::new(),
            })
            .collect();
        let refs: Vec<&Row> = rows.iter().collect();
        let (a, b) = halve(&refs);
        let low = |v: &[&Row]| {
            v.iter()
                .filter(|r| r.truth.as_deref() == Some("low"))
                .count()
        };
        assert_eq!(low(&a), 3);
        assert_eq!(low(&b), 3);
        assert_eq!(a.len() + b.len(), 10);
    }

    #[test]
    fn take_class_is_a_prefix_so_the_size_sweep_nests() {
        // The size sweep's whole claim is that a bigger index contains the
        // smaller one. If this resampled, a flip would be an artifact of the draw.
        let rows: Vec<Row> = (0..6)
            .map(|i| Row {
                hash: format!("h{i}"),
                vector: Vec::new(),
                tags: Vec::new(),
                truth: Some("low".to_string()),
                origin: String::new(),
                name: String::new(),
            })
            .collect();
        let refs: Vec<&Row> = rows.iter().collect();
        let small = take_class(&refs, "low", 2);
        let big = take_class(&refs, "low", 4);
        assert_eq!(small.len(), 2);
        assert_eq!(big.len(), 4);
        assert!(
            small.iter().zip(&big).all(|(s, b)| std::ptr::eq(*s, *b)),
            "the smaller index must be a prefix of the larger"
        );
    }

    #[test]
    fn the_filename_rules_are_name_only() {
        // A rule reading spectral centroid would be the classifier's own input
        // deciding what counts as the population the classifier adds value to.
        let rules = filename_rules().expect("the starter pack seeds");
        assert!(!rules.is_empty());
        for r in &rules {
            assert!(
                r.conditions.iter().all(|c| c.field == RuleField::Name),
                "{} reads a field other than the name",
                r.name
            );
        }
    }
}