rete-core 0.5.0

//! **External (chunked, memory-bounded) build**: assemble an arbitrarily large
//! single `.rete` file inside a caller-chosen memory budget by spilling sorted
//! intermediate artifacts to disk and merging them, instead of holding the whole
//! dictionary + id-triples + permutations in RAM (SPEC-compatible output;
//! byte-identical to [`crate::ingest::assemble_dataset_streaming_algo`] with the
//! pyramid disabled, text index included or not).
//!
//! ## How the budget bounds memory
//!
//! The input is consumed **once** and cut into K *chunks*, sealed whenever the
//! buffered quads' estimated resident bytes reach half the budget, or the
//! quads and the chunk's dictionary together reach 85% of it — so the budget
//! directly decides K (the "number of iterations"). Each chunk is assembled
//! with the ordinary in-RAM machinery (its own small dictionary, observed as
//! the quads arrive), then reduced to disk artifacts and dropped:
//!
//! ```text
//! input ─┬─ chunk 0: dict₀ → {shared,subj,obj,pred}₀ term files + triples₀ (local ids,
//!        │                    one sorted run)
//!        ├─ chunk 1: …          (+ graph₀ name file + quads₀ when named graphs exist)
//!        └─ chunk K: …          (+ c<i>.ocnt object counts with --frequent-objects)
//! top-N:  (`--frequent-objects N` only) k-way walk of the chunks' node terms
//!           → the N most frequent object-only terms (see "Frequent objects")
//! merge:  k-way term merge over all chunks
//!           → global front-coded dict sections (streamed to disk)
//!           → per-chunk id remap tables (local id → global id), appended to
//!             disk as the merge reaches each local id (`extbuild_remap`)
//!           → k-way merge of the graph-name files → global graph ordinals
//! remap:  one chunk's remap tables loaded at a time: its default triples,
//!           remapped to global ids, go straight into the first permutation's
//!           sorted runs; its named quads into (g, s, p, o) runs
//! index:  per permutation: k-way merge of its runs (dedup) → streaming tiler →
//!           independently compressed tiles on disk, and → the NEXT
//!           permutation's runs (the permutations are chained, see "Spill")
//! named:  the (g, s, p, o) runs merged, one graph-leading order after another
//!           (chained like the permutations) → the quad index, one spill file
//! text:   (`--text-index` only) each chunk's literal words → c<i>.tok, remapped
//!           with the same tables → global.tok, external sort by (token,
//!           subject) → the TEXT_INDEX section (`crate::text_index_spill`)
//! labels: (`--label-index`) the remap tees (object, slot, subject) ids; once
//!           the sections are done their dictionary, read through the spill
//!           files as one address space (`extbuild_v06`), names them, and the
//!           LABEL_INDEX and SCHEMA sections join the image (`extbuild_sections`)
//! card:   (the CLI: `build --memory-budget-mb --card…`, `merge`) the full card is
//!           derived by streaming the same image, before any output exists
//!           (`ExternalBuildOptions::card_from_image`)
//! write:  format 0x06, once: the directories are read out of the image, the
//!           payload is copied to its final offset and hashed on the way, and
//!           each spill file is deleted as soon as its data is in the output
//!           (`crate::v06::write::assemble_with`); the output is written beside
//!           its name and renamed into place
//! ```
//!
//! ## Spill
//!
//! Every spill file is compressed (`extbuild_spill`, G23): the statements as
//! sorted runs of delta-varint id tuples, the term files front-coded, the rest
//! as plain byte streams, all in zstd blocks at a fast level. On the
//! OpenCitations Meta sample the chunk statements take 1.6 B per statement
//! instead of 12, a sort run about 2 B, the term files 16% of their raw size.
//!
//! The permutations are **chained**: the remap writes the first permutation's
//! sorted runs directly (there is no `global.tri`), and each permutation's
//! merge, which reads its runs segment by segment and deletes each segment once
//! read, writes the next permutation's runs as it goes. So the statements are
//! on disk once at any time — up to 0.4.1 the permutation phase held a raw
//! `global.tri` plus one permutation's raw runs, 24 B per statement before any
//! index byte. The finished sections are copied into the output and deleted one
//! by one, so the write does not hold the index twice either.
//!
//! The merge deletes each chunk's term files once it has read them, the remap
//! each chunk's statements and remap tables once rewritten. Up to 0.4.1 the
//! peak was at the end of the dictionary merge; now it is usually in the last
//! permutations, where the finished sections — the index itself — are most of
//! it. Nothing of the label pass is on disk before the file is written.
//!
//! A build killed in the permutation phase can still be finished from its
//! spill (the `resume_from_spill` harness): every finished permutation's tiles
//! hold all the statements, sorted, so the chain restarts from the last one.
//! A kill in the first permutation cannot be: its runs are consumed as they
//! are merged, where up to 0.4.1 a raw `global.tri` survived to the end.
//!
//! Peak RAM ≈ max(one chunk's working set, the merge's read and write buffers,
//! one chunk's remap tables, one sort run, one named graph's index) — all
//! sized from `memory_budget`, never from the dataset:
//!
//! - a chunk is sealed when its quads reach half the budget, or earlier when
//!   its quads plus its dictionary — charged at twice the term bytes plus
//!   96 B per entry, what building it takes — reach 85% (G24: a term-dense
//!   stretch, two new terms per statement, used to take the chunk phase a
//!   quarter past the budget); its remap tables, 4 B per local id, are at
//!   most a quarter of the half (`extbuild_remap`);
//! - the merge reads every chunk's term files at once, one 64 KiB decoded
//!   block of each, and appends to every chunk's remap files at once, with
//!   buffers that shrink as the chunk count grows: a sixteenth of the budget,
//!   floored at 4 KiB per writer;
//! - a dictionary section's restart offsets (8 B per 16 terms) and chunk
//!   directory are spilled too, and its header streamed from them;
//! - a sort run holds a quarter of the budget in records (the text index's
//!   vectors may grow to half); a merge holds one decoded block per run, and
//!   while it runs, the next permutation's run being filled;
//! - what does grow with the data are the tile directories, about 30 B per
//!   64 KiB tile — tens of MB at billions of statements.
//!
//! Up to 0.4.0 every chunk's remap tables stayed resident from the merge to
//! the end of the remap, 4 B per chunk-local id: 15.0 GB at 1.73 B chunk-local
//! terms, which capped one OpenCitations Meta build at half the graph.
//!
//! ## Named graphs
//!
//! Graph names are **not** dictionary terms (the format stores them verbatim in
//! the named-graphs section), so they get their own per-chunk name file and
//! their own k-way merge, yielding a global **graph ordinal**. That ordinal is
//! the leading column of the named-quad spill, which makes the external sort do
//! the grouping for free: after the merge, each graph's triples arrive together
//! and its index is built from that run alone. One dictionary, one extra spill,
//! no per-graph files.
//!
//! The default graph keeps its own triple runs and its own sort, untouched —
//! so an input with no named graphs runs exactly the code (and the record
//! widths) it ran before this existed.
//!
//! ## Quad index (`GraphLayout::Quad`)
//!
//! Graph names **are** dictionary terms there (subject role), so the chunker
//! observes each name into its chunk dictionary and spills a named quad as
//! `(local subject id of g, s, p, o)`: no name file, no graph-ordinal merge;
//! the remap phase maps `g` like `s`. Then, per graph-leading order, one
//! external sort of `(g, a, b, c)` records (chained like the default graph's),
//! a k-way merge with dedup,
//! and the same `quad::QuadTiler` the in-RAM builder uses — so the two write
//! identical bytes. The `G+SPO` pass also emits the graph directory. The
//! section is assembled in a spill file (its digest goes into the index
//! container's seal) and streamed after the index. Peak RAM: one sort run plus
//! one tile batch.
//!
//!
//! ## Frequent objects (`--frequent-objects N`, G3)
//!
//! The in-RAM builder moves the `N` object-only terms carried by the most
//! statements (ties by term order) into the shared section. A chunk cannot
//! decide that: a term's frequency, and whether it is a subject somewhere, are
//! only known over the whole input. So with `N > 0` each chunk counts its
//! objects' statements and spills the counts of its object-only terms beside
//! its term file (`c<i>.ocnt`); before the merge, one more k-way walk over
//! the chunks' term files sums them, drops every term some chunk has as a
//! subject, and keeps the best `N` in a bounded heap. The merge then files
//! those terms in the shared section, in term order, as the in-RAM builder
//! does — so the file is the same bytes. The walk costs one more read of the
//! node term files and holds `N` terms.
//!
//! Limits (clear errors, not silent degradation): no community pyramid.
//! SPARQL/joins/verify are unaffected (they never require the pyramid). The
//! full-text index is supported, in any scope, and its section is byte-identical
//! to the in-RAM build's.

#![cfg(not(target_arch = "wasm32"))]

use std::collections::BinaryHeap;
use std::fs::File;
use std::io::{BufReader, BufWriter, Read, Seek, Write};
use std::path::{Path, PathBuf};
use std::sync::Arc;

use crate::build_progress::{self as progress, Stat, Ticker, Unit};
use crate::build_trace::{self as trace, note_merge, Phase};
use crate::dict::env_restart_interval;

use crate::extbuild_remap::{
    remap_buf_bytes, role_only_id, shared_id, RemapStore, RemapWriter, Seg, SegCounts,
};
use crate::extbuild_sections::{ExtLabels, ExtSchema};
use crate::extbuild_spill::{
    self as spill, RunFile, RunReader, RunSpiller, TermReader, TermWriter, ZReader, ZWriter,
};
use crate::index::GroupSizer;
use crate::ingest::{BuildStats, GraphLayout, IngestError, RawQuad};
use crate::triples::TripleBlockBuilder;
use crate::varint::write_uvarint;
use crate::DictionaryBuilder;

/// An id-triple in a single permutation's ordering.
type IdTriple = (u32, u32, u32);
/// A tile synopsis: `(min_b, max_b, min_c, max_c)` over the tile's triples.
type Synopsis = (u32, u32, u32, u32);
/// Builds the metadata payload (the Dataset Card) once the counts are known.
/// Return an empty `Vec` for none — byte-identical to a metadata-free build.
pub type MetadataFn = Box<dyn FnOnce(&BuildStats) -> Vec<u8>>;
/// A distinct term with the `(chunk, local id)` pairs that carry it.
type TermCarriers = (String, Vec<(usize, u32)>);
/// One tile's directory entry: `(min_a, max_a, compressed length, synopsis)`.
type TileDirEntry = (u32, u32, u64, Synopsis);
/// One encoded tile: `(min_a, max_a, compressed bytes, synopsis)`.
type EncodedTile = (u32, u32, Vec<u8>, Synopsis);

/// Options for [`build_external`].
pub struct ExternalBuildOptions {
    /// Approximate peak-RAM target in bytes. Controls how many chunks the input
    /// is cut into and how large each external-sort run is. Not a hard cgroup
    /// limit — a working-set target the phases size themselves against.
    pub memory_budget: u64,
    /// Where intermediate spill files live (a fresh
    /// `.rete-extbuild-<pid>-<seq>-<random>` directory is created inside, never
    /// an existing one). Defaults to the output file's parent directory — same
    /// filesystem, no surprise `/tmp` exhaustion.
    pub tmp_dir: Option<PathBuf>,
    /// Metadata payload (the Dataset Card), derived after counts are known.
    pub metadata: MetadataFn,
    /// Build-conditions payload ([`crate::header::SectionKind::BuildInfo`]),
    /// written verbatim after the metadata section and **excluded from the
    /// content hash** — per-build facts (timestamp, builder, parameters) that
    /// must not make two builds of identical data hash differently. Empty =
    /// no section (byte-identical to a pre-build-info file).
    pub build_info: Vec<u8>,
    /// Which permutations to sort and write ([`crate::index::PermSet::ALL`] by
    /// default). Each one is a full external sort of every triple plus its own
    /// section on disk, so this is the single biggest lever on both build time
    /// and file size that does not touch the data.
    pub perms: crate::index::PermSet,
    /// Write the LABEL_INDEX section (kind 8). See `extbuild_sections.rs`.
    pub label_index: bool,
    /// Write the SCHEMA section (kind 9). See `extbuild_sections.rs`.
    pub schema: bool,
    /// The schema's type predicate; `rdf:type` when `None`.
    pub type_override: Option<String>,
    /// Build the full-text index over the literals of this scope (`None` = no
    /// TEXT_INDEX section). A second external sort, on the literal words, inside
    /// the same budget; see `text_index_spill`.
    pub text_index: Option<crate::text_index::TextIndexScope>,
    /// How the named graphs are stored: only [`GraphLayout::Quad`] (the default), the one
    /// layout format 0x06 has — the quad index (SPEC.md §6.5), through its own external
    /// sorts, see "Quad index" in the module docs. Anything else is refused.
    pub graphs: GraphLayout,
    /// Bytes of `memory_budget` the input stream itself holds while it is
    /// read — `rete merge`'s open shards, their dictionary, tile and block
    /// caches capped to it (G22). The chunk phase, the only one that runs
    /// while the stream is being read, sizes itself to the budget less this;
    /// every later phase has the whole budget. `0` (the default) for a
    /// parser, whose buffers are small.
    pub input_memory: u64,
    /// Derive the card (and the build record) from the data itself, before the
    /// file is written: called with the finished sections as a reader — the
    /// `0x05`-shaped view the spill makes up, the same bytes the `0x06` file
    /// gets — and the final counts; returns `(card, build record)`. Overrides
    /// `metadata` and `build_info`. The CLI's `--card` runs the card derivation
    /// and the starter queries here, so the card lands in its region of the
    /// one file the build writes.
    pub card_from_image: Option<CardFromImage>,
}

/// See [`ExternalBuildOptions::card_from_image`].
pub type CardFromImage = Box<
    dyn FnOnce(
        Arc<dyn crate::RangeReader + Send + Sync>,
        &BuildStats,
    ) -> Result<(Vec<u8>, Vec<u8>), ExtBuildError>,
>;

impl Default for ExternalBuildOptions {
    fn default() -> Self {
        Self {
            memory_budget: 4 << 30,
            tmp_dir: None,
            metadata: Box::new(|_| Vec::new()),
            build_info: Vec::new(),
            perms: crate::index::PermSet::ALL,
            label_index: false,
            schema: false,
            type_override: None,
            text_index: None,
            graphs: GraphLayout::Quad,
            input_memory: 0,
            card_from_image: None,
        }
    }
}

#[cfg(test)]
thread_local! {
    /// `(chunk budget, text-index run bytes)` for tests that drive a small
    /// graph through [`build_external`] as many chunks and many sort runs —
    /// the public budget floor would otherwise make it one of each.
    pub(crate) static TEST_SPLIT: std::cell::Cell<Option<(u64, usize)>> = const { std::cell::Cell::new(None) };
    /// `(triple run, quad run)` lengths, in records, for tests that need many
    /// sort runs through [`build_external`].
    pub(crate) static TEST_RUNS: std::cell::Cell<Option<(usize, usize)>> = const { std::cell::Cell::new(None) };
    /// The chunk working-set cap ([`CHUNK_WORKING_FRACTION`] of the budget),
    /// for tests that make a small graph's dictionary seal its chunks.
    pub(crate) static TEST_WORKING: std::cell::Cell<Option<u64>> = const { std::cell::Cell::new(None) };
}

