feat: add Sankoff parsimony model and directory locking

Implements a calibrated 16-state Sankoff substitution cost matrix and CLI pipeline for evolutionary distance computation, including empirical calibration via saturation-filtered SNP counts. Refactors the sibling scanning stage to use batched transforms for improved synchronization efficiency. Introduces an OS-level advisory directory lock across all index-modifying commands to prevent concurrent write corruption. Updates dependencies and exposes new Sankoff utilities in the public API.

src/obikindex/src/lib.rs

@@ -8,6 +8,7 @@ mod merge;
 mod numa;
 mod rebuild;
 mod reindex;
+mod sankoff;
 mod select;
 mod siblings;
 mod stats;
@@ -19,4 +20,8 @@ pub use merge::MergeMode;
 pub use meta::{validate_label, GenomeInfo, IndexConfig, IndexMeta, META_FILENAME};
 pub use state::{IndexState, SENTINEL_COUNTED, SENTINEL_INDEXED, SENTINEL_SCATTERED};
 pub use stats::IndexBitsPerKmer;
-pub use siblings::{RawSnpDistanceOutput, SiblingAnnexStats, SnpAlignment};
+pub use siblings::{BasePairTally, RawSnpDistanceOutput, SiblingAnnexStats, SnpAlignment};
+pub use sankoff::{
+    build_cost_matrix, c_ctx_from_p_hat, calibrate_p_hat, mean_substitution_cost,
+    substitution_costs_from_tally, PHatEstimate, SankoffWeights,
+};

src/obikindex/src/sankoff.rs

