tilezz 0.2.0

//! Classify a `tilezz-ratdb` DAFSA asset into a certified verdict store.
//!
//! Two drivers over the SAME funnel machinery (one executor, two stage tables;
//! see [`cascade`](crate::classify::cascade)), named after their CLI
//! modes:
//!
//! The **fast pass** ([`run_fast`]) screens a whole perimeter (ZZ12 has
//! 597,581 rats at perimeter 12 alone), sweeping every tile through the
//! cheap-first [`fast_stages`] funnel: cheap Heesch rejects cut the non-tiler
//! bulk before the expensive periodic accepts ever run, and every budget is
//! small. Every accept mints a verified, provenance (`via`)-tagged
//! [`crate::classify::cert::PeriodicCert`], and a tile no stage
//! settles is [`Classified::Undecided`] (an aperiodic candidate). The store is
//! a JSONL of `<dafsa index>\t<Classified>` lines, written+flushed batchwise
//! so the run is inspectable and resumable.
//!
//! The **deep pass** ([`run_deep`]) re-classifies only the small Undecided
//! residue with the high-budget
//! [`deep_stages`](crate::classify::cascade::deep_stages) funnel (deep
//! Heesch, high torus kmax, seeded aniso restarts), so the bulk stays fast and
//! only the hard tail pays for the escalation. Its results append to a
//! `<store>.deep.jsonl` overlay as they land (resumable like the fast pass);
//! [`crate::classify::store::run_merge`] then applies them onto the store -- a cert overrides
//! Undecided, never the reverse.
//!
//! [`run_verify`] re-checks a finished store from disk: every cert re-verifies and
//! every present perimeter block is COMPLETE against the dataset's counts.
//!
//! Input is a blocked DAFSA asset (`block_index.json` + `blocks/<sha>.bin`), read
//! lazily via [`LazyRatDafsa`]. Rats are stored length-prefixed in `(length, lex)`
//! order, so each perimeter is a contiguous index range. The `classify_tiles` binary
//! (`src/bin/classify_tiles.rs`) is the command-line front end: the fast pass is the
//! default (`--fast`, restrictable with `--periodic-only` / `--heesch-only`), the
//! deep pass is `--deep`, and [`run_verify`] is `--verify`.

use std::io;
use std::path::Path;
use std::sync::Mutex;
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::time::Instant;

pub use crate::classify::cascade::StageMode;
use crate::classify::cascade::{
    AcceptBounds, Stage, StageOutcome, classify, fast_stages, run_stages,
};
use crate::classify::cert::{Classified, Verdict};
use crate::classify::reptile::{RepScreen, reptile_screen_one};
use crate::classify::store::{
    CoveredRange, append_range, covered_union, deep_overlay_path, in_covered_union, read_ranges,
    read_resume,
};
use crate::cyclotomic::IsRing;
use crate::dataset::LazyRatDafsa;
use crate::geom::rat::Rat;

/// Free ZZ12 rat counts by perimeter (OEIS A316192), index = perimeter.
/// Submitted/validated; `count.py` reproduces these off the asset index. Used to
/// derive perimeter index ranges (and completeness targets) without walking the
/// automaton.
pub const FREE_ZZ12: [u64; 17] = [
    0, 0, 0, 1, 3, 4, 22, 69, 418, 2210, 14024, 89075, 597581, 4076855, 28499301, 202464580,
    1460982297,
];

/// The `Dataset` node of an asset's `ro-crate-metadata.json`: the single
/// `@graph` entry whose `@type` is (or contains) `"Dataset"`. This node carries
/// the ring order and per-perimeter free counts the classifier keys off of.
/// Panics with a clear message if the file, its `@graph`, or the Dataset node
/// is missing.
fn ro_crate_dataset(dir: &str) -> serde_json::Value {
    let path = format!("{dir}/ro-crate-metadata.json");
    let txt = std::fs::read_to_string(&path).unwrap_or_else(|e| panic!("read {path}: {e}"));
    let v: serde_json::Value =
        serde_json::from_str(&txt).unwrap_or_else(|e| panic!("parse {path}: {e}"));
    let graph = v["@graph"]
        .as_array()
        .unwrap_or_else(|| panic!("{path}: no @graph array"));
    for node in graph {
        let is_dataset = match &node["@type"] {
            serde_json::Value::String(s) => s == "Dataset",
            serde_json::Value::Array(a) => a.iter().any(|t| t.as_str() == Some("Dataset")),
            _ => false,
        };
        if is_dataset {
            return node.clone();
        }
    }
    panic!("{path}: no @type==Dataset node in @graph");
}

/// The effective cyclotomic ring order of an asset (10 or 12), read from the
/// `effectiveRing` `additionalProperty` of its RO-Crate Dataset node. This is
/// the discriminant the classifier dispatches its ring-generic pipeline on.
/// Panics with a clear message if the property is missing.
pub fn asset_ring(dir: &str) -> usize {
    let node = ro_crate_dataset(dir);
    let props = node["additionalProperty"].as_array().unwrap_or_else(|| {
        panic!("{dir}/ro-crate-metadata.json: Dataset has no additionalProperty")
    });
    for p in props {
        if p["name"].as_str() == Some("effectiveRing") {
            return p["value"]
                .as_u64()
                .unwrap_or_else(|| panic!("{dir}: effectiveRing value is not an integer"))
                as usize;
        }
    }
    panic!("{dir}/ro-crate-metadata.json: no effectiveRing additionalProperty");
}

/// Free rat counts by perimeter for an asset, read from the `free`
/// `variableMeasured` of its RO-Crate Dataset node. The stored value is the
/// comma-separated counts 1-indexed by perimeter; this PREPENDS a single 0 so
/// the returned vec is indexed BY PERIMETER (`counts[perimeter]`), matching the
/// [`FREE_ZZ12`] const convention exactly (`counts[0] == 0`, `counts[3]` is the
/// first tileable perimeter's count). Feeds [`perimeter_range`] /
/// [`perimeter_of`] and the completeness targets for any ring. Panics with a
/// clear message if the variable is missing or malformed.
pub fn asset_free_counts(dir: &str) -> Vec<u64> {
    let node = ro_crate_dataset(dir);
    let vars = node["variableMeasured"]
        .as_array()
        .unwrap_or_else(|| panic!("{dir}/ro-crate-metadata.json: Dataset has no variableMeasured"));
    for v in vars {
        if v["name"].as_str() == Some("free") {
            let s = v["value"]
                .as_str()
                .unwrap_or_else(|| panic!("{dir}: free variableMeasured value is not a string"));
            // Prepend 0 so index == perimeter (the stored list is 1-indexed).
            let mut counts = vec![0u64];
            for tok in s.split(',') {
                counts.push(
                    tok.trim().parse::<u64>().unwrap_or_else(|e| {
                        panic!("{dir}: free value has non-integer {tok:?}: {e}")
                    }),
                );
            }
            return counts;
        }
    }
    panic!("{dir}/ro-crate-metadata.json: no free variableMeasured");
}

/// `[start, end)` DAFSA index range of the perimeter-`n` rats under the
/// `(length, lex)` ordering: everything of smaller perimeter sorts first.
/// `counts` is the per-perimeter free tally (index == perimeter), e.g.
/// [`FREE_ZZ12`] or [`asset_free_counts`].
pub fn perimeter_range(counts: &[u64], n: usize) -> (u64, u64) {
    let start: u64 = counts[..n].iter().sum();
    (start, start + counts[n])
}

/// Perimeter of a DAFSA index via the `counts` prefix sums (index ==
/// perimeter, e.g. [`FREE_ZZ12`]), or `None` if the index is past the tabulated
/// range.
pub fn perimeter_of(counts: &[u64], idx: u64) -> Option<usize> {
    let mut cum = 0u64;
    for (p, &c) in counts.iter().enumerate() {
        if idx >= cum && idx < cum + c {
            return Some(p);
        }
        cum += c;
    }
    None
}

/// Open a local blocked ratdb asset directory (`block_index.json` +
/// `blocks/<sha>.bin`) for lazy reading.
pub fn open_ratdb(dir: &str) -> LazyRatDafsa<impl Fn(u32) -> io::Result<Vec<u8>>> {
    let manifest = std::fs::read_to_string(format!("{dir}/block_index.json"))
        .unwrap_or_else(|e| panic!("read {dir}/block_index.json: {e}"));
    let v: serde_json::Value = serde_json::from_str(&manifest).unwrap();
    let shas: Vec<String> = v["blocks"]
        .as_array()
        .unwrap()
        .iter()
        .map(|b| b["sha256"].as_str().unwrap().to_string())
        .collect();
    let base = dir.to_string();
    LazyRatDafsa::open(&manifest, move |idx| {
        std::fs::read(format!("{base}/blocks/{}.bin", shas[idx as usize]))
    })
    .unwrap()
}