#[derive(Debug, thiserror::Error)]
pub enum ExtBuildError {
    #[error("ingest: {0}")]
    Ingest(#[from] IngestError),
    #[error("io: {0}")]
    Io(#[from] std::io::Error),
    #[error("internal: {0}")]
    Internal(&'static str),
    /// The request asks for something this build does not do (a named-graph
    /// layout other than the quad index).
    #[error("{0}")]
    Unsupported(String),
    /// The dataset does not fit the format: a position (subject or object) has
    /// more than `u32::MAX` distinct terms.
    #[error("{0}")]
    Limit(String),
}

/// Slack on the HyperLogLog estimate (relative error about 0.8 %, so 3 % is
/// nearly four standard errors) before [`check_position_estimate`] refuses.
const POSITION_ESTIMATE_SLACK: u64 = u32::MAX as u64 / 32;

/// Refuse a build early when the distinct subject or object terms seen so far
/// clearly exceed what the format addresses (`u32::MAX` ids per position),
/// instead of after hours of merging. The counts are HyperLogLog estimates,
/// updated as each chunk seals; only a count past the limit by the estimator's
/// margin refuses, and the exact check (on final ids) still runs after the merge.
pub(crate) fn check_position_estimate(subjects: u64, objects: u64) -> Result<(), ExtBuildError> {
    for (name, n) in [("subject", subjects), ("object", objects)] {
        if n > u32::MAX as u64 + POSITION_ESTIMATE_SLACK {
            return Err(ExtBuildError::Limit(format!(
                "about {n} distinct {name} terms so far (HyperLogLog estimate), and a .rete file                  addresses at most {} per position: this build cannot succeed, so it stops                  now rather than after the merge",
                u32::MAX
            )));
        }
    }
    Ok(())
}

/// Fraction of the budget a chunk's buffered raw quads may occupy before the
/// chunk is sealed. The rest covers the chunk's dictionary and its id-encode
/// working set, which [`CHUNK_WORKING_FRACTION`] bounds.
const CHUNK_BUDGET_FRACTION: f64 = 0.5;
/// Fraction of the budget a chunk's whole working set may reach before the
/// chunk is sealed (G24): its buffered quads plus its dictionary at the size
/// [`DictionaryBuilder::build`] briefly takes it to. On most graphs the quads
/// reach [`CHUNK_BUDGET_FRACTION`] first and nothing changes; on a term-dense
/// stretch — two new terms per statement, the OpenCitations Meta
/// `oc:doiLink` tail — the dictionary is the larger half, and up to 0.4.1 a
/// chunk sealed by bytes alone took the build 25% past its budget (14.99 GB
/// under a 16 GB cap at budget 12000; 1,348 MiB at 1,000 on a synthetic).
/// The rest of the budget is the program and the input's buffers.
const CHUNK_WORKING_FRACTION: f64 = 0.85;
/// Bytes charged per distinct dictionary entry (a term once for every role
/// set it is in), on top of twice its length, for a chunk's dictionary: the
/// hash-set slot it is observed into (24 B at most 7/8 full, in a
/// power-of-two table), the allocator's rounding, and what `build` adds while
/// it sorts and front-codes the terms — a `Vec` slot, the section builder's
/// copy of the string and its slot. Measured: 165 B per entry of 37-byte
/// terms, which `2 × 37 + 96` covers.
const DICT_ENTRY_OVERHEAD: u64 = 96;
/// Overhead charged per buffered quad on top of its term byte lengths (String
/// headers, Vec slot, allocator slack) when estimating chunk residency.
const PER_QUAD_OVERHEAD: u64 = 96;
/// Per-tile-batch size for parallel tile compression.
const TILE_COMPRESS_BATCH: usize = 512;

/// Build a `.rete` at `output` from a **single-pass** quad stream within
/// `opts.memory_budget`. Returns the same [`BuildStats`] as the in-RAM paths.
///
/// `stream` is invoked once; it must call the visitor for every quad.
pub fn build_external<S>(
    mut stream: S,
    output: &Path,
    opts: ExternalBuildOptions,
) -> Result<BuildStats, ExtBuildError>
where
    S: FnMut(&mut dyn FnMut(RawQuad) -> Result<(), ExtBuildError>) -> Result<(), ExtBuildError>,
{
    let tmp_parent = opts
        .tmp_dir
        .clone()
        .or_else(|| output.parent().map(|p| p.to_path_buf()))
        .unwrap_or_else(|| PathBuf::from("."));
    let tmp = TmpDir::create(&tmp_parent)?;
    progress::set_spill_dir(Some(&tmp.dir));
    let budget = opts.memory_budget.max(64 << 20); // floor: 64 MiB

    // ---- Phase 1: chunk the input ------------------------------------------
    // The input's own share (a merge's shard readers) is out of the chunks'.
    let chunk_phase = budget.saturating_sub(opts.input_memory).max(32 << 20);
    let chunk_budget = (chunk_phase as f64 * CHUNK_BUDGET_FRACTION) as u64;
    // A text-index sort run holds at most a quarter of the budget in records;
    // its vectors may grow to twice that, which is the half the phase may use.
    let text_run_bytes = (budget / 4) as usize;
    #[cfg(test)]
    let (chunk_budget, text_run_bytes) = TEST_SPLIT
        .with(|c| c.get())
        .unwrap_or((chunk_budget, text_run_bytes));
    let working_cap = (chunk_phase as f64 * CHUNK_WORKING_FRACTION) as u64;
    #[cfg(test)]
    let working_cap = TEST_WORKING.with(|c| c.get()).unwrap_or(working_cap);
    if opts.input_memory > 0 {
        eprintln!(
            "extbuild: budget {} MiB, {} MiB of it for reading the input -> chunk target {} MiB, \
             {} MiB with the chunk's dictionary",
            budget >> 20,
            opts.input_memory >> 20,
            chunk_budget >> 20,
            working_cap >> 20
        );
    } else {
        eprintln!(
            "extbuild: budget {} MiB -> chunk target {} MiB, {} MiB with the chunk's dictionary",
            budget >> 20,
            chunk_budget >> 20,
            working_cap >> 20
        );
    }
    // Format 0x06 keeps named graphs in the quad index only; the per-graph
    // layout (kind 5) is not a thing it can carry.
    if opts.graphs != GraphLayout::Quad {
        return Err(ExtBuildError::Unsupported(
            "the memory-bounded build writes format 0x06, which stores named graphs in the \
             quad index only; the per-graph layout is not available"
                .into(),
        ));
    }
    // Fail now, not after hours: the output must be writable and not a
    // directory. The finished file is written beside it under a name of its own.
    let partial = PartialFile::create(output)?;
    let quad_layout = true;
    crate::quad::check_perms(opts.perms)
        .map_err(|_| ExtBuildError::Internal("the quad index needs the 3- or 6-permutation set"))?;
    let mut text = opts
        .text_index
        .clone()
        .map(crate::text_index_spill::TextSpill::new);
    // `--frequent-objects N` (G3): the chunks count their objects, and the
    // N most frequent object-only terms of the whole input are picked before
    // the merge, which files them in the shared section as the in-RAM
    // builder does.
    let promote = crate::write_opts::frequent_objects();
    let mut chunker = Chunker::new(&tmp, chunk_budget, opts.graphs)
        .with_working_cap(working_cap)
        .with_object_counts(promote > 0);
    chunker.text = text.as_mut();
    // LABEL_INDEX / SCHEMA (kinds 8, 9): the passes live in extbuild_sections;
    // each phase below calls one hook. Both work on global ids from the remap
    // on, so nothing of them is spilled beside the chunk files. A sort run
    // gets an eighth of the budget while the remap runs, a quarter once the
    // file is written.
    let mut ext_labels = opts
        .label_index
        .then(|| ExtLabels::new(&tmp.dir, (budget / 8) as usize));
    let mut ext_schema = opts.schema.then(|| {
        ExtSchema::new(
            &tmp.dir,
            (budget / 8) as usize,
            opts.type_override.as_deref(),
        )
    });
    // The caller's stream knows what bounds the parse (input bytes, or a
    // merge's statements) and sets the phase's total.
    progress::begin(Phase::ParseChunk, None, Unit::None);
    stream(&mut |q: RawQuad| chunker.push(q))?;
    let chunks = chunker.finish()?;
    let statements: u64 = chunks
        .iter()
        .map(|c| c.triple_count + c.quad_count)
        .sum::<u64>();
    progress::set_stat(Stat::Statements, statements);
    let has_named = chunks.iter().any(|c| c.quad_count > 0);
    eprintln!(
        "extbuild: {} chunk(s), {} statement(s) spilled",
        chunks.len(),
        statements
    );
    trace::add("chunks", chunks.len() as u64);
    note_scratch(&tmp);
    trace::lap(Phase::ParseChunk);

    // ---- Phase 2: merge chunk dictionaries into the global dictionary -------
    let promoted = if promote > 0 {
        let carriers = chunks.iter().map(|c| {
            let [shared, subj, obj, _] = c.section_terms;
            shared as u64 + subj as u64 + obj as u64
        });
        progress::begin(Phase::FrequentObjects, Some(carriers.sum()), Unit::Terms);
        let p = select_frequent_objects(&tmp, &chunks, promote)?;
        trace::lap(Phase::FrequentObjects);
        p
    } else {
        Vec::new()
    };
    // Every chunk-local term the merge meets: a shared one in the subject and
    // the object walks both.
    let carriers = chunks.iter().map(|c| {
        let [shared, subj, obj, pred] = c.section_terms;
        2 * shared as u64 + subj as u64 + obj as u64 + pred as u64 + c.graph_terms as u64
    });
    progress::begin(Phase::Dictionary, Some(carriers.sum()), Unit::Terms);
    let schema_preds = crate::extbuild_sections::schema_predicates(opts.type_override.as_deref());
    let (merged, remaps) =
        merge_dictionaries_with(&tmp, &chunks, &schema_preds, budget, &promoted)?;
    drop(promoted);
    if let Some(x) = ext_schema.as_mut() {
        x.set_merge_facts(&merged.schema_facts);
    }
    if let Some(l) = ext_labels.as_mut() {
        l.set_predicates(merged.schema_facts.label_preds.clone());
    }
    note_scratch(&tmp);
    trace::lap(Phase::Dictionary);
    progress::set_stat(Stat::Terms, merged.term_count);
    eprintln!(
        "extbuild: merged dictionary — {} term(s)",
        merged.term_count
    );

    // ---- Phase 3: remap chunk statements into the first sort runs -----------
    // A sort run holds R triples (12 B) or quads (16 B); with the vector's
    // growth slack that is up to twice over, so R·12·2 = budget/2 — and the
    // same for quads. Reusing `run_len` for quads would have quietly spent
    // 4/3 of the intended half-budget on the quad sort.
    let run_len = ((budget / 2) / 24).clamp(1 << 16, u32::MAX as u64) as usize;
    let quad_run_len = ((budget / 2) / 32).clamp(1 << 16, u32::MAX as u64) as usize;
    #[cfg(test)]
    let (run_len, quad_run_len) = TEST_RUNS
        .with(|c| c.get())
        .unwrap_or((run_len, quad_run_len));
    let first = opts
        .perms
        .iter()
        .next()
        .ok_or(ExtBuildError::Internal("no permutation to index"))?;
    if let Some(text) = text.as_mut() {
        text.begin_remap(tmp.path("global.tok"))?;
    }
    progress::begin(Phase::Encode, Some(statements), Unit::Statements);
    // The remap fills the triple and the quad runs at once: half each.
    let (remap_run_len, remap_quad_run_len) = if has_named {
        (run_len / 2, quad_run_len / 2)
    } else {
        (run_len, quad_run_len)
    };
    let (tri_runs, quad_runs) = remap_chunks(
        &tmp,
        &chunks,
        &remaps,
        quad_layout,
        first,
        remap_run_len,
        remap_quad_run_len,
        text.as_mut(),
        ext_schema.as_mut(),
        ext_labels.as_mut(),
    )?;
    note_scratch(&tmp);
    trace::lap(Phase::Encode);

    // ---- Phase 3b: the full-text index (`--text-index`) -----------------------
    let codec = crate::file::writer_codec();
    let text_section = match text {
        Some(t) => {
            progress::begin(Phase::TextIndex, None, Unit::None);
            let s = t.finish(&|name| tmp.path(name), text_run_bytes, codec)?;
            note_scratch(&tmp);
            trace::lap(Phase::TextIndex);
            s
        }
        None => None,
    };

    // ---- Phase 4: the permutations, chained ---------------------------------
    if let Some(x) = ext_schema.as_mut() {
        x.begin_spo()?;
    }
    // Each permutation merges the spilled default triples once (fewer after
    // the first, which drops duplicates): `done` steps to k·N as the k-th ends.
    let spilled: u64 = chunks.iter().map(|c| c.triple_count).sum();
    let perm_count = opts.perms.len() as u64;
    progress::begin(
        Phase::IndexPermutations,
        Some(perm_count * spilled),
        Unit::Records,
    );
    let mut perms_done = 0u64;
    let (perm_sections, default_count) = index_permutations(
        &tmp,
        tri_runs,
        opts.perms,
        run_len,
        codec,
        "",
        // The schema pass reads the deduplicated SPO and POS merge streams.
        &mut |perm, t| match (ext_schema.as_mut(), perm) {
            (Some(x), crate::index::IndexPermutation::Spo) => x.observe_spo(t),
            (Some(x), crate::index::IndexPermutation::Pos) => {
                x.observe_pos(t);
                Ok(())
            }
            _ => Ok(()),
        },
        &mut |perm, n| {
            eprintln!(
                "extbuild: permutation {} indexed ({} unique triple(s))",
                perm.name(),
                n
            );
            trace::lap(Phase::IndexPermutations);
            perms_done += 1;
            progress::set_done(perms_done * spilled);
        },
    )?;
    if let Some(x) = ext_schema.as_mut() {
        x.finish_links()?;
    }

    // ---- Phase 4b: the quad index — graph-leading orders over the quad runs --
    if has_named {
        let quads: u64 = chunks.iter().map(|c| c.quad_count).sum();
        progress::begin(Phase::NamedGraphs, Some(perm_count * quads), Unit::Records);
    }
    let quad = if has_named {
        let q = build_quad_section(&tmp, quad_runs, quad_run_len, opts.perms, codec)?;
        eprintln!(
            "extbuild: quad index — {} named graph(s), {} unique quad(s), {} byte(s)",
            q.graph_count, q.quad_count, q.len
        );
        note_scratch(&tmp);
        trace::lap(Phase::NamedGraphs);
        Some(q)
    } else {
        None
    };
    let named_triples = quad.as_ref().map(|q| q.quad_count).unwrap_or(0);
    let named_graphs = quad.as_ref().map(|q| q.graph_count).unwrap_or(0);
    let quad_count = default_count + named_triples;

    // ---- Phases 5 and 6, which a failure leaves the whole spill for ---------
    // Up to the first spill file deleted, the spill is complete: a failure in the
    // label or schema pass, the card, or the start of the write keeps the
    // directory (and says where) instead of losing hours of work with it.
    let image_slot: std::cell::RefCell<Option<Arc<crate::extbuild_v06::SpillImage>>> =
        std::cell::RefCell::new(None);
    let tail = (|| -> Result<BuildStats, ExtBuildError> {
        // ---- Phase 5: LABEL_INDEX / SCHEMA, from the spill's dictionary ---------
        // The two passes resolve the strings they name through the dictionary of
        // the image the spill makes up (see `spill_image`), before any output
        // exists; the sections are then part of the same image.
        let base = Arc::new(spill_image(
            &merged,
            &perm_sections,
            text_section.as_ref(),
            quad.as_ref(),
            quad_count,
            codec,
            opts.perms,
            None,
            &[],
        )?);
        let (label_section, schema_section) = if ext_labels.is_some() || ext_schema.is_some() {
            progress::begin(Phase::LabelsSchema, None, Unit::None);
            let shared = || crate::extbuild_v06::SharedImage(base.clone());
            let label_section = match ext_labels {
                Some(l) => {
                    let s = l
                        .finish(shared(), (budget / 4) as usize, budget / 8, codec)
                        .map_err(file_err)?;
                    note_scratch(&tmp);
                    s
                }
                None => None,
            };
            let schema_section = match ext_schema {
                Some(x) => x.finish(shared()).map_err(file_err)?,
                None => Vec::new(),
            };
            if let Some(l) = &label_section {
                eprintln!("extbuild: label index — {} entries", l.entries);
            }
            if !schema_section.is_empty() {
                eprintln!("extbuild: schema section — {} bytes", schema_section.len());
            }
            trace::lap(Phase::LabelsSchema);
            (label_section, schema_section)
        } else {
            (None, Vec::new())
        };
        drop(base);
        let image = Arc::new(spill_image(
            &merged,
            &perm_sections,
            text_section.as_ref(),
            quad.as_ref(),
            quad_count,
            codec,
            opts.perms,
            label_section.as_ref(),
            &schema_section,
        )?);
        *image_slot.borrow_mut() = Some(image.clone());

        // ---- Phase 6: the card, then the 0x06 file ------------------------------
        // `statements` is what was INGESTED; `quad_count` is what the permutation
        // merge actually kept after dedup, and it is what the header records. The
        // metadata callback (the Dataset Card) has to see the latter, or a build
        // from input containing duplicates publishes a card that disagrees with its
        // own file. They are equal whenever the input has no duplicates, which is
        // why the gap went unnoticed.
        let mut stats = BuildStats {
            statements: quad_count as usize,
            default_triples: default_count as usize,
            named_graphs: named_graphs as usize,
            terms: merged.term_count as usize,
            pyramid_levels: 0,
            label_index: false,
        };
        // Every spill file is still there, and the output not yet started: the
        // scratch high-water mark of the build.
        note_scratch(&tmp);
        let (metadata, build_info) = match opts.card_from_image {
            // A caller that derives its card from the data (the CLI's full card,
            // with starter queries) gets the image to read it through, and laps
            // the phases itself.
            Some(derive) => derive(
                Arc::new(crate::extbuild_v06::SharedImage(image.clone())),
                &stats,
            )?,
            None => {
                let metadata = (opts.metadata)(&stats);
                if !metadata.is_empty() {
                    trace::lap(Phase::CardDerive);
                }
                (metadata, opts.build_info)
            }
        };
        note_scratch(&tmp);
        // The total is the image's size until the write knows its data bytes.
        progress::begin(
            Phase::Write,
            Some(crate::reader::RangeReader::len(&*image)),
            Unit::Bytes,
        );
        // Each spill file is deleted as soon as every range of it is in the
        // output, so the output grows as the spill shrinks.
        write_v06(output, &image, &metadata, &build_info, true, &partial)?;
        stats.pyramid_levels = 0;
        note_scratch(&tmp);
        trace::lap(Phase::Write);
        progress::set_spill_dir(None);
        Ok(stats)
    })();
    match tail {
        Ok(stats) => Ok(stats),
        Err(e) => {
            let released = image_slot
                .borrow()
                .as_ref()
                .is_some_and(|i| i.any_released());
            if released {
                eprintln!(
                    "extbuild: the write failed after it began deleting spill files; the spill \
                     is not whole and is removed"
                );
            } else {
                tmp.keep.set(true);
                eprintln!(
                    "extbuild: the build failed after its sections were done ({e}); the spill \
                     is kept, whole, at {}. It holds the dictionary and index sections: \
                     rete-core's `resume_from_spill` harness (extbuild tests, RETE_RESUME_*) \
                     writes the file from it for a default-graph, label-less build; otherwise \
                     rerun the build and delete that directory.",
                    tmp.dir.display()
                );
            }
            Err(e)
        }
    }
}

/// Raise the `scratch_peak_bytes` counter to what the spill directory holds
/// now. Sampled at phase boundaries (and once each external sort has written
/// its runs, its high-water mark), so it is a lower bound on the true peak.
/// Skipped — no directory walk — when no build trace is recording.
fn note_scratch(tmp: &TmpDir) {
    if !trace::active() {
        return;
    }
    fn walk(dir: &Path) -> u64 {
        let Ok(rd) = std::fs::read_dir(dir) else {
            return 0;
        };
        rd.flatten()
            .map(|e| match e.metadata() {
                Ok(m) if m.is_dir() => walk(&e.path()),
                Ok(m) => m.len(),
                Err(_) => 0,
            })
            .sum()
    }
    trace::max("scratch_peak_bytes", walk(&tmp.dir));
}

// ---------------------------------------------------------------------------
// tmp-dir guard
// ---------------------------------------------------------------------------

struct TmpDir {
    dir: PathBuf,
    /// Leave the directory behind on drop (a failed build's whole spill).
    keep: std::cell::Cell<bool>,
}

impl TmpDir {
    /// A fresh spill directory in `parent`, owned by this build alone.
    ///
    /// The name was `.rete-extbuild-<pid>-<seq>`, made with `create_dir_all`.
    /// In a container the pid is usually 1, so every container's first build
    /// got the same name: two builds sharing a `--tmp-dir` wrote into one
    /// directory, and the first to finish deleted the other's spill. A
    /// directory left behind by a killed build was adopted the same way. Now
    /// the name also carries 64 random bits, and `create_dir` refuses a name
    /// that exists, so a directory is never shared or reused; on a clash the
    /// next name is tried. Nothing here enters or deletes a directory this
    /// process did not create, so a build still running (any version) keeps
    /// its spill.
    fn create(parent: &Path) -> Result<Self, std::io::Error> {
        static SEQ: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
        Self::create_named(parent, || {
            let seq = SEQ.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
            let pid = std::process::id();
            format!(".rete-extbuild-{pid}-{seq}-{:016x}", spill_nonce())
        })
    }

    /// [`create`](Self::create) with the names to try supplied by `name`.
    fn create_named(
        parent: &Path,
        mut name: impl FnMut() -> String,
    ) -> Result<Self, std::io::Error> {
        std::fs::create_dir_all(parent)?;
        for _ in 0..TMP_DIR_ATTEMPTS {
            let dir = parent.join(name());
            match std::fs::create_dir(&dir) {
                Ok(()) => {
                    return Ok(TmpDir {
                        dir,
                        keep: std::cell::Cell::new(false),
                    })
                }
                Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => continue,
                Err(e) => return Err(e),
            }
        }
        Err(std::io::Error::new(
            std::io::ErrorKind::AlreadyExists,
            format!(
                "no fresh spill directory in {}: {TMP_DIR_ATTEMPTS} names were taken",
                parent.display()
            ),
        ))
    }
    fn path(&self, name: &str) -> PathBuf {
        self.dir.join(name)
    }
}

impl Drop for TmpDir {
    fn drop(&mut self) {
        if !self.keep.get() {
            let _ = std::fs::remove_dir_all(&self.dir);
        }
    }
}

/// Names [`TmpDir::create`] tries before it gives up. With 64 random bits in
/// each, a second attempt already means something else is wrong.
const TMP_DIR_ATTEMPTS: u32 = 16;

/// 64 random bits for a spill directory name: std's per-process hash keys
/// (seeded from the OS, different on every call) over the clock. Exclusivity
/// does not rest on these bits — `create_dir` refuses a name that exists —
/// they only make a clash, and so a retry, rare.
fn spill_nonce() -> u64 {
    use std::hash::{BuildHasher, Hasher};
    let mut h = std::collections::hash_map::RandomState::new().build_hasher();
    let now = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_nanos())
        .unwrap_or(0);
    h.write_u128(now);
    h.finish()
}

// ---------------------------------------------------------------------------
// Phase 1: chunking
// ---------------------------------------------------------------------------

struct ChunkInfo {
    /// default-graph statements in this chunk
    triple_count: u64,
    /// named-graph statements in this chunk
    quad_count: u64,
    /// term counts per section file, in file order (shared, subj, obj, pred)
    section_terms: [u32; 4],
    /// distinct graph names in this chunk (`c<i>.graph` term count)
    graph_terms: u32,
    /// the default-graph statements in chunk-local ids, one sorted run
    tri: RunFile,
    /// the named quads, `(graph, s, p, o)` in chunk-local ids, one sorted run
    /// (empty when the chunk has none)
    qtri: RunFile,
}

impl ChunkInfo {
    /// Local ids per remap segment ([`Seg`]): the section files' term counts,
    /// then the graph names.
    fn seg_counts(&self) -> SegCounts {
        let [shared, subj, obj, pred] = self.section_terms;
        [shared, subj, obj, pred, self.graph_terms]
    }
}

struct Chunker<'a> {
    tmp: &'a TmpDir,
    chunk_budget: u64,
    acc_bytes: u64,
    /// The chunk is also sealed once its quads plus its dictionary's charge
    /// ([`Chunker::dict_charge`]) reach this (G24).
    working_cap: u64,
    /// The chunk's dictionary, observed quad by quad as it is pushed, so its
    /// size is known before it is built.
    db: DictionaryBuilder,
    /// Chunks sealed by [`Chunker::working_cap`] before their quads reached
    /// `chunk_budget`.
    term_sealed: u64,
    /// `--frequent-objects` (G3): count every object's statements and spill
    /// the counts of the chunk's object-only terms (`c<i>.ocnt`), for
    /// [`select_frequent_objects`].
    count_objects: bool,
    quads: Vec<(String, String, String)>,
    /// named-graph statements: `(chunk-local graph id, s, p, o)`. The graph is
    /// interned rather than cloned per quad — fedlex averages ~113 quads per
    /// graph, so storing the name once instead of 113 times is the difference
    /// between the buffer being bounded by distinct graphs and by statements.
    named: Vec<(u32, String, String, String)>,
    /// graph name → chunk-local *insertion* id (ranked at seal)
    graph_ids: std::collections::BTreeMap<String, u32>,
    chunks: Vec<ChunkInfo>,
    has_quoted: bool,
    /// `--text-index`: each sealed chunk spills its literal words here.
    text: Option<&'a mut crate::text_index_spill::TextSpill>,
    /// Distinct terms of the subject and the object space over the sealed
    /// chunks, for the early size refusal ([`check_position_estimate`]).
    subj_hll: crate::hll::Hll,
    obj_hll: crate::hll::Hll,
    /// Quad index: graph names are subject terms of the chunk dictionary and
    /// a named quad's graph id is the name's local subject id.
    quad: bool,
}

impl<'a> Chunker<'a> {
    fn new(tmp: &'a TmpDir, chunk_budget: u64, graphs: GraphLayout) -> Self {
        Chunker {
            tmp,
            chunk_budget,
            acc_bytes: 0,
            working_cap: u64::MAX,
            db: Self::new_builder(false),
            term_sealed: 0,
            count_objects: false,
            quads: Vec::new(),
            named: Vec::new(),
            graph_ids: std::collections::BTreeMap::new(),
            chunks: Vec::new(),
            has_quoted: false,
            text: None,
            subj_hll: crate::hll::Hll::new(),
            obj_hll: crate::hll::Hll::new(),
            quad: graphs == GraphLayout::Quad,
        }
    }

    /// Seal a chunk once its quads and its dictionary together reach `cap`
    /// bytes, whichever of the two limits comes first.
    fn with_working_cap(mut self, cap: u64) -> Self {
        self.working_cap = cap;
        self
    }

    /// Count the objects' statements for `--frequent-objects` (G3).
    fn with_object_counts(mut self, on: bool) -> Self {
        self.count_objects = on;
        self.db = Self::new_builder(on);
        self
    }

    /// A chunk's dictionary is built from role-classified terms only:
    /// frequent-object promotion is decided over the whole input, once every
    /// chunk's counts are in ([`select_frequent_objects`]); a chunk only
    /// counts.
    fn new_builder(count_objects: bool) -> DictionaryBuilder {
        let b = DictionaryBuilder::default();
        if count_objects {
            b.with_object_counts()
        } else {
            b
        }
    }

    /// What the chunk's dictionary costs at its largest, while `build` turns
    /// the observed sets into sections: about twice the term bytes plus
    /// [`DICT_ENTRY_OVERHEAD`] per entry (16 B more for a counted entry's
    /// count, in the map and in the sorted pairs).
    fn dict_charge(&self) -> u64 {
        let (entries, bytes) = self.db.footprint();
        let per_entry = DICT_ENTRY_OVERHEAD + if self.count_objects { 16 } else { 0 };
        2 * bytes + per_entry * entries
    }

    fn push(&mut self, q: RawQuad) -> Result<(), ExtBuildError> {
        let (s, p, o, g) = q;
        self.acc_bytes += (s.len() + p.len() + o.len()) as u64 + PER_QUAD_OVERHEAD;
        // Graph NAMES are not dictionary terms — the format stores them
        // verbatim in the named-graphs section — but the s/p/o of a named
        // quad are, and the in-RAM builder observes them from the same single
        // pass.
        self.db.observe(&s, &p, &o);
        match g {
            None => self.quads.push((s, p, o)),
            Some(graph) => {
                let next = self.graph_ids.len() as u32 + 1;
                let gid = match self.graph_ids.get(&graph) {
                    Some(&id) => id,
                    None => {
                        // first sight in this chunk: the name itself is charged
                        // once, the per-quad cost is the 4-byte id above
                        self.acc_bytes += graph.len() as u64 + PER_QUAD_OVERHEAD;
                        // ...except in the quad index, where it is a subject
                        // term
                        if self.quad {
                            self.db.observe_graph(&graph);
                        }
                        self.graph_ids.insert(graph, next);
                        next
                    }
                };
                self.named.push((gid, s, p, o));
            }
        }
        if self.acc_bytes >= self.chunk_budget {
            self.seal()?;
        } else if self.acc_bytes + self.dict_charge() >= self.working_cap {
            self.term_sealed += 1;
            self.seal()?;
        }
        Ok(())
    }