@@ -0,0 +1,402 @@
+//! Sankoff parsimony cost matrix for the 16-state (powerset of `{A,C,G,T}`)
+//! family alphabet, and calibration of its parameters from real pairwise
+//! SNP/shared counts. See `docmd/theory/evolutionary_distances.md`,
+//! "Sankoff parsimony as the resolution of the 16-state model problem".
+
+use crate::{BasePairTally, RawSnpDistanceOutput};
+
+/// Transition base pairs in the project's fixed bit convention (see
+/// `siblings::central_base`: bit 0=A, 1=C, 2=G, 3=T). All other single-bit
+/// swaps are transversions.
+const TRANSITION_PAIRS: [(u8, u8); 2] = [(0, 2), (1, 3)]; // A<->G, C<->T
+
+/// Tunable weights for the 16-state Sankoff cost matrix — see "A concrete
+/// Sankoff cost matrix for the 16-state alphabet" in the design doc.
+///
+/// Only two costs, not three: an earlier version had a separate `c_gl` for
+/// "gain/loss between two nonempty states" alongside `c_ctx` for "collapse
+/// to `∅`", but presence/absence tracking never observes "the flanks"
+/// independently of a whole 31-mer — losing one member of a family while
+/// others remain (`{A,C}->{A}`) and losing the last one (`{A}->∅`) are the
+/// same event: a specific, complete, homologous 31-mer that used to be
+/// observed no longer is. Both are exactly what `c_ctx` (see
+/// `c_ctx_from_p_hat`) prices, so it is used for every gain/loss uniformly
+/// — including a compound family losing several members at once, `∅`
+/// included, which costs `|X|*c_ctx` (no flat shortcut for `∅`: see
+/// `build_cost_matrix`'s docs for why an earlier version's flat special
+/// case there didn't hold up).
+#[derive(Debug, Clone, Copy)]
+pub struct SankoffWeights {
+    /// Cost of a single-base substitution `a <-> b` (`a != b`), symmetric
+    /// (`sub_cost[a][b] == sub_cost[b][a]`) and only meaningful off the
+    /// diagonal. The full 6-category (AC, AG, AT, CG, CT, GT) resolution a
+    /// symmetric cost matrix allows — see `substitution_costs_from_tally`
+    /// for calibrating it from real data, or `SankoffWeights::ts_tv` for
+    /// the coarser 2-category (transition/transversion) convenience
+    /// constructor.
+    pub sub_cost: [[f64; 4]; 4],
+    /// Cost of losing or gaining one family member — the same constant
+    /// everywhere a member is gained or lost, member count and target state
+    /// (`∅` or not) included: see the struct docs.
+    pub c_ctx: f64,
+}
+
+impl SankoffWeights {
+    /// Convenience constructor for the coarser 2-category
+    /// (transition/transversion) substitution model: every transition pair
+    /// (A<->G, C<->T) costs `c_ts`, every transversion pair costs `c_tv`.
+    pub fn ts_tv(c_ts: f64, c_tv: f64, c_ctx: f64) -> Self {
+        let mut sub_cost = [[0.0; 4]; 4];
+        for a in 0..4usize {
+            for b in 0..4usize {
+                if a == b {
+                    continue;
+                }
+                let is_ts = TRANSITION_PAIRS.contains(&(a as u8, b as u8));
+                sub_cost[a][b] = if is_ts { c_ts } else { c_tv };
+            }
+        }
+        Self { sub_cost, c_ctx }
+    }
+}
+
+/// Build the 16x16 Sankoff step-cost matrix, states indexed `0..=15` by
+/// their bitmask (matching `siblings::iupac_code`'s convention — state `0`
+/// is `∅`).
+///
+/// All 16 states, `∅` included, form a single graph: an edge of weight
+/// `c_ts`/`c_tv` between two states of equal cardinality differing by
+/// exactly one element (classified by whether the swapped bases form a
+/// transition or transversion pair), and an edge of weight `c_ctx` between
+/// a state and one that is its strict subset plus exactly one element
+/// (`∅` is a strict subset of every singleton state, so it connects
+/// directly to each of them this way — no special case). `cost(X, Y)` is
+/// the shortest-path distance in that graph (16 nodes — Floyd-Warshall,
+/// trivial at this size), uniformly, including `X <-> ∅`: a compound
+/// family losing several members at once, `∅` included, costs `|X|*c_ctx`
+/// via that many single-element steps — no cheaper flat alternative for
+/// `∅` specifically, since nothing distinguishes it from any other
+/// multi-element loss (see design doc — an earlier flat special case here
+/// was inconsistent with how every *other* multi-element transformation is
+/// already priced, and had no principled justification once that was
+/// noticed).
+pub fn build_cost_matrix(w: &SankoffWeights) -> [[f64; 16]; 16] {
+    const INF: f64 = f64::INFINITY;
+    let mut dist = [[INF; 16]; 16];
+    for (i, row) in dist.iter_mut().enumerate() {
+        row[i] = 0.0;
+    }
+
+    for x in 0u8..16 {
+        for y in (x + 1)..16 {
+            let (xu, yu) = (x as usize, y as usize);
+            let diff = x ^ y;
+            let card_x = x.count_ones();
+            let card_y = y.count_ones();
+            let edge = if card_x == card_y && diff.count_ones() == 2 {
+                // Exactly one element swapped: the bit only in x is the base
+                // leaving, the bit only in y is the base entering.
+                let a = (x & diff).trailing_zeros() as u8;
+                let b = (y & diff).trailing_zeros() as u8;
+                Some(w.sub_cost[a as usize][b as usize])
+            } else if card_x.abs_diff(card_y) == 1 && (x & y) == x.min(y) {
+                // True subset relationship: a pure gain/loss of one element.
+                Some(w.c_ctx)
+            } else {
+                None
+            };
+            if let Some(cost) = edge {
+                dist[xu][yu] = cost;
+                dist[yu][xu] = cost;
+            }
+        }
+    }
+
+    for k in 0..16 {
+        for i in 0..16 {
+            if dist[i][k].is_infinite() {
+                continue;
+            }
+            for j in 0..16 {
+                let via = dist[i][k] + dist[k][j];
+                if via < dist[i][j] {
+                    dist[i][j] = via;
+                }
+            }
+        }
+    }
+
+    dist
+}
+
+/// `p_hat` calibrated from real pairwise SNP/shared counts, restricted to
+/// pairs below `ratio_ceiling`, and its variance as a pooled Bernoulli
+/// proportion.
+#[derive(Debug, Clone, Copy)]
+pub struct PHatEstimate {
+    pub p_hat: f64,
+    /// `p_hat*(1-p_hat) / n_loci_total` — the pooled estimator's variance is
+    /// governed by the total number of loci across all included pairs, not
+    /// by any single pair's count (see design doc: this is why the
+    /// exclusion criterion below is the per-pair *ratio*, not a per-pair
+    /// minimum-count floor — a low-count pair barely moves the pool either
+    /// way, but a saturated pair with plenty of loci would bias it).
+    pub variance: f64,
+    pub n_pairs_included: usize,
+    pub n_loci_total: u64,
+}
+
+/// Pool `snp`/`shared` counts across all genome pairs whose per-pair ratio
+/// `snp/(snp+shared)` is at most `ratio_ceiling`, then estimate
+/// `p_hat = sum(snp) / sum(snp+shared)` over the included pairs.
+///
+/// A pair with no eligible locus at all (`snp+shared == 0`) is always
+/// excluded (nothing to pool). `ratio_ceiling` should be well below 1.0 —
+/// pairs at or near saturation (e.g. cross-domain comparisons, where nearly
+/// every shared central-position family already differs) carry no
+/// information about `p_hat` and bias it upward if pooled in.
+pub fn calibrate_p_hat(raw: &RawSnpDistanceOutput, ratio_ceiling: f64) -> PHatEstimate {
+    let n = raw.snp.nrows();
+    let mut snp_sum: u64 = 0;
+    let mut total_sum: u64 = 0;
+    let mut n_pairs = 0usize;
+    for i in 0..n {
+        for j in (i + 1)..n {
+            let snp = raw.snp[[i, j]];
+            let shared = raw.shared[[i, j]];
+            let total = snp + shared;
+            if total == 0 {
+                continue;
+            }
+            let ratio = snp as f64 / total as f64;
+            if ratio > ratio_ceiling {
+                continue;
+            }
+            snp_sum += snp;
+            total_sum += total;
+            n_pairs += 1;
+        }
+    }
+    let p_hat = if total_sum > 0 { snp_sum as f64 / total_sum as f64 } else { 0.0 };
+    let variance = if total_sum > 0 {
+        p_hat * (1.0 - p_hat) / total_sum as f64
+    } else {
+        0.0
+    };
+    PHatEstimate { p_hat, variance, n_pairs_included: n_pairs, n_loci_total: total_sum }
+}
+
+/// `c_ctx(p) = mean_sub_cost * [(2m*p) / (1 - (1-p)^(2m)) + p]` — see
+/// "Context, detectability, and a 3-way ordinal distance per pair" in the
+/// design doc for the bracketed term's derivation. `m` is the flank length
+/// on *each* side of the central position (`k = 2m+1`); pass `(k-1)/2`, not
+/// the project's minimizer-size parameter of the same name.
+///
+/// The bracketed term is `E[mutation count | at least one occurred]` — a
+/// *count* of mutations, not a cost. An earlier version used it directly as
+/// the cost, implicitly pricing every mutation at a flat `1` regardless of
+/// type. That stopped making sense once substitution costs were calibrated
+/// per base-pair type (`substitution_costs_from_tally`): transitions and
+/// transversions aren't equally likely (roughly 5-6x apart in this
+/// project's own data) nor equally costly, so "one mutation" isn't worth a
+/// flat unit — it's worth whatever the *empirical mix* of mutation types is
+/// worth on average. `mean_sub_cost` (see `mean_substitution_cost`) is that
+/// weighted average, turning the expected mutation *count* into an actual
+/// expected cost.
+pub fn c_ctx_from_p_hat(p_hat: f64, m: usize, mean_sub_cost: f64) -> f64 {
+    if p_hat <= 0.0 {
+        return 0.0;
+    }
+    let two_m = 2.0 * m as f64;
+    let expected_mutations = (two_m * p_hat) / (1.0 - (1.0 - p_hat).powf(two_m)) + p_hat;
+    expected_mutations * mean_sub_cost
+}
+
+/// Mean substitution cost, weighted by each of the 6 base-pair categories'
+/// observed frequency in `tally` — the empirical average cost of "one
+/// mutation" under the calibrated substitution spectrum. Falls back to
+/// `1.0` (a flat per-mutation cost) if `tally` has no signal at all (no
+/// substitution ever observed), rather than dividing by zero.
+pub fn mean_substitution_cost(tally: &BasePairTally, sub_cost: &[[f64; 4]; 4]) -> f64 {
+    let mut weighted_sum = 0.0;
+    let mut total = 0u64;
+    for a in 0..4 {
+        for b in (a + 1)..4 {
+            let count = tally.counts[a][b];
+            weighted_sum += count as f64 * sub_cost[a][b];
+            total += count;
+        }
+    }
+    if total == 0 {
+        return 1.0;
+    }
+    weighted_sum / total as f64
+}
+
+/// Derive the 4x4 symmetric substitution cost table from an observed
+/// [`BasePairTally`]: `cost(a,b) = -ln(rate(a,b))`, normalised so the most
+/// frequently observed substitution type costs exactly `1.0` — the standard
+/// generalised-parsimony step-weighting heuristic (see design doc,
+/// "Transition/transversion refinement"), generalised here from 2
+/// categories (Ts/Tv) to the full 6-category resolution a symmetric matrix
+/// allows: a rarer substitution type is treated as less parsimonious to
+/// invoke, and therefore costs more.
+///
+/// A pair type never observed at all (`counts[a][b] == 0`) gets an infinite
+/// cost — parsimony should never spend a mutation on something with no
+/// empirical support in this data.
+pub fn substitution_costs_from_tally(tally: &BasePairTally) -> [[f64; 4]; 4] {
+    let total: u64 = (0..4)
+        .flat_map(|a| (a + 1..4).map(move |b| (a, b)))
+        .map(|(a, b)| tally.counts[a][b])
+        .sum();
+
+    let mut raw = [[0.0f64; 4]; 4];
+    let mut min_cost = f64::INFINITY;
+    for a in 0..4 {
+        for b in (a + 1)..4 {
+            let count = tally.counts[a][b];
+            let cost = if count > 0 && total > 0 {
+                -((count as f64 / total as f64).ln())
+            } else {
+                f64::INFINITY
+            };
+            raw[a][b] = cost;
+            raw[b][a] = cost;
+            if cost < min_cost {
+                min_cost = cost;
+            }
+        }
+    }
+
+    let mut out = [[0.0f64; 4]; 4];
+    for a in 0..4 {
+        for b in 0..4 {
+            if a != b {
+                out[a][b] = raw[a][b] / min_cost;
+            }
+        }
+    }
+    out
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use ndarray::Array2;
+
+    fn weights() -> SankoffWeights {
+        SankoffWeights::ts_tv(1.0, 2.0, 10.0)
+    }
+
+    #[test]
+    fn identity_is_zero() {
+        let m = build_cost_matrix(&weights());
+        for row in m.iter().enumerate() {
+            assert_eq!(m[row.0][row.0], 0.0);
+        }
+    }
+
+    #[test]
+    fn single_base_transition_costs_c_ts() {
+        let m = build_cost_matrix(&weights());
+        // A = 0b0001 (state 1), G = 0b0100 (state 4): A<->G is a transition.
+        assert_eq!(m[1][4], 1.0);
+        assert_eq!(m[4][1], 1.0);
+    }
+
+    #[test]
+    fn single_base_transversion_costs_c_tv() {
+        let m = build_cost_matrix(&weights());
+        // A = 0b0001 (state 1), C = 0b0010 (state 2): A<->C is a transversion.
+        assert_eq!(m[1][2], 2.0);
+    }
+
+    #[test]
+    fn gain_loss_costs_c_ctx() {
+        let m = build_cost_matrix(&weights());
+        // A = 0b0001 (state 1) -> {A,C} = 0b0011 (state 3): pure gain of C.
+        assert_eq!(m[1][3], 10.0);
+    }
+
+    #[test]
+    fn empty_set_costs_scale_with_cardinality() {
+        let m = build_cost_matrix(&weights());
+        assert_eq!(m[0][1], 10.0); // {A} -> ∅: one member, one c_ctx step.
+        // N (all four) -> ∅: no flat shortcut — four single-element steps,
+        // same as losing four members down to a nonempty state would cost.
+        assert_eq!(m[0][15], 40.0);
+        assert_eq!(m[0][0], 0.0);
+    }
+
+    #[test]
+    fn matrix_is_symmetric() {
+        let m = build_cost_matrix(&weights());
+        for i in 0..16 {
+            for j in 0..16 {
+                assert_eq!(m[i][j], m[j][i], "asymmetry at ({i},{j})");
+            }
+        }
+    }
+
+    #[test]
+    fn calibrate_p_hat_excludes_saturated_pairs() {
+        // 3 genomes: pair (0,1) informative (10% SNP), pair (0,2) saturated
+        // (100% SNP) — must be excluded from the pooled estimate.
+        let mut snp = Array2::<u64>::zeros((3, 3));
+        let mut shared = Array2::<u64>::zeros((3, 3));
+        snp[[0, 1]] = 10;
+        shared[[0, 1]] = 90;
+        snp[[1, 0]] = 10;
+        shared[[1, 0]] = 90;
+        snp[[0, 2]] = 50;
+        shared[[0, 2]] = 0;
+        snp[[2, 0]] = 50;
+        shared[[2, 0]] = 0;
+        let raw = RawSnpDistanceOutput { snp, shared };
+
+        let est = calibrate_p_hat(&raw, 0.5);
+        assert_eq!(est.n_pairs_included, 1);
+        assert_eq!(est.n_loci_total, 100);
+        assert!((est.p_hat - 0.1).abs() < 1e-9);
+    }
+
+    #[test]
+    fn c_ctx_from_p_hat_zero_is_zero() {
+        assert_eq!(c_ctx_from_p_hat(0.0, 15, 1.5), 0.0);
+    }
+
+    #[test]
+    fn c_ctx_from_p_hat_matches_known_value() {
+        // p=0.1, m=15: [(30*0.1)/(1-(0.9)^30) + 0.1] * mean_sub_cost
+        let m = 15usize;
+        let p = 0.1;
+        let mean_sub_cost = 1.5;
+        let expected = ((2.0 * m as f64 * p) / (1.0 - (1.0 - p).powf(2.0 * m as f64)) + p) * mean_sub_cost;
+        assert!((c_ctx_from_p_hat(p, m, mean_sub_cost) - expected).abs() < 1e-12);
+    }
+
+    #[test]
+    fn mean_substitution_cost_is_weighted_average() {
+        let mut counts = [[0u64; 4]; 4];
+        counts[0][1] = 3; // A/C
+        counts[1][0] = 3;
+        counts[0][2] = 1; // A/G
+        counts[2][0] = 1;
+        let tally = BasePairTally { counts };
+        let mut sub_cost = [[0.0f64; 4]; 4];
+        sub_cost[0][1] = 2.0;
+        sub_cost[1][0] = 2.0;
+        sub_cost[0][2] = 1.0;
+        sub_cost[2][0] = 1.0;
+        // (3*2.0 + 1*1.0) / 4 = 1.75
+        assert!((mean_substitution_cost(&tally, &sub_cost) - 1.75).abs() < 1e-12);
+    }
+
+    #[test]
+    fn mean_substitution_cost_falls_back_to_one_with_no_data() {
+        let tally = BasePairTally { counts: [[0u64; 4]; 4] };
+        let sub_cost = [[0.0f64; 4]; 4];
+        assert_eq!(mean_substitution_cost(&tally, &sub_cost), 1.0);
+    }
+}