/// The per-verdict re-check policy shared by the fast pass's emit and run_deep:
/// replay a reject's corona witness (verify_lower_bound -- its ONLY
/// release-mode check, heesch_cert only debug_asserts it, and cheap: empty
/// build for Heesch 0, a short glue replay otherwise); trust the
/// minter-gated periodic certs in release (every minter gates on
/// PeriodicCert::verify -- a second run of the identical function is kept as
/// a debug_assert, and --verify re-checks the store independently);
/// Undecided claims nothing. Returns (tally slot, check passed) with slots
/// [periodic, cannot_tile, undecided].
fn verdict_check<T: IsRing>(base: &Rat<T>, c: &Classified) -> (usize, bool) {
    match c {
        Classified::Decided(Verdict::Periodic(pc)) => {
            debug_assert!(pc.verify(base), "minted cert must be pre-verified");
            (0, true)
        }
        Classified::Decided(Verdict::CannotTile(hc)) => (1, hc.verify_lower_bound(base)),
        Classified::Undecided { .. } => (2, true),
    }
}

/// Resolve a worker count: the given value, else all available cores.
fn workers_or_all(workers: usize) -> usize {
    if workers > 0 {
        workers
    } else {
        crate::util::available_workers()
    }
}

// ---------------------------------------------------------------------------
// The fast pass (`--fast`): screen a whole perimeter.
// ---------------------------------------------------------------------------

/// Fast-pass (`--fast`) parameters. The defaults are the calibrated production values.
#[derive(Debug, Clone)]
pub struct FastConfig {
    /// Which stages to run (default: the full funnel).
    pub mode: StageMode,
    /// Deep-Heesch corona bound (keeps genuine Heesch-3+ rejects sound).
    pub deep_bound: usize,
    /// Deep-Heesch node budget. 500k: a Heesch-4/5/6 non-tiler that rejects under
    /// this settles in the fast pass; anything needing more defers to the
    /// `--deep` residue pass ([`run_deep`]). The fast pass must not grind the
    /// hard tail, that is exactly what `--deep` is for.
    pub deep_budget: usize,
    /// torus kmax for the residue-only deep-torus stage (large-period catcher; a
    /// period-18 n=13 tile needed kmax 18, so 32 has margin).
    pub deep_torus_kmax: usize,
    /// Acceptance bounds for the periodic detectors (see [`AcceptBounds`]).
    pub bounds: AcceptBounds,
    /// Worker threads (0 = all cores).
    pub workers: usize,
    /// Index-window size: the perimeter is processed in windows of this many
    /// tiles, each fully classified before the next is
    /// loaded -- bounding the resident rat set. `0` = one window for the whole
    /// perimeter. Materializing every rat stops scaling around n = 15 (202M
    /// rats) and is impossible at n = 16 (1.46G), so big perimeters must
    /// chunk; verdicts are unaffected (the funnel is per-tile).
    pub chunk: usize,
}

impl Default for FastConfig {
    fn default() -> Self {
        Self {
            mode: StageMode::Full,
            deep_bound: 6,
            deep_budget: 500_000,
            deep_torus_kmax: 32,
            // FAST accept bounds: aniso bails at 5k examined clusters (~4s), so a
            // tile whose cluster is not found cheaply defers to run_deep, which
            // retries with seeded restarts at a far higher budget (12 * 30k). (Was
            // 300k -- aniso then ground ~75s on every tile it could NOT detect;
            // even 20k was ~17s/failed-tile.)
            bounds: AcceptBounds {
                aniso_budget: 5_000,
                ..AcceptBounds::default()
            },
            workers: 0,
            // Whole perimeter up front: right through n <= 14 (28.5M rats ~ 1GB);
            // n >= 15 runs must set a window (e.g. 2_000_000).
            chunk: 0,
        }
    }
}

// ---------------------------------------------------------------------------
// Coverage sidecar: "scanned" decoupled from "kept".
//
// The store doubles as the progress ledger (resume = which indices have
// lines), which breaks the moment boring certs are FILTERED away to keep big
// perimeters manageable. The `<store>.ranges` sidecar fixes that: one line
// per fully-processed window, `start<TAB>end<TAB>P<TAB>N<TAB>U`, appended by
// the fast pass at each window boundary. The load-bearing INVARIANT: every
// tile is accounted for either by a kept store LINE or by a covered RANGE
// whose tallies counted it at (verified) emission time. Hence a window gets a
// range record only when every one of its tiles was emitted fresh in that run
// (a window containing line-resumed tiles stays line-accounted -- see
// run_fast), and a filter may only drop lines INSIDE covered ranges
// (tools/filter_store.py enforces this).
// ---------------------------------------------------------------------------

/// Run `work(j)` for every `j in 0..total` on `workers` threads (work-stealing
/// via an atomic cursor). Big work-lists get a 15s `name`-labelled heartbeat
/// (done/total + rate) -- on the million-tile perimeters a single stage runs
/// for minutes and would otherwise be a silent black box; small lists skip it,
/// sparing the ~1s poll-join tail per stage. Returns when all items are done.
fn parallel_sweep(name: &str, total: usize, workers: usize, work: impl Fn(usize) + Sync) {
    use crate::util::parallel::parallel_drain;
    const PROGRESS_MIN: usize = 2_000;

    // Small lists: just drain, no heartbeat thread (sparing its ~1s poll-join
    // tail on stages that finish quickly).
    if total < PROGRESS_MIN {
        parallel_drain(total, workers, || (), |_, j| work(j), |_, _| ());
        return;
    }

    // Big lists: a 15s heartbeat runs alongside the drain. `done_ct` is bumped
    // per item; `stop` is raised by the drop guard once the drain returns --
    // whether normally or by unwinding a worker panic through `parallel_drain`
    // -- so the enclosing scope never blocks forever on the monitor (a panic
    // then keeps propagating instead of hanging).
    let st = Instant::now();
    let done_ct = AtomicUsize::new(0);
    let stop = AtomicBool::new(false);
    let (done_ct, stop) = (&done_ct, &stop);
    std::thread::scope(|sc| {
        sc.spawn(move || {
            let mut printed_at = 0u64;
            while !stop.load(Ordering::Relaxed) {
                std::thread::sleep(std::time::Duration::from_secs(1));
                let secs = st.elapsed().as_secs();
                if secs >= printed_at + 15 {
                    printed_at = secs;
                    let dn = done_ct.load(Ordering::Relaxed);
                    let rate = dn as f64 / st.elapsed().as_secs_f64().max(1e-3);
                    eprintln!("    {name}: {dn}/{total} ({rate:.0}/s)");
                }
            }
        });
        struct StopOnDrop<'a>(&'a AtomicBool);
        impl Drop for StopOnDrop<'_> {
            fn drop(&mut self) {
                self.0.store(true, Ordering::Relaxed);
            }
        }
        let _stop_guard = StopOnDrop(stop);
        parallel_drain(
            total,
            workers,
            || (),
            |_, j| {
                work(j);
                done_ct.fetch_add(1, Ordering::Relaxed);
            },
            |_, _| (),
        );
    });
}

/// Tally of a fast-pass classification (or a resumed store's running totals).
#[derive(Debug, Clone, Copy, Default)]
pub struct ClassifyStats {
    pub periodic: usize,
    pub cannot_tile: usize,
    pub undecided: usize,
    /// Certs that failed to self-verify on emit (must be 0 -- a bug if not).
    pub verify_fail: usize,
    pub total: usize,
}

/// Run the fast pass over perimeter `perim` -- sugar for
/// [`run_fast_range`] on [`perimeter_range`]`(counts, perim)`. `counts` is the
/// asset's per-perimeter free tally (index == perimeter, e.g. [`FREE_ZZ12`] or
/// [`asset_free_counts`]).
pub fn run_fast<T: IsRing, F>(
    d: &LazyRatDafsa<F>,
    counts: &[u64],
    perim: usize,
    out: &Path,
    cfg: &FastConfig,
) -> ClassifyStats
where
    F: Fn(u32) -> io::Result<Vec<u8>>,
{
    let (start, end) = perimeter_range(counts, perim);
    run_fast_range::<T, F>(d, start, end, out, cfg)
}