    /// Build this chunk's private dictionary, spill its four sorted section term
    /// files + local-id triples (+ the graph name file and named quads when the
    /// chunk saw any), then drop everything.
    fn seal(&mut self) -> Result<(), ExtBuildError> {
        if self.quads.is_empty() && self.named.is_empty() {
            return Ok(());
        }
        let ci = self.chunks.len();
        let quads = std::mem::take(&mut self.quads);
        let named = std::mem::take(&mut self.named);
        let graph_ids = std::mem::take(&mut self.graph_ids);
        self.acc_bytes = 0;

        // The terms were observed as the quads were pushed (see `push`).
        let db = std::mem::replace(&mut self.db, Self::new_builder(self.count_objects));
        let (dict, object_counts) = db.build_counted();
        // `--frequent-objects`: the statements carrying each object-only term,
        // in the order of `c<i>.obj`.
        if self.count_objects {
            if object_counts.len() != dict.object_only_count() as usize {
                return Err(ExtBuildError::Internal(
                    "object counts do not match the object-only section",
                ));
            }
            let mut w = ZWriter::create(&self.tmp.path(&format!("c{ci}.ocnt")))?;
            let mut rec = Vec::with_capacity(10);
            for n in object_counts {
                rec.clear();
                write_uvarint(&mut rec, n);
                w.write_all(&rec)?;
            }
            w.finish()?;
        }
        if dict.has_quoted_triples() {
            self.has_quoted = true;
        }
        // `--text-index`: the chunk's literal words, keyed by chunk-local subject
        // ids, while the raw strings and their dictionary are both in hand.
        if let Some(text) = self.text.as_mut() {
            text.spill_chunk(&self.tmp.path(&format!("c{ci}.tok")), &dict, &quads, &named)?;
        }

        // Local id-triples (subject-space id, predicate id, object-space id),
        // spilled as one sorted run: sorted ids delta-encode to a fraction of
        // their 12 bytes, and nothing downstream depends on the input order.
        // Consuming `quads` frees each statement's strings as its ids are
        // taken, so the id vector never stacks on top of them.
        let triple_count = quads.len() as u64;
        let mut ids: Vec<IdTriple> = Vec::with_capacity(quads.len());
        for (s, p, o) in quads {
            ids.push(
                dict.encode(&s, &p, &o)
                    .ok_or(ExtBuildError::Internal("chunk term missing from own dict"))?,
            );
        }
        let tri = spill::spill_sorted_run(&self.tmp.path(&format!("c{ci}.tri")), &mut ids)?;
        drop(ids);

        // Named quads: the chunk-local graph id is the name's RANK in this
        // chunk's sorted name file, so `c<i>.graph` and the ids in `c<i>.qtri`
        // agree and the same k-way machinery that merges terms merges graphs.
        // (Graph names are NT tokens — an IRI or a blank node — so they never
        // contain a raw \n, the same property the term files rely on.)
        let quad_count = named.len() as u64;
        let graphs_seen = graph_ids.len();
        let graph_terms = if self.quad { 0 } else { graph_ids.len() as u32 };
        // The graph column of a named quad: in the quad index the name's
        // local subject id, remapped later exactly like a subject; otherwise
        // the name's RANK in this chunk's sorted name file, so `c<i>.graph`
        // and the ids in `c<i>.qtri` agree and the same k-way machinery that
        // merges terms merges graphs.
        let mut gcol = vec![0u32; graph_ids.len() + 1];
        if !named.is_empty() && self.quad {
            for (name, insertion_id) in &graph_ids {
                gcol[*insertion_id as usize] = dict
                    .subject_id(name)
                    .ok_or(ExtBuildError::Internal("graph name missing from own dict"))?;
            }
        } else if !named.is_empty() {
            let mut gw = TermWriter::create(&self.tmp.path(&format!("c{ci}.graph")))?;
            for (rank, (name, insertion_id)) in graph_ids.iter().enumerate() {
                gcol[*insertion_id as usize] = rank as u32 + 1;
                gw.push(name)?;
            }
            gw.finish()?;
        }
        let mut qids: Vec<IdQuad> = Vec::with_capacity(named.len());
        for (gid, s, p, o) in named {
            let (si, pi, oi) = dict
                .encode(&s, &p, &o)
                .ok_or(ExtBuildError::Internal("chunk term missing from own dict"))?;
            qids.push((gcol[gid as usize], si, pi, oi));
        }
        let qtri = spill::spill_sorted_run(&self.tmp.path(&format!("c{ci}.qtri")), &mut qids)?;
        drop(qids);

        // four sorted section term files, front-coded
        let shared = dict.shared_count();
        let subj_only = dict.subject_only_count();
        let obj_only = dict.object_only_count();
        let preds = {
            let mut n = 1u32;
            while dict.predicate_term(n).is_some() {
                n += 1;
            }
            n - 1
        };
        let write_terms = |name: &str,
                           count: u32,
                           term_of: &dyn Fn(u32) -> Option<String>,
                           seen: &mut dyn FnMut(&str)|
         -> Result<(), ExtBuildError> {
            let mut w = TermWriter::create(&self.tmp.path(name))?;
            for i in 1..=count {
                let t = term_of(i).ok_or(ExtBuildError::Internal("dict id out of range"))?;
                seen(&t);
                w.push(&t)?;
            }
            w.finish()?;
            Ok(())
        };
        let (subj_hll, obj_hll) = (&mut self.subj_hll, &mut self.obj_hll);
        write_terms(
            &format!("c{ci}.shared"),
            shared,
            &|i| dict.subject_term(i),
            &mut |t| {
                subj_hll.add(t);
                obj_hll.add(t);
            },
        )?;
        write_terms(
            &format!("c{ci}.subj"),
            subj_only,
            &|i| dict.subject_term(shared + i),
            &mut |t| subj_hll.add(t),
        )?;
        write_terms(
            &format!("c{ci}.obj"),
            obj_only,
            &|i| dict.object_term(shared + i),
            &mut |t| obj_hll.add(t),
        )?;
        write_terms(
            &format!("c{ci}.pred"),
            preds,
            &|i| dict.predicate_term(i),
            &mut |_| {},
        )?;
        // Refuse a dataset that cannot fit as soon as it is clear, not after the merge.
        check_position_estimate(self.subj_hll.count(), self.obj_hll.count())?;

        progress::set_stat(Stat::Chunks, ci as u64 + 1);
        eprintln!(
            "extbuild: chunk {ci} sealed — {} statement(s), {} term(s), {graphs_seen} graph(s)",
            triple_count + quad_count,
            shared + subj_only + obj_only + preds
        );
        self.chunks.push(ChunkInfo {
            triple_count,
            quad_count,
            section_terms: [shared, subj_only, obj_only, preds],
            graph_terms,
            tri,
            qtri,
        });
        Ok(())
    }

    fn finish(mut self) -> Result<Vec<ChunkInfo>, ExtBuildError> {
        self.seal()?;
        if self.term_sealed > 0 {
            eprintln!(
                "extbuild: {} of {} chunk(s) sealed by their dictionary's size before their quads' \
                 (a term-dense stretch)",
                self.term_sealed,
                self.chunks.len()
            );
            trace::add("chunks_term_sealed", self.term_sealed);
        }
        if self.chunks.is_empty() {
            // an empty input still produces one (empty) chunk so downstream
            // phases have a well-formed shape to merge
            self.chunks.push(ChunkInfo {
                triple_count: 0,
                quad_count: 0,
                section_terms: [0, 0, 0, 0],
                graph_terms: 0,
                tri: spill::write_run::<IdTriple>(&self.tmp.path("c0.tri"), &[])?,
                qtri: spill::write_run::<IdQuad>(&self.tmp.path("c0.qtri"), &[])?,
            });
            for name in ["c0.shared", "c0.subj", "c0.obj", "c0.pred"] {
                File::create(self.tmp.path(name))?;
            }
        }
        Ok(self.chunks)
    }
}

// ---------------------------------------------------------------------------
// Phase 2: global dictionary merge
// ---------------------------------------------------------------------------

struct MergedDict {
    /// The four chunked-encoded dict section payloads, spilled to tmp files
    /// (shared, subjects, objects, predicates — the container order).
    section_files: [SectionFile; 4],
    term_count: u64,
    has_quoted: bool,
    /// What the SCHEMA pass needs from the merge.
    schema_facts: crate::extbuild_sections::MergeFacts,
}

/// A finished section payload living in a tmp file.
struct SectionFile {
    path: PathBuf,
    len: u64,
}

/// One chunk's contribution to a term-space merge: an iterator over
/// `(term, local_space_id)` in ascending term order. Subject space enumerates
/// shared then subject-only ids interleaved back into term order.
struct SpaceStream {
    a: TermFileReader, // shared section
    b: TermFileReader, // role-only section
    a_base: u32,       // shared ids are 1..=shared
    b_base: u32,       // role-only ids are shared+1..
    a_next: Option<String>,
    b_next: Option<String>,
    a_rank: u32,
    b_rank: u32,
}

impl SpaceStream {
    fn new(a: TermFileReader, b: TermFileReader, shared: u32) -> Result<Self, std::io::Error> {
        let mut s = SpaceStream {
            a,
            b,
            a_base: 0,
            b_base: shared,
            a_next: None,
            b_next: None,
            a_rank: 0,
            b_rank: 0,
        };
        s.a_next = s.a.next()?;
        s.b_next = s.b.next()?;
        Ok(s)
    }

    /// Next `(term, local_space_id)` in ascending term order.
    fn next(&mut self) -> Result<Option<(String, u32)>, std::io::Error> {
        let take_a = match (&self.a_next, &self.b_next) {
            (None, None) => return Ok(None),
            (Some(_), None) => true,
            (None, Some(_)) => false,
            // sections are disjoint, so strict ordering decides
            (Some(x), Some(y)) => x < y,
        };
        if take_a {
            let t = self.a_next.take().unwrap();
            self.a_rank += 1;
            let id = self.a_base + self.a_rank;
            self.a_next = self.a.next()?;
            Ok(Some((t, id)))
        } else {
            let t = self.b_next.take().unwrap();
            self.b_rank += 1;
            let id = self.b_base + self.b_rank;
            self.b_next = self.b.next()?;
            Ok(Some((t, id)))
        }
    }
}

/// Reader over a chunk's sorted, front-coded term file ([`TermWriter`]).
struct TermFileReader(TermReader);

impl TermFileReader {
    fn open(path: &Path) -> Result<Self, std::io::Error> {
        TermReader::open(path).map(TermFileReader)
    }

    /// Deleting the file once it has been read to its end: the dictionary
    /// merge opens every chunk's term files at once and gives a chunk's space
    /// back as soon as the merge is past its last term rather than when every
    /// chunk is done. (A shared-section file is read twice, by the subject and
    /// the object stream; its space comes back when the second one closes.)
    /// Each reader holds one decoded block, [`spill::BYTE_BLOCK`].
    fn open_consuming(path: &Path) -> Result<Self, std::io::Error> {
        TermReader::open_consuming(path).map(TermFileReader)
    }

    fn next(&mut self) -> Result<Option<String>, std::io::Error> {
        self.0.next()
    }
}

/// Heap entry for the k-way term merge (min-heap by term, then chunk index for
/// determinism).
struct HeapEntry {
    term: String,
    chunk: usize,
    local_id: u32,
}
impl PartialEq for HeapEntry {
    fn eq(&self, other: &Self) -> bool {
        self.term == other.term && self.chunk == other.chunk
    }
}
impl Eq for HeapEntry {}
impl PartialOrd for HeapEntry {
    fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
        Some(self.cmp(other))
    }
}
impl Ord for HeapEntry {
    fn cmp(&self, other: &Self) -> std::cmp::Ordering {
        // BinaryHeap is a max-heap; invert for ascending term order
        other
            .term
            .cmp(&self.term)
            .then_with(|| other.chunk.cmp(&self.chunk))
    }
}

/// K-way merged view over one term space (subject or object) of every chunk:
/// yields each distinct term once, ascending, with every `(chunk, local_id)`
/// that carries it.
struct KWayTerms {
    heap: BinaryHeap<HeapEntry>,
    streams: Vec<SpaceStream>,
    /// Counts the carriers met, for `--progress`.
    ticker: Ticker,
}

impl KWayTerms {
    fn new(mut streams: Vec<SpaceStream>) -> Result<Self, std::io::Error> {
        note_merge(streams.len());
        let mut heap = BinaryHeap::new();
        for (ci, s) in streams.iter_mut().enumerate() {
            if let Some((term, id)) = s.next()? {
                heap.push(HeapEntry {
                    term,
                    chunk: ci,
                    local_id: id,
                });
            }
        }
        Ok(KWayTerms {
            heap,
            streams,
            ticker: Ticker::new(),
        })
    }

    /// Next distinct term with its (chunk, local_id) carriers.
    fn next(&mut self) -> Result<Option<TermCarriers>, std::io::Error> {
        let first = match self.heap.pop() {
            Some(e) => e,
            None => return Ok(None),
        };
        let term = first.term;
        let mut carriers = vec![(first.chunk, first.local_id)];
        if let Some((t, id)) = self.streams[first.chunk].next()? {
            self.heap.push(HeapEntry {
                term: t,
                chunk: first.chunk,
                local_id: id,
            });
        }
        while let Some(top) = self.heap.peek() {
            if top.term != term {
                break;
            }
            let e = self.heap.pop().unwrap();
            carriers.push((e.chunk, e.local_id));
            if let Some((t, id)) = self.streams[e.chunk].next()? {
                self.heap.push(HeapEntry {
                    term: t,
                    chunk: e.chunk,
                    local_id: id,
                });
            }
        }
        self.ticker.tick(carriers.len() as u64);
        Ok(Some((term, carriers)))
    }
}

#[cfg(test)]
fn merge_dictionaries(
    tmp: &TmpDir,
    chunks: &[ChunkInfo],
) -> Result<(MergedDict, RemapStore), ExtBuildError> {
    merge_dictionaries_with(tmp, chunks, &[], 64 << 20, &[])
}

/// A candidate of [`select_frequent_objects`], ordered **worst first**: fewer
/// statements, then the later term — so a max-heap of them keeps the one to
/// evict on top.
struct Frequent {
    count: u64,
    term: String,
}

impl Frequent {
    /// Whether a term carried by `count` statements comes ahead of this one
    /// in the in-RAM builder's order: more statements, ties by term order
    /// (`DictionaryBuilder::build`).
    fn is_beaten_by(&self, count: u64, term: &str) -> bool {
        count > self.count || (count == self.count && term < self.term.as_str())
    }
}

impl PartialEq for Frequent {
    fn eq(&self, other: &Self) -> bool {
        self.count == other.count && self.term == other.term
    }
}
impl Eq for Frequent {}
impl PartialOrd for Frequent {
    fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
        Some(self.cmp(other))
    }
}
impl Ord for Frequent {
    fn cmp(&self, other: &Self) -> std::cmp::Ordering {
        other
            .count
            .cmp(&self.count)
            .then_with(|| self.term.cmp(&other.term))
    }
}

/// `--frequent-objects n` in the memory-bounded build (G3): the `n` object-only
/// terms of the whole input carried by the most statements, ties by term
/// order — the terms [`DictionaryBuilder::build`] promotes into the shared
/// section, so the merge can file them there and the file is the in-RAM
/// build's. Returned in term order.
///
/// A term's count is the sum of its chunks' counts (`c<i>.ocnt`, one per
/// chunk-local object-only term): every statement was observed in exactly one
/// chunk, duplicates included, as the in-RAM builder observes them. A term
/// object-only in its chunks may still be a subject in another, so the walk
/// over the object-only terms runs beside one over every chunk's subject
/// space, and a term found there is no candidate. One more read of the
/// chunks' node term files, one decoded block per file; `n` terms resident.
fn select_frequent_objects(
    tmp: &TmpDir,
    chunks: &[ChunkInfo],
    n: usize,
) -> Result<Vec<String>, ExtBuildError> {
    let empty = tmp.path("ocnt.empty");
    File::create(&empty)?;
    let mut subj_streams = Vec::with_capacity(chunks.len());
    let mut obj_streams = Vec::with_capacity(chunks.len());
    let mut counts = Vec::with_capacity(chunks.len());
    for (ci, c) in chunks.iter().enumerate() {
        let open = |suffix: &str| TermFileReader::open(&tmp.path(&format!("c{ci}.{suffix}")));
        subj_streams.push(SpaceStream::new(
            open("shared")?,
            open("subj")?,
            c.section_terms[0],
        )?);
        obj_streams.push(SpaceStream::new(
            open("obj")?,
            TermFileReader::open(&empty)?,
            c.section_terms[2],
        )?);
        let path = tmp.path(&format!("c{ci}.ocnt"));
        counts.push(if c.section_terms[2] > 0 {
            Some(ZReader::open_consuming(&path)?)
        } else {
            let _ = std::fs::remove_file(&path);
            None
        });
    }
    // The next local id each chunk's object-only stream must yield: the
    // counts are matched to the terms by position.
    let mut next_local = vec![1u32; chunks.len()];
    let mut subjects = KWayTerms::new(subj_streams)?;
    let mut objects = KWayTerms::new(obj_streams)?;
    let mut subject = subjects.next()?;
    let mut top: BinaryHeap<Frequent> = BinaryHeap::with_capacity(n + 1);
    let mut candidates = 0u64;
    while let Some((term, carriers)) = objects.next()? {
        // Every carrier's count, in order, whether or not the term qualifies:
        // the count files are read in step with the term files.
        let mut count = 0u64;
        for (ci, local) in carriers {
            let r = counts[ci]
                .as_mut()
                .ok_or(ExtBuildError::Internal("object counts missing for a chunk"))?;
            let c = spill::read_stream_varint(r)?.ok_or(ExtBuildError::Internal(
                "object counts end before the object-only terms",
            ))?;
            if local != next_local[ci] {
                return Err(ExtBuildError::Internal(
                    "object-only terms out of step with their counts",
                ));
            }
            next_local[ci] += 1;
            count += c;
        }
        while subject
            .as_ref()
            .is_some_and(|(s, _)| s.as_str() < term.as_str())
        {
            subject = subjects.next()?;
        }
        if subject.as_ref().is_some_and(|(s, _)| *s == term) {
            continue; // a subject somewhere: shared, not object-only
        }
        candidates += 1;
        if top.len() < n {
            top.push(Frequent { count, term });
        } else if top
            .peek()
            .is_some_and(|worst| worst.is_beaten_by(count, &term))
        {
            top.pop();
            top.push(Frequent { count, term });
        }
    }
    for r in counts.iter_mut().flatten() {
        if spill::read_stream_varint(r)?.is_some() {
            return Err(ExtBuildError::Internal(
                "object counts outlast the object-only terms",
            ));
        }
    }
    let _ = std::fs::remove_file(&empty);
    let mut promoted: Vec<Frequent> = top.into_vec();
    promoted.sort_unstable_by(|a, b| a.term.cmp(&b.term));
    let least = promoted.iter().map(|f| f.count).min().unwrap_or(0);
    eprintln!(
        "extbuild: frequent objects — {} of {candidates} object-only term(s) promoted to the \
         shared section (each in at least {least} statement(s))",
        promoted.len()
    );
    Ok(promoted.into_iter().map(|f| f.term).collect())
}

/// [`merge_dictionaries`], also recording what the SCHEMA pass needs: the
/// global ids of `schema_preds`, the shared-term count and the literal
/// prefixes of the shared and object-only sections. `budget` sizes the
/// buffers of the merge's per-chunk readers and remap writers.
///
/// Returns the merged dictionary and every chunk's id remap tables, which are
/// written to disk while the merge reaches each chunk-local id (see
/// [`crate::extbuild_remap`]): within one remap segment of one chunk the
/// merge meets the local ids in increasing order, so each table is a file of
/// sequential appends.
///
/// `promoted` ([`select_frequent_objects`], in term order) are object-only
/// terms filed in the shared section, where their order puts them, as the
/// in-RAM builder files the frequent objects it promotes.
fn merge_dictionaries_with(
    tmp: &TmpDir,
    chunks: &[ChunkInfo],
    schema_preds: &[String],
    budget: u64,
    promoted: &[String],
) -> Result<(MergedDict, RemapStore), ExtBuildError> {
    let k = chunks.len();
    // The next promoted term the object-only merge will meet.
    let mut next_promoted = promoted.iter().peekable();
    let mut schema_facts = crate::extbuild_sections::MergeFacts {
        pred_ids: vec![None; schema_preds.len()],
        ..Default::default()
    };

    // The node merge reads four term files per chunk at once (shared twice,
    // subject-only, object-only), one decoded block of each in memory.
    let counts: Vec<SegCounts> = chunks.iter().map(ChunkInfo::seg_counts).collect();
    let mut remaps = RemapWriter::new(&tmp.dir, &counts, remap_buf_bytes(budget, k));
    // A chunk-local id of the subject or object space at or below this is in
    // the chunk's shared section, the same id in both spaces.
    let shared_of = |ci: usize| chunks[ci].section_terms[0];

    // subject-space + object-space k-way streams
    let mut subj_streams = Vec::with_capacity(k);
    let mut obj_streams = Vec::with_capacity(k);
    for (ci, c) in chunks.iter().enumerate() {
        let open =
            |suffix: &str| TermFileReader::open_consuming(&tmp.path(&format!("c{ci}.{suffix}")));
        // Both readers of the shared file open before either stream reads:
        // a stream deletes its file at the end, and an empty one ends at once.
        let (shared_s, shared_o) = (open("shared")?, open("shared")?);
        subj_streams.push(SpaceStream::new(
            shared_s,
            open("subj")?,
            c.section_terms[0],
        )?);
        obj_streams.push(SpaceStream::new(
            shared_o,
            open("obj")?,
            c.section_terms[0],
        )?);
    }
    let mut subjects = KWayTerms::new(subj_streams)?;
    let mut objects = KWayTerms::new(obj_streams)?;

    // The three node sections are written as raw front-coded bodies first; the
    // final chunked encoding needs term counts, so bodies spill to tmp.
    let mut shared_sec = RawSectionWriter::create(tmp.path("g.shared.raw"))?;
    let mut subj_sec = RawSectionWriter::create(tmp.path("g.subj.raw"))?;
    let mut obj_sec = RawSectionWriter::create(tmp.path("g.obj.raw"))?;
    let mut has_quoted = false;

    let mut ranks = [0u32; 3]; // shared, subj-only, obj-only
    let mut s_item = subjects.next()?;
    let mut o_item = objects.next()?;
    loop {
        enum Class {
            Shared,
            SubjOnly,
            ObjOnly,
        }
        let class = match (&s_item, &o_item) {
            (None, None) => break,
            (Some(_), None) => Class::SubjOnly,
            (None, Some(_)) => Class::ObjOnly,
            (Some((st, _)), Some((ot, _))) => match st.cmp(ot) {
                std::cmp::Ordering::Less => Class::SubjOnly,
                std::cmp::Ordering::Greater => Class::ObjOnly,
                std::cmp::Ordering::Equal => Class::Shared,
            },
        };
        match class {
            Class::Shared => {
                let (term, s_carriers) = s_item.take().unwrap();
                let (_, o_carriers) = o_item.take().unwrap();
                // A shared id is final at once: shared ids lead both spaces.
                let id = shared_id(ranks[0])?;
                for (ci, lid) in s_carriers {
                    match lid.checked_sub(shared_of(ci)) {
                        None | Some(0) => remaps.push(ci, Seg::Shared, lid, id)?,
                        Some(rel) => remaps.push(ci, Seg::SubjOnly, rel, id)?,
                    }
                }
                for (ci, lid) in o_carriers {
                    match lid.checked_sub(shared_of(ci)) {
                        // shared in its chunk too: written from the subject side
                        None | Some(0) => {}
                        Some(rel) => remaps.push(ci, Seg::ObjOnly, rel, id)?,
                    }
                }
                has_quoted |= term.starts_with("<<");
                if term.starts_with('"') {
                    schema_facts.shared_literals += 1;
                }
                shared_sec.push(&term)?;
                ranks[0] += 1;
                s_item = subjects.next()?;
                o_item = objects.next()?;
            }
            Class::SubjOnly => {
                let (term, carriers) = s_item.take().unwrap();
                let id = role_only_id(ranks[1])?;
                for (ci, lid) in carriers {
                    match lid.checked_sub(shared_of(ci)) {
                        Some(rel) if rel > 0 => remaps.push(ci, Seg::SubjOnly, rel, id)?,
                        _ => {
                            return Err(ExtBuildError::Internal(
                                "a chunk-shared term merged as subject-only",
                            ))
                        }
                    }
                }
                has_quoted |= term.starts_with("<<");
                subj_sec.push(&term)?;
                ranks[1] += 1;
                s_item = subjects.next()?;
            }
            Class::ObjOnly => {
                let (term, carriers) = o_item.take().unwrap();
                // A promoted frequent object (G3) takes the next shared id, in
                // term order among the shared terms, and never a subject's.
                let promote = next_promoted.next_if(|p| **p == term).is_some();
                let id = if promote {
                    shared_id(ranks[0])?
                } else {
                    role_only_id(ranks[2])?
                };
                for (ci, lid) in carriers {
                    match lid.checked_sub(shared_of(ci)) {
                        Some(rel) if rel > 0 => remaps.push(ci, Seg::ObjOnly, rel, id)?,
                        _ => {
                            return Err(ExtBuildError::Internal(
                                "a chunk-shared term merged as object-only",
                            ))
                        }
                    }
                }
                has_quoted |= term.starts_with("<<");
                if promote {
                    if term.starts_with('"') {
                        schema_facts.shared_literals += 1;
                    }
                    shared_sec.push(&term)?;
                    ranks[0] += 1;
                } else {
                    if term.starts_with('"') {
                        schema_facts.objonly_literals += 1;
                    }
                    obj_sec.push(&term)?;
                    ranks[2] += 1;
                }
                o_item = objects.next()?;
            }
        }
    }
    if next_promoted.next().is_some() {
        return Err(ExtBuildError::Internal(
            "a promoted frequent object was not met as an object-only term",
        ));
    }
    // The node merge is over: release its readers and write buffers.
    drop((subjects, objects));
    for seg in [Seg::Shared, Seg::SubjOnly, Seg::ObjOnly] {
        remaps.flush(seg)?;
    }
    let (n_shared, n_subj, n_obj) = (ranks[0], ranks[1], ranks[2]);
    if n_shared.checked_add(n_subj.max(n_obj)).is_none() {
        return Err(ExtBuildError::Internal(
            "more than 2^32 subject or object terms: the dictionary is too large for one file",
        ));
    }
    schema_facts.n_shared = n_shared;

    // predicates: plain k-way merge over the pred term files
    let mut pred_sec = RawSectionWriter::create(tmp.path("g.pred.raw"))?;
    {
        let mut streams = Vec::with_capacity(k);
        for (ci, c) in chunks.iter().enumerate() {
            // real pred reader + empty role-only reader => a plain single stream
            let empty = tmp.path(&format!("c{ci}.pred.empty"));
            File::create(&empty)?;
            streams.push(SpaceStream::new(
                TermFileReader::open_consuming(&tmp.path(&format!("c{ci}.pred")))?,
                TermFileReader::open_consuming(&empty)?,
                c.section_terms[3],
            )?);
        }
        let mut kway = KWayTerms::new(streams)?;
        let mut rank = 0u32;
        while let Some((term, carriers)) = kway.next()? {
            for (ci, lid) in carriers {
                remaps.push(ci, Seg::Pred, lid, rank as u64 + 1)?; // pred ids are final now
            }
            if let Some(i) = schema_preds.iter().position(|w| *w == term) {
                schema_facts.pred_ids[i] = Some(rank + 1);
            }
            if let Some(slot) = crate::label_index::slot_of_predicate(&term) {
                schema_facts.label_preds.push((rank + 1, slot));
            }
            pred_sec.push(&term)?;
            rank += 1;
        }
    }
    remaps.flush(Seg::Pred)?;

    // Subject- and object-only ids stay provisional until `n_shared` is
    // known: they are fixed up as each chunk's tables are loaded.
    let remaps = remaps.finish(n_shared)?;

    // chunk-encode each raw section (identical to encode_chunked_dict_section)
    let codec = crate::file::writer_codec();
    let section_files = [
        shared_sec.finish_chunked(tmp, "g.shared.sec", codec)?,
        subj_sec.finish_chunked(tmp, "g.subj.sec", codec)?,
        obj_sec.finish_chunked(tmp, "g.obj.sec", codec)?,
        pred_sec.finish_chunked(tmp, "g.pred.sec", codec)?,
    ];
    // chunk term files are no longer needed
    for ci in 0..k {
        for suffix in [
            "shared",
            "subj",
            "obj",
            "pred",
            "pred.empty",
            "graph",
            "graph.empty",
        ] {
            let _ = std::fs::remove_file(tmp.path(&format!("c{ci}.{suffix}")));
        }
    }

    let term_count =
        n_shared as u64 + n_subj as u64 + n_obj as u64 + section_files[3].term_count as u64;

    Ok((
        MergedDict {
            section_files: section_files.map(|s| s.file),
            term_count,
            has_quoted,
            schema_facts,
        },
        remaps,
    ))
}