src/obikindex/src/siblings.rs

@@ -14,18 +14,25 @@
 //! mask by callers, not stored (see `FamilyMask` and
 //! [`sibling_annex_stats`](KmerIndex::sibling_annex_stats) below).
 //!
-//! Per layer, an `obipipeline` `Flat` stage (throttled — see
-//! `obipipeline::throttle`) generates each k-mer's 3 central variants,
-//! interleaved across many in-flight k-mers by the scheduler's shared
-//! worker pool rather than processed on a single thread. The actual
-//! cross-partition lookup, though, reuses
-//! `KmerPartition::query_partition_with` — the same partition-batching
-//! mechanism `obikmer query` already uses (open a partition's files once,
-//! answer a whole batch of queries against it) — rather than a per-item
-//! pipeline stage: an earlier per-item design reopened/re-mmap'd every
-//! target partition's files on every single lookup, which was fine at
-//! toy scale but manifested as ~90% system time against a real index.
-//! See `docmd/theory/evolutionary_distances.md`, Step 2b, "Mechanism".
+//! Per layer, an `obipipeline` batch transform (throttled — see
+//! `obipipeline::throttle`) generates a whole batch's central variants at
+//! once (`BATCH_SIZE` source k-mers in, that batch's variants out as one
+//! pipeline message), interleaved across many in-flight batches by the
+//! scheduler's shared worker pool rather than processed on a single
+//! thread. The actual cross-partition lookup reuses a `PartitionCache` of
+//! every partition's already-open MPHF layers, built once for the whole
+//! `build_sibling_annex` run, rather than reopening files per lookup or
+//! per source layer. Two earlier, coarser-grained designs were tried and
+//! measured (not guessed) to be worse, in order: (1) reopening/re-mmap'ing
+//! every target partition's files on every single lookup — fine at toy
+//! scale, ~90% system time against a real index; (2) a `Flat` pipeline
+//! stage pushing one message per generated *variant* (up to 3 per source
+//! k-mer) — cheaper than reopening files, but sampling a real run showed
+//! most wall-clock time going into per-message channel send/notify
+//! syscalls rather than the lookup itself, because a single k-mer's ≤3
+//! variants is far too fine a granularity to amortise a pipeline's
+//! synchronisation cost over. See `docmd/theory/evolutionary_distances.md`,
+//! Step 2b, "Mechanism".
 
 use std::path::Path;
 use std::sync::atomic::{AtomicU8, Ordering};