/// Run the fast pass over the explicit dafsa index range `[start, end)`,
/// appending certified verdict lines to `out`. Resumable two ways: store
/// lines (fine-grained, survives filtering only for kept lines) and the
/// coverage sidecar (`<store>.ranges` -- whole windows recorded as covered
/// with their run-time tallies, surviving any filtering). A window that was
/// processed entirely fresh in this run is appended to the sidecar; a window
/// containing line-resumed tiles is NOT (its tiles stay accounted by their
/// lines -- the conservative half of the coverage invariant, see the sidecar
/// section). Returns the running tally (new + resumed, lines and ranges).
///
/// The funnel itself -- stage order, depths, budgets, and their measured
/// rationale -- is [`fast_stages`]; this driver only sweeps it (via the
/// shared [`run_stages`] executor, per tile) over the range's windows and
/// reports per-stage tallies.
///
/// Panics if any minted cert fails to self-verify.
pub fn run_fast_range<T: IsRing, F>(
    d: &LazyRatDafsa<F>,
    start: u64,
    end: u64,
    out: &Path,
    cfg: &FastConfig,
) -> ClassifyStats
where
    F: Fn(u32) -> io::Result<Vec<u8>>,
{
    use std::io::Write;
    use std::sync::Arc;
    use std::sync::atomic::AtomicU64;

    assert!(start <= end, "empty or inverted range");
    let t0 = Instant::now();
    let n = (end - start) as usize;
    let workers = workers_or_all(cfg.workers);
    let b = cfg.bounds;
    let (deep_bound, deep_budget, deep_torus_kmax) =
        (cfg.deep_bound, cfg.deep_budget, cfg.deep_torus_kmax);

    // Resume, coarse: the coverage sidecar (whole windows, filter-proof).
    let ranges = read_ranges(out);
    let union = covered_union(&ranges);
    // Resumed-range tallies: ranges fully inside [start, end). A range
    // straddling the launch boundary cannot be split (its tallies are not
    // positional), so it is skipped with a warning -- align launch ranges
    // with past window boundaries to keep the stats exact (coverage and
    // skipping are unaffected either way).
    let mut rr = [0usize; 3];
    for r in &ranges {
        if r.start >= start && r.end <= end {
            for (acc, t) in rr.iter_mut().zip(r.tally) {
                *acc += t;
            }
        } else if r.start < end && r.end > start {
            eprintln!(
                "fast: sidecar range [{},{}) straddles the launch range -- its tallies are excluded from the reported stats",
                r.start, r.end
            );
        }
    }
    // Resume, fine: store lines (partial windows, kept certs). Lines inside
    // covered ranges are already counted by the sidecar, so they are excluded
    // from the line tally to avoid double counting.
    let (done, [lp, ln, lu]) = read_resume(out, start, end, &union);
    let (rp, rn, ru) = (rr[0] + lp, rr[1] + ln, rr[2] + lu);
    // Tiles of [start, end) already accounted by coverage (for the final
    // every-rat-accounted check; line-done tiles inside the union are part of
    // this count, not double-counted -- done is only consulted for OPEN tiles).
    let covered_ct: u64 = union
        .iter()
        .map(|&(s, e)| e.min(end).saturating_sub(s.max(start)))
        .sum();
    // Batched flush: the explicit flush is what makes a killed run resumable
    // (BufWriter alone flushes at buffer granularity), but one flush syscall
    // PER LINE serializes the worker pool on the cheap stages of a
    // million-tile perimeter. Flush every FLUSH_LINES lines or FLUSH_SECS
    // seconds instead: crash-resume then re-does at most that window (the
    // resume reader re-runs missing tiles and truncates a torn tail), and a
    // watcher still sees fresh lines on slow stages via the time bound.
    const FLUSH_LINES: usize = 256;
    const FLUSH_SECS: u64 = 1;
    let writer = Mutex::new((
        std::io::BufWriter::new(
            std::fs::OpenOptions::new()
                .create(true)
                .append(true)
                .open(out)
                .unwrap(),
        ),
        0usize,
        Instant::now(),
    ));
    let chunk = if cfg.chunk == 0 { n.max(1) } else { cfg.chunk };
    eprintln!(
        "fast [{start},{end}) = {n} rats (windows of {chunk}), {workers} workers -> {} (resuming {} line-done + {} covered ranges, setup {:?})",
        out.display(),
        done.len(),
        ranges.len(),
        t0.elapsed(),
    );

    // P, N, Undecided, verify_fail -- tallied as tiles are emitted.
    let cnt = [
        AtomicUsize::new(0),
        AtomicUsize::new(0),
        AtomicUsize::new(0),
        AtomicUsize::new(0),
    ];
    // Check each verdict's cert before it hits the store ([`verdict_check`]),
    // tally, and append its line (serialized by the writer, batched flush).
    let emit = |seq: &[i8], abs_idx: u64, c: &Classified| {
        let base = Rat::<T>::from_slice_trusted(seq);
        let (slot, ok) = verdict_check(&base, c);
        cnt[slot].fetch_add(1, Ordering::Relaxed);
        if !ok {
            cnt[3].fetch_add(1, Ordering::Relaxed);
            eprintln!("VERIFY FAIL idx {abs_idx} seq {seq:?}");
        }
        let line = format!("{abs_idx}\t{}\n", serde_json::to_string(c).unwrap());
        let mut w = writer.lock().unwrap();
        w.0.write_all(line.as_bytes()).unwrap();
        w.1 += 1;
        if w.1 >= FLUSH_LINES || w.2.elapsed().as_secs() >= FLUSH_SECS {
            w.0.flush().unwrap();
            w.1 = 0;
            w.2 = Instant::now();
        }
    };

    // The --fast funnel is DATA ([`fast_stages`], next to --deep's table in
    // cascade.rs); here each stage is only decorated with per-stage tallies
    // (settled count + busy time summed across workers) -- the readout the
    // stage ordering is tuned with.
    struct Tally {
        settled: AtomicUsize,
        busy_ns: AtomicU64,
    }
    let mut tallies: Vec<(&'static str, Arc<Tally>)> = Vec::new();
    let stages: Vec<Stage> = fast_stages::<T>(b, deep_torus_kmax, deep_bound, deep_budget)
        .into_iter()
        .filter(|st| st.kind.selected_by(cfg.mode))
        .map(|st| {
            let t = Arc::new(Tally {
                settled: AtomicUsize::new(0),
                busy_ns: AtomicU64::new(0),
            });
            tallies.push((st.name, t.clone()));
            let inner = st.check;
            let check = move |s: &[i8]| {
                let t0 = Instant::now();
                let out = inner(s);
                t.busy_ns
                    .fetch_add(t0.elapsed().as_nanos() as u64, Ordering::Relaxed);
                if matches!(out, StageOutcome::Settled(_)) {
                    t.settled.fetch_add(1, Ordering::Relaxed);
                }
                out
            };
            Stage {
                name: st.name,
                kind: st.kind,
                check: Box::new(check),
            }
        })
        .collect();

    let mut wstart = start;
    while wstart < end {
        let wend = (wstart + chunk as u64).min(end);
        if chunk < n {
            eprintln!("  window [{wstart},{wend})");
        }
        // Coverage-aware skip BEFORE materializing: a fully-covered window
        // costs nothing on resume (no dafsa reads at all).
        let open: Vec<u64> = (wstart..wend)
            .filter(|&i| !in_covered_union(&union, i))
            .collect();
        if open.is_empty() {
            wstart = wend;
            continue;
        }
        // Only this window's open rats are resident.
        let rats: Vec<Vec<i8>> = open
            .iter()
            .map(|&i| d.get(i).expect("index in range"))
            .collect();
        let todo: Vec<usize> = (0..open.len())
            .filter(|&j| !done.contains(&open[j]))
            .collect();
        let line_resumed = open.len() - todo.len();
        let before = [
            cnt[0].load(Ordering::Relaxed),
            cnt[1].load(Ordering::Relaxed),
            cnt[2].load(Ordering::Relaxed),
        ];
        let st = Instant::now();
        // Tile-major: each tile runs the whole funnel to a verdict (the executor
        // settles EVERY tile -- an unsettled one becomes Undecided with the
        // funnel's banked witness), so there is no per-stage barrier and no
        // separate residue pass.
        parallel_sweep("classify", todo.len(), workers, |j| {
            let t = todo[j];
            emit(
                &rats[t],
                open[t],
                &run_stages::<T>(&rats[t], &stages, cfg.mode),
            );
        });
        for (name, t) in &tallies {
            let settled = t.settled.swap(0, Ordering::Relaxed);
            let busy = t.busy_ns.swap(0, Ordering::Relaxed);
            eprintln!(
                "  stage {name}: settled {settled}, busy {:.1}s",
                busy as f64 / 1e9
            );
        }
        eprintln!(
            "  window: {} tiles classified ({:?})",
            todo.len(),
            st.elapsed()
        );
        // Sidecar: record the window as covered ONLY when every open tile was
        // emitted fresh in this run -- then the record's tallies count each
        // not-previously-covered tile of the window exactly once (overlaps
        // with older records are legal; tallies SUM to one count per tile)
        // and its store lines may later be filtered. A window with
        // line-resumed tiles stays line-accounted.
        if line_resumed == 0 && !todo.is_empty() {
            writer.lock().unwrap().0.flush().unwrap(); // lines land before the coverage claim
            let after = [
                cnt[0].load(Ordering::Relaxed),
                cnt[1].load(Ordering::Relaxed),
                cnt[2].load(Ordering::Relaxed),
            ];
            append_range(
                out,
                CoveredRange {
                    start: wstart,
                    end: wend,
                    tally: [
                        after[0] - before[0],
                        after[1] - before[1],
                        after[2] - before[2],
                    ],
                },
            );
        }
        wstart = wend;
    }
    writer.lock().unwrap().0.flush().unwrap();

    let load = |i: usize| cnt[i].load(Ordering::Relaxed);
    let (p, nn, u, vf) = (load(0), load(1), load(2), load(3));
    let stats = ClassifyStats {
        periodic: rp + p,
        cannot_tile: rn + nn,
        undecided: ru + u,
        verify_fail: vf,
        total: covered_ct as usize + done.len() + p + nn + u,
    };
    eprintln!(
        "fast [{start},{end}) done in {:?}: Periodic={} CannotTile={} Undecided={} (resumed {}, new {}); verify_fail={vf}; total={}",
        t0.elapsed(),
        stats.periodic,
        stats.cannot_tile,
        stats.undecided,
        covered_ct as usize + done.len(),
        p + nn + u,
        stats.total,
    );
    assert_eq!(stats.total, n, "every rat accounted for (store + this run)");
    assert_eq!(vf, 0, "every minted cert must self-verify");
    stats
}

// ---------------------------------------------------------------------------
// The deep pass (`--deep`): crunch the Undecided residue.
// ---------------------------------------------------------------------------

/// Deep-pass (`--deep`) parameters (the "crunch the rest hard" bounds).
#[derive(Debug, Clone)]
pub struct DeepConfig {
    /// Which halves of the cascade to run (default: full). `PeriodicOnly` /
    /// `HeeschOnly` chase one verdict on the residue when you know what to expect.
    pub mode: StageMode,
    /// Deep-Heesch corona bound.
    pub res_bound: usize,
    /// Deep-Heesch node budget. 100M is calibrated to the MEASURED worst case, not
    /// a round guess: there is a "budget-limited Heesch family" -- genuine non-tilers
    /// with a small Heesch number but an EXPENSIVE proof (exhausting "no corona k+1").
    /// Across the whole n<=13 residue the worst is n=10 idx 12336 (Heesch 2) at
    /// 90,679,656 nodes (~58min); idx 14908 is 28.8M; the n=13 stragglers are only
    /// ~0.5-1.7M. So 100M is ~10% above the true worst -- do NOT lower it (idx 12336
    /// would then stay Undecided, breaking the bit-exact le13). This is a CEILING:
    /// free for tiles that resolve below it (every n<=13 tile does), so it never
    /// slows the classification -- it only bounds a genuine aperiodic-candidate grind at higher
    /// n. Measure with HEESCH_SPENT=1 before changing it.
    pub res_budget: usize,
    /// Acceptance bounds (higher kmax + seeded aniso restarts than the fast pass).
    pub bounds: AcceptBounds,
    /// Worker threads (0 = all cores).
    pub workers: usize,
}

impl Default for DeepConfig {
    fn default() -> Self {
        // Generous multi-copy reach for the small residue, and the aniso search is
        // seeded RESTARTS (heavy-tailed -> a hard cluster surfaces early under some
        // seed) bounded at restarts * per_budget, so 12 * 30k = 360k examined --
        // comfortably above the old --fast single-search ceiling (300k), so every
        // aniso periodic the fast pass now defers (it bails at 5k) is caught here. Torus
        // grows deeper coronas to expose large-period lattices directly.
        let bounds = AcceptBounds {
            torus_kmax: 40,
            // 4 -> 5 (2026-07-07, measured): the two related n=14 tiles with
            // k=18 domains (idx 16856076 / 23665356) hide from the coronas-4
            // patch but expose their lattice at coronas 5 in 0.1s.
            torus_coronas: 5,
            aniso_kmax: 6,
            aniso_budget: 30_000,
            aniso_restarts: 12,
            ..AcceptBounds::default()
        };
        Self {
            mode: StageMode::Full,
            res_bound: 6,
            res_budget: 100_000_000,
            bounds,
            workers: 0,
        }
    }
}

/// Result of a deep pass: how many Undecided tiles moved where.
#[derive(Debug, Clone, Copy, Default)]
pub struct DeepStats {
    pub to_periodic: usize,
    pub to_cannot_tile: usize,
    pub still_undecided: usize,
}

/// Re-classify the store's Undecided residue (parallel), APPENDING each
/// result to the `<store>.deep.jsonl` overlay as it lands -- the store itself
/// is never touched (apply the overlay with [`crate::classify::store::run_merge`]). This makes the
/// deep pass RESUMABLE: a killed run loses only its in-flight tiles, and a
/// restart reads the overlay back (the same self-healing `read_resume` the
/// fast pass uses) and skips everything already resolved.
///
/// Each Undecided tile runs [`classify`] -- the deep funnel: cheap
/// accepts, ONE full-budget Heesch reject, then the expensive accepts (seeded
/// aniso restarts / iso / deep torus) -- so a periodic tile the fast pass
/// missed is caught here with the escalated bounds. For LATENCY the cheap
/// accepts sweep the whole residue on their own first (verdict-identical --
/// they are classify's own deterministic step 1), so instant certifications
/// are never head-of-line blocked behind hours-long grinders.
/// `cfg.mode` restricts the cascade to one half (`--periodic-only`
/// / `--heesch-only`), which is the usual deep-pass shape when the residue's likely
/// verdict is known. Panics if any newly-minted cert fails to verify.
pub fn run_deep<T: IsRing, F>(d: &LazyRatDafsa<F>, store: &Path, cfg: &DeepConfig) -> DeepStats
where
    F: Fn(u32) -> io::Result<Vec<u8>>,
{
    use std::io::Write;

    let workers = workers_or_all(cfg.workers);
    let (bounds, res_bound, res_budget, mode) =
        (cfg.bounds, cfg.res_bound, cfg.res_budget, cfg.mode);
    let overlay = deep_overlay_path(store);
    // Resume: results already in the overlay are skipped (and pre-tallied).
    let (done, [rp, rn, ru]) = read_resume(&overlay, 0, u64::MAX, &[]);

    // Gather the Undecided work (dafsa index, seq, banked corona witness) up
    // front -- `d.get` runs on this thread -- so the re-classify runs in
    // parallel on the gathered data alone.
    let txt = std::fs::read_to_string(store).expect("store readable");
    let mut work: Vec<(u64, Vec<i8>, Vec<crate::geom::matches::PatchMatch>)> = Vec::new();
    for line in txt.lines() {
        let Some((idx_s, json)) = line.split_once('\t') else {
            continue;
        };
        let Ok(idx) = idx_s.parse::<u64>() else {
            continue;
        };
        if done.contains(&idx) {
            continue;
        }
        let Ok(c) = serde_json::from_str::<Classified>(json) else {
            continue;
        };
        if let Classified::Undecided { corona, .. } = c {
            work.push((idx, d.get(idx).expect("idx in range"), corona));
        }
    }
    let total = work.len();
    eprintln!(
        "deep: {total} Undecided tiles to re-classify ({} already in {}), {workers} workers",
        done.len(),
        overlay.display()
    );

    let writer = Mutex::new(std::io::BufWriter::new(
        std::fs::OpenOptions::new()
            .create(true)
            .append(true)
            .open(&overlay)
            .unwrap(),
    ));
    let cnt = [
        AtomicUsize::new(0),
        AtomicUsize::new(0),
        AtomicUsize::new(0),
        AtomicUsize::new(0),
    ];
    // Per-tile eprintln below is the fine-grained progress; parallel_sweep adds
    // the rate heartbeat on a big residue. Flushed per line: deep tiles cost
    // seconds to hours, so the syscall is noise and every finished tile is
    // durable immediately.
    let emit = |idx: u64, seq: &[i8], nc: &Classified, secs: f64, tag: &str| {
        let base = Rat::<T>::from_slice_trusted(seq);
        let (slot, ok) = verdict_check(&base, nc);
        cnt[slot].fetch_add(1, Ordering::Relaxed);
        if !ok {
            cnt[3].fetch_add(1, Ordering::Relaxed);
            eprintln!("CERT-FAIL idx {idx}");
        }
        let new = ["Periodic", "CannotTile", "Undecided"][slot];
        eprintln!("idx {idx} Undecided->{new} ({secs:.2}s{tag}) seq {seq:?}");
        let mut wtr = writer.lock().unwrap();
        writeln!(wtr, "{idx}\t{}", serde_json::to_string(nc).unwrap()).unwrap();
        wtr.flush().unwrap();
    };

    // PHASE 1 (latency, not verdicts): sweep the CHEAP steps alone over the
    // whole residue first, so every easily-certified tile is emitted within
    // seconds. Without this, a run of expensive tiles at the front of the
    // list head-of-line blocks hours of instant certifications behind it.
    // Two cheap steps per tile: the funnel's own cheap accepts, then the
    // WITNESS replay (the tile's banked corona is routinely deeper than any
    // funnel grow -- if the lattice shows in it, the tile certifies without
    // any growth or search at all). Verdicts are unchanged: the cheap accepts
    // are classify's own deterministic step 1, and the witness step only ADDS
    // verified certs for tiles the funnel would grind much longer for.
    let survivors: Vec<usize> = if mode.runs_accepts() {
        let kept: Mutex<Vec<usize>> = Mutex::new(Vec::new());
        parallel_sweep("deep-cheap", total, workers, |w| {
            let (idx, seq, witness) = &work[w];
            let t = Instant::now();
            let base = Rat::<T>::from_slice_trusted(seq);
            let pc =
                crate::classify::cascade::certify_periodic_cheap::<T>(seq, &bounds).or_else(|| {
                    crate::classify::mint::witness_torus_cert(&base, witness, bounds.torus_kmax)
                });
            match pc {
                Some(pc) => {
                    let nc = Classified::Decided(Verdict::Periodic(pc));
                    emit(*idx, seq, &nc, t.elapsed().as_secs_f64(), ", cheap");
                }
                None => kept.lock().unwrap().push(w),
            }
        });
        let mut kept = kept.into_inner().unwrap();
        kept.sort_unstable();
        eprintln!(
            "deep: cheap accepts settled {}, {} remain for the full funnel",
            total - kept.len(),
            kept.len()
        );
        kept
    } else {
        (0..total).collect()
    };

    // PHASE 2: the full deep funnel on the hard survivors.
    parallel_sweep("deep", survivors.len(), workers, |s| {
        let (idx, seq, _) = &work[survivors[s]];
        let t = Instant::now();
        let nc = classify::<T>(seq, &bounds, res_bound, res_budget, mode);
        emit(*idx, seq, &nc, t.elapsed().as_secs_f64(), "");
    });
    assert_eq!(
        cnt[3].load(Ordering::Relaxed),
        0,
        "every newly-minted cert must verify"
    );

    let stats = DeepStats {
        to_periodic: rp + cnt[0].load(Ordering::Relaxed),
        to_cannot_tile: rn + cnt[1].load(Ordering::Relaxed),
        still_undecided: ru + cnt[2].load(Ordering::Relaxed),
    };
    eprintln!(
        "deep: Undecided -> Periodic {} / CannotTile {} / still-Undecided {} (incl. resumed); apply with --merge {}",
        stats.to_periodic,
        stats.to_cannot_tile,
        stats.still_undecided,
        overlay.display()
    );
    stats
}

// ---------------------------------------------------------------------------

/// Result of [`run_verify`]. A clean store has `bad_lines == 0`,
/// `reverify_fail == 0`, `incomplete == 0`, `dups == 0`.
#[derive(Debug, Clone, Default)]
pub struct VerifyReport {
    pub perimeters: usize,
    pub bad_lines: usize,
    pub reverify_fail: usize,
    pub incomplete: usize,
    pub dups: usize,
}

impl VerifyReport {
    /// Whether the store is fully sound + complete.
    pub fn is_clean(&self) -> bool {
        self.perimeters > 0
            && self.bad_lines == 0
            && self.reverify_fail == 0
            && self.incomplete == 0
            && self.dups == 0
    }
}

/// Screen a cert `store`'s PERIODIC tiles for rep-tiles: does each tile tile a
/// `k`-scaled copy of itself with `k*k` copies? Only periodics are candidates --
/// a rep-tile tiles the plane, so it is necessarily one of the store's periodic
/// verdicts, never a proven non-tiler. Writes one `<index>\t<RepTileCert>` line
/// per hit to `out` and prints a by-order tally.
///
/// The store is STREAMED (never fully resident -- it can be gigabytes): pass 1
/// filters the periodic indices with a cheap substring test (a full serde parse
/// of every verdict just to bucket would dominate), pass 2 resolves each seq
/// from the dataset and runs the exact rep-tile fill in parallel.
pub fn run_reptile_screen<T: IsRing, F: Fn(u32) -> io::Result<Vec<u8>>>(
    d: &LazyRatDafsa<F>,
    store: &Path,
    out: &Path,
    kmax: usize,
    budget: usize,
) {
    use std::collections::BTreeMap;
    use std::io::{BufRead, Write};

    let rdr = std::io::BufReader::new(std::fs::File::open(store).expect("open cert store"));
    let mut periodics: Vec<u64> = Vec::new();
    for line in rdr.lines() {
        let line = line.expect("read cert store line");
        let Some((idx_s, json)) = line.split_once('\t') else {
            continue;
        };
        // Cheap candidate filter: "Periodic" appears only in the periodic-verdict
        // tag, so this picks the rep-tile candidates without a full serde parse.
        if !json.contains("\"Periodic\"") {
            continue;
        }
        if let Ok(idx) = idx_s.parse::<u64>() {
            periodics.push(idx);
        }
    }
    // Resolve seqs serially (the dataset reader is not shared across threads);
    // the fill, not the decode, is the work.
    let cands: Vec<(u64, Vec<i8>)> = periodics
        .iter()
        .filter_map(|&idx| d.get(idx).map(|s| (idx, s)))
        .collect();
    let n = cands.len();
    eprintln!(
        "reptile-screen: {} periodic tiles ({n} resolved), kmax={kmax}, budget={budget}",
        periodics.len()
    );

    // Work-stealing fill (the crate parallelizes with `thread::scope`, no
    // rayon): each worker runs the exact rep-tile fill on candidates pulled from
    // a shared cursor and accumulates `(idx, cert json)` hit lines, a by-order
    // tally, and the budget-capped (inconclusive) indices. Hits are keyed by
    // index and sorted before writing, so the output is deterministic despite
    // the nondeterministic completion order.
    type Acc = (Vec<(u64, String)>, BTreeMap<usize, usize>, Vec<u64>);
    let workers = crate::util::available_workers();
    let done = AtomicUsize::new(0);
    let (mut hit_lines, counts, mut inconclusive): Acc = crate::util::parallel::parallel_drain(
        cands.len(),
        workers,
        || (Vec::new(), BTreeMap::new(), Vec::new()),
        |acc, i| {
            let (idx, seq) = &cands[i];
            let base = Rat::<T>::from_slice_trusted(seq);
            match reptile_screen_one(&base, kmax, budget) {
                RepScreen::RepTile(cert) => {
                    let json = serde_json::to_string(&cert).unwrap();
                    acc.0.push((*idx, json));
                    *acc.1.entry(cert.k * cert.k).or_insert(0) += 1;
                }
                RepScreen::Inconclusive => acc.2.push(*idx),
                RepScreen::Anomaly(k) => {
                    eprintln!(
                        "  ANOMALY idx {idx}: scale-{k} tiling found but cert re-verify failed (build-reconstruction bug)"
                    );
                    acc.2.push(*idx);
                }
                RepScreen::No => {}
            }
            let c = done.fetch_add(1, Ordering::Relaxed) + 1;
            if c.is_multiple_of(50_000) {
                eprintln!("  {c}/{} screened", cands.len());
            }
        },
        |mut a, b| {
            a.0.extend(b.0);
            for (order, c) in b.1 {
                *a.1.entry(order).or_insert(0) += c;
            }
            a.2.extend(b.2);
            a
        },
    );

    hit_lines.sort_unstable_by_key(|(idx, _)| *idx);
    let mut w = std::io::BufWriter::new(std::fs::File::create(out).expect("create reptile out"));
    for (idx, json) in &hit_lines {
        writeln!(w, "{idx}\t{json}").unwrap();
    }
    w.flush().unwrap();
    let hits: usize = counts.values().sum();
    eprintln!(
        "reptile-screen: {hits} rep-tiles among {n} periodics ({} inconclusive/budget-capped)",
        inconclusive.len()
    );
    for (order, c) in &counts {
        eprintln!("  order {order}: {c}");
    }
    if !inconclusive.is_empty() {
        inconclusive.sort_unstable();
        let side = out.with_extension("inconclusive");
        let text: String = inconclusive.iter().map(|i| format!("{i}\n")).collect();
        if std::fs::write(&side, text).is_ok() {
            eprintln!(
                "  wrote {} budget-capped indices to {} (re-run deeper if wanted)",
                inconclusive.len(),
                side.display()
            );
        }
    }
}

/// Re-verify a finished store from disk: replay every cert (PeriodicCert::verify /
/// HeeschCert::verify_lower_bound; Undecided passes trivially) and check
/// COMPLETENESS -- every perimeter present must be fully accounted for,
/// exactly once. With a coverage sidecar present (a store whose boring certs
/// were filtered away), completeness means kept lines PLUS covered ranges
/// jointly cover the whole block, and the reported P/N/U totals combine the
/// range tallies with the tallies of lines OUTSIDE the covered union (each
/// tile counted exactly once -- the coverage invariant); the caveat is
/// inherent: discarded tiles' tallies rest on their at-emission verification,
/// they are no longer re-replayable. Without a sidecar this is the strict
/// all-lines check. `only_perim` restricts to one perimeter; `None` checks
/// every perimeter block present in a (possibly combined) store. Prints a
/// per-perimeter summary and returns the report (the caller decides whether a
/// dirty store is fatal).
pub fn run_verify<T: IsRing, F>(
    d: &LazyRatDafsa<F>,
    counts: &[u64],
    store: &Path,
    only_perim: Option<usize>,
) -> VerifyReport
where
    F: Fn(u32) -> io::Result<Vec<u8>>,
{
    use std::collections::{BTreeMap, HashSet};

    #[derive(Default)]
    struct Stats {
        lines: usize,
        seen: HashSet<u64>,
        /// Line tallies OUTSIDE the covered union (inside it, the range
        /// tallies are authoritative and the lines are the kept subset).
        p: usize,
        nn: usize,
        u: usize,
        vf: usize,
        /// Range-tally totals + covered index count for this perimeter.
        rp: usize,
        rn: usize,
        ru: usize,
        covered_ct: u64,
    }
    let t0 = Instant::now();
    let ranges = read_ranges(store);
    let union = covered_union(&ranges);
    let txt = std::fs::read_to_string(store).expect("store readable");
    let mut per: BTreeMap<usize, Stats> = BTreeMap::new();
    let mut bad = 0usize;
    // Fold range tallies + covered counts into their perimeters (a sidecar
    // range straddling a perimeter boundary would mis-attribute tallies; the
    // fast pass never emits such windows for --perim launches, and --from/--to
    // launches should align with perimeter boundaries for exact per-perimeter
    // stats).
    for r in &ranges {
        let Some(pm) = perimeter_of(counts, r.start) else {
            bad += 1;
            continue;
        };
        if only_perim.is_some_and(|op| op != pm) {
            continue;
        }
        let e = per.entry(pm).or_default();
        e.rp += r.tally[0];
        e.rn += r.tally[1];
        e.ru += r.tally[2];
    }
    // Covered index counts per perimeter (clip union intervals to blocks).
    for &(s0, e0) in &union {
        let mut pos = s0;
        while pos < e0 {
            let Some(pm) = perimeter_of(counts, pos) else {
                break;
            };
            let (_, block_end) = perimeter_range(counts, pm);
            let seg_end = e0.min(block_end);
            if only_perim.is_none_or(|op| op == pm) {
                per.entry(pm).or_default().covered_ct += seg_end - pos;
            }
            pos = seg_end;
        }
    }
    for line in txt.lines() {
        let Some((idx_s, json)) = line.split_once('\t') else {
            bad += 1;
            continue;
        };
        let Ok(idx) = idx_s.parse::<u64>() else {
            bad += 1;
            continue;
        };
        let Some(pm) = perimeter_of(counts, idx) else {
            bad += 1;
            continue;
        };
        if only_perim.is_some_and(|op| op != pm) {
            continue;
        }
        let Ok(c) = serde_json::from_str::<Classified>(json) else {
            bad += 1;
            continue;
        };
        let in_union = in_covered_union(&union, idx);
        let e = per.entry(pm).or_default();
        e.lines += 1;
        e.seen.insert(idx);
        let base = Rat::<T>::from_slice_trusted(&d.get(idx).expect("index in range"));
        // Kept lines always re-verify; they join the P/N/U totals only when
        // OUTSIDE the covered union (inside it the range tallies count them).
        let ok = match &c {
            Classified::Decided(Verdict::Periodic(pc)) => {
                if !in_union {
                    e.p += 1;
                }
                pc.verify(&base)
            }
            Classified::Decided(Verdict::CannotTile(hc)) => {
                if !in_union {
                    e.nn += 1;
                }
                hc.verify_lower_bound(&base)
            }
            Classified::Undecided { .. } => {
                if !in_union {
                    e.u += 1;
                }
                true
            }
        };
        if !ok {
            e.vf += 1;
            eprintln!("RE-VERIFY FAIL idx {idx} (perimeter {pm})");
        }
    }

    let mut report = VerifyReport {
        perimeters: per.len(),
        bad_lines: bad,
        ..Default::default()
    };
    for (&pm, e) in &per {
        let expect = counts[pm] as usize;
        // Accounted = covered indices + kept lines outside the union (an
        // in-union kept line is the retained subset of its range, not extra).
        let outside = e
            .seen
            .iter()
            .filter(|&&i| !in_covered_union(&union, i))
            .count();
        let accounted = e.covered_ct as usize + outside;
        let complete = accounted == expect;
        let nodup = e.lines == e.seen.len();
        if !complete {
            report.incomplete += 1;
        }
        if !nodup {
            report.dups += 1;
        }
        report.reverify_fail += e.vf;
        eprintln!(
            "  perimeter {pm}: {accounted}/{expect} {}{}{} | P={} N={} U={} | reverify_fail={}",
            if complete { "COMPLETE" } else { "INCOMPLETE" },
            if nodup { "" } else { " +DUPLICATES" },
            if e.covered_ct > 0 {
                format!(" ({} by coverage, {} kept lines)", e.covered_ct, e.lines)
            } else {
                String::new()
            },
            e.p + e.rp,
            e.nn + e.rn,
            e.u + e.ru,
            e.vf,
        );
    }
    eprintln!(
        "verify: {} perimeter(s); bad_lines {}; reverify_fail {}; incomplete {}; with_dups {}; {:?}",
        report.perimeters,
        report.bad_lines,
        report.reverify_fail,
        report.incomplete,
        report.dups,
        t0.elapsed()
    );
    report
}

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

    /// `parallel_sweep` must run every index exactly once and return -- with
    /// `total` above `PROGRESS_MIN` so the heartbeat monitor + drop-guard path
    /// is exercised (a broken stop signal would deadlock the enclosing scope).
    #[test]
    fn parallel_sweep_runs_every_item_with_heartbeat() {
        use std::sync::atomic::{AtomicUsize, Ordering};
        let total = 3_000usize;
        let count = AtomicUsize::new(0);
        let sum = AtomicUsize::new(0);
        parallel_sweep("test", total, 4, |j| {
            count.fetch_add(1, Ordering::Relaxed);
            sum.fetch_add(j, Ordering::Relaxed);
        });
        assert_eq!(count.load(Ordering::Relaxed), total);
        assert_eq!(sum.load(Ordering::Relaxed), total * (total - 1) / 2);
    }

    /// Positive control for the Rust classify path (the runner/filter glue is
    /// separate -- see the end-to-end check): the EXACT --fast funnel must flag
    /// the known interesting tiles as "kept", never silently bucket them as a
    /// heesch<2 reject or (falsely) periodic.
    ///
    /// - SPECTRE (aperiodic monotile) -> Undecided.
    /// - A known Heesch-3 tile -> CannotTile(heesch=3) [kept, >=2] OR, if the
    ///   fast bound-3 budget cannot exhaust the 4th corona, AtLeast/Undecided
    ///   [also kept]. We ALSO assert heesch_number reports exactly 3, so the
    ///   reject path's Heesch value itself is verified, not just "interesting".
    #[cfg(feature = "cli")]
    #[test]
    #[ignore = "positive control (~min): known aperiodic + Heesch-3 tiles must be kept"]
    fn fast_catches_known_interesting_tiles() {
        use crate::classify::cert::{Classified, HeeschStatus, Verdict};
        use crate::classify::heesch::{Heesch, heesch_number};
        use crate::cyclotomic::ZZ12;
        use crate::geom::rat::Rat;
        use crate::geom::tiles;
        use crate::geom::tileset::TileSet;

        let cfg = FastConfig::default();
        let stages = fast_stages::<ZZ12>(
            cfg.bounds,
            cfg.deep_torus_kmax,
            cfg.deep_bound,
            cfg.deep_budget,
        );
        let classify_one = |seq: &[i8]| run_stages::<ZZ12>(seq, &stages, StageMode::Full);
        // Mirror tools/filter_store.py::interesting (Periodic excluded here).
        let interesting = |c: &Classified| match c {
            Classified::Undecided { .. } => true,
            Classified::Decided(Verdict::CannotTile(hc)) => {
                hc.heesch >= 2 || matches!(hc.status, HeeschStatus::Unknown)
            }
            Classified::Decided(Verdict::Periodic(_)) => false,
        };

        // SPECTRE: minimal chiral aperiodic ZZ12 monotile -> MUST be Undecided.
        let spectre_seq = Rat::<ZZ12>::from_snake_trusted(&tiles::spectre::<ZZ12>())
            .seq()
            .to_vec();
        let sc = classify_one(&spectre_seq);
        eprintln!("SPECTRE -> {sc:?}");
        assert!(
            matches!(sc, Classified::Undecided { .. }),
            "SPECTRE must classify Undecided, got {sc:?}"
        );

        // A known Heesch-3 tile (12-edge; 3-corona patch of 39 tiles).
        let h3: &[i8] = &[-4, 2, 4, -4, 2, 4, -2, 4, -2, 2, 2, 4];
        let ts = TileSet::single(Rat::<ZZ12>::from_slice_trusted(h3));
        let hn = heesch_number(ts, 0, 4, 20_000_000);
        eprintln!("Heesch-3 tile heesch_number(bound=4) -> {hn:?}");
        assert!(
            matches!(hn, Heesch::Finite(3)),
            "known Heesch-3 tile must report Finite(3), got {hn:?}"
        );
        let hc = classify_one(h3);
        eprintln!("Heesch-3 tile --fast -> {hc:?}");
        assert!(
            interesting(&hc),
            "known Heesch-3 tile must be kept (heesch>=2/Undecided), got {hc:?}"
        );
    }

    /// perimeter_range partitions the DAFSA index space by the FREE_ZZ12 counts:
    /// consecutive, non-overlapping, each block sized FREE_ZZ12[n].
    #[test]
    fn perimeter_ranges_partition_the_index_space() {
        let mut prev_end = 0u64;
        for (n, &count) in FREE_ZZ12.iter().enumerate().skip(1) {
            let (s, e) = perimeter_range(&FREE_ZZ12, n);
            assert_eq!(s, prev_end, "perimeter {n} starts where {} ended", n - 1);
            assert_eq!(e - s, count, "perimeter {n} block size");
            prev_end = e;
        }
        // perimeter_of is the inverse on a few interior points.
        for n in 3..=13 {
            let (s, e) = perimeter_range(&FREE_ZZ12, n);
            assert_eq!(perimeter_of(&FREE_ZZ12, s), Some(n));
            assert_eq!(perimeter_of(&FREE_ZZ12, e - 1), Some(n));
        }
    }

    /// The RO-Crate metadata readers agree with the hardcoded ZZ12 convention
    /// and read the ZZ10 asset correctly (needs the ratdb assets, so opt-in):
    /// `asset_free_counts` for the ZZ12 asset must reproduce [`FREE_ZZ12`]
    /// byte-for-byte (index == perimeter, leading 0 prepended), and the ZZ10
    /// asset's ring/counts must match its published totals.
    #[test]
    #[ignore = "metadata readers: needs the ratdb assets (ro-crate-metadata.json)"]
    fn asset_metadata_readers_match_conventions() {
        assert_eq!(asset_ring("web/ratdb/data/zz12_n16_free"), 12);
        assert_eq!(
            asset_free_counts("web/ratdb/data/zz12_n16_free"),
            FREE_ZZ12.to_vec(),
            "zz12 asset counts must equal the FREE_ZZ12 const"
        );
        assert_eq!(asset_ring("web/ratdb/data/zz10_n18_free"), 10);
        let z10 = asset_free_counts("web/ratdb/data/zz10_n18_free");
        assert_eq!(z10[0], 0, "leading 0 prepended (index == perimeter)");
        assert_eq!(z10[16], 66_850_339, "zz10 perimeter-16 free count");
        assert_eq!(z10.iter().sum::<u64>(), 2_875_831_850, "zz10 total");
    }

    /// The self-healing resume reader: good lines are kept and tallied,
    /// malformed lines and duplicate indices are DROPPED by an atomic rewrite
    /// (their tiles re-run), and a torn trailing line disappears with them.
    #[test]
    fn read_resume_drops_bad_and_duplicate_lines() {
        let undecided = r#"{"Undecided":{"depth":0,"corona":[]}}"#;
        let out = std::env::temp_dir().join("read_resume_test.jsonl");
        // Line 2 malformed, line 4 duplicates idx 5, the idx-8 line is torn
        // (no trailing newline).
        let txt =
            format!("5\t{undecided}\n6\tGARBAGE\n7\t{undecided}\n5\t{undecided}\n8\t{undecided}");
        std::fs::write(&out, &txt).unwrap();
        let (done, tally) = read_resume(&out, 0, 100, &[]);
        assert_eq!(tally, [0, 0, 2], "idx 5 and 7 tallied once each");
        assert!(done.contains(&5) && done.contains(&7) && !done.contains(&6) && !done.contains(&8));
        // The rewrite kept exactly the two good complete lines.
        let back = std::fs::read_to_string(&out).unwrap();
        assert_eq!(back, format!("5\t{undecided}\n7\t{undecided}\n"));
        // Idempotent: a clean store is left untouched.
        let (done2, tally2) = read_resume(&out, 0, 100, &[]);
        assert_eq!((done2.len(), tally2), (2, [0, 0, 2]));
        assert_eq!(std::fs::read_to_string(&out).unwrap(), back);
        let _ = std::fs::remove_file(&out);
    }

    /// Coverage-sidecar primitives: parse (torn tail skipped), union merge of
    /// overlapping/adjacent records, O(log n) membership, and read_resume's
    /// covered-lines-excluded tally semantics.
    #[test]
    fn coverage_sidecar_primitives() {
        use std::io::Write;
        let dir = std::env::temp_dir().join(format!("cov_test_{}", std::process::id()));
        std::fs::create_dir_all(&dir).unwrap();
        let store = dir.join("s.jsonl");

        // Sidecar: two adjacent records, one overlapping, one disjoint, a torn tail.
        let mut f = std::fs::File::create(ranges_path(&store)).unwrap();
        writeln!(f, "10\t20\t1\t8\t1").unwrap();
        writeln!(f, "20\t30\t2\t8\t0").unwrap();
        writeln!(f, "25\t35\t0\t5\t0").unwrap();
        writeln!(f, "50\t60\t3\t7\t0").unwrap();
        write!(f, "70\t80").unwrap(); // torn tail: skipped
        drop(f);
        let ranges = read_ranges(&store);
        assert_eq!(ranges.len(), 4, "torn tail skipped");
        let union = covered_union(&ranges);
        assert_eq!(union, vec![(10, 35), (50, 60)]);
        for (idx, want) in [
            (9, false),
            (10, true),
            (34, true),
            (35, false),
            (49, false),
            (50, true),
            (59, true),
            (60, false),
        ] {
            assert_eq!(in_covered_union(&union, idx), want, "idx {idx}");
        }

        // read_resume: a line inside the union joins neither done nor the
        // tally; a line outside joins both.
        let undec = serde_json::to_string(&Classified::Undecided {
            depth: 0,
            corona: vec![],
        })
        .unwrap();
        let mut f = std::fs::File::create(&store).unwrap();
        writeln!(f, "12\t{undec}").unwrap(); // covered
        writeln!(f, "40\t{undec}").unwrap(); // uncovered
        drop(f);
        let (done, tally) = read_resume(&store, 0, 100, &union);
        assert!(!done.contains(&12) && done.contains(&40));
        assert_eq!(tally, [0, 0, 1], "only the uncovered line tallied");
        std::fs::remove_dir_all(&dir).ok();
    }

    /// End-to-end coverage roundtrip on real perimeters (needs the ratdb
    /// asset): classify perimeter 8 as two --from/--to half-range launches
    /// (sidecar records per window), FILTER the boring lines away (keeping
    /// only Undecided/high-Heesch/large-k -- here: simulate the Python filter
    /// by dropping every covered CannotTile line), re-launch over the full
    /// perimeter (must do zero work -- everything coverage-resumed), and
    /// verify: completeness via kept-lines + covered ranges, exact combined
    /// P/N/U totals, every kept cert re-verified.
    #[test]
    #[ignore = "coverage roundtrip: range launches + filter + verify (needs the ratdb asset)"]
    fn range_launch_filter_verify_roundtrip() {
        use crate::cyclotomic::ZZ12;
        use std::io::Write;
        let asset = std::env::var("ASSET").unwrap_or("web/ratdb/data/zz12_n16_free".into());
        let d = open_ratdb(&asset);
        let dir = std::env::temp_dir().join(format!("covrt_{}", std::process::id()));
        std::fs::create_dir_all(&dir).unwrap();
        let store = dir.join("s.jsonl");
        let (s8, e8) = perimeter_range(&FREE_ZZ12, 8);
        let mid = (s8 + e8) / 2;
        let cfg = FastConfig {
            chunk: 100,
            ..FastConfig::default()
        };

        // Two half-range launches; windows of 100 -> several sidecar records.
        let st1 = run_fast_range::<ZZ12, _>(&d, s8, mid, &store, &cfg);
        let st2 = run_fast_range::<ZZ12, _>(&d, mid, e8, &store, &cfg);
        assert_eq!(st1.total as u64, mid - s8);
        assert_eq!(st2.total as u64, e8 - mid);
        let ranges = read_ranges(&store);
        assert!(!ranges.is_empty(), "windows recorded");
        let union = covered_union(&ranges);
        assert!(in_covered_union(&union, s8) && in_covered_union(&union, e8 - 1));
        let total_tallied: [usize; 3] = ranges.iter().fold([0; 3], |mut a, r| {
            for (acc, t) in a.iter_mut().zip(r.tally) {
                *acc += t;
            }
            a
        });
        assert_eq!(
            total_tallied.iter().sum::<usize>() as u64,
            e8 - s8,
            "tallies cover the block"
        );

        // Filter: drop covered CannotTile lines (the boring bulk), like
        // tools/filter_store.py with a high --min-heesch.
        let txt = std::fs::read_to_string(&store).unwrap();
        let mut f = std::fs::File::create(&store).unwrap();
        let mut dropped = 0;
        for line in txt.lines() {
            let (idx_s, json) = line.split_once('\t').unwrap();
            let idx: u64 = idx_s.parse().unwrap();
            let c: Classified = serde_json::from_str(json).unwrap();
            if in_covered_union(&union, idx)
                && matches!(c, Classified::Decided(Verdict::CannotTile(_)))
            {
                dropped += 1;
                continue;
            }
            writeln!(f, "{line}").unwrap();
        }
        drop(f);
        assert!(dropped > 0, "the boring bulk existed and was dropped");

        // Full-perimeter relaunch: everything is coverage-resumed, zero new work.
        let st3 = run_fast::<ZZ12, _>(&d, &FREE_ZZ12, 8, &store, &cfg);
        assert_eq!(st3.total as u64, e8 - s8, "all accounted");
        assert_eq!(
            (st3.periodic + st3.cannot_tile + st3.undecided) as u64,
            e8 - s8,
            "tallies intact after filtering (ranges carry the dropped lines)"
        );
        assert_eq!(
            read_ranges(&store).len(),
            ranges.len(),
            "no duplicate coverage appended"
        );

        // Verify: coverage-based completeness + kept certs re-verify.
        let report = run_verify::<ZZ12, _>(&d, &FREE_ZZ12, &store, Some(8));
        assert!(
            report.is_clean(),
            "filtered store verifies clean via coverage: {report:?}"
        );
        std::fs::remove_dir_all(&dir).ok();
    }

    /// End-to-end pipeline smoke test on the small perimeters: classify a fresh store,
    /// then verify it is complete + all certs re-verify. Runs WINDOWED (chunk 50)
    /// so the windowing path is exercised too. Needs the ratdb asset, so opt-in.
    /// Env ASSET overrides the default path.
    #[test]
    #[ignore = "pipeline smoke: windowed fast pass + verify on small perimeters (needs the ratdb asset)"]
    #[allow(clippy::needless_range_loop)] // 3..=8 is a perimeter range, not an index loop
    fn fast_then_verify_small_perimeters() {
        use crate::cyclotomic::ZZ12;
        let asset =
            std::env::var("ASSET").unwrap_or_else(|_| "web/ratdb/data/zz12_n16_free".into());
        let d = open_ratdb(&asset);
        let out = std::env::temp_dir().join("classify_smoke.jsonl");
        let _ = std::fs::remove_file(&out);
        let cfg = FastConfig {
            chunk: 50,
            ..FastConfig::default()
        };
        for perim in 3..=8 {
            let s = run_fast::<ZZ12, _>(&d, &FREE_ZZ12, perim, &out, &cfg);
            assert_eq!(s.verify_fail, 0);
            assert_eq!(s.total as u64, FREE_ZZ12[perim]);
        }
        let report = run_verify::<ZZ12, _>(&d, &FREE_ZZ12, &out, None);
        assert!(report.is_clean(), "smoke store must be clean: {report:?}");
    }

    /// run_pack sorts a shuffled store by index, and store_lookup's byte
    /// bisection then finds every present index (verdicts intact) and misses
    /// every absent one -- across line lengths spanning the bisection's scan
    /// window boundaries.
    #[test]
    fn pack_then_lookup_roundtrip() {
        use std::io::Write;
        let dir = std::env::temp_dir().join(format!("pack_lookup_{}", std::process::id()));
        std::fs::create_dir_all(&dir).unwrap();
        let store = dir.join("s.jsonl");
        // Shuffled indices with gaps; corona padding varies the line lengths.
        let idxs: Vec<u64> = (0..500u64).map(|i| (i * 7919) % 100_000).collect();
        {
            let mut f = std::io::BufWriter::new(std::fs::File::create(&store).unwrap());
            for (n, &idx) in idxs.iter().enumerate() {
                let c = Classified::Undecided {
                    depth: (n % 7),
                    corona: vec![
                        crate::geom::matches::PatchMatch::new(
                            crate::geom::matches::EdgeRange::new(n % 13, 1),
                            crate::geom::matches::Segment::new(
                                0,
                                crate::geom::matches::EdgeRange::new(n % 5, 1)
                            ),
                        );
                        n % 40
                    ],
                };
                writeln!(f, "{idx}\t{}", serde_json::to_string(&c).unwrap()).unwrap();
            }
        }
        let n = run_pack(&store);
        assert_eq!(n, idxs.len());
        for (i, &idx) in idxs.iter().enumerate() {
            let c = store_lookup(&store, idx)
                .unwrap()
                .unwrap_or_else(|| panic!("idx {idx} found"));
            match c {
                Classified::Undecided { depth, corona } => {
                    assert_eq!(depth, i % 7, "idx {idx} verdict intact");
                    assert_eq!(corona.len(), i % 40);
                }
                other => panic!("unexpected verdict {other:?}"),
            }
        }
        // Absent indices (7919 is coprime to 100_000, so exactly the 500
        // written residues are present).
        let present: std::collections::HashSet<u64> = idxs.iter().copied().collect();
        for probe in [1u64, 3, 99_999, 50_001] {
            if !present.contains(&probe) {
                assert!(
                    store_lookup(&store, probe).unwrap().is_none(),
                    "idx {probe} absent"
                );
            }
        }
        std::fs::remove_dir_all(&dir).ok();
    }

    /// run_merge applies a deep overlay with the cert-overrides-unknown
    /// policy: a Decided overlay verdict replaces the store's Undecided line;
    /// an Undecided overlay entry and a Decided store line are both left
    /// alone; an overlay entry with no store line is appended. Store line
    /// order (and thus packing) is preserved for pure replacements.
    #[test]
    fn merge_overrides_undecided_only() {
        use std::io::Write;
        let dir = std::env::temp_dir().join(format!("merge_test_{}", std::process::id()));
        std::fs::create_dir_all(&dir).unwrap();
        let (store, overlay) = (dir.join("s.jsonl"), dir.join("s.deep.jsonl"));

        let undec = |d: usize| {
            serde_json::to_string(&Classified::Undecided {
                depth: d,
                corona: vec![],
            })
            .unwrap()
        };
        let decided = serde_json::to_string(&Classified::Decided(Verdict::CannotTile(
            crate::classify::cert::HeeschCert {
                heesch: 0,
                status: crate::classify::cert::HeeschStatus::Finite,
                build: vec![],
                bound: 1,
                budget: 1,
            },
        )))
        .unwrap();

        // Store: idx 10 Undecided, 20 Decided, 30 Undecided, 40 Undecided.
        let mut f = std::fs::File::create(&store).unwrap();
        writeln!(f, "10\t{}", undec(1)).unwrap();
        writeln!(f, "20\t{decided}").unwrap();
        writeln!(f, "30\t{}", undec(2)).unwrap();
        writeln!(f, "40\t{}", undec(3)).unwrap();
        // Overlay: 10 -> Decided (override), 20 -> Decided (kept: store already
        // decided), 30 -> Undecided (no override), 99 -> Decided (append).
        let mut f = std::fs::File::create(&overlay).unwrap();
        writeln!(f, "10\t{decided}").unwrap();
        writeln!(f, "20\t{decided}").unwrap();
        writeln!(f, "30\t{}", undec(6)).unwrap();
        writeln!(f, "99\t{decided}").unwrap();

        let stats = run_merge(&store, &overlay);
        assert_eq!(
            stats,
            MergeStats {
                overridden: 1,
                kept_decided: 1,
                still_undecided: 1,
                appended: 1
            }
        );
        let txt = std::fs::read_to_string(&store).unwrap();
        let lines: Vec<&str> = txt.lines().collect();
        assert_eq!(lines.len(), 5);
        assert!(
            lines[0].starts_with("10\t") && lines[0].contains("CannotTile"),
            "10 overridden"
        );
        assert!(lines[1].starts_with("20\t"), "20 kept in place");
        assert!(
            lines[2].starts_with("30\t") && lines[2].contains(r#""depth":2"#),
            "30 untouched"
        );
        assert!(
            lines[3].starts_with("40\t"),
            "40 untouched (no overlay entry)"
        );
        assert!(lines[4].starts_with("99\t"), "99 appended");
        std::fs::remove_dir_all(&dir).ok();
    }
}