/// A raw front-coded dict-section body being streamed to a tmp file, exactly as
/// [`crate::dict::DictSectionBuilder::build`] would lay it out — plus the
/// restart-offset bookkeeping needed to emit the header and (later) the chunked
/// encoding without ever holding the body in RAM.
///
/// The restart offsets go to disk too (8 B per `restart_interval` terms: a
/// section of a billion terms has 62.5 M of them), and so does the chunk
/// directory [`finish_chunked`](Self::finish_chunked) builds from them, so a
/// section of any size costs this writer one 64 KiB chunk of RAM.
struct RawSectionWriter {
    /// the raw body, compressed on the way to disk (read back in order)
    w: ZWriter,
    path: PathBuf,
    prev: String,
    n: u64,
    body_len: u64,
    /// every restart offset, `u64` little-endian, in order
    restarts: BufWriter<File>,
    restarts_path: PathBuf,
    n_restarts: u64,
    /// the restart offsets' varint bytes, summed: the header's length,
    /// known without reading them back
    restart_varint_bytes: u64,
}

struct FinishedSection {
    file: SectionFile,
    term_count: u32,
}

/// Reads back the `u64` restart offsets a [`RawSectionWriter`] spilled.
struct OffsetReader {
    rd: BufReader<File>,
}

impl OffsetReader {
    fn open(path: &Path) -> Result<Self, std::io::Error> {
        Ok(OffsetReader {
            rd: BufReader::with_capacity(1 << 16, File::open(path)?),
        })
    }

    fn next(&mut self) -> Result<Option<u64>, std::io::Error> {
        let mut b = [0u8; 8];
        match self.rd.read_exact(&mut b) {
            Ok(()) => Ok(Some(u64::from_le_bytes(b))),
            Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => Ok(None),
            Err(e) => Err(e),
        }
    }
}

impl RawSectionWriter {
    fn create(path: PathBuf) -> Result<Self, std::io::Error> {
        let restarts_path = path.with_extension("restarts");
        Ok(RawSectionWriter {
            w: ZWriter::create(&path)?,
            path,
            prev: String::new(),
            n: 0,
            body_len: 0,
            restarts: BufWriter::with_capacity(1 << 16, File::create(&restarts_path)?),
            restarts_path,
            n_restarts: 0,
            restart_varint_bytes: 0,
        })
    }

    fn push(&mut self, term: &str) -> Result<(), std::io::Error> {
        let r = env_restart_interval() as u64;
        let mut entry = Vec::with_capacity(term.len() + 8);
        if self.n.is_multiple_of(r) {
            self.restarts.write_all(&self.body_len.to_le_bytes())?;
            self.n_restarts += 1;
            self.restart_varint_bytes += uvarint_len(self.body_len);
            write_uvarint(&mut entry, 0);
            write_uvarint(&mut entry, term.len() as u64);
            entry.extend_from_slice(term.as_bytes());
        } else {
            let shared = common_prefix_len(&self.prev, term);
            let suffix = &term.as_bytes()[shared..];
            write_uvarint(&mut entry, shared as u64);
            write_uvarint(&mut entry, suffix.len() as u64);
            entry.extend_from_slice(suffix);
        }
        self.w.write_all(&entry)?;
        self.body_len += entry.len() as u64;
        self.prev.clear();
        self.prev.push_str(term);
        self.n += 1;
        Ok(())
    }

    /// Emit the chunked section payload (byte-identical to running
    /// `encode_chunked_dict_section` over the equivalent raw section) into a new
    /// tmp file, deleting the raw body. Streams: the restart offsets are read
    /// back once to cut the chunks and once more to write the header, and the
    /// chunk directory is spilled as it is built.
    fn finish_chunked(
        mut self,
        tmp: &TmpDir,
        out_name: &str,
        codec: u8,
    ) -> Result<FinishedSection, ExtBuildError> {
        self.w.finish()?;
        self.restarts.flush()?;
        drop(self.restarts);
        let term_count = self.n as u32;
        let interval = env_restart_interval() as u64;

        // the raw header exactly as DictSectionBuilder writes it:
        // [n][restart interval][restart count][restart offsets…]
        let header_len = uvarint_len(self.n)
            + uvarint_len(interval)
            + uvarint_len(self.n_restarts)
            + self.restart_varint_bytes;

        // Chunk bounds over the body by run offsets (whole runs, 64 KiB
        // budget), exactly as `encode_chunked_dict_section` cuts them: a chunk
        // starts at a run and takes the following runs while they start less
        // than the budget past its own start.
        let budget = crate::write_opts::dict_chunk_budget() as u64;
        let mut offs = OffsetReader::open(&self.restarts_path)?;
        // The chunks tile the body from offset 0, so it is read straight
        // through: `at` is where the reader stands.
        let mut body = ZReader::open(&self.path)?;
        let mut at = 0u64;
        // compressed chunks and the encoded directory, each spilled
        let comp_path = tmp.path(&format!("{out_name}.chunks"));
        let mut comp_out = BufWriter::with_capacity(1 << 20, File::create(&comp_path)?);
        let dir_path = tmp.path(&format!("{out_name}.dir"));
        let mut dir_out = BufWriter::with_capacity(1 << 16, File::create(&dir_path)?);
        let mut dir_len = 0u64;
        let mut num_chunks = 0u64;
        let mut prev_run = 0u64;
        // The previous chunk's last term — the separator's lower bound. Kept in
        // lockstep with `encode_chunked_dict_section`; the two writers must emit
        // byte-identical directories for the same input.
        let mut prev_last: Option<Vec<u8>> = None;
        let mut entry = Vec::new();
        // `next` is the offset of run `run`, the first run not yet in a chunk
        let mut run = 0u64;
        let mut next = offs.next()?;
        while let Some(start) = next {
            let first_run = run;
            let mut last_run_start = start;
            run += 1;
            next = offs.next()?;
            while let Some(o) = next {
                if o - start >= budget {
                    break;
                }
                last_run_start = o;
                run += 1;
                next = offs.next()?;
            }
            let end = next.unwrap_or(self.body_len);
            if start != at {
                return Err(ExtBuildError::Internal(
                    "dictionary chunks do not tile the body",
                ));
            }
            let mut raw = vec![0u8; (end - start) as usize];
            body.read_exact(&mut raw)?;
            at = end;
            // the chunk's first run entry is a restart: [0][len][full term]
            let first_term = read_restart_term(&raw)
                .ok_or(ExtBuildError::Internal("restart entry unreadable"))?;
            // Routing key = the shortest separator, not the first term; chunk 0
            // needs none. See `crate::dict::SectionChunk::key`.
            let key = if num_chunks == 0 {
                Vec::new()
            } else {
                match &prev_last {
                    None => first_term,
                    Some(pl) => crate::dict::shortest_separator(pl, &first_term),
                }
            };
            if next.is_some() {
                // This chunk's last run is the one before the next chunk's
                // first; `raw` already holds the bytes, so decoding it is free
                // of I/O.
                let last_run_rel = (last_run_start - start) as usize;
                prev_last = crate::dict::run_last_term(&raw, last_run_rel, raw.len());
            }
            let comp = crate::file::compress(codec, &raw);
            comp_out.write_all(&comp)?;
            entry.clear();
            write_uvarint(&mut entry, first_run - prev_run);
            write_uvarint(&mut entry, key.len() as u64);
            entry.extend_from_slice(&key);
            write_uvarint(&mut entry, comp.len() as u64);
            dir_out.write_all(&entry)?;
            dir_len += entry.len() as u64;
            prev_run = first_run;
            num_chunks += 1;
        }
        comp_out.flush()?;
        drop(comp_out);
        dir_out.flush()?;
        drop(dir_out);
        drop(body);
        let _ = std::fs::remove_file(&self.path);

        // assemble: [header_len][header][num_chunks][dir…][compressed chunks…]
        let out_path = tmp.path(out_name);
        let mut out = BufWriter::with_capacity(1 << 20, File::create(&out_path)?);
        let mut head = Vec::new();
        write_uvarint(&mut head, header_len);
        write_uvarint(&mut head, self.n);
        write_uvarint(&mut head, interval);
        write_uvarint(&mut head, self.n_restarts);
        out.write_all(&head)?;
        let mut written = head.len() as u64;
        let mut offs = OffsetReader::open(&self.restarts_path)?;
        while let Some(off) = offs.next()? {
            head.clear();
            write_uvarint(&mut head, off);
            out.write_all(&head)?;
            written += head.len() as u64;
        }
        drop(offs);
        let _ = std::fs::remove_file(&self.restarts_path);
        if written != uvarint_len(header_len) + header_len {
            return Err(ExtBuildError::Internal(
                "dictionary section header length mismatch",
            ));
        }
        head.clear();
        write_uvarint(&mut head, num_chunks);
        out.write_all(&head)?;
        written += head.len() as u64;
        let mut dir_in = File::open(&dir_path)?;
        written += std::io::copy(&mut dir_in, &mut out)?;
        drop(dir_in);
        let _ = std::fs::remove_file(&dir_path);
        let mut comp_in = File::open(&comp_path)?;
        let copied = std::io::copy(&mut comp_in, &mut out)?;
        drop(comp_in);
        out.flush()?;
        let _ = std::fs::remove_file(&comp_path);
        let len = written + copied;
        debug_assert_eq!(
            len,
            uvarint_len(header_len) + header_len + uvarint_len(num_chunks) + dir_len + copied
        );
        Ok(FinishedSection {
            file: SectionFile {
                path: out_path,
                len,
            },
            term_count,
        })
    }
}

fn common_prefix_len(a: &str, b: &str) -> usize {
    a.as_bytes()
        .iter()
        .zip(b.as_bytes())
        .take_while(|(x, y)| x == y)
        .count()
}

/// Read the full term of the restart entry at the start of `raw`.
fn read_restart_term(raw: &[u8]) -> Option<Vec<u8>> {
    let (shared, n1) = crate::varint::read_uvarint(raw)?;
    if shared != 0 {
        return None;
    }
    let (len, n2) = crate::varint::read_uvarint(&raw[n1..])?;
    let start = n1 + n2;
    raw.get(start..start + len as usize).map(|s| s.to_vec())
}

// ---------------------------------------------------------------------------
// Phase 3: remap every chunk to global ids
// ---------------------------------------------------------------------------

/// Rewrite every chunk's statements in global ids, loading one chunk's remap
/// tables at a time, straight into the external sort's first runs: default
/// triples in `first`'s order (sorted runs of `run_len`), named quads as
/// `(graph, s, p, o)` (runs of `quad_run_len`; the graph a subject id in the
/// quad layout, an ordinal in the per-graph one). The default triples are teed
/// into the SCHEMA and LABEL_INDEX passes and each chunk's text-index words
/// are remapped onto the same subject ids. A chunk's spills are deleted as
/// they are read, so the chunks' statements turn into the runs without the
/// two ever being on disk whole at once (there is no `global.tri`).
///
/// Returns the triple runs and the quad runs.
#[allow(clippy::too_many_arguments)]
fn remap_chunks(
    tmp: &TmpDir,
    chunks: &[ChunkInfo],
    remaps: &RemapStore,
    quad_layout: bool,
    first: crate::index::IndexPermutation,
    run_len: usize,
    quad_run_len: usize,
    mut text: Option<&mut crate::text_index_spill::TextSpill>,
    mut schema: Option<&mut ExtSchema>,
    mut labels: Option<&mut ExtLabels>,
) -> Result<(Vec<RunFile>, Vec<RunFile>), ExtBuildError> {
    let mut tri = RunSpiller::<IdTriple>::new(&tmp.dir, first.name(), run_len);
    let mut quads: Option<RunSpiller<IdQuad>> = None;
    let mut ticker = Ticker::new();
    for (ci, chunk) in chunks.iter().enumerate() {
        let maps = remaps.load(ci)?;
        // The chunk's literal words, onto the same global subject ids.
        if let Some(text) = text.as_deref_mut() {
            text.remap_chunk(&tmp.path(&format!("c{ci}.tok")), &maps.subj)?;
        }
        let mut rd = RunReader::<IdTriple>::open(chunk.tri.clone());
        while let Some((s, p, o)) = rd.next()? {
            let gs = maps.subj[(s - 1) as usize];
            let gp = maps.pred[(p - 1) as usize];
            let go = maps.obj[(o - 1) as usize];
            tri.push(first.forward((gs, gp, go)))?;
            ticker.tick(1);
            if let Some(x) = schema.as_deref_mut() {
                x.tee(gs, gp, go)?;
            }
            if let Some(x) = labels.as_deref_mut() {
                x.tee(gs, gp, go)?;
            }
        }
        drop(rd);
        if chunk.quad_count > 0 {
            let q = quads.get_or_insert_with(|| {
                RunSpiller::new(&tmp.dir, &format!("q{}", QUAD_FIRST.name()), quad_run_len)
            });
            // Quad index: the graph is a subject id; else an ordinal.
            let gmap = if quad_layout { &maps.subj } else { &maps.graph };
            let mut rd = RunReader::<IdQuad>::open(chunk.qtri.clone());
            while let Some((g, s, p, o)) = rd.next()? {
                q.push((
                    gmap[(g - 1) as usize],
                    maps.subj[(s - 1) as usize],
                    maps.pred[(p - 1) as usize],
                    maps.obj[(o - 1) as usize],
                ))?;
                ticker.tick(1);
            }
        } else {
            chunk.qtri.remove();
        }
        note_scratch(tmp);
    }
    let total = remaps.total_table_bytes();
    trace::add("remap_total_bytes", total);
    eprintln!(
        "extbuild: remapped {} chunk(s) one at a time — {} B of id tables in all, \
         at most {} B resident",
        chunks.len(),
        total,
        remaps.largest_table_bytes()
    );
    let tri = tri.finish()?;
    let quads = match quads {
        Some(q) => q.finish()?,
        None => Vec::new(),
    };
    Ok((tri, quads))
}

// ---------------------------------------------------------------------------
// Phase 4: the permutations, one external sort chained into the next
// ---------------------------------------------------------------------------

/// The order of the remap's quad runs, `(graph, s, p, o)`: the per-graph
/// layout's grouping key and the quad index's first order, G+SPO.
const QUAD_FIRST: crate::index::IndexPermutation = crate::index::IndexPermutation::Spo;

/// Records between two samples of the spill directory inside a long merge
/// (tests sample often, to see the peak of a small graph).
#[cfg(not(test))]
const SCRATCH_SAMPLE_EVERY: u64 = 1 << 24;
#[cfg(test)]
const SCRATCH_SAMPLE_EVERY: u64 = 1 << 9;

/// Index the triples of `runs` — sorted runs in the order of `perms`' first
/// permutation — under every permutation of `perms`, in container order, each
/// section byte-identical to `encode_tiled_section` over the same triples.
///
/// The permutations are **chained**: the k-way merge of one permutation's
/// runs (with dedup) feeds its streaming tiler and, mapped into the next
/// permutation's order, that permutation's runs. The merge deletes the runs it
/// reads segment by segment, so the next permutation's runs take the space as
/// it is freed: one copy of the statements is on disk at a time, never the
/// two of a separate `global.tri` plus a permutation's runs. `observe` sees
/// every deduplicated triple of each permutation, in its order (the SCHEMA
/// pass); `done` is called as each permutation finishes, with its count.
#[allow(clippy::too_many_arguments)]
fn index_permutations(
    tmp: &TmpDir,
    runs: Vec<RunFile>,
    perms: crate::index::PermSet,
    run_len: usize,
    codec: u8,
    prefix: &str,
    observe: &mut dyn FnMut(crate::index::IndexPermutation, IdTriple) -> std::io::Result<()>,
    done: &mut dyn FnMut(crate::index::IndexPermutation, u64),
) -> Result<(Vec<TiledSection>, u64), ExtBuildError> {
    let order: Vec<crate::index::IndexPermutation> = perms.iter().collect();
    index_chain(tmp, runs, &order, run_len, codec, prefix, observe, done)
}

/// [`index_permutations`] over an explicit order: `runs` are sorted in
/// `order[0]`'s order. (A resumed build starts the chain part-way.)
#[allow(clippy::too_many_arguments)]
fn index_chain(
    tmp: &TmpDir,
    runs: Vec<RunFile>,
    order: &[crate::index::IndexPermutation],
    run_len: usize,
    codec: u8,
    prefix: &str,
    observe: &mut dyn FnMut(crate::index::IndexPermutation, IdTriple) -> std::io::Result<()>,
    done: &mut dyn FnMut(crate::index::IndexPermutation, u64),
) -> Result<(Vec<TiledSection>, u64), ExtBuildError> {
    let mut runs = runs;
    let mut sections = Vec::with_capacity(order.len());
    let mut count: Option<u64> = None;
    for (k, &perm) in order.iter().enumerate() {
        let next = order.get(k + 1).copied();
        let mut spill = next.map(|np| {
            RunSpiller::<IdTriple>::new(&tmp.dir, &format!("{prefix}{}", np.name()), run_len)
        });
        let mut tiler = StreamingTiler::new(tmp, &format!("{prefix}{}", perm.name()), codec)?;
        note_scratch(tmp);
        let mut seen = 0u64;
        // `--progress` counts the default graph's chain (a spilled named
        // graph's runs inside its own phase are not this phase's records).
        let mut ticker = prefix.is_empty().then(Ticker::new);
        let n = spill::merge_runs(std::mem::take(&mut runs), |t: IdTriple| {
            observe(perm, t)?;
            tiler.push(t)?;
            if let Some(tk) = ticker.as_mut() {
                tk.tick(1);
            }
            if let (Some(sp), Some(np)) = (spill.as_mut(), next) {
                sp.push(np.forward(perm.back(t)))?;
            }
            seen += 1;
            if seen.is_multiple_of(SCRATCH_SAMPLE_EVERY) {
                note_scratch(tmp);
            }
            Ok::<(), ExtBuildError>(())
        })?;
        note_scratch(tmp);
        let section = tiler.finish()?;
        if let Some(sp) = spill {
            runs = sp.finish()?;
        }
        // every permutation dedups the same multiset — counts must agree
        if count.is_some_and(|c| c != n) {
            return Err(ExtBuildError::Internal("permutation dedup counts diverge"));
        }
        count = Some(n);
        sections.push(section);
        drop(ticker);
        done(perm, n);
    }
    for r in &runs {
        r.remove(); // an empty permutation set: nothing reads them
    }
    Ok((sections, count.unwrap_or(0)))
}

// ---------------------------------------------------------------------------
// Phase 4b: named graphs — the graph ordinal leads the sort key
// ---------------------------------------------------------------------------

/// A spilled quad in `(graph ordinal, s, p, o)` order.
type IdQuad = (u32, u32, u32, u32);

/// Bytes [`write_uvarint`] takes for `v`.
fn uvarint_len(mut v: u64) -> u64 {
    let mut n = 1;
    while v >= 0x80 {
        v >>= 7;
        n += 1;
    }
    n
}

// ---------------------------------------------------------------------------
// Phase 4b (quad index): the graph-leading orders of `GraphLayout::Quad`
// ---------------------------------------------------------------------------

/// The finished QUAD_INDEX section, spilled, with the digest the index
/// container's seal carries.
struct QuadSectionFile {
    path: PathBuf,
    len: u64,
    digest: [u8; 16],
    graph_count: u64,
    quad_count: u64,
}

/// Encode a batch of finished quad tiles into an order's directory + payload.
fn drain_quad_tiles(
    pending: &mut Vec<Vec<crate::quad::QuadRow>>,
    codec: u8,
    perm: crate::index::IndexPermutation,
    dir: &mut crate::quad::PermDirWriter,
    payload: &mut BufWriter<File>,
) -> Result<(), ExtBuildError> {
    for t in crate::quad::encode_tiles(pending, codec, perm) {
        dir.add(&t);
        payload.write_all(&t.bytes)?;
    }
    pending.clear();
    Ok(())
}