@@ -107,34 +114,30 @@ fn partition_of(kmer: CanonicalKmer, n_partitions: usize) -> usize {
 
 // ── obipipeline data types ─────────────────────────────────────────────────
 
-/// One distinct k-mer of the layer currently being processed, at its local
-/// MPHF slot — the pipeline's source item. Carries a throttle slot (shared,
-/// `Arc`-wrapped so it can be cloned into each of the (up to 3) items this
-/// one fans out into via the `Flat` stage) that is only released once every
-/// one of those descendants has been fully processed — see the module docs'
-/// "Throttling" note for why this is required, not optional, once a `Flat`
-/// stage is in the pipeline.
-struct SourceItem {
-    slot: usize,
-    kmer: CanonicalKmer,
-    _permit: Arc<ThrottleGuard>,
+/// A batch of this layer's distinct k-mers (local MPHF slot + k-mer), the
+/// pipeline's source item — batched, not one k-mer per item, so that
+/// pipeline messages and their synchronisation cost stay amortised over
+/// thousands of lookups (see `build_layer_sibling_annex`'s comment on
+/// `BATCH_SIZE`). Carries the throttle permit for the whole batch, moved
+/// (not cloned) into the corresponding `VariantBatch` — a 1-to-1 transform,
+/// unlike the fan-out `Flat` stage this replaced, needs no `Arc` sharing.
+struct SourceBatch {
+    items: Vec<(usize, CanonicalKmer)>,
+    _permit: ThrottleGuard,
 }
 
-/// One of a source k-mer's (up to 3) central-substitution variants, already
-/// routed to its destination partition and carrying its own central base
-/// (0=A/1=C/2=G/3=T) — the mask bit it will set on a hit. Carries a clone of
-/// the source item's throttle permit.
-struct VariantQuery {
-    source_slot: usize,
-    dest_partition: usize,
-    variant: CanonicalKmer,
-    base: u8,
-    _permit: Arc<ThrottleGuard>,
+/// One batch's worth of central-substitution variants (up to 3 per source
+/// k-mer), each already routed to its destination partition and carrying
+/// its own central base (0=A/1=C/2=G/3=T) — the mask bit it will set on a
+/// hit. `(dest_partition, variant, source_slot, base)` per entry.
+struct VariantBatch {
+    items: Vec<(usize, CanonicalKmer, usize, u8)>,
+    _permit: ThrottleGuard,
 }
 
 enum SibData {
-    Item(SourceItem),
-    Query(VariantQuery),
+    Batch(SourceBatch),
+    Variants(VariantBatch),
 }
 
 /// Every partition's already-open MPHF layers, built **once** for the whole
@@ -341,57 +344,72 @@ impl KmerIndex {
             mask[slot].fetch_or(1 << central_base(kmer, k), Ordering::Relaxed);
         }
 
-        // ── obipipeline: Flat stage generates variants only — the actual
-        // cross-partition lookup reuses `KmerPartition::query_partition_with`
-        // (the same batching mechanism `obikmer query` already uses: open a
-        // partition's files once, answer a whole batch of queries against
-        // it) instead of one lookup per pipeline item. A per-item lookup
-        // (tried first) reopened/re-mmap'd every target partition's files on
-        // every single variant — fine at the scale of a handful of test
-        // k-mers, but with billions of lookups against a real index this
-        // manifested as ~90% system time, observed in practice. ───────────
+        // ── obipipeline: a *batch* transform, not a per-k-mer `Flat` one —
+        // the actual cross-partition lookup reuses
+        // `KmerPartition::query_partition_with` (the same batching mechanism
+        // `obikmer query` already uses: open a partition's files once,
+        // answer a whole batch of queries against it) instead of one lookup
+        // per pipeline item. A per-item lookup (tried first) reopened/
+        // re-mmap'd every target partition's files on every single variant
+        // — fine at toy scale, but ~90% system time against a real index,
+        // observed in practice. A *later* attempt still pushed one pipeline
+        // message per generated variant (a `Flat` stage, `SourceItem` =>
+        // `VariantQuery`, one k-mer in => up to 3 variants out as separate
+        // messages) — cheaper than reopening files, but sampling a real run
+        // showed most wall-clock time going into per-message channel
+        // send/notify syscalls instead of the lookup itself: the pipeline's
+        // whole point is amortising synchronisation over a batch, and a
+        // single k-mer's ≤3 variants is far too fine a granularity for
+        // that. Batching `BATCH_SIZE` source k-mers into one pipeline item
+        // — a plain 1-to-1 (`|`, not `||`) transform, batch in, batch of
+        // variants out, one message either way — keeps the per-message
+        // synchronisation cost amortised over thousands of lookups instead
+        // of one to three. ──────────────────────────────────────────────
+        const BATCH_SIZE: usize = 4096;
         let n_workers = obisys::effective_parallelism();
         let capacity = 256;
 
-        // Throttling is not optional once a `Flat` stage is in the pipeline
-        // (see `obipipeline::throttle`'s docs): without it, every worker can
-        // become a simultaneous `Flat` producer, saturate the shared output
-        // channel, and deadlock against the scheduler's own dispatch loop —
-        // also observed in practice. The permit acquired here for a source
-        // k-mer is held (via the `Arc`-shared guard carried through
-        // `SourceItem` -> `VariantQuery`) until every one of its (up to 3)
-        // descendants has been read out of the pipeline by the accumulation
-        // loop below, not just until the `Flat` stage itself returns.
+        // Throttling limits how many *batches* are in flight at once — the
+        // permit is acquired per batch (not per k-mer) in the source
+        // thread, and released once its `VariantBatch` has been read out of
+        // the pipeline by the accumulation loop below. See
+        // `obipipeline::throttle`'s docs for why this is required, not
+        // optional, once a `Flat`-style stage sits in the pipeline.
         let sources: Vec<(usize, CanonicalKmer)> = slot_kmer
             .iter()
             .enumerate()
             .filter_map(|(slot, maybe_kmer)| maybe_kmer.map(|kmer| (slot, kmer)))
             .collect();
-        let throttled = obipipeline::throttle(sources.into_iter(), n_workers).map(|t| SourceItem {
-            slot: t.item.0,
-            kmer: t.item.1,
-            _permit: Arc::new(t.guard),
+        let batches: Vec<Vec<(usize, CanonicalKmer)>> = sources
+            .chunks(BATCH_SIZE)
+            .map(|chunk| chunk.to_vec())
+            .collect();
+        let throttled = obipipeline::throttle(batches.into_iter(), n_workers).map(|t| SourceBatch {
+            items: t.item,
+            _permit: t.guard,
         });
 
         let pipe = obipipeline::make_pipe! {
-            SibData : SourceItem => VariantQuery,
-            || {
-                move |item: SourceItem| -> Vec<VariantQuery> {
-                    let kmer = item.kmer;
-                    let permit = item._permit;
-                    kmer.central_canonical_neighbors()
-                        .into_iter()
-                        .filter(|variant| *variant != kmer)
-                        .map(|variant| VariantQuery {
-                            source_slot: item.slot,
-                            dest_partition: partition_of(variant, n_parts),
-                            variant,
-                            base: central_base(variant, k),
-                            _permit: Arc::clone(&permit),
-                        })
-                        .collect::<Vec<_>>()
+            SibData : SourceBatch => VariantBatch,
+            | {
+                move |batch: SourceBatch| -> VariantBatch {
+                    let mut items = Vec::with_capacity(batch.items.len() * 3);
+                    for (slot, kmer) in batch.items {
+                        for variant in kmer.central_canonical_neighbors() {
+                            if variant == kmer {
+                                continue;
+                            }
+                            items.push((
+                                partition_of(variant, n_parts),
+                                variant,
+                                slot,
+                                central_base(variant, k),
+                            ));
+                        }
+                    }
+                    VariantBatch { items, _permit: batch._permit }
                 }
-            } : Item => Query,
+            } : Batch => Variants,
         };
 
         // ── Group generated variants by destination partition. `cache`
@@ -400,11 +418,13 @@ impl KmerIndex {
         // lookup is not free just because the file isn't reopened. Grouping
         // keeps one partition's pages hot while its whole batch is resolved,
         // instead of faulting pages in and out as lookups jump between
-        // partitions in whatever order the `Flat` stage happens to produce
-        // them. The throttle permit drops here, once accumulated. ─────────
+        // partitions in whatever order the pipeline happens to produce
+        // them. Each batch's throttle permit drops here, once accumulated.
         let mut outgoing: Vec<Vec<(CanonicalKmer, usize, u8)>> = (0..n_parts).map(|_| Vec::new()).collect();
-        for vq in pipe.apply(throttled, n_workers, capacity) {
-            outgoing[vq.dest_partition].push((vq.variant, vq.source_slot, vq.base));
+        for vb in pipe.apply(throttled, n_workers, capacity) {
+            for (dest_partition, variant, source_slot, base) in vb.items {
+                outgoing[dest_partition].push((variant, source_slot, base));
+            }
         }
 
         // ── Resolve each partition's batch against the cache in one
@@ -635,9 +655,27 @@ pub struct RawSnpDistanceOutput {
 }
 
 impl KmerIndex {
-    /// Compute [`RawSnpDistanceOutput`] from an already-built sibling annex
-    /// (run [`build_sibling_annex`](Self::build_sibling_annex) first).
-    pub fn raw_snp_distance(&self) -> OKIResult<RawSnpDistanceOutput> {
+    /// Shared traversal behind [`raw_snp_distance`](Self::raw_snp_distance)
+    /// and [`base_pair_tally`](Self::base_pair_tally): for every family
+    /// (tallied once, at its minorant) of every layer of the already-built
+    /// sibling annex, resolves each genome's single observed form (`None`
+    /// if absent or ambiguous/multi-copy), then calls `on_pair(acc, i, j,
+    /// bi, bj)` for every genome pair `(i, j)` where both are unambiguous
+    /// and single-copy (`bi == bj` means shared at that locus, `bi != bj`
+    /// means a SNP). Layers are processed in parallel (rayon); each gets
+    /// its own accumulator from `zero()`, combined pairwise via `combine`.
+    fn scan_family_pairs<Acc, F, C>(
+        &self,
+        label: &str,
+        zero: impl Fn() -> Acc + Sync,
+        on_pair: F,
+        combine: C,
+    ) -> OKIResult<Acc>
+    where
+        Acc: Send,
+        F: Fn(&mut Acc, usize, usize, u8, u8) + Sync,
+        C: Fn(Acc, Acc) -> Acc,
+    {
         let n_parts = self.n_partitions();
         let n_genomes = self.meta.genomes.len();
         let with_counts = self.meta.config.with_counts;
@@ -673,12 +711,11 @@ impl KmerIndex {
             }
         }
 
-        let pb = progress_bar("raw_snp_distance", layer_dirs.len() as u64, "layers");
-        let partials: Vec<(Array2<u64>, Array2<u64>)> = layer_dirs
+        let pb = progress_bar(label, layer_dirs.len() as u64, "layers");
+        let partials: Vec<Acc> = layer_dirs
             .par_iter()
-            .map(|layer_dir| -> OKIResult<(Array2<u64>, Array2<u64>)> {
-                let mut snp = Array2::<u64>::zeros((n_genomes, n_genomes));
-                let mut shared = Array2::<u64>::zeros((n_genomes, n_genomes));
+            .map(|layer_dir| -> OKIResult<Acc> {
+                let mut acc = zero();
 
                 let index_dir = layer_dir.parent().expect("layer_dir has a parent index dir");
                 let meta = PartitionMeta::load(index_dir).map_err(olm_to_ok)?;
@@ -754,31 +791,102 @@ impl KmerIndex {
                                 continue;
                             }
                             let Some(bj) = single_form[j] else { continue };
-                            if bi == bj {
-                                shared[[i, j]] += 1;
-                                shared[[j, i]] += 1;
-                            } else {
-                                snp[[i, j]] += 1;
-                                snp[[j, i]] += 1;
-                            }
+                            on_pair(&mut acc, i, j, bi, bj);
                         }
                     }
                 }
 
                 pb.inc(1);
-                Ok((snp, shared))
+                Ok(acc)
             })
             .collect::<OKIResult<Vec<_>>>()?;
         pb.finish_and_clear();
 
-        let mut snp = Array2::<u64>::zeros((n_genomes, n_genomes));
-        let mut shared = Array2::<u64>::zeros((n_genomes, n_genomes));
-        for (s, sh) in partials {
-            snp += &s;
-            shared += &sh;
+        let mut total = zero();
+        for partial in partials {
+            total = combine(total, partial);
         }
+        Ok(total)
+    }
+
+    /// Compute [`RawSnpDistanceOutput`] from an already-built sibling annex
+    /// (run [`build_sibling_annex`](Self::build_sibling_annex) first).
+    pub fn raw_snp_distance(&self) -> OKIResult<RawSnpDistanceOutput> {
+        let n_genomes = self.meta.genomes.len();
+        let (snp, shared) = self.scan_family_pairs(
+            "raw_snp_distance",
+            || (Array2::<u64>::zeros((n_genomes, n_genomes)), Array2::<u64>::zeros((n_genomes, n_genomes))),
+            |(snp, shared), i, j, bi, bj| {
+                if bi == bj {
+                    shared[[i, j]] += 1;
+                    shared[[j, i]] += 1;
+                } else {
+                    snp[[i, j]] += 1;
+                    snp[[j, i]] += 1;
+                }
+            },
+            |(mut snp, mut shared), (s, sh)| {
+                snp += &s;
+                shared += &sh;
+                (snp, shared)
+            },
+        )?;
         Ok(RawSnpDistanceOutput { snp, shared })
     }
+
+    /// Symmetric 6-category base-pair substitution tally (AC, AG, AT, CG,
+    /// CT, GT — indexed `0=A,1=C,2=G,3=T`), pooled only over genome pairs
+    /// whose overall SNP ratio in `raw` is at or below `ratio_ceiling` —
+    /// same saturation-exclusion discipline as `calibrate_p_hat`, for the
+    /// same reason: a saturated pair's observed base-pair mix trends toward
+    /// neutral base composition, not the true point-mutation spectrum.
+    ///
+    /// A second full pass over the annex, sharing
+    /// [`raw_snp_distance`](Self::raw_snp_distance)'s traversal (guided by
+    /// it, not a blind re-scan) — needed because `raw_snp_distance` only
+    /// keeps aggregate SNP/shared counts per genome pair, not which bases
+    /// were actually involved at each locus, and the ratio-ceiling filter
+    /// can only be evaluated once the aggregate counts are known.
+    pub fn base_pair_tally(&self, raw: &RawSnpDistanceOutput, ratio_ceiling: f64) -> OKIResult<BasePairTally> {
+        let n_genomes = self.meta.genomes.len();
+        let included = Array2::from_shape_fn((n_genomes, n_genomes), |(i, j)| {
+            if i == j {
+                return false;
+            }
+            let snp = raw.snp[[i, j]];
+            let total = snp + raw.shared[[i, j]];
+            total > 0 && (snp as f64 / total as f64) <= ratio_ceiling
+        });
+
+        let counts = self.scan_family_pairs(
+            "base_pair_tally",
+            || [[0u64; 4]; 4],
+            |counts, i, j, bi, bj| {
+                if bi != bj && included[[i, j]] {
+                    counts[bi as usize][bj as usize] += 1;
+                    counts[bj as usize][bi as usize] += 1;
+                }
+            },
+            |mut total, partial| {
+                for a in 0..4 {
+                    for b in 0..4 {
+                        total[a][b] += partial[a][b];
+                    }
+                }
+                total
+            },
+        )?;
+        Ok(BasePairTally { counts })
+    }
+}
+
+/// See [`KmerIndex::base_pair_tally`].
+pub struct BasePairTally {
+    /// `counts[a][b] == counts[b][a]` = number of eligible loci, pooled
+    /// over included genome pairs, where the two genomes' single forms are
+    /// `a` and `b` (0=A, 1=C, 2=G, 3=T). Diagonal always `0` (an `a == b`
+    /// locus is `shared`, not tallied here).
+    pub counts: [[u64; 4]; 4],
 }
 
 /// IUPAC ambiguity code for a per-genome family presence mask (bit `b` set