/// Build the QUAD_INDEX section from the remap's quad runs (sorted by
/// `(g, s, p, o)`, the first order): per graph-leading order, a k-way merge
/// into the shared [`crate::quad::QuadTiler`] — byte-identical to
/// `quad::encode_quad_index` — chained into the next order's runs exactly as
/// [`index_permutations`] chains the default graph's.
fn build_quad_section(
    tmp: &TmpDir,
    runs: Vec<RunFile>,
    run_len: usize,
    perms: crate::index::PermSet,
    codec: u8,
) -> Result<QuadSectionFile, ExtBuildError> {
    let order: Vec<crate::index::IndexPermutation> = perms.iter().collect();
    if order.first() != Some(&QUAD_FIRST) {
        return Err(ExtBuildError::Internal(
            "the quad index's first order must be G+SPO",
        ));
    }
    let gblocks_path = tmp.path("quad.gblocks");
    let mut gblocks = BufWriter::with_capacity(1 << 20, File::create(&gblocks_path)?);
    let mut gdir = crate::quad::GraphDirWriter::new(codec);
    let mut dirs: Vec<Vec<u8>> = Vec::with_capacity(order.len());
    let mut payloads: Vec<(PathBuf, u64)> = Vec::with_capacity(order.len());
    let mut quad_count: Option<u64> = None;
    let mut runs = runs;
    for (k, &perm) in order.iter().enumerate() {
        let spo = perm == crate::index::IndexPermutation::Spo;
        let next = order.get(k + 1).copied();
        let mut spill =
            next.map(|np| RunSpiller::<IdQuad>::new(&tmp.dir, &format!("q{}", np.name()), run_len));
        let ppath = tmp.path(&format!("q{}.tiles", perm.name()));
        let mut payload = BufWriter::with_capacity(1 << 20, File::create(&ppath)?);
        let mut dir = crate::quad::PermDirWriter::default();
        let mut tiler = crate::quad::QuadTiler::new();
        let mut pending: Vec<Vec<crate::quad::QuadRow>> = Vec::new();
        // G+SPO also yields the graph directory: (graph, deduplicated count).
        let mut cur: Option<(u32, u64)> = None;
        note_scratch(tmp);
        let mut ticker = Ticker::new();
        spill::merge_runs(std::mem::take(&mut runs), |q: IdQuad| {
            ticker.tick(1);
            tiler.push(q, &mut |t| pending.push(t));
            if pending.len() >= TILE_COMPRESS_BATCH {
                drain_quad_tiles(&mut pending, codec, perm, &mut dir, &mut payload)?;
            }
            if spo {
                cur = match cur {
                    Some((g, n)) if g == q.0 => Some((g, n + 1)),
                    Some((g, n)) => {
                        gdir.push(g, n, &mut gblocks)?;
                        Some((q.0, 1))
                    }
                    None => Some((q.0, 1)),
                };
            }
            if let (Some(sp), Some(np)) = (spill.as_mut(), next) {
                let (a, b, c) = np.forward(perm.back((q.1, q.2, q.3)));
                sp.push((q.0, a, b, c))?;
            }
            Ok::<(), ExtBuildError>(())
        })?;
        note_scratch(tmp);
        let n = tiler.finish(&mut |t| pending.push(t));
        drain_quad_tiles(&mut pending, codec, perm, &mut dir, &mut payload)?;
        if let (true, Some((g, k))) = (spo, cur) {
            gdir.push(g, k, &mut gblocks)?;
        }
        payload.flush()?;
        drop(payload);
        if let Some(sp) = spill {
            runs = sp.finish()?;
        }
        if quad_count.is_some_and(|prev| prev != n) {
            return Err(ExtBuildError::Internal("quad orders' dedup counts diverge"));
        }
        quad_count = Some(n);
        payloads.push((ppath, dir.payload_len()));
        dirs.push(dir.finish());
        eprintln!(
            "extbuild: quad order G+{} indexed ({n} unique quad(s))",
            perm.name()
        );
    }
    let (gindex, gblocks_len, graph_count) = gdir.finish(&mut gblocks)?;
    gblocks.flush()?;
    drop(gblocks);
    let quad_count = quad_count.unwrap_or(0);

    let mut part_lens = vec![gindex.len() as u64];
    part_lens.extend(dirs.iter().map(|d| d.len() as u64));
    part_lens.push(gblocks_len);
    part_lens.extend(payloads.iter().map(|(_, l)| *l));
    let preamble = crate::quad::encode_preamble(perms, graph_count, quad_count, &part_lens);

    let path = tmp.path("quad.sec");
    let mut out = BufWriter::with_capacity(1 << 20, File::create(&path)?);
    let mut hasher = blake3::Hasher::new();
    let mut len = 0u64;
    for bytes in std::iter::once(&preamble)
        .chain(std::iter::once(&gindex))
        .chain(dirs.iter())
    {
        out.write_all(bytes)?;
        hasher.update(bytes);
        len += bytes.len() as u64;
    }
    for p in std::iter::once(&gblocks_path).chain(payloads.iter().map(|(p, _)| p)) {
        len += copy_hashed(p, &mut out, Some(&mut hasher))?;
        let _ = std::fs::remove_file(p);
    }
    out.flush()?;
    let mut digest = [0u8; 16];
    digest.copy_from_slice(&hasher.finalize().as_bytes()[..16]);
    Ok(QuadSectionFile {
        path,
        len,
        digest,
        graph_count,
        quad_count,
    })
}

/// Streaming re-implementation of [`crate::index::build_tiles`]'s grouping (same
/// byte-budget accounting, same whole-a-group boundaries) that never holds more
/// than the current a-group + the current tile in RAM. Tiles are compressed in
/// parallel batches and spilled; the section payload (directory + tiles +
/// synopsis trailer) is assembled at `finish`, byte-identical to
/// `encode_tiled_section`.
struct StreamingTiler {
    tile_budget: usize,
    codec: u8,
    /// triples of the tile being accumulated
    tile: Vec<(u32, u32, u32)>,
    tile_size: usize,
    /// the a-group slice currently being measured — bounded: a group whose
    /// running encoded size alone exceeds the tile budget is cut into
    /// consecutive tiles sharing the leading id (same rule as `build_tiles`,
    /// via the shared [`GroupSizer`]), so mega-groups (a 2B-triple predicate
    /// in POS/PSO, a hot class object in OSP/OPS) can no longer grow this
    /// buffer to tens of GB
    group: Vec<(u32, u32, u32)>,
    sizer: GroupSizer,
    gtotal: usize,
    prev_a: u32,
    /// finished-but-uncompressed tiles awaiting a parallel compress batch
    pending: Vec<Vec<(u32, u32, u32)>>,
    /// spilled compressed tiles
    comp_out: BufWriter<File>,
    comp_path: PathBuf,
    /// per-tile directory: (min_a, max_a, comp_len, synopsis)
    dir: Vec<TileDirEntry>,
    count: u64,
}

impl StreamingTiler {
    fn new(tmp: &TmpDir, name: &str, codec: u8) -> Result<Self, ExtBuildError> {
        let comp_path = tmp.path(&format!("{name}.tiles"));
        Ok(StreamingTiler {
            tile_budget: crate::write_opts::tile_budget(),
            codec,
            tile: Vec::new(),
            tile_size: 0,
            group: Vec::new(),
            sizer: GroupSizer::start(0, 0),
            gtotal: 0,
            prev_a: 0,
            pending: Vec::new(),
            comp_out: BufWriter::with_capacity(1 << 20, File::create(&comp_path)?),
            comp_path,
            dir: Vec::new(),
            count: 0,
        })
    }

    /// Push the next triple (already permuted, sorted, deduped). Runs the
    /// SAME boundary rules as `build_tiles` (shared `GroupSizer` accounting):
    /// whole-group packing to the byte budget, plus a mid-group cut whenever
    /// the current group's own running size exceeds the budget — the cut is
    /// what bounds this tiler's memory under mega-group skew.
    fn push(&mut self, t: (u32, u32, u32)) -> Result<(), ExtBuildError> {
        self.count += 1;
        if let Some(&(ga, _, _)) = self.group.first() {
            if t.0 != ga {
                self.close_group()?;
            }
        }
        if self.group.is_empty() {
            self.sizer = GroupSizer::start(t.0, self.prev_a);
        }
        self.group.push(t);
        self.gtotal = self.sizer.push(t.1, t.2);
        if self.gtotal > self.tile_budget {
            // Mega-group cut — mirror `build_tiles`: completed groups plus
            // the slice measured so far become ONE tile; the group continues
            // in a fresh chain with `a` as its own delta base.
            self.tile.append(&mut self.group);
            self.flush_tile()?;
            self.tile_size = 0;
            self.prev_a = t.0;
            self.gtotal = 0;
        }
        Ok(())
    }

    /// The current a-group (or its final slice after mid-group cuts) is
    /// complete: pack it into the current tile, flushing first on overflow —
    /// identical to `build_tiles`' end-of-group rule.
    fn close_group(&mut self) -> Result<(), ExtBuildError> {
        if self.group.is_empty() {
            return Ok(());
        }
        let a = self.group[0].0;
        let gsize = self.sizer.total();
        if !self.tile.is_empty() && self.tile_size + gsize > self.tile_budget {
            self.flush_tile()?;
        }
        self.tile_size += gsize;
        self.prev_a = a;
        self.tile.append(&mut self.group);
        Ok(())
    }

    fn flush_tile(&mut self) -> Result<(), ExtBuildError> {
        if self.tile.is_empty() {
            return Ok(());
        }
        self.pending.push(std::mem::take(&mut self.tile));
        self.tile_size = 0;
        if self.pending.len() >= TILE_COMPRESS_BATCH {
            self.compress_pending()?;
        }
        Ok(())
    }

    fn compress_pending(&mut self) -> Result<(), ExtBuildError> {
        if self.pending.is_empty() {
            return Ok(());
        }
        let batch = std::mem::take(&mut self.pending);
        let codec = self.codec;
        // encode + synopsis + compress each tile; par_iter preserves order
        let encode_one = |run: &Vec<(u32, u32, u32)>| -> (u32, u32, Vec<u8>, (u32, u32, u32, u32)) {
            let mut b = TripleBlockBuilder::new();
            for &t in run {
                b.push(t);
            }
            let bytes = b.build();
            let syn = match crate::triples::TripleBlock::parse(&bytes) {
                Ok(blk) => {
                    let z = blk.zone();
                    (z.min_b, z.max_b, z.min_c, z.max_c)
                }
                Err(_) => (0, u32::MAX, 0, u32::MAX),
            };
            let comp = crate::file::compress(codec, &bytes);
            (run[0].0, run[run.len() - 1].0, comp, syn)
        };
        #[cfg(feature = "parallel")]
        let encoded: Vec<EncodedTile> = {
            use rayon::prelude::*;
            batch.par_iter().map(encode_one).collect()
        };
        #[cfg(not(feature = "parallel"))]
        let encoded: Vec<EncodedTile> = batch.iter().map(encode_one).collect();
        for (min_a, max_a, comp, syn) in encoded {
            self.comp_out.write_all(&comp)?;
            self.dir.push((min_a, max_a, comp.len() as u64, syn));
        }
        Ok(())
    }

    /// Close out: the directory and the synopsis trailer in memory, the
    /// compressed tiles in their spill file. The three are concatenated only
    /// when the section is copied into the output (or a graph's container),
    /// so the tiles are never copied into a second spill file first.
    fn finish(mut self) -> Result<TiledSection, ExtBuildError> {
        self.close_group()?;
        self.flush_tile()?;
        self.compress_pending()?;
        self.comp_out.flush()?;
        drop(self.comp_out);
        Ok(TiledSection::from_dir(&self.dir, self.comp_path))
    }
}

/// A finished permutation section, spilled as its three parts, `head ‖ tiles
/// ‖ trailer`: the tile directory, the compressed tiles (a file) and the
/// synopses — byte-identical, once concatenated, to `encode_tiled_section`.
struct TiledSection {
    head: Vec<u8>,
    tiles: PathBuf,
    tiles_len: u64,
    trailer: Vec<u8>,
}

impl TiledSection {
    /// The section of the tiles in `tiles`, described by `dir`:
    /// `[num_tiles][per tile: delta(min_a), max_a-min_a, comp_len][tiles…][synopses…]`.
    fn from_dir(dir: &[TileDirEntry], tiles: PathBuf) -> Self {
        let mut head = Vec::new();
        write_uvarint(&mut head, dir.len() as u64);
        let mut prev_min = 0u32;
        let mut tiles_len = 0u64;
        for &(min_a, max_a, comp_len, _) in dir {
            write_uvarint(&mut head, (min_a - prev_min) as u64);
            write_uvarint(&mut head, (max_a - min_a) as u64);
            write_uvarint(&mut head, comp_len);
            prev_min = min_a;
            tiles_len += comp_len;
        }
        let mut trailer = Vec::new();
        for &(_, _, _, (min_b, max_b, min_c, max_c)) in dir {
            write_uvarint(&mut trailer, min_b as u64);
            write_uvarint(&mut trailer, (max_b - min_b) as u64);
            write_uvarint(&mut trailer, min_c as u64);
            write_uvarint(&mut trailer, (max_c - min_c) as u64);
        }
        TiledSection {
            head,
            tiles,
            tiles_len,
            trailer,
        }
    }

    /// The section's length in the file.
    fn len(&self) -> u64 {
        self.head.len() as u64 + self.tiles_len + self.trailer.len() as u64
    }
}

// ---------------------------------------------------------------------------
// Phase 5/6: the labels and schema, the card, and the 0x06 file
// ---------------------------------------------------------------------------

/// The `0x05`-shaped address space over the build's spill files (see
/// [`crate::extbuild_v06`]): what the build used to write to disk as a file, as
/// `header | dict container | index container | text index | quad index |
/// label index | schema | footer`, with the payload still in its spill files.
/// `labels` and `schema` are the two sections computed from the dictionary of
/// the image without them.
#[allow(clippy::too_many_arguments)]
fn spill_image(
    dict: &MergedDict,
    perm_sections: &[TiledSection],
    text: Option<&crate::text_index_spill::TextSection>,
    quad: Option<&QuadSectionFile>,
    quad_count: u64,
    codec: u8,
    perms: crate::index::PermSet,
    labels: Option<&crate::extbuild_sections::ExtLabelSection>,
    schema: &[u8],
) -> Result<crate::extbuild_v06::SpillImage, ExtBuildError> {
    use crate::extbuild_v06::ImageBuilder;
    use crate::header::{
        Header, Section, SectionKind, FLAG_HAS_QUADS, FLAG_HAS_QUOTED_TRIPLES,
        FLAG_TEXT_INDEX_NAMED_GRAPHS, FLAG_TILE_SYNOPSIS, HEADER_LEN,
    };
    use crate::v06::kind;
    // The index container ends with the quad index's seal (the section's
    // digest), which every pre-quad-index reader refuses on the section count.
    let seal = quad.map(|q| crate::quad::encode_seal(q.digest));
    let varint = |v: u64| {
        let mut b = Vec::new();
        write_uvarint(&mut b, v);
        b
    };
    let mut b = ImageBuilder::default();
    b.bytes(&[0u8; HEADER_LEN]);

    let dict_offset = b.at();
    b.bytes(&varint(4));
    for s in &dict.section_files {
        b.bytes(&varint(s.len));
        b.file(&s.path, s.len, Some(kind::DICT_DATA));
    }
    let dict_len = b.at() - dict_offset;

    let index_offset = b.at();
    b.bytes(&varint(
        (perm_sections.len() + usize::from(seal.is_some())) as u64,
    ));
    for s in perm_sections {
        b.bytes(&varint(s.len()));
        b.bytes(&s.head);
        b.file(&s.tiles, s.tiles_len, Some(kind::INDEX_DATA));
        b.bytes(&s.trailer);
    }
    if let Some(seal) = &seal {
        b.bytes(&varint(seal.len() as u64));
        b.bytes(seal);
    }
    let index_len = b.at() - index_offset;

    // Text index: a kind-11 (predicate-restricted) section starts with its
    // own 16-byte digest, which the assembler recomputes; zeros stand in.
    let text_offset = b.at();
    if let Some(t) = text {
        if t.partial {
            b.bytes(&[0u8; crate::header::TEXT_INDEX_PARTIAL_DIGEST_LEN as usize]);
        }
        b.bytes(&t.head);
        b.file(&t.table, t.table_len, Some(kind::TEXT_DATA));
        b.file(&t.postings, t.postings_len, Some(kind::TEXT_DATA));
    }
    let text_len = b.at() - text_offset;

    let mut extra = Vec::new();
    let quad_offset = b.at();
    if let Some(q) = quad {
        b.file(&q.path, q.len, Some(kind::QUAD_DATA));
        extra.push(Section {
            kind: SectionKind::QuadIndex,
            flags: 0,
            offset: quad_offset,
            length: q.len,
        });
    }
    if let Some(l) = labels {
        let at = b.at();
        let body = std::fs::metadata(&l.body)?.len();
        b.bytes(&l.digest_and_head);
        b.file(&l.body, body, Some(kind::LABEL_DATA));
        extra.push(Section {
            kind: SectionKind::LabelIndex,
            flags: 0,
            offset: at,
            length: l.digest_and_head.len() as u64 + body,
        });
    }
    if !schema.is_empty() {
        extra.push(Section {
            kind: SectionKind::Schema,
            flags: 0,
            offset: b.at(),
            length: schema.len() as u64,
        });
        b.bytes(schema);
    }
    b.bytes(&crate::header::MAGIC);

    let header = Header {
        version: crate::header::LEGACY_FORMAT_VERSION,
        flags: FLAG_TILE_SYNOPSIS
            | if quad.is_some() {
                FLAG_HAS_QUADS | crate::header::FLAG_QUAD_INDEX
            } else {
                0
            }
            | if dict.has_quoted {
                FLAG_HAS_QUOTED_TRIPLES
            } else {
                0
            }
            | if text.is_some_and(|t| t.named) {
                FLAG_TEXT_INDEX_NAMED_GRAPHS
            } else {
                0
            },
        metadata_offset: HEADER_LEN as u64,
        metadata_len: 0,
        dictionary_offset: dict_offset,
        dictionary_len: dict_len,
        root_dir_offset: index_offset,
        root_dir_len: index_len,
        pyramid_meta_offset: 0,
        pyramid_meta_len: 0,
        dict_codec: codec,
        block_codec: codec,
        pyramid_levels: 0,
        id_widths: 0,
        perms,
        quad_count,
        term_count: dict.term_count,
        content_hash: [0; 16],
        named_graphs_offset: 0,
        named_graphs_len: 0,
        schema_meta_len: 0,
        text_index_offset: if text_len > 0 { text_offset } else { 0 },
        text_index_len: text_len,
        text_index_partial: text.is_some_and(|t| t.partial),
        build_info_offset: 0,
        build_info_len: 0,
        extra_sections: extra,
    };
    b.patch_head(header.to_bytes().to_vec());
    Ok(b.finish())
}

/// The finished file's working name: beside `output`, unique to this build
/// (pid and a random nonce, as the spill directory's), created exclusively. It
/// is made at the start of the build, so an output that cannot be written (a
/// directory, a read-only place) is found before the long run, not after it.
/// Dropped, it removes the file — a failure or a panic leaves nothing — unless
/// [`keep`](Self::keep) was called: the finished file whose rename failed.
struct PartialFile {
    path: PathBuf,
    keep: std::cell::Cell<bool>,
}

impl PartialFile {
    fn create(output: &Path) -> Result<Self, ExtBuildError> {
        if output.is_dir() {
            return Err(ExtBuildError::Io(std::io::Error::other(format!(
                "{} is a directory: give a file name for the output",
                output.display()
            ))));
        }
        let mut name = output.as_os_str().to_os_string();
        name.push(format!(
            ".{}-{:016x}.partial",
            std::process::id(),
            spill_nonce()
        ));
        let path = PathBuf::from(name);
        std::fs::OpenOptions::new()
            .write(true)
            .create_new(true)
            .open(&path)
            .map_err(|e| {
                ExtBuildError::Io(std::io::Error::new(
                    e.kind(),
                    format!(
                        "cannot create the output file beside {}: {e}",
                        output.display()
                    ),
                ))
            })?;
        Ok(PartialFile {
            path,
            keep: std::cell::Cell::new(false),
        })
    }
}

impl Drop for PartialFile {
    fn drop(&mut self) {
        if !self.keep.get() {
            let _ = std::fs::remove_file(&self.path);
        }
    }
}

/// The output file, shared between the buffered writer above it and the
/// code that makes it durable. In tests it can fail or panic after a given
/// number of bytes have reached the file (what a full disk looks like, below
/// the buffer, so a flush can fail).
struct SharedFile {
    file: std::rc::Rc<File>,
    #[cfg(test)]
    written: u64,
}

impl Write for SharedFile {
    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
        #[cfg(test)]
        {
            self.written += buf.len() as u64;
            if let Some(limit) = TEST_FAIL_AFTER.with(|c| c.get()) {
                if self.written > limit {
                    return Err(std::io::Error::other("injected: no space left on device"));
                }
            }
            if let Some(limit) = TEST_PANIC_AFTER.with(|c| c.get()) {
                if self.written > limit {
                    panic!("injected panic in the write");
                }
            }
        }
        (&*self.file).write(buf)
    }
    fn flush(&mut self) -> std::io::Result<()> {
        (&*self.file).flush()
    }
}

impl Seek for SharedFile {
    fn seek(&mut self, pos: std::io::SeekFrom) -> std::io::Result<u64> {
        (&*self.file).seek(pos)
    }
}

#[cfg(test)]
thread_local! {
    /// Fail the write with an I/O error once this many bytes have reached the file.
    static TEST_FAIL_AFTER: std::cell::Cell<Option<u64>> = const { std::cell::Cell::new(None) };
    /// Panic in the write once this many bytes have reached the file.
    static TEST_PANIC_AFTER: std::cell::Cell<Option<u64>> = const { std::cell::Cell::new(None) };
}

/// An `ExtBuildError` from a `FileError`, without a second `io:` prefix.
fn file_err(e: crate::file::FileError) -> ExtBuildError {
    match e {
        crate::file::FileError::Io(e) => ExtBuildError::Io(e),
        other => ExtBuildError::Io(std::io::Error::other(other.to_string())),
    }
}

/// Write the `0x06` file at `output` from `image`: the directories are read out
/// of the spill, the payload is copied once to its final offset, and each spill
/// file goes once every range of it is in the output **and durable**: before a
/// file is deleted the buffer is flushed and the output `sync_data`-ed, so a
/// full disk (which may only show at the flush or the sync) finds the spill
/// still whole. The file is written to `partial` (beside `output`), synced,
/// then renamed into place, so a failed build leaves no half file under the
/// final name. If only the rename fails, the finished file is kept and the
/// error says where it is.
fn write_v06(
    output: &Path,
    image: &Arc<crate::extbuild_v06::SpillImage>,
    card: &[u8],
    record: &[u8],
    consume: bool,
    partial: &PartialFile,
) -> Result<(), ExtBuildError> {
    let src = crate::extbuild_v06::SharedImage(image.clone());
    let mut parts = crate::v06::parts::from_v05(&src).map_err(file_err)?;
    parts.card = (!card.is_empty()).then(|| card.to_vec());
    parts.build = (!record.is_empty()).then(|| record.to_vec());
    {
        let file = std::rc::Rc::new(
            std::fs::OpenOptions::new()
                .write(true)
                .truncate(true)
                .open(&partial.path)?,
        );
        let mut out = BufWriter::with_capacity(
            8 << 20,
            SharedFile {
                file: file.clone(),
                #[cfg(test)]
                written: 0,
            },
        );
        // `--progress`: the data bytes copied, out of the data bytes in all.
        let mut sized = false;
        let written = crate::v06::write::assemble_with(
            &parts,
            &src,
            &mut out,
            crate::v06::write::WriteOpts::default(),
            &mut crate::v06::write::Hooks {
                copied: &mut |kind| {
                    if consume {
                        image.release(kind)
                    }
                },
                progress: &mut |done, total| {
                    if !sized {
                        sized = true;
                        progress::set_total(Some(total), Unit::Bytes);
                    }
                    progress::set_done(done);
                },
                needed: &mut |off, len| {
                    if consume {
                        image.need(off, len)
                    }
                },
                consumed: &mut |off, len| {
                    if consume {
                        image.consumed(off, len)
                    }
                },
                settle: &mut |out| {
                    if consume && image.has_ready() {
                        out.flush()?;
                        file.sync_data()?;
                        image.delete_ready();
                    }
                    Ok(())
                },
            },
        )
        .map_err(file_err)?;
        out.flush()?;
        // Durable before it takes the final name.
        drop(out.into_inner().map_err(|e| e.into_error())?);
        file.sync_all()?;
        eprintln!(
            "extbuild: wrote format 0x06, {} bytes",
            written.header.file_len
        );
    }
    if let Err(e) = std::fs::rename(&partial.path, output) {
        partial.keep.set(true);
        return Err(ExtBuildError::Io(std::io::Error::new(
            e.kind(),
            format!(
                "the finished file is complete at {} but could not be renamed to {}: {e}",
                partial.path.display(),
                output.display()
            ),
        )));
    }
    // The new name is durable too (Unix: the directory entry).
    #[cfg(unix)]
    if let Some(dir) = output.parent() {
        let dir = if dir.as_os_str().is_empty() {
            Path::new(".")
        } else {
            dir
        };
        if let Ok(d) = File::open(dir) {
            let _ = d.sync_all();
        }
    }
    Ok(())
}

/// Append the file at `path` to `out` (and to `hasher`); returns its length.
fn copy_hashed(
    path: &Path,
    out: &mut BufWriter<File>,
    mut hasher: Option<&mut blake3::Hasher>,
) -> Result<u64, std::io::Error> {
    let mut rd = File::open(path)?;
    let mut buf = vec![0u8; 1 << 20];
    let mut len = 0u64;
    loop {
        let n = rd.read(&mut buf)?;
        if n == 0 {
            break;
        }
        out.write_all(&buf[..n])?;
        if let Some(h) = hasher.as_deref_mut() {
            h.update(&buf[..n]);
        }
        len += n as u64;
    }
    Ok(len)
}

#[cfg(test)]
mod tests {
    #[test]
    fn a_dataset_clearly_past_u32_per_position_is_refused_early() {
        use super::check_position_estimate as check;
        assert!(check(0, 0).is_ok());
        assert!(check(u32::MAX as u64, u32::MAX as u64).is_ok());
        // within the estimator's margin: the exact check after the merge decides
        assert!(check(u32::MAX as u64 + 1_000_000, 5).is_ok());
        let e = check(5_000_000_000, 10).unwrap_err().to_string();
        assert!(e.contains("subject") && e.contains("4294967295"), "{e}");
        let e = check(10, 5_000_000_000).unwrap_err().to_string();
        assert!(e.contains("object"), "{e}");
        // the estimator the build feeds is accurate to well within that margin
        let mut h = crate::hll::Hll::new();
        for i in 0..300_000u32 {
            h.add(&format!("<http://ex/{i}>"));
        }
        assert!((h.count() as f64 - 300_000.0).abs() / 300_000.0 < 0.03);
    }

    use super::*;
    use crate::ingest;

    /// The old test entry point: the image of these sections, written as `0x06`.
    #[allow(clippy::too_many_arguments)]
    fn write_final_file(
        output: &Path,
        metadata: &[u8],
        build_info: &[u8],
        dict: &MergedDict,
        perm_sections: &[TiledSection],
        text: Option<&crate::text_index_spill::TextSection>,
        _named: Option<&()>,
        quad: Option<&QuadSectionFile>,
        quad_count: u64,
        codec: u8,
        perms: crate::index::PermSet,
        consume: bool,
    ) -> Result<(), ExtBuildError> {
        let image = Arc::new(spill_image(
            dict,
            perm_sections,
            text,
            quad,
            quad_count,
            codec,
            perms,
            None,
            &[],
        )?);
        let partial = PartialFile::create(output)?;
        write_v06(output, &image, metadata, build_info, consume, &partial)
    }

    /// Generate a graph that exercises the merge machinery: terms shared across
    /// chunks, subject-only/object-only/shared classification that differs
    /// per-chunk vs globally, langs, datatypes, bnodes, duplicate triples.
    fn test_quads(n: usize) -> Vec<RawQuad> {
        let mut quads = Vec::new();
        for i in 0..n {
            let s = format!("<http://ex/s{}>", i % (n / 7 + 1));
            let p = format!("<http://ex/p{}>", i % 13);
            let o = match i % 5 {
                0 => format!("<http://ex/s{}>", (i + 3) % (n / 7 + 1)), // makes shared terms
                1 => format!("\"lit {i}\""),
                2 => format!("\"{}\"^^<http://www.w3.org/2001/XMLSchema#integer>", i),
                3 => format!("\"v{}\"@en", i % 50),
                _ => format!("_:b{}", i % 97),
            };
            quads.push((s, p, o, None));
        }
        // duplicates, far apart so they land in different chunks
        for i in 0..(n / 20) {
            let j = i * 17 % n;
            quads.push(quads[j].clone());
        }
        quads
    }

    /// The in-RAM build of the same quads, as the bounded build writes it:
    /// format 0x06, named graphs in the quad index, no pyramid / text / card.
    fn build_reference(quads: Vec<RawQuad>) -> Vec<u8> {
        build_reference_quad(quads, crate::index::PermSet::ALL)
    }

    /// An in-RAM `0x05` image as the `0x06` file the writers emit.
    fn as_v06(image: Vec<u8>) -> Vec<u8> {
        crate::v06::write::convert_image(&image, crate::v06::write::WriteOpts::default()).unwrap()
    }

    /// A scratch parent for the spill-directory tests, removed by the caller.
    fn spill_parent(tag: &str) -> PathBuf {
        let p = std::env::temp_dir().join(format!(
            "rete-spill-parent-{tag}-{}-{:016x}",
            std::process::id(),
            spill_nonce()
        ));
        std::fs::create_dir_all(&p).unwrap();
        p
    }

    /// Two builds in two containers that share a `--tmp-dir`: both pid 1, both
    /// their process's first build. They used to get the same directory, and
    /// the first to finish deleted the other's spill.
    #[test]
    fn spill_directories_are_exclusive_even_with_one_pid() {
        let parent = spill_parent("pid1");
        let first_build = || format!(".rete-extbuild-1-0-{:016x}", spill_nonce());
        let a = TmpDir::create_named(&parent, first_build).unwrap();
        let b = TmpDir::create_named(&parent, first_build).unwrap();
        assert_ne!(a.dir, b.dir);
        std::fs::write(a.path("c0.tri"), b"spill").unwrap();
        drop(b);
        assert_eq!(std::fs::read(a.path("c0.tri")).unwrap(), b"spill");
        drop(a);
        assert_eq!(std::fs::read_dir(&parent).unwrap().count(), 0);
        let _ = std::fs::remove_dir_all(&parent);
    }

    /// The production names: distinct within one process too, and recognisable.
    #[test]
    fn spill_directory_names_carry_pid_seq_and_random_bits() {
        let parent = spill_parent("names");
        let dirs: Vec<TmpDir> = (0..8).map(|_| TmpDir::create(&parent).unwrap()).collect();
        let mut names: Vec<String> = dirs
            .iter()
            .map(|d| d.dir.file_name().unwrap().to_string_lossy().into_owned())
            .collect();
        let prefix = format!(".rete-extbuild-{}-", std::process::id());
        for n in &names {
            assert!(n.starts_with(&prefix), "{n}");
            let rand = n.rsplit('-').next().unwrap();
            assert_eq!(rand.len(), 16, "{n}");
            assert!(u64::from_str_radix(rand, 16).is_ok(), "{n}");
        }
        names.sort();
        names.dedup();
        assert_eq!(names.len(), 8);
        drop(dirs);
        let _ = std::fs::remove_dir_all(&parent);
    }

    /// A directory that already exists — a live build's, or one a killed build
    /// left — is never adopted, entered or deleted: the next name is tried, and
    /// when every name is taken the build fails instead of sharing one.
    #[test]
    fn an_existing_spill_directory_is_never_reused() {
        let parent = spill_parent("taken");
        for name in ["taken-0", "taken-1"] {
            std::fs::create_dir(parent.join(name)).unwrap();
            std::fs::write(parent.join(name).join("owner"), name).unwrap();
        }
        let mut names = ["taken-0", "taken-1", "fresh"].into_iter();
        let fresh = TmpDir::create_named(&parent, || names.next().unwrap().to_string()).unwrap();
        assert_eq!(fresh.dir, parent.join("fresh"));
        drop(fresh);
        assert!(!parent.join("fresh").exists());

        let err = TmpDir::create_named(&parent, || "taken-0".to_string())
            .err()
            .expect("every name taken must be an error, not a shared directory");
        assert_eq!(err.kind(), std::io::ErrorKind::AlreadyExists);

        for name in ["taken-0", "taken-1"] {
            assert_eq!(
                std::fs::read_to_string(parent.join(name).join("owner")).unwrap(),
                name
            );
        }
        let _ = std::fs::remove_dir_all(&parent);
    }

    fn build_ext(quads: Vec<RawQuad>, budget: u64) -> (Vec<u8>, BuildStats) {
        static TEST_SEQ: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
        let seq = TEST_SEQ.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        let dir = std::env::temp_dir().join(format!(
            "rete-extbuild-test-{}-{budget}-{seq}",
            std::process::id()
        ));
        std::fs::create_dir_all(&dir).unwrap();
        let out = dir.join("out.rete");
        let stats = build_external(
            |visit| {
                for q in quads.iter().cloned() {
                    visit(q)?;
                }
                Ok(())
            },
            &out,
            ExternalBuildOptions {
                memory_budget: budget,
                tmp_dir: Some(dir.clone()),
                metadata: Box::new(|_| Vec::new()),
                build_info: Vec::new(),
                perms: crate::index::PermSet::ALL,
                ..Default::default()
            },
        )
        .unwrap();
        let bytes = std::fs::read(&out).unwrap();
        let _ = std::fs::remove_dir_all(&dir);
        (bytes, stats)
    }

    /// A build-info payload is written as a kind-7 section, readable back, and
    /// leaves the content hash identical to a build without one — the external
    /// builder's version of the outside-the-hash property.
    #[test]
    fn external_build_info_is_outside_the_hash() {
        let quads = test_quads(500);
        let (plain, _) = build_ext(quads.clone(), 0);

        let dir = std::env::temp_dir().join(format!("rete-extbuild-bi-{}", std::process::id()));
        std::fs::create_dir_all(&dir).unwrap();
        let out = dir.join("out.rete");
        let info = br#"{"built_at":"2026-08-04T00:00:00Z"}"#.to_vec();
        build_external(
            |visit| {
                for q in quads.iter().cloned() {
                    visit(q)?;
                }
                Ok(())
            },
            &out,
            ExternalBuildOptions {
                memory_budget: 0,
                tmp_dir: Some(dir.clone()),
                metadata: Box::new(|_| Vec::new()),
                build_info: info.clone(),
                perms: crate::index::PermSet::ALL,
                ..Default::default()
            },
        )
        .unwrap();
        let with = std::fs::read(&out).unwrap();
        let _ = std::fs::remove_dir_all(&dir);

        assert_eq!(crate::read_build_info(&with).unwrap().unwrap(), info);
        let h_with = crate::Header::from_bytes(&with).unwrap();
        let h_plain = crate::Header::from_bytes(&plain).unwrap();
        assert_eq!(h_with.content_hash, h_plain.content_hash);
        assert!(crate::verify(&with).unwrap());
        // Stripping the section restores the plain image byte-for-byte.
        assert_eq!(crate::attach_build_info(&with, &[]).unwrap(), plain);
        // And the graph still opens.
        crate::Rete::open(&with).unwrap();
    }

    /// The heart of the feature: a tiny budget (forcing many chunks, many sort
    /// runs) must produce **byte-identical** output to the in-RAM build.
    #[test]
    fn external_build_is_byte_identical_to_streaming() {
        let quads = test_quads(3000);
        let reference = build_reference(quads.clone());
        // 64 MiB is the floor; to force chunking on a small graph we can't rely
        // on the floor — instead check both a floor build (1 chunk) and verify
        // chunk-count > 1 via a direct Chunker probe below.
        let (bytes_floor, stats) = build_ext(quads.clone(), 0);
        // `test_quads` repeats some statements, so this fixture is exactly the
        // case #128 was about: the reported count is what the build WROTE (and
        // what the header records), not what it read.
        let header = crate::Header::from_bytes(&bytes_floor).unwrap();
        assert_eq!(stats.statements as u64, header.quad_count);
        assert!(
            stats.statements < quads.len(),
            "fixture must contain duplicates for this to mean anything"
        );
        assert_eq!(
            bytes_floor, reference,
            "single-chunk external build must be byte-identical"
        );
    }

    /// Mega-group inputs (one predicate dominating the graph — the Crossref
    /// `cites` shape) exercise the mid-group tile cuts; the external and
    /// in-RAM builds must STILL be byte-identical, proving both tilers cut at
    /// the same boundaries (shared `GroupSizer`).
    #[test]
    fn skewed_external_build_is_byte_identical() {
        let mut quads: Vec<RawQuad> = Vec::new();
        for i in 0..30_000usize {
            quads.push((
                format!("<http://ex/s{}>", i % 500),
                "<http://ex/cites>".to_string(),
                format!("<http://ex/o{i}>"),
                None,
            ));
        }
        for i in 0..500usize {
            quads.push((
                format!("<http://ex/s{i}>"),
                format!("<http://ex/p{}>", i % 7),
                format!("\"lit {i}\""),
                None,
            ));
        }
        let reference = build_reference(quads.clone());
        let (bytes, stats) = build_ext(quads.clone(), 0);
        assert_eq!(stats.statements, quads.len());
        assert_eq!(
            bytes, reference,
            "mega-group external build must be byte-identical"
        );
    }

    /// Multi-chunk path: shrink the chunk budget directly (bypassing the public
    /// floor) so a small graph is split across many chunks + runs, and the
    /// merged output must STILL be byte-identical.
    #[test]
    fn multi_chunk_external_build_is_byte_identical() {
        let quads = test_quads(3000);
        let reference = build_reference(quads.clone());

        let dir = std::env::temp_dir().join(format!("rete-extbuild-mc-{}", std::process::id()));
        std::fs::create_dir_all(&dir).unwrap();
        let out = dir.join("out.rete");

        // drive the internals directly with a tiny chunk budget
        let tmp = TmpDir::create(&dir).unwrap();
        let mut chunker = Chunker::new(&tmp, 4 * 1024, GraphLayout::PerGraph); // ~4 KiB chunks
        for q in quads.iter().cloned() {
            chunker.push(q).unwrap();
        }
        let chunks = chunker.finish().unwrap();
        assert!(
            chunks.len() >= 4,
            "expected many chunks, got {}",
            chunks.len()
        );
        let statements: u64 = chunks.iter().map(|c| c.triple_count).sum();
        let (merged, remaps) = merge_dictionaries(&tmp, &chunks).unwrap();

        // tiny runs to force multi-run merging, chained across permutations
        let codec = crate::file::writer_codec();
        let (runs, _) = remap_chunks(
            &tmp,
            &chunks,
            &remaps,
            false,
            crate::index::IndexPermutation::Spo,
            256,
            256,
            None,
            None,
            None,
        )
        .unwrap();
        assert!(runs.len() >= 4, "expected many runs, got {}", runs.len());
        let (sections, count) = index_permutations(
            &tmp,
            runs,
            crate::index::PermSet::ALL,
            256,
            codec,
            "",
            &mut |_, _| Ok(()),
            &mut |_, _| {},
        )
        .unwrap();
        write_final_file(
            &out,
            &[],
            &[],
            &merged,
            &sections,
            None,
            None,
            None,
            count,
            codec,
            crate::index::PermSet::ALL,
            true,
        )
        .unwrap();

        let bytes = std::fs::read(&out).unwrap();
        assert_eq!(statements as usize, quads.len());
        assert_eq!(
            bytes, reference,
            "multi-chunk external build must be byte-identical"
        );
        drop(tmp);
        let _ = std::fs::remove_dir_all(&dir);
    }

    /// [`build_ext`] with the chunk budget forced down to `chunk` bytes, so a
    /// small graph becomes many chunks.
    fn build_ext_split(quads: &[RawQuad], chunk: u64) -> Vec<u8> {
        TEST_SPLIT.with(|c| c.set(Some((chunk, 1 << 10))));
        let (bytes, _) = build_ext(quads.to_vec(), 0);
        TEST_SPLIT.with(|c| c.set(None));
        bytes
    }

    /// The id remap tables were the part of the memory-bounded build that grew
    /// with the dataset: every chunk's stayed resident from the dictionary
    /// merge to the end of the remap. Now they go to disk as the merge writes
    /// them and one chunk's are loaded at a time. Cutting four times the
    /// graph into four times the chunks quadruples what the old build held
    /// and leaves what this one holds where it was: one chunk's tables, at
    /// most a quarter of the chunk budget. The files stay byte-identical.
    #[test]
    fn remap_memory_does_not_grow_with_the_chunk_count() {
        const CHUNK: u64 = 4 * 1024;
        let run = |n: usize| -> (u64, u64, u64) {
            let quads = test_quads(n);
            let reference = build_reference(quads.clone());
            let collector = crate::build_trace::Collector::start();
            let bytes = build_ext_split(&quads, CHUNK);
            let t = collector.snapshot();
            drop(collector);
            assert_eq!(bytes, reference, "{n} quads: external build drifted");
            (
                t.counter("chunks").unwrap(),
                t.counter("remap_total_bytes").unwrap(),
                t.counter("remap_peak_bytes").unwrap(),
            )
        };
        let (k1, total1, peak1) = run(1_000);
        let (k4, total4, peak4) = run(4_000);
        assert!(k1 >= 20 && k4 >= 3 * k1, "chunks: {k1}, then {k4}");
        // What the old merge held resident grows with the chunks…
        assert!(total4 >= 3 * total1, "tables: {total1} B, then {total4} B");
        assert!(
            total4 >= 50 * peak4,
            "{total4} B in all, {peak4} B resident"
        );
        // …what this one holds does not: every buffered statement is charged
        // at least 96 B against the chunk budget and brings at most six ids.
        assert!(
            peak4 <= peak1 * 3 / 2,
            "resident: {peak1} B, then {peak4} B"
        );
        assert!(
            peak4 <= (CHUNK + 1024) / 4,
            "{peak4} B for a {CHUNK} B chunk"
        );
    }

    /// G24: a term-dense stretch — every statement bringing two new terms,
    /// the OpenCitations Meta `oc:doiLink` tail — is sealed by its
    /// dictionary's size, not its bytes, so it takes more and smaller chunks
    /// than the bytes alone would cut; the rest of the graph is chunked as
    /// before. Chunk boundaries never reach the file: it is the in-RAM
    /// build's, and the one-chunk build's, byte for byte.
    #[test]
    fn term_dense_tail_seals_by_dictionary_and_stays_identical() {
        let mut quads = test_quads(2_000);
        for i in 0..3_000usize {
            quads.push((
                format!("<http://ex/s{}>", i % 300),
                "<https://w3id.org/oc/ontology/doiLink>".to_string(),
                format!("<https://doi.org/10.{}/x{i}>", 1000 + i % 7),
                None,
            ));
        }
        for i in 0..3_000usize {
            quads.push((
                format!("<http://ex/id/{i}>"),
                "<http://ex/hasLiteralValue>".to_string(),
                format!("\"10.{}/x{i}\"", 1000 + i % 7),
                None,
            ));
        }
        let reference = build_reference(quads.clone());
        let (one_chunk, _) = build_ext(quads.clone(), 0);
        assert_eq!(one_chunk, reference, "the one-chunk build drifted");

        let chunked = |working: Option<u64>| -> (Vec<u8>, u64, u64) {
            TEST_SPLIT.with(|c| c.set(Some((64 << 10, 1 << 16))));
            TEST_WORKING.with(|c| c.set(working));
            let collector = crate::build_trace::Collector::start();
            let (bytes, _) = build_ext(quads.clone(), 0);
            let t = collector.snapshot();
            drop(collector);
            TEST_SPLIT.with(|c| c.set(None));
            TEST_WORKING.with(|c| c.set(None));
            (
                bytes,
                t.counter("chunks").unwrap(),
                t.counter("chunks_term_sealed").unwrap_or(0),
            )
        };
        // By bytes alone (the cap never reached)...
        let (by_bytes, k_bytes, sealed_bytes) = chunked(Some(u64::MAX));
        assert_eq!(sealed_bytes, 0);
        // ...and with the cap at 2.4 times the byte budget: the head (about
        // 1.7 times its bytes with its dictionary) still reaches the byte
        // budget first, the dense tail (about 2.8 times) does not.
        let (by_terms, k_terms, sealed_terms) = chunked(Some(154 << 10));
        eprintln!("{k_bytes} chunks by bytes; {k_terms} with the cap, {sealed_terms} sealed by it");
        assert!(k_bytes >= 4, "{k_bytes} chunk(s) by bytes");
        // The tail is sealed by its dictionary, the head still by its bytes.
        assert!(
            sealed_terms >= 8 && k_terms - sealed_terms >= 4,
            "{sealed_terms} of {k_terms} chunk(s) sealed by the dictionary"
        );
        assert!(k_terms > k_bytes, "{k_terms} chunks, {k_bytes} by bytes");
        assert_eq!(by_bytes, reference, "the byte-chunked build drifted");
        assert_eq!(by_terms, reference, "the term-chunked build drifted");
    }

    /// G23: the spill is compressed and the permutations are chained, so the
    /// build never holds its id statements raw, let alone twice. Up to 0.4.1
    /// the permutation phase held `global.tri` plus one permutation's runs —
    /// 24 B per statement — beside the finished sections, and the dictionary
    /// merge the 12 B per statement of the chunk spills beside the raw term
    /// files. Here the recorded peak, finished sections included, stays under
    /// the output's size plus 6 B per statement; and the file is the same.
    #[test]
    fn compressed_spill_peak_stays_under_one_raw_copy() {
        let quads = test_quads(40_000);
        let n = quads.len() as u64;
        let reference = build_reference(quads.clone());
        TEST_SPLIT.with(|c| c.set(Some((256 << 10, 1 << 16))));
        TEST_RUNS.with(|c| c.set(Some((6_000, 6_000))));
        let collector = crate::build_trace::Collector::start();
        let (bytes, _) = build_ext(quads, 0);
        let t = collector.snapshot();
        drop(collector);
        TEST_SPLIT.with(|c| c.set(None));
        TEST_RUNS.with(|c| c.set(None));
        assert_eq!(bytes, reference, "the compressed spill changed the file");
        let peak = t.counter("scratch_peak_bytes").unwrap();
        let chunks = t.counter("chunks").unwrap();
        let out = bytes.len() as u64;
        eprintln!(
            "{n} statements, {chunks} chunks: spill peak {peak} B = {:.2} B per statement (output {out} B = {:.2})",
            peak as f64 / n as f64,
            out as f64 / n as f64
        );
        assert!(chunks >= 4, "{chunks} chunk(s)");
        // A raw copy of the statements alone would be 12 B each.
        assert!(
            peak <= out + 6 * n,
            "spill peak {peak} B for {n} statements and a {out} B file"
        );
    }

    /// The two dictionary writers must agree on the **separator keys**, not
    /// just on everything else. `finish_chunked` and
    /// `file::encode_chunked_dict_section` compute the previous chunk's last
    /// term from different sources — a spill file read back in slices, versus
    /// one contiguous buffer — so this is the place they can drift.
    ///
    /// The literals here are long enough that the object-only section really
    /// splits into chunks; a short-IRI graph would compare two one-chunk
    /// directories and prove nothing.
    #[test]
    fn separator_keyed_dictionary_is_byte_identical_across_writers() {
        let blob = "w".repeat(1200);
        let quads: Vec<RawQuad> = (0..900usize)
            .map(|i| {
                (
                    format!("<http://ex/s{i:04}>"),
                    "<http://ex/note>".to_string(),
                    // half share a long prefix (long separators), half diverge
                    // in their first bytes (one-byte separators)
                    if i % 2 == 0 {
                        format!("\"{blob}{i:04}\"")
                    } else {
                        format!("\"{i:04}{blob}\"")
                    },
                    None,
                )
            })
            .collect();

        let reference = build_reference(quads.clone());
        let (bytes, _) = build_ext(quads, 0);
        assert_eq!(
            bytes, reference,
            "the external builder's chunk directory drifted from the in-RAM one"
        );

        // …and the directory they agree on really is multi-chunk and keyed by
        // separators, not by the 1,206-byte literals it used to copy.
        let keys = crate::file::dict_chunk_keys_for_test(&bytes);
        let obj = &keys[2]; // section 2 = object-only, the one with the literals
        assert!(
            obj.len() > 4,
            "object-only section has {} chunks",
            obj.len()
        );
        assert!(obj[0].is_empty(), "chunk 0 must carry no separator");
        let longest = obj.iter().map(Vec::len).max().unwrap();
        assert!(
            longest < 1_200,
            "longest key is {longest} B — that is a stored term, not a separator"
        );

        // Every object term still resolves: the routing works in the merged file.
        let rete = crate::Rete::open(&bytes).unwrap();
        for i in (0..900usize).step_by(7) {
            let o = if i % 2 == 0 {
                format!("\"{blob}{i:04}\"")
            } else {
                format!("\"{i:04}{blob}\"")
            };
            assert!(
                rete.dictionary().object_id(&o).is_some(),
                "object {i} lost by the chunk directory"
            );
        }
    }

    /// …and the same, with the terms spread across **named graphs**. The graph
    /// column changes which chunk a term is first seen in, which changes what
    /// the per-chunk dictionaries look like — so the separator directory the two
    /// writers agree on has to survive quads, not just triples. Same shape as
    /// `separator_keyed_dictionary_is_byte_identical_across_writers` (#222),
    /// with the literals distributed over 30 graphs.
    #[test]
    fn separator_keyed_dictionary_survives_named_graphs() {
        let blob = "w".repeat(1200);
        let quads: Vec<RawQuad> = (0..900usize)
            .map(|i| {
                (
                    format!("<http://ex/s{i:04}>"),
                    "<http://ex/note>".to_string(),
                    if i % 2 == 0 {
                        format!("\"{blob}{i:04}\"")
                    } else {
                        format!("\"{i:04}{blob}\"")
                    },
                    // graph names generated out of order, so the graph merge has
                    // to sort them and the ordinals are not the insertion order
                    Some(format!("<http://ex/g{:02}>", (i * 7) % 30)),
                )
            })
            .collect();

        let reference = build_reference(quads.clone());
        let (bytes, stats) = build_ext(quads, 0);
        assert_eq!(
            bytes, reference,
            "the external builder's chunk directory drifted from the in-RAM one \
             once named graphs were in the input"
        );
        assert_eq!(stats.named_graphs, 30);

        let keys = crate::file::dict_chunk_keys_for_test(&bytes);
        let obj = &keys[2]; // section 2 = object-only, the one with the literals
        assert!(
            obj.len() > 4,
            "object-only section has {} chunks",
            obj.len()
        );
        assert!(obj[0].is_empty(), "chunk 0 must carry no separator");
        let longest = obj.iter().map(Vec::len).max().unwrap();
        assert!(
            longest < 1_200,
            "longest key is {longest} B — that is a stored term, not a separator"
        );
    }

    /// The external file must open and answer queries like any other build.
    #[test]
    fn external_build_output_is_queryable() {
        let quads = test_quads(1200);
        let (bytes, _) = build_ext(quads.clone(), 0);
        let rete = crate::Rete::open(&bytes).unwrap();
        // spot-check a known triple through the SPARQL engine
        let out = crate::eval_query(
            &rete,
            "SELECT (COUNT(*) AS ?n) WHERE { ?s <http://ex/p1> ?o }",
        )
        .unwrap();
        match out {
            crate::QueryOutput::Select(_, rows) => assert_eq!(rows.len(), 1),
            other => panic!("expected select, got {other:?}"),
        }
    }

    /// A finished permutation's spilled tiles, read back: the section (its
    /// directory and synopses rebuilt from the tiles, exactly as the tiler
    /// wrote them), every triple handed to `f` in the permutation's order.
    /// Tiles are independent zstd frames, so the file splits without its
    /// directory — which lived in the killed process's memory.
    #[cfg(feature = "compression")]
    fn section_from_tiles(
        path: &Path,
        codec: u8,
        f: &mut dyn FnMut(IdTriple) -> Result<(), ExtBuildError>,
    ) -> Result<TiledSection, ExtBuildError> {
        let mut file = File::open(path)?;
        let (mut buf, mut pos, mut eof) = (Vec::new(), 0usize, false);
        let mut dir: Vec<TileDirEntry> = Vec::new();
        loop {
            if !eof && buf.len() - pos < (4 << 20) {
                buf.drain(..pos);
                pos = 0;
                let at = buf.len();
                buf.resize(at + (16 << 20), 0);
                let n = file.read(&mut buf[at..])?;
                buf.truncate(at + n);
                eof = n == 0;
                continue;
            }
            if pos == buf.len() {
                break;
            }
            let len = zstd::zstd_safe::find_frame_compressed_size(&buf[pos..])
                .map_err(|_| ExtBuildError::Internal("a spilled tile is cut short"))?;
            let raw = crate::file::decompress(codec, &buf[pos..pos + len])
                .map_err(|_| ExtBuildError::Internal("a spilled tile does not decode"))?;
            let blk = crate::triples::TripleBlock::parse(&raw)
                .map_err(|_| ExtBuildError::Internal("a spilled tile does not parse"))?;
            let triples = blk.triples();
            let z = blk.zone();
            let (Some(first), Some(last)) = (triples.first(), triples.last()) else {
                return Err(ExtBuildError::Internal("an empty spilled tile"));
            };
            dir.push((
                first.0,
                last.0,
                len as u64,
                (z.min_b, z.max_b, z.min_c, z.max_c),
            ));
            for t in triples {
                f(t)?;
            }
            pos += len;
        }
        Ok(TiledSection::from_dir(&dir, path.to_path_buf()))
    }

    /// Restart the permutation chain of a spill whose build was killed in
    /// the permutation phase (see [`resume_from_spill`]): the sections of the
    /// finished permutations of `order`, rebuilt from their tiles, followed by
    /// the rest of the chain, re-run from the last finished one's tiles.
    #[cfg(feature = "compression")]
    fn resume_chain(
        tmp: &TmpDir,
        order: &[crate::index::IndexPermutation],
        statements: u64,
        run_len: usize,
        codec: u8,
    ) -> Result<Vec<TiledSection>, ExtBuildError> {
        let tiles = |p: crate::index::IndexPermutation| tmp.path(&format!("{}.tiles", p.name()));
        // A permutation's tiles file is created as its merge starts, after
        // the previous one's section is finished: the last one present was
        // cut short, every one before it is whole.
        let started = order.iter().take_while(|p| tiles(**p).exists()).count();
        if order.iter().skip(started).any(|p| tiles(*p).exists()) {
            return Err(ExtBuildError::Internal(
                "the permutations' tiles files have gaps",
            ));
        }
        if started < 2 {
            return Err(ExtBuildError::Internal(
                "killed in the first permutation: its runs are half consumed, rerun the build",
            ));
        }
        let cut = started - 1;
        // whatever the cut-short permutation left: its tiles and runs, and
        // the runs it was writing for the next one
        for entry in std::fs::read_dir(&tmp.dir)?.flatten() {
            let n = entry.file_name().to_string_lossy().into_owned();
            let doomed = order[cut..]
                .iter()
                .take(2)
                .any(|p| n.starts_with(&format!("{}.", p.name())));
            if doomed {
                let _ = std::fs::remove_file(entry.path());
            }
        }
        let mut sections = Vec::with_capacity(order.len());
        for &p in &order[..cut - 1] {
            sections.push(section_from_tiles(&tiles(p), codec, &mut |_| Ok(()))?);
        }
        let (prev, next) = (order[cut - 1], order[cut]);
        let mut spill = RunSpiller::<IdTriple>::new(&tmp.dir, next.name(), run_len);
        let mut n = 0u64;
        sections.push(section_from_tiles(&tiles(prev), codec, &mut |t| {
            n += 1;
            spill.push(next.forward(prev.back(t)))?;
            Ok(())
        })?);
        if n != statements {
            return Err(ExtBuildError::Internal(
                "the last finished section does not hold every statement",
            ));
        }
        let (rest, count) = index_chain(
            tmp,
            spill.finish()?,
            &order[cut..],
            run_len,
            codec,
            "",
            &mut |_, _| Ok(()),
            &mut |p, _| eprintln!("resume: {} done", p.name()),
        )?;
        if count != statements {
            return Err(ExtBuildError::Internal("permutation dedup counts diverge"));
        }
        sections.extend(rest);
        Ok(sections)
    }

    /// **Operational resume harness** (ignored; run explicitly with env vars):
    /// finish an external build whose process was killed in the permutation
    /// phase, from its surviving spill directory. Requires:
    ///   RETE_RESUME_SPILL   the .rete-extbuild-* dir
    ///   RETE_RESUME_OUT     output .rete path
    ///   RETE_RESUME_TERMS   merged dictionary term count
    ///   RETE_RESUME_QUADS   deduped triple count (statements when no dups)
    ///   RETE_RESUME_CARD    optional JSON card file to embed
    ///   RETE_RESUME_PERMS   the build's `--permutations`, 6 (default) or 3
    ///   RETE_RESUME_BUDGET_MB  sizes the sort runs (default 16384)
    ///
    /// The permutations are chained and each one's runs are consumed as they
    /// are merged, so no `global.tri` survives to restart from (up to 0.4.1
    /// one did). What survives is every finished permutation's tiles file,
    /// which holds every statement, sorted: [`resume_chain`] reads the last
    /// finished one back and runs the chain again from the permutation that
    /// was cut short. Not resumable: a kill in the first permutation (its
    /// runs are half consumed) or in the final write (each section is deleted
    /// once copied); a kill before the permutations never was.
    ///
    /// **Default graph only.** A spill from a build that had named graphs
    /// carries quad runs (`qSPO.run*`) and `g.graphs`, and this harness
    /// ignores both — it would resume as a file with the named graphs
    /// silently dropped. That is checked rather than commented.
    ///
    /// **A text index is carried over when it was finished.** Phase 3b leaves
    /// `tok.table` + `tok.postings` behind before any permutation is merged,
    /// and they are reused as they are. Text-index files of an unfinished
    /// phase (`global.tok`, `tok.run*`, `c<i>.tok`) are a hard stop for the
    /// same reason as the quad runs.
    #[test]
    #[ignore = "operational tool, driven by RETE_RESUME_* env vars"]
    #[cfg(feature = "compression")]
    fn resume_from_spill() {
        let spill = PathBuf::from(std::env::var("RETE_RESUME_SPILL").expect("RETE_RESUME_SPILL"));
        let names: Vec<String> = std::fs::read_dir(&spill)
            .unwrap()
            .flatten()
            .map(|e| e.file_name().to_string_lossy().into_owned())
            .collect();
        assert!(
            !names
                .iter()
                .any(|n| n.starts_with('q') || n == "g.graphs" || n.starts_with("named.")),
            "this spill has named quads; resuming it here would write a file with \
             the named graphs silently missing — rerun the build instead"
        );
        assert!(
            !names
                .iter()
                .any(|n| n == "global.tok" || n.starts_with("tok.run") || n.ends_with(".tok")),
            "this spill has an unfinished text index (global.tok / tok.run* / c<i>.tok); \
             resuming it here would write a file without it — rerun the build instead"
        );
        let out = PathBuf::from(std::env::var("RETE_RESUME_OUT").expect("RETE_RESUME_OUT"));
        let term_count: u64 = std::env::var("RETE_RESUME_TERMS").unwrap().parse().unwrap();
        let quad_count: u64 = std::env::var("RETE_RESUME_QUADS").unwrap().parse().unwrap();
        let metadata: Vec<u8> = std::env::var("RETE_RESUME_CARD")
            .ok()
            .map(|p| std::fs::read(p).unwrap())
            .unwrap_or_default();
        let perms = match std::env::var("RETE_RESUME_PERMS").as_deref() {
            Ok("3") => crate::index::PermSet::CORE,
            _ => crate::index::PermSet::ALL,
        };
        let budget_mb: u64 = std::env::var("RETE_RESUME_BUDGET_MB")
            .ok()
            .and_then(|v| v.parse().ok())
            .unwrap_or(16384);
        let run_len = ((budget_mb << 20) / 2 / 24) as usize;

        let sec = |name: &str| -> SectionFile {
            let path = spill.join(name);
            let len = std::fs::metadata(&path)
                .unwrap_or_else(|_| panic!("{name} is gone: the final write had begun"))
                .len();
            SectionFile { path, len }
        };
        let merged = MergedDict {
            section_files: [
                sec("g.shared.sec"),
                sec("g.subj.sec"),
                sec("g.obj.sec"),
                sec("g.pred.sec"),
            ],
            term_count,
            has_quoted: false,
            schema_facts: Default::default(),
        };
        let tmp = TmpDir {
            dir: spill.clone(),
            keep: std::cell::Cell::new(false),
        };
        let codec = crate::file::writer_codec();
        let order: Vec<_> = perms.iter().collect();
        let sections = resume_chain(&tmp, &order, quad_count, run_len, codec).unwrap();
        let (table, postings) = (spill.join("tok.table"), spill.join("tok.postings"));
        let text = if table.exists() && postings.exists() {
            let table_len = std::fs::metadata(&table).unwrap().len();
            let mut head = Vec::new();
            write_uvarint(&mut head, table_len);
            // The scope block in the token table decides kind 6 or 11, and
            // the header's named-graphs bit.
            let mut prefix = head.clone();
            prefix.extend(std::fs::read(&table).unwrap());
            let scope = crate::text_index::TextIndex::from_token_table(
                &prefix,
                codec,
                Box::new(|_, _| None),
            )
            .expect("tok.table parses")
            .scope()
            .clone();
            eprintln!("resume: reusing the text index");
            Some(crate::text_index_spill::TextSection {
                head,
                table,
                table_len,
                postings_len: std::fs::metadata(&postings).unwrap().len(),
                postings,
                partial: scope.restricts_predicates(),
                named: scope.covers_named_graphs(),
            })
        } else {
            None
        };
        write_final_file(
            &out,
            &metadata,
            &[],
            &merged,
            &sections,
            text.as_ref(),
            None,
            None,
            quad_count,
            codec,
            perms,
            false, // keep the spill until the file is verified
        )
        .unwrap();
        eprintln!("resume: wrote {}", out.display());
        std::mem::forget(tmp);
    }

    /// The resume path, end to end: a build "killed" as its fourth
    /// permutation starts (its tiles half written, its runs and the fifth's
    /// left as they were) resumes from the third's tiles to the same file.
    #[test]
    #[cfg(feature = "compression")]
    fn a_build_killed_mid_chain_resumes_to_the_same_file() {
        let quads = test_quads(5000);
        let reference = build_reference(quads.clone());
        let dir = std::env::temp_dir().join(format!("rete-extbuild-resume-{}", std::process::id()));
        std::fs::create_dir_all(&dir).unwrap();
        let out = dir.join("out.rete");
        let tmp = TmpDir::create(&dir).unwrap();
        let mut chunker = Chunker::new(&tmp, 16 * 1024, GraphLayout::PerGraph);
        for q in quads.iter().cloned() {
            chunker.push(q).unwrap();
        }
        let chunks = chunker.finish().unwrap();
        let (merged, remaps) = merge_dictionaries(&tmp, &chunks).unwrap();
        let order: Vec<_> = crate::index::PermSet::ALL.iter().collect();
        let codec = crate::file::writer_codec();
        let (runs, _) = remap_chunks(
            &tmp, &chunks, &remaps, false, order[0], 700, 700, None, None, None,
        )
        .unwrap();
        // The first three permutations finish; the chain stops there, so the
        // fourth's runs are what its merge would have read…
        let (_, n) = index_chain(
            &tmp,
            runs,
            &order[..3],
            700,
            codec,
            "",
            &mut |_, _| Ok(()),
            &mut |_, _| {},
        )
        .unwrap();
        // …and the fourth was cut short with half a tile on disk.
        std::fs::write(
            tmp.path(&format!("{}.tiles", order[3].name())),
            [0x28, 0xb5],
        )
        .unwrap();
        let sections = resume_chain(&tmp, &order, n, 700, codec).unwrap();
        write_final_file(
            &out,
            &[],
            &[],
            &merged,
            &sections,
            None,
            None,
            None,
            n,
            codec,
            crate::index::PermSet::ALL,
            true,
        )
        .unwrap();
        assert_eq!(
            std::fs::read(&out).unwrap(),
            reference,
            "resumed file differs"
        );
        drop(tmp);
        let _ = std::fs::remove_dir_all(&dir);
    }

    /// A faithful reduction of the one published named-graph dataset
    /// (`switzerland-fedlex`: 56.3 M quads across 497,905 graphs, most of them
    /// tiny). Includes the three awkward shapes on purpose:
    ///   * default-graph triples living alongside named ones,
    ///   * the same triple asserted in two different graphs,
    ///   * a graph holding exactly one statement.
    fn fedlex_shaped_quads(graphs: usize, per_graph: usize) -> Vec<RawQuad> {
        let mut quads: Vec<RawQuad> = Vec::new();
        // the ontology triples fedlex keeps in the default graph
        for i in 0..40usize {
            quads.push((
                format!("<http://data.europa.eu/eli/ontology#p{i}>"),
                "<http://www.w3.org/1999/02/22-rdf-syntax-ns#type>".to_string(),
                "<http://www.w3.org/2002/07/owl#FunctionalProperty>".to_string(),
                None,
            ));
        }
        for g in 0..graphs {
            // the graph names are NOT generated in sorted order — the k-way
            // merge really has to order them
            let name = format!(
                "<https://fedlex.data.admin.ch/eli/cc/{:05}/graph>",
                g * 7 % 9973
            );
            let n = if g % 11 == 0 { 1 } else { per_graph };
            for i in 0..n {
                quads.push((
                    format!("<https://fedlex.data.admin.ch/eli/cc/{g}/art_{i}>"),
                    format!("<http://data.europa.eu/eli/ontology#pred{}>", i % 6),
                    match i % 4 {
                        0 => format!("<https://fedlex.data.admin.ch/eli/cc/{g}>"),
                        1 => format!("\"Artikel {i}\"@de"),
                        2 => format!("\"{i}\"^^<http://www.w3.org/2001/XMLSchema#integer>"),
                        _ => format!("_:b{}", i % 13),
                    },
                    Some(name.clone()),
                ));
            }
            // the SAME triple in a second graph — must survive in both
            quads.push((
                "<https://fedlex.data.admin.ch/eli/cc/shared>".to_string(),
                "<http://data.europa.eu/eli/ontology#in_force>".to_string(),
                "\"true\"^^<http://www.w3.org/2001/XMLSchema#boolean>".to_string(),
                Some(name.clone()),
            ));
            // …and in the default graph too
            if g == 0 {
                quads.push((
                    "<https://fedlex.data.admin.ch/eli/cc/shared>".to_string(),
                    "<http://data.europa.eu/eli/ontology#in_force>".to_string(),
                    "\"true\"^^<http://www.w3.org/2001/XMLSchema#boolean>".to_string(),
                    None,
                ));
            }
        }
        // duplicates, far apart, so dedup has something to do in both builders
        for i in 0..(quads.len() / 20) {
            let j = i * 17 % quads.len();
            quads.push(quads[j].clone());
        }
        quads
    }

    /// The acceptance criterion from #139: the same named-graph input built both
    /// ways must be **byte-identical**, not merely equivalent.
    #[test]
    fn named_graph_external_build_is_byte_identical() {
        let quads = fedlex_shaped_quads(60, 9);
        let reference = build_reference(quads.clone());
        let (bytes, stats) = build_ext(quads.clone(), 0);
        assert_eq!(
            bytes, reference,
            "named-graph external build must be byte-identical"
        );

        let header = crate::Header::from_bytes(&bytes).unwrap();
        assert_eq!(stats.statements as u64, header.quad_count);
        assert!(
            stats.statements < quads.len(),
            "fixture must contain duplicates for this to mean anything"
        );
        assert!(stats.named_graphs > 1);
        assert!(stats.default_triples > 0, "default graph must be populated");

        // …and the file really carries the graphs, not just the same bytes.
        let rete = crate::Rete::open(&bytes).unwrap();
        assert_eq!(rete.named_graph_count(), stats.named_graphs);
    }

    /// Named graphs with an EMPTY default graph: the default index is still
    /// written (zero triples) and the header still says quads.
    #[test]
    fn named_only_build_is_byte_identical() {
        let quads: Vec<RawQuad> = (0..50usize)
            .map(|i| {
                (
                    format!("<http://ex/s{i}>"),
                    "<http://ex/p>".to_string(),
                    format!("\"v{i}\""),
                    Some(format!("<http://ex/g{}>", i % 5)),
                )
            })
            .collect();
        let reference = build_reference(quads.clone());
        let (bytes, stats) = build_ext(quads, 0);
        assert_eq!(bytes, reference, "named-only build must be byte-identical");
        assert_eq!(stats.default_triples, 0);
        assert_eq!(stats.named_graphs, 5);
        let rete = crate::Rete::open(&bytes).unwrap();
        assert_eq!(rete.named_graph_count(), 5);
    }

    /// Named graphs survive the trip: every distinct quad comes back out of the
    /// file, in the right graph, and `GRAPH ?g` still discriminates.
    #[test]
    fn named_graph_output_is_queryable() {
        let quads = fedlex_shaped_quads(12, 5);
        let (bytes, _) = build_ext(quads.clone(), 0);
        let rete = crate::Rete::open(&bytes).unwrap();

        type Quad = (String, String, String, Option<String>);
        let expected: std::collections::BTreeSet<Quad> = quads.iter().cloned().collect();
        let mut got: std::collections::BTreeSet<Quad> = std::collections::BTreeSet::new();
        for (s, p, o) in rete.dump(None) {
            got.insert((s, p, o, None));
        }
        let names: Vec<String> = rete.graph_names().iter().map(|g| g.to_string()).collect();
        for name in &names {
            for (s, p, o) in rete.dump(Some(name)) {
                got.insert((s, p, o, Some(name.clone())));
            }
        }
        assert_eq!(
            got, expected,
            "quads lost or misfiled by the external build"
        );

        let out = crate::eval_query(
            &rete,
            "SELECT (COUNT(DISTINCT ?g) AS ?n) WHERE { GRAPH ?g { ?s ?p ?o } }",
        )
        .unwrap();
        match out {
            crate::QueryOutput::Select(_, rows) => assert_eq!(rows.len(), 1),
            other => panic!("expected select, got {other:?}"),
        }
    }

    // ---- `--text-index` in the memory-bounded build -------------------------

    /// Literal-rich quads: titles and provenance-style descriptions in the
    /// default graph (multi-byte words included), provenance snapshots in
    /// named graphs, on top of the fedlex shape (its `"Artikel …"@de` literals
    /// live in named graphs too).
    fn text_quads() -> Vec<RawQuad> {
        let mut q = fedlex_shaped_quads(40, 9);
        for i in 0..600usize {
            let s = format!("<http://ex/w{}>", i % 150);
            q.push((
                s.clone(),
                "<http://purl.org/dc/terms/title>".to_string(),
                format!("\"Glucose Études {} on ünïcode café\"@en", i % 37),
                None,
            ));
            q.push((
                s.clone(),
                "<http://purl.org/dc/terms/description>".to_string(),
                format!("\"The entity {} has been modified.\"", i % 150),
                None,
            ));
            q.push((
                format!("<http://ex/w{}/prov/se/1>", i % 150),
                "<http://purl.org/dc/terms/description>".to_string(),
                format!("\"The entity has been created ({i}).\""),
                Some(format!("<http://ex/w{}/prov/>", i % 150)),
            ));
        }
        q
    }

    fn text_scopes() -> Vec<crate::text_index::TextIndexScope> {
        use crate::text_index::{TextIndexGraphs, TextIndexScope};
        let none: [&str; 0] = [];
        vec![
            TextIndexScope::default(),
            TextIndexScope::from_options(TextIndexGraphs::All, none, none).unwrap(),
            TextIndexScope::from_options(
                TextIndexGraphs::Default,
                none,
                ["http://purl.org/dc/terms/description"],
            )
            .unwrap(),
            TextIndexScope::from_options(
                TextIndexGraphs::All,
                ["http://purl.org/dc/terms/description"],
                none,
            )
            .unwrap(),
        ]
    }

    fn build_reference_text(
        quads: &[RawQuad],
        scope: &crate::text_index::TextIndexScope,
    ) -> Vec<u8> {
        let (bytes, _stats) = ingest::assemble_dataset_streaming_with_layout(
            |visit: &mut dyn FnMut(RawQuad)| {
                for q in quads.iter().cloned() {
                    visit(q);
                }
                Ok(())
            },
            false,
            scope.clone(),
            None,
            crate::PyramidAlgo::Louvain,
            crate::index::PermSet::ALL,
            GraphLayout::Quad,
            |_, _, _| Vec::new(),
        )
        .unwrap();
        as_v06(bytes)
    }

    /// The public entry point with a text index; `split` forces many chunks and
    /// many text sort runs on a small graph.
    fn build_ext_text(
        quads: &[RawQuad],
        scope: &crate::text_index::TextIndexScope,
        split: Option<(u64, usize)>,
    ) -> Vec<u8> {
        static SEQ: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
        let seq = SEQ.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        let dir =
            std::env::temp_dir().join(format!("rete-extbuild-text-{}-{seq}", std::process::id()));
        std::fs::create_dir_all(&dir).unwrap();
        let out = dir.join("out.rete");
        TEST_SPLIT.with(|c| c.set(split));
        let res = build_external(
            |visit| {
                for q in quads.iter().cloned() {
                    visit(q)?;
                }
                Ok(())
            },
            &out,
            ExternalBuildOptions {
                memory_budget: 0,
                tmp_dir: Some(dir.clone()),
                text_index: Some(scope.clone()),
                ..Default::default()
            },
        );
        TEST_SPLIT.with(|c| c.set(None));
        res.unwrap();
        let bytes = std::fs::read(&out).unwrap();
        // The spill directory is gone; only the output remains.
        let left: Vec<_> = std::fs::read_dir(&dir)
            .unwrap()
            .map(|e| e.unwrap().file_name())
            .collect();
        assert_eq!(left, [std::ffi::OsString::from("out.rete")]);
        let _ = std::fs::remove_dir_all(&dir);
        bytes
    }

    /// B1's requirement: with `--text-index` the memory-bounded build writes the
    /// same TEXT_INDEX section as the in-RAM build — and, the pyramid being off
    /// on both sides, the same file — in one chunk and in many, for every scope.
    #[test]
    fn external_text_index_is_byte_identical_to_in_ram() {
        let quads = text_quads();
        for scope in text_scopes() {
            let reference = build_reference_text(&quads, &scope);
            let rh = crate::Header::from_bytes(&reference).unwrap();
            assert!(
                rh.text_index_len > 0,
                "{scope:?}: the fixture must index something"
            );
            assert_eq!(rh.text_index_partial, scope.restricts_predicates());
            assert_eq!(
                rh.text_index_covers_named_graphs(),
                scope.covers_named_graphs()
            );
            for split in [None, Some((4 * 1024, 1024))] {
                let bytes = build_ext_text(&quads, &scope, split);
                assert_eq!(
                    bytes, reference,
                    "{scope:?}, split {split:?}: external build must be byte-identical"
                );
            }
        }
    }

    /// The external file's text index answers searches, verifies, and carries
    /// its scope: the provenance text is found only when named graphs are in
    /// scope, the descriptions only when their predicate is.
    #[test]
    fn external_text_index_is_searchable_in_its_scope() {
        let quads = text_quads();
        let scopes = text_scopes();
        let search = |scope: &crate::text_index::TextIndexScope, word: &str| -> usize {
            let bytes = build_ext_text(&quads, scope, Some((4 * 1024, 1024)));
            assert!(crate::verify(&bytes).unwrap());
            let rete = crate::Rete::open(&bytes).unwrap();
            assert!(rete.has_text_index());
            assert_eq!(rete.text_index().unwrap().scope(), scope);
            rete.text_search(&[word], None, 0).len()
        };
        // "glucose": 150 titled works, in every scope that keeps titles.
        assert_eq!(search(&scopes[0], "glucose"), 150);
        assert_eq!(search(&scopes[2], "glucose"), 150);
        assert_eq!(
            search(&scopes[3], "glucose"),
            0,
            "an include-only scope drops titles"
        );
        // "modified": the default-graph descriptions.
        assert_eq!(search(&scopes[0], "modified"), 150);
        assert_eq!(search(&scopes[2], "modified"), 0, "excluded predicate");
        // "created": only in the named provenance graphs.
        assert_eq!(
            search(&scopes[0], "created"),
            0,
            "named graphs not in scope"
        );
        assert_eq!(search(&scopes[1], "created"), 150);
        assert_eq!(search(&scopes[3], "created"), 150);
        // A multi-byte word survives the spill's byte-level front coding.
        assert_eq!(search(&scopes[0], "études"), 150);
    }

    /// No literal in scope: no section at all, as the in-RAM build writes none.
    #[test]
    fn external_text_index_with_nothing_in_scope_writes_no_section() {
        let quads = test_quads(400);
        let none: [&str; 0] = [];
        let scope = crate::text_index::TextIndexScope::from_options(
            crate::text_index::TextIndexGraphs::Default,
            ["http://ex/never"],
            none,
        )
        .unwrap();
        let reference = build_reference_text(&quads, &scope);
        let bytes = build_ext_text(&quads, &scope, None);
        assert_eq!(bytes, reference);
        assert_eq!(crate::Header::from_bytes(&bytes).unwrap().text_index_len, 0);
    }

    // ---- quad index (`GraphLayout::Quad`) --------------------------------

    fn build_reference_quad(quads: Vec<RawQuad>, perms: crate::index::PermSet) -> Vec<u8> {
        let (bytes, _) = ingest::assemble_dataset_streaming_with_layout(
            move |visit: &mut dyn FnMut(RawQuad)| {
                for q in quads.iter().cloned() {
                    visit(q);
                }
                Ok(())
            },
            false,
            false,
            None,
            crate::PyramidAlgo::Louvain,
            perms,
            GraphLayout::Quad,
            |_, _, _| Vec::new(),
        )
        .unwrap();
        as_v06(bytes)
    }

    fn build_ext_quad(quads: Vec<RawQuad>, perms: crate::index::PermSet) -> (Vec<u8>, BuildStats) {
        static SEQ: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
        let seq = SEQ.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        let dir =
            std::env::temp_dir().join(format!("rete-extbuild-quad-{}-{seq}", std::process::id()));
        std::fs::create_dir_all(&dir).unwrap();
        let out = dir.join("out.rete");
        let stats = build_external(
            |visit| {
                for q in quads.iter().cloned() {
                    visit(q)?;
                }
                Ok(())
            },
            &out,
            ExternalBuildOptions {
                memory_budget: 0,
                tmp_dir: Some(dir.clone()),
                perms,
                graphs: GraphLayout::Quad,
                ..Default::default()
            },
        )
        .unwrap();
        let bytes = std::fs::read(&out).unwrap();
        let _ = std::fs::remove_dir_all(&dir);
        (bytes, stats)
    }

    /// The external quad build writes the in-RAM build's bytes, for both
    /// permutation sets — the shared `QuadTiler` is what guarantees it.
    #[test]
    fn quad_external_build_is_byte_identical() {
        for perms in [crate::index::PermSet::ALL, crate::index::PermSet::CORE] {
            let quads = fedlex_shaped_quads(60, 9);
            let reference = build_reference_quad(quads.clone(), perms);
            let (bytes, stats) = build_ext_quad(quads.clone(), perms);
            assert_eq!(bytes, reference, "quad external build, {perms:?}");
            let header = crate::Header::from_bytes(&bytes).unwrap();
            assert!(header.has_quad_index());
            assert_eq!(stats.statements as u64, header.quad_count);
            let rete = crate::Rete::open(&bytes).unwrap();
            assert_eq!(rete.named_graph_count(), stats.named_graphs);
            assert!(crate::file::verify(&bytes).unwrap());
        }
    }

    /// Many chunks and many sort runs: graph names first seen in different
    /// chunks must merge into one dictionary id, and runs must merge back into
    /// the same tiles.
    #[test]
    fn multi_chunk_multi_run_quad_build_is_byte_identical() {
        let quads = fedlex_shaped_quads(80, 7);
        let reference = build_reference_quad(quads.clone(), crate::index::PermSet::ALL);

        let dir = std::env::temp_dir().join(format!("rete-extbuild-mcq-{}", std::process::id()));
        std::fs::create_dir_all(&dir).unwrap();
        let out = dir.join("out.rete");
        let tmp = TmpDir::create(&dir).unwrap();
        let mut chunker = Chunker::new(&tmp, 4 * 1024, GraphLayout::Quad);
        for q in quads.iter().cloned() {
            chunker.push(q).unwrap();
        }
        let chunks = chunker.finish().unwrap();
        assert!(
            chunks.len() >= 8,
            "expected many chunks, got {}",
            chunks.len()
        );
        let (merged, remaps) = merge_dictionaries(&tmp, &chunks).unwrap();
        // 64-quad runs: many runs per order, chained and merged back.
        let (tri_runs, quad_runs) = remap_chunks(
            &tmp,
            &chunks,
            &remaps,
            true,
            crate::index::IndexPermutation::Spo,
            256,
            64,
            None,
            None,
            None,
        )
        .unwrap();
        assert!(quad_runs.len() >= 4, "{} quad run(s)", quad_runs.len());
        let codec = crate::file::writer_codec();
        let perms = crate::index::PermSet::ALL;
        let (sections, default_count) = index_permutations(
            &tmp,
            tri_runs,
            perms,
            256,
            codec,
            "",
            &mut |_, _| Ok(()),
            &mut |_, _| {},
        )
        .unwrap();
        let quad = build_quad_section(&tmp, quad_runs, 64, perms, codec).unwrap();
        write_final_file(
            &out,
            &[],
            &[],
            &merged,
            &sections,
            None,
            None,
            Some(&quad),
            default_count + quad.quad_count,
            codec,
            perms,
            true,
        )
        .unwrap();
        let bytes = std::fs::read(&out).unwrap();
        assert_eq!(bytes, reference, "multi-chunk, multi-run quad build");
        drop(tmp);
        let _ = std::fs::remove_dir_all(&dir);
    }

    /// Frequent object-only terms with ties at every count, named graphs whose
    /// names are also objects, and the most frequent object of all, which is
    /// a subject in one statement at the very end.
    fn frequent_object_quads() -> Vec<RawQuad> {
        let rdf_type = "<http://www.w3.org/1999/02/22-rdf-syntax-ns#type>";
        let mut q = test_quads(1_500);
        for i in 0..3_000usize {
            let s = format!("<http://ex/item/{i:05}>");
            q.push((
                s.clone(),
                rdf_type.into(),
                format!("<http://vocab/C{}>", i % 9),
                None,
            ));
            // forty statuses of 75 statements each: the cut falls in a tie
            q.push((
                s.clone(),
                "<http://ex/status>".into(),
                format!("\"st{}\"", i % 40),
                None,
            ));
            q.push((
                s.clone(),
                "<http://ex/doi>".into(),
                format!("<https://doi.org/10.1/x{i}>"),
                None,
            ));
            // a graph name is a term of the quad index (a subject), not
            // otherwise: there it is an object-only term like any other
            let g = format!("<http://ex/g{}>", i % 3);
            q.push((s, "<http://ex/inGraph>".into(), g.clone(), Some(g)));
        }
        for i in 0..4_000usize {
            q.push((
                format!("<http://ex/item/{:05}>", i % 3_000),
                "<http://ex/about>".into(),
                "<http://ex/hub>".into(),
                None,
            ));
        }
        q.push((
            "<http://ex/hub>".into(),
            "<http://ex/label>".into(),
            "\"hub\"".into(),
            None,
        ));
        q
    }

    /// G3: `--frequent-objects N` in the memory-bounded build. The chunks
    /// count, the walk before the merge picks the same `N` terms as the in-RAM
    /// builder — the same counts, duplicates included, the same tie-break —
    /// and the merge files them where it does: the file is the in-RAM
    /// build's, byte for byte, in one chunk or many, in either named-graph
    /// layout, for a cut inside a tie and for `N` past every candidate.
    #[test]
    fn frequent_objects_external_build_is_byte_identical() {
        use crate::write_opts::TEST_FREQUENT_OBJECTS;
        let quads = frequent_object_quads();
        let plain = build_reference(quads.clone());
        for n in [1usize, 9, 64, 100_000] {
            TEST_FREQUENT_OBJECTS.with(|c| c.set(Some(n)));
            let reference = build_reference(quads.clone());
            let (one_chunk, _) = build_ext(quads.clone(), 0);
            TEST_SPLIT.with(|c| c.set(Some((16 << 10, 1 << 16))));
            let collector = crate::build_trace::Collector::start();
            let (chunked, _) = build_ext(quads.clone(), 0);
            let chunks = collector.snapshot().counter("chunks").unwrap();
            drop(collector);
            let (quad_chunked, _) = build_ext_quad(quads.clone(), crate::index::PermSet::ALL);
            TEST_SPLIT.with(|c| c.set(None));
            let (quad_one, _) = build_ext_quad(quads.clone(), crate::index::PermSet::ALL);
            let quad_reference = build_reference_quad(quads.clone(), crate::index::PermSet::ALL);
            TEST_FREQUENT_OBJECTS.with(|c| c.set(None));

            assert!(chunks >= 8, "{chunks} chunk(s)");
            assert_ne!(reference, plain, "N = {n}: nothing was promoted");
            assert_eq!(one_chunk, reference, "N = {n}: one chunk");
            assert_eq!(chunked, reference, "N = {n}: {chunks} chunks");
            assert_eq!(quad_one, quad_reference, "N = {n}: quad index, one chunk");
            assert_eq!(quad_chunked, quad_reference, "N = {n}: quad index, chunked");

            let rete = crate::Rete::open(&chunked).unwrap();
            let dict = rete.dictionary();
            let shared = |t: &str| dict.object_id(t).unwrap() <= dict.shared_count();
            // The hub is a subject once: shared by role, never a candidate.
            assert!(shared("<http://ex/hub>"));
            if n >= 9 {
                assert!(
                    shared("<http://ex/g0>"),
                    "N = {n}: the busiest object-only term"
                );
            }
            if n == 64 {
                // three graphs, nine classes, forty statuses, then a tie
                assert!(shared("\"st39\"") && shared("<http://vocab/C8>"));
            }
        }
    }

    /// The quad layout needs the 3- or 6-permutation set (see `quad::check_perms`).
    #[test]
    fn quad_external_build_refuses_other_permutation_sets() {
        let dir = std::env::temp_dir().join(format!("rete-extbuild-q5-{}", std::process::id()));
        std::fs::create_dir_all(&dir).unwrap();
        let r = build_external(
            |visit| visit(("<a>".into(), "<p>".into(), "<b>".into(), Some("<g>".into()))),
            &dir.join("out.rete"),
            ExternalBuildOptions {
                tmp_dir: Some(dir.clone()),
                perms: crate::index::PermSet::from_bits(0b0001_1111).unwrap(),
                graphs: GraphLayout::Quad,
                ..Default::default()
            },
        );
        assert!(r.is_err());
        let _ = std::fs::remove_dir_all(&dir);
    }

    // ---- failure paths of the 0x06 write --------------------------------------

    /// A scratch directory and a build into it that can be made to fail.
    struct Failing {
        dir: PathBuf,
        out: PathBuf,
    }

    impl Failing {
        fn new(tag: &str) -> Self {
            let dir = spill_parent(tag);
            Failing {
                out: dir.join("out.rete"),
                dir,
            }
        }

        fn build(
            &self,
            opts: impl FnOnce(&mut ExternalBuildOptions),
        ) -> Result<BuildStats, ExtBuildError> {
            let quads = test_quads(400);
            let mut o = ExternalBuildOptions {
                memory_budget: 0,
                tmp_dir: Some(self.dir.clone()),
                ..Default::default()
            };
            opts(&mut o);
            build_external(
                |visit| {
                    for q in quads.iter().cloned() {
                        visit(q)?;
                    }
                    Ok(())
                },
                &self.out,
                o,
            )
        }

        /// File names directly in the directory, and the sub-directories.
        fn entries(&self) -> (Vec<String>, Vec<PathBuf>) {
            let mut files = Vec::new();
            let mut dirs = Vec::new();
            for e in std::fs::read_dir(&self.dir).unwrap().flatten() {
                if e.path().is_dir() {
                    dirs.push(e.path());
                } else {
                    files.push(e.file_name().to_string_lossy().into_owned());
                }
            }
            (files, dirs)
        }
    }

    impl Drop for Failing {
        fn drop(&mut self) {
            let _ = std::fs::remove_dir_all(&self.dir);
        }
    }

    /// A card step that fails: the sections are done, so the spill is kept
    /// whole, there is no output, and no working file is left behind.
    #[test]
    fn a_failing_card_step_keeps_the_whole_spill() {
        let f = Failing::new("cardfail");
        let r = f.build(|o| {
            o.card_from_image = Some(Box::new(|_, _| Err(ExtBuildError::Internal("card boom"))));
        });
        assert!(matches!(r, Err(ExtBuildError::Internal("card boom"))));
        let (files, dirs) = f.entries();
        assert!(files.is_empty(), "no output, no .partial: {files:?}");
        assert_eq!(dirs.len(), 1, "the spill directory is kept");
        let names: Vec<String> = std::fs::read_dir(&dirs[0])
            .unwrap()
            .flatten()
            .map(|e| e.file_name().to_string_lossy().into_owned())
            .collect();
        assert!(
            names.iter().any(|n| n == "g.shared.sec") && names.len() > 8,
            "dictionary and index files are all there: {names:?}"
        );
    }

    /// The first flush fails (injected below the buffer, as a full disk shows
    /// it): the first spill file has been copied into the buffer but must not
    /// have been deleted, so the whole spill is kept; the working file is not.
    #[test]
    fn a_failing_first_flush_keeps_the_spill_whole() {
        let f = Failing::new("iofail-early");
        TEST_FAIL_AFTER.with(|c| c.set(Some(0)));
        let r = f.build(|_| {});
        TEST_FAIL_AFTER.with(|c| c.set(None));
        let e = r.unwrap_err().to_string();
        assert!(e.contains("injected"), "{e}");
        let (files, dirs) = f.entries();
        assert!(files.is_empty(), "no output and no working file: {files:?}");
        assert_eq!(dirs.len(), 1, "the whole spill is kept");
        let names: Vec<String> = std::fs::read_dir(&dirs[0])
            .unwrap()
            .flatten()
            .map(|e| e.file_name().to_string_lossy().into_owned())
            .collect();
        assert!(
            ["g.shared.sec", "g.subj.sec", "g.obj.sec", "g.pred.sec"]
                .iter()
                .all(|n| names.iter().any(|m| m == n)),
            "every dictionary section is still there, the predicates one included: {names:?}"
        );
        assert!(!e.contains("io: io:"), "{e}");
    }

    /// A write that fails part way (a full disk, injected): an error, no
    /// output under the name, no working file, and the half-consumed spill is
    /// removed rather than left as a trap.
    #[test]
    fn an_io_error_mid_write_leaves_nothing_behind() {
        let size = {
            let g = Failing::new("iofail-size");
            g.build(|_| {}).unwrap();
            std::fs::metadata(&g.out).unwrap().len()
        };
        let f = Failing::new("iofail");
        TEST_FAIL_AFTER.with(|c| c.set(Some(size / 2)));
        let r = f.build(|_| {});
        TEST_FAIL_AFTER.with(|c| c.set(None));
        let e = r.unwrap_err().to_string();
        assert!(e.contains("injected"), "{e}");
        let (files, dirs) = f.entries();
        assert!(files.is_empty(), "{files:?}");
        assert!(dirs.is_empty(), "{dirs:?}");
    }

    /// A panic in the write removes the working file (and the spill).
    #[test]
    fn a_panic_in_the_write_removes_the_working_file() {
        let f = Failing::new("panicfail");
        TEST_PANIC_AFTER.with(|c| c.set(Some(0)));
        let r = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| f.build(|_| {})));
        TEST_PANIC_AFTER.with(|c| c.set(None));
        assert!(r.is_err(), "the injected panic propagates");
        let (files, dirs) = f.entries();
        assert!(files.is_empty(), "{files:?}");
        assert!(dirs.is_empty(), "{dirs:?}");
    }

    /// A rename that fails keeps the finished file and names it.
    #[test]
    fn a_failed_rename_keeps_the_finished_file_and_says_where() {
        let f = Failing::new("renamefail");
        let out = f.out.clone();
        let r = f.build(|o| {
            // After the probe, before the write: the output name becomes a
            // directory, which a rename of a file onto it refuses.
            o.card_from_image = Some(Box::new(move |_, _| {
                std::fs::create_dir(&out).unwrap();
                Ok((Vec::new(), Vec::new()))
            }));
        });
        let e = r.unwrap_err().to_string();
        assert!(e.contains("complete at") && e.contains(".partial"), "{e}");
        let (files, _) = f.entries();
        let kept: Vec<&String> = files.iter().filter(|n| n.ends_with(".partial")).collect();
        assert_eq!(kept.len(), 1, "{files:?}");
        let bytes = std::fs::read(f.dir.join(kept[0])).unwrap();
        assert!(crate::v06::open::is_v06(&bytes), "a whole 0x06 file");
        assert!(crate::verify(&bytes).unwrap());
    }

    /// An output that is a directory is refused before anything is built.
    #[test]
    fn an_output_that_is_a_directory_is_refused_up_front() {
        let f = Failing::new("outdir");
        std::fs::create_dir(&f.out).unwrap();
        let e = f.build(|_| {}).unwrap_err().to_string();
        assert!(e.contains("is a directory"), "{e}");
        // Nothing was spilled: the build stopped before it began.
        let (_, dirs) = f.entries();
        assert_eq!(dirs, vec![f.out.clone()]);
    }

    /// The per-graph layout is a plain refusal, not an internal error.
    #[test]
    fn the_per_graph_layout_is_refused_by_name() {
        let f = Failing::new("pergraph");
        match f.build(|o| o.graphs = GraphLayout::PerGraph) {
            Err(ExtBuildError::Unsupported(m)) => assert!(m.contains("quad index"), "{m}"),
            other => panic!("expected a refusal, got {other:?}"),
        }
    }

    /// Two builds into one directory get different working names.
    #[test]
    fn working_names_are_unique() {
        let dir = spill_parent("partial-names");
        let out = dir.join("o.rete");
        let a = PartialFile::create(&out).unwrap();
        let b = PartialFile::create(&out).unwrap();
        assert_ne!(a.path, b.path);
        drop(a);
        drop(b);
        assert_eq!(std::fs::read_dir(&dir).unwrap().count(), 0);
        let _ = std::fs::remove_dir_all(&dir);
    }

    /// Every range the copy announces is consumed exactly once, whatever the
    /// order: what lets each spill file go as soon as it is read through.
    #[test]
    fn the_copy_consumes_every_range_it_announces() {
        let quads = fedlex_shaped_quads(30, 6);
        let opts = ingest::AssembleOptions {
            with_pyramid: false,
            label_index: true,
            schema: true,
            graph_layout: GraphLayout::Quad,
            ..ingest::AssembleOptions::sectioned()
        };
        let image = ingest::assemble_dataset_streaming_opts(
            |visit: &mut dyn FnMut(RawQuad)| {
                for q in quads.iter().cloned() {
                    visit(q);
                }
                Ok(())
            },
            &opts,
            |_, _, _| Vec::new(),
        )
        .unwrap()
        .0;
        let r = crate::reader::SliceReader::new(&image);
        let parts = crate::v06::parts::from_v05(&r).unwrap();
        let (mut needed, mut consumed) = (Vec::new(), Vec::new());
        let mut out = std::io::Cursor::new(Vec::new());
        crate::v06::write::assemble_with(
            &parts,
            &r,
            &mut out,
            crate::v06::write::WriteOpts::default(),
            &mut crate::v06::write::Hooks {
                copied: &mut |_| {},
                progress: &mut |_, _| {},
                needed: &mut |o, l| needed.push((o, l)),
                consumed: &mut |o, l| consumed.push((o, l)),
                settle: &mut |_| Ok(()),
            },
        )
        .unwrap();
        assert!(needed.len() > 10);
        let mut sorted = consumed.clone();
        sorted.sort_unstable();
        let mut n = needed.clone();
        n.sort_unstable();
        assert_eq!(n, sorted);
        assert_ne!(
            consumed, sorted,
            "the dictionary is copied predicates first: not in input order"
        );
    }
}