//! A state: a dependency-closed set of changes, applied to a tree graph and
//! per-node file graphs (spec §8.3), with materialization (spec §9.5, §11).

use crate::cbor::{self, Value};
use crate::graph::{AtomKey, DepOracle, FileGraph, Materialized, Vertex};
use crate::hash::KeyId;
use crate::hash::{ChangeHash, ObjectRef};
use crate::ids::NodeId;
use crate::objects::{
    self as o, AtomRef, Change, ChangeKind, Edit, Hlc, Mode, NodeKind, ObjectError, TreeOp,
};
use crate::tree::{TreeGraph, TreeView};
use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet};
use std::sync::Arc;

/// Checkpoint payload version 2 (spec §11): file graphs by reference.
pub const CHECKPOINT_INDEX_VERSION: u64 = 2;

/// A file graph encoded in fewer bytes stays inline in a version 2 index.
pub const INLINE_GRAPH_LIMIT: usize = 2048;

#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
pub enum StateError {
    #[error("change {1} depends on {0}, which is not in the state")]
    MissingDep(ChangeHash, ChangeHash),
}

/// One `Replace` edit: `(change, old blob, new blob)`.
type ReplaceRecord = (ChangeHash, Option<ObjectRef>, Option<ObjectRef>);

/// What a state remembers about a change: everything but its operations,
/// which live in the graphs once applied (and in the store).
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct ChangeMeta {
    pub change_id: crate::ids::ChangeId,
    pub deps: Vec<ChangeHash>,
    pub author: KeyId,
    pub hlc: Hlc,
    pub message: String,
    pub kind: ChangeKind,
}

/// The log's order of changes: newest first by clock, then by hash, so
/// it is total.
pub fn log_order(a: (&ChangeHash, &Hlc), b: (&ChangeHash, &Hlc)) -> std::cmp::Ordering {
    b.1.cmp(a.1).then(b.0.cmp(a.0))
}

impl ChangeMeta {
    /// In the flat layout of local indexes and caches ([`crate::flat`]).
    pub(crate) fn write_flat(&self, w: &mut crate::flat::Writer) {
        w.fixed(&self.change_id.0);
        w.count(self.deps.len());
        for d in &self.deps {
            w.object(d);
        }
        w.object(&self.author);
        w.u64(self.hlc.wall_ms);
        w.u32(self.hlc.logical);
        w.bytes(self.message.as_bytes());
        w.u8(self.kind.tag() as u8);
    }

    pub(crate) fn read_flat(r: &mut crate::flat::Reader) -> Result<ChangeMeta, ObjectError> {
        let change_id = crate::ids::ChangeId(r.fixed::<16>()?);
        let mut deps = Vec::new();
        for _ in 0..r.count()? {
            deps.push(r.object()?);
        }
        Ok(ChangeMeta {
            change_id,
            deps,
            author: r.object()?,
            hlc: Hlc {
                wall_ms: r.u64()?,
                logical: r.u32()?,
            },
            message: r.text()?,
            kind: ChangeKind::from_tag(r.u8()? as u64)?,
        })
    }

    pub fn of(c: &Change) -> ChangeMeta {
        ChangeMeta {
            change_id: c.change_id,
            deps: c.deps.clone(),
            author: c.author,
            hlc: c.hlc,
            message: c.message.clone(),
            kind: c.kind,
        }
    }
}

#[derive(Default)]
pub struct State {
    /// Shared between clones, like `files` and `deferred`: a state kept
    /// beside another that differs by a few changes shares the rest.
    changes: crate::fasthash::FastMap<ChangeHash, Arc<ChangeMeta>>,
    tree: TreeGraph,
    /// Shared between clones: a clone copies a graph only when an edit
    /// changes it (`Arc::make_mut`), so keeping several states that differ
    /// by a few changes costs little more than one.
    files: crate::fasthash::FastMap<NodeId, Arc<FileGraph>>,
    replaces: HashMap<NodeId, Vec<ReplaceRecord>>,
    /// Memoised transitive dependencies, bounded by the hashes they hold
    /// (each set is the whole history below a change, so keeping one for
    /// every change asked about grew with the square of history: #25). A
    /// mutex, not a `RefCell`, so a loaded state can be shared across
    /// threads by a host that caches it.
    closure: std::sync::RwLock<ClosureMemo>,
    /// Each change's level: 0 with no dependencies, else one more than its
    /// highest dependency. A change can only depend on changes of lower
    /// level, which bounds the walk [`DepOracle::depends_on`] does.
    levels: std::sync::RwLock<crate::fasthash::FastMap<ChangeHash, u32>>,
    /// `None` for a full state. For a history-only state (loaded for its
    /// changes and tree), the file graphs not built yet, by node: `files`
    /// holds only those loaded with [`State::load_graphs`].
    deferred: Option<crate::fasthash::FastMap<NodeId, Arc<Deferred>>>,
    /// The tree as readers see it ([`State::view`]), once worked out for
    /// a tree with name conflicts; cleared by every change added.
    unified_view: std::sync::OnceLock<TreeView>,
}

use crate::objects::TextRef;

/// A file graph a history-only state has not built: its base from a
/// version 2 checkpoint, if any, and the edits of later changes in the
/// order they were applied (spec §11). Building it later gives the same
/// graph, since graph construction does not depend on order (§8).
///
/// Read from a state index, it stays in the index, unread, until first
/// used: a command that builds a few files reads and checks only their
/// histories (oxygen#94).
#[derive(Clone, Default)]
struct Deferred {
    /// The history: `raw` read and decoded, with `more` after it. An
    /// error when `raw` does not match its hash.
    body: std::sync::OnceLock<Result<DeferredBody, String>>,
    /// The history as an index holds it, not read yet.
    raw: Option<RawHistory>,
    /// Changes added since the index was read, kept beside `raw`, so that
    /// adding one neither reads nor decodes the history.
    more: Vec<(ChangeHash, Vec<Edit>)>,
    more_texts: Vec<TextRef>,
    /// Its graph key, once worked out (or read from an index); reset by
    /// any edit added.
    key: std::sync::OnceLock<[u8; 32]>,
}

/// Where a state index's file histories are: the index file itself, read
/// on first use, or bytes in memory.
#[derive(Clone)]
pub enum Histories {
    Bytes(Arc<Vec<u8>>),
    /// The file, where its histories start, and how many bytes they take.
    File(Arc<std::fs::File>, u64, u64),
}

impl Histories {
    fn len(&self) -> u64 {
        match self {
            Histories::Bytes(b) => b.len() as u64,
            Histories::File(_, _, len) => *len,
        }
    }

    fn read(&self, at: u64, len: usize) -> std::io::Result<Vec<u8>> {
        match self {
            Histories::Bytes(b) => Ok(b[at as usize..at as usize + len].to_vec()),
            Histories::File(f, start, _) => {
                let mut buf = vec![0u8; len];
                read_exact_at(f, &mut buf, start + at)?;
                Ok(buf)
            }
        }
    }
}

#[cfg(unix)]
fn read_exact_at(f: &std::fs::File, buf: &mut [u8], at: u64) -> std::io::Result<()> {
    std::os::unix::fs::FileExt::read_exact_at(f, buf, at)
}

#[cfg(windows)]
fn read_exact_at(f: &std::fs::File, mut buf: &mut [u8], mut at: u64) -> std::io::Result<()> {
    while !buf.is_empty() {
        match std::os::windows::fs::FileExt::seek_read(f, buf, at)? {
            0 => return Err(std::io::ErrorKind::UnexpectedEof.into()),
            n => {
                buf = &mut buf[n..];
                at += n as u64;
            }
        }
    }
    Ok(())
}

/// One file's history in a state index: where it is and its hash.
#[derive(Clone)]
struct RawHistory {
    from: Histories,
    at: u64,
    len: usize,
    sum: [u8; 32],
}

impl RawHistory {
    /// The history's bytes, checked against the hash the index holds.
    fn bytes(&self) -> Result<Vec<u8>, ObjectError> {
        let damaged = |why: String| ObjectError::DamagedIndex(why);
        let b = self
            .from
            .read(self.at, self.len)
            .map_err(|e| damaged(format!("a file's history cannot be read: {e}")))?;
        if *blake3::hash(&b).as_bytes() != self.sum {
            return Err(damaged("a file's history does not match its hash".into()));
        }
        Ok(b)
    }
}

/// The closures kept, and how many hashes they hold in all.
#[derive(Default)]
struct ClosureMemo {
    sets: HashMap<ChangeHash, Arc<HashSet<ChangeHash>>>,
    held: usize,
}

/// The most hashes the closure memo holds before it starts again: a few
/// hundred megabytes at most.
const CLOSURE_MEMO_HASHES: usize = 4_000_000;

#[derive(Clone, Default)]
struct DeferredBody {
    base: Option<DeferredBase>,
    edits: Vec<(ChangeHash, Vec<Edit>)>,
    /// The text blob and line count of each insert in `edits` left without
    /// its lines, in order: read when the graph is built.
    texts: Vec<TextRef>,
}

impl Deferred {
    fn from_body(body: DeferredBody) -> Deferred {
        Deferred {
            body: std::sync::OnceLock::from(Ok(body)),
            ..Deferred::default()
        }
    }

    /// The history, read and checked on first use.
    fn body(&self) -> Result<&DeferredBody, ObjectError> {
        self.body
            .get_or_init(|| {
                let mut b = match &self.raw {
                    // Matching its hash, it is what `write_index` wrote: it
                    // decodes.
                    Some(raw) => raw
                        .bytes()
                        .and_then(|b| State::read_deferred_body(&b))
                        .map_err(|e| e.to_string())?,
                    None => DeferredBody::default(),
                };
                b.edits.extend(self.more.iter().cloned());
                b.texts.extend(self.more_texts.iter().cloned());
                Ok(b)
            })
            .as_ref()
            .map_err(|e| ObjectError::DamagedIndex(e.clone()))
    }

    /// Add a change's edits to the history: beside the index's encoding
    /// while it is unread, so that nothing is read or decoded to add them.
    fn add(&mut self, h: ChangeHash, edits: Vec<Edit>, texts: Vec<TextRef>) {
        if self.raw.is_some() && !matches!(self.body.get(), Some(Ok(_))) {
            self.more.push((h, edits));
            self.more_texts.extend(texts);
            // A body found damaged stays so: what was added cannot mend it.
            return;
        }
        // No encoding to read, or decoded already: this cannot fail.
        let b = self.body_mut().expect("a history in memory");
        b.edits.push((h, edits));
        b.texts.extend(texts);
    }

    fn body_mut(&mut self) -> Result<&mut DeferredBody, ObjectError> {
        self.body()?;
        self.raw = None;
        self.more.clear();
        self.more_texts.clear();
        Ok(self
            .body
            .get_mut()
            .and_then(|b| b.as_mut().ok())
            .expect("decoded above"))
    }

    fn into_body(mut self) -> Result<DeferredBody, ObjectError> {
        self.body()?;
        Ok(self
            .body
            .take()
            .and_then(|b| b.ok())
            .expect("decoded above"))
    }
}

#[derive(Clone)]
enum DeferredBase {
    Inline(Value),
    Blob(ObjectRef),
}

/// A local index of a history-only state (oxygen#41), in the flat layout
/// of [`crate::flat`]: its change metadata, tree (with its view), replace
/// records and every file's deferred history with its graph key, so a
/// command reads it rather than replaying history. Derived only: written
/// from a state loaded the long way, and read back as the same state.
impl State {
    const INDEX_MAGIC: &'static [u8] = b"oxy state index 4\n";

    /// The index of this state, or `None` when it has anything an index
    /// does not hold: a built graph, or text in memory (a change added
    /// whole rather than as its skeleton).
    pub fn write_index(&self) -> Option<Vec<u8>> {
        use crate::flat::Writer;
        let deferred = self.deferred.as_ref()?;
        if !self.files.is_empty() {
            return None;
        }
        // The changes first, as a block a reader can take on its own (see
        // [`State::read_index_changes`]).
        let mut w = Writer::default();
        w.fixed(Self::INDEX_MAGIC);
        // In log order, newest first (oxygen#94): the log takes them as
        // they come, with no map and no sort.
        let mut changes: Vec<(&ChangeHash, &Arc<ChangeMeta>)> = self.changes.iter().collect();
        changes.sort_unstable_by(|a, b| log_order((a.0, &a.1.hlc), (b.0, &b.1.hlc)));
        w.count(changes.len());
        for (h, m) in changes {
            w.object(h);
            m.write_flat(&mut w);
        }
        let changes_block = std::mem::take(&mut w);
        w.bytes(&changes_block.0);
        // The tree and the files' histories as blocks of their own, so a
        // reader decodes them side by side.
        let mut tw = Writer::default();
        self.tree.write_flat(&mut tw);
        w.bytes(&tw.0);
        let mut replaces: Vec<(&NodeId, &Vec<ReplaceRecord>)> = self.replaces.iter().collect();
        replaces.sort_by_key(|(n, _)| **n);
        w.count(replaces.len());
        for (n, recs) in replaces {
            w.fixed(&n.0);
            w.count(recs.len());
            // In any order: which record is live does not depend on it.
            let mut recs: Vec<&ReplaceRecord> = recs.iter().collect();
            recs.sort();
            for (c, old, new) in recs {
                w.object(c);
                w.opt_object(old);
                w.opt_object(new);
            }
        }
        let mut front = w;
        // The table: per file its key and where its history is, with the
        // history's hash, which is checked when it is read (oxygen#94).
        let mut table = Writer::default();
        let mut histories: Vec<u8> = Vec::new();
        let mut nodes: Vec<(&NodeId, &Arc<Deferred>)> = deferred.iter().collect();
        nodes.sort_by_key(|(n, _)| **n);
        table.count(nodes.len());
        for (n, d) in nodes {
            let key = match d.key.get() {
                Some(k) => *k,
                None => {
                    let mut h = blake3::Hasher::new();
                    h.update(b"oxy graph key 1\0");
                    h.update(&n.0);
                    let k = d.graph_key(h).ok()?;
                    let _ = d.key.set(k);
                    k
                }
            };
            // Unchanged since read: its bytes as they are, checked.
            let bytes = match &d.raw {
                Some(raw) if d.more.is_empty() => raw.bytes().ok()?,
                _ => Self::write_deferred_body(d.body().ok()?)?,
            };
            table.fixed(&n.0);
            table.fixed(&key);
            table.u64(histories.len() as u64);
            table.u32(u32::try_from(bytes.len()).ok()?);
            table.fixed(blake3::hash(&bytes).as_bytes());
            histories.extend_from_slice(&bytes);
        }
        front.bytes(&table.0);
        let mut out = Writer::default();
        out.u64(front.0.len() as u64);
        out.fixed(&front.0);
        out.fixed(&histories);
        Some(out.0)
    }

    /// One file's history as an index encodes it; `None` when it holds
    /// text in memory, which an index does not.
    fn write_deferred_body(d: &DeferredBody) -> Option<Vec<u8>> {
        use crate::flat::Writer;
        let atom = |w: &mut Writer, a: &AtomRef| match a {
            AtomRef::Start => w.u8(0),
            AtomRef::End => w.u8(1),
            AtomRef::Atom { change, index } => {
                w.u8(2);
                w.object(change);
                w.u32(*index);
            }
        };
        let mut w = Writer::default();
        match &d.base {
            None => w.u8(0),
            Some(DeferredBase::Blob(r)) => {
                w.u8(1);
                w.object(r);
            }
            Some(DeferredBase::Inline(v)) => {
                w.u8(2);
                w.bytes(&cbor::encode(v));
            }
        }
        // Each change's edits with the texts of its inserts, sorted by
        // change: a graph does not depend on the order changes applied
        // in, and the index is then the same however history was read.
        let mut texts = d.texts.iter();
        let mut per_change: Vec<(&ChangeHash, &Vec<Edit>, Vec<&TextRef>)> = Vec::new();
        for (c, edits) in &d.edits {
            let n = edits
                .iter()
                .filter(|e| matches!(e, Edit::Insert { lines, .. } if lines.is_empty()))
                .count();
            per_change.push((c, edits, texts.by_ref().take(n).collect()));
        }
        per_change.sort_by_key(|(c, _, _)| **c);
        w.count(per_change.len());
        for (c, edits, _) in &per_change {
            w.object(c);
            w.count(edits.len());
            for e in edits.iter() {
                match e {
                    Edit::Insert {
                        node,
                        after,
                        before,
                        lines,
                    } => {
                        if !lines.is_empty() {
                            return None;
                        }
                        w.u8(0);
                        w.fixed(&node.0);
                        atom(&mut w, after);
                        atom(&mut w, before);
                    }
                    Edit::Delete { node, atoms } => {
                        w.u8(1);
                        w.fixed(&node.0);
                        w.count(atoms.len());
                        for r in atoms {
                            w.object(&r.change);
                            w.u32(r.from);
                            w.u32(r.to);
                        }
                    }
                    Edit::Replace { .. } => return None,
                }
            }
        }
        let texts: Vec<&TextRef> = per_change
            .iter()
            .flat_map(|(_, _, t)| t.iter().copied())
            .collect();
        w.count(texts.len());
        for t in texts {
            match t {
                TextRef::Blob(r, count, form) => {
                    w.u8(0);
                    w.object(r);
                    w.u32(*count);
                    w.u8(match form {
                        crate::objects::TextForm::Cbor => 0,
                        crate::objects::TextForm::Raw => 1,
                        crate::objects::TextForm::CborFallback => 2,
                    });
                }
                TextRef::Inline { change, start, end } => {
                    w.u8(1);
                    w.object(change);
                    w.u32(*start);
                    w.u32(*end);
                }
            }
        }
        Some(w.0)
    }

    /// The state an index holds (see [`State::write_index`]).
    pub fn read_index(bytes: &[u8]) -> Result<State, ObjectError> {
        let bad = || ObjectError::Schema("state index: truncated".into());
        let front_len =
            usize::try_from(crate::flat::Reader::new(bytes).u64()?).map_err(|_| bad())?;
        let front = bytes.get(8..8 + front_len).ok_or_else(bad)?;
        let histories = Histories::Bytes(Arc::new(bytes[8 + front_len..].to_vec()));
        Self::read_index_front(front, histories)
    }

    /// An index's front, everything but the files' histories, with where
    /// those are: each is read, and checked against the hash the front
    /// holds, only when its graph is built (oxygen#94).
    pub fn read_index_front(bytes: &[u8], histories: Histories) -> Result<State, ObjectError> {
        use crate::flat::Reader;
        let bad = |what: &str| ObjectError::Schema(format!("state index: {what}"));
        let mut r = Reader::new(bytes);
        let mut st = State::new_history_only();
        {
            let _t = crate::timing::phase("index: changes");
            st.changes = Self::read_index_changes(r.bytes()?)?;
        }
        let tree_bytes = r.bytes()?;
        for _ in 0..r.count()? {
            let n = NodeId(r.fixed::<16>()?);
            let mut recs = Vec::new();
            for _ in 0..r.count()? {
                recs.push((r.object()?, r.opt_object()?, r.opt_object()?));
            }
            st.replaces.insert(n, recs);
        }
        let deferred_bytes = r.bytes()?;
        if !r.at_end() {
            return Err(bad("trailing bytes"));
        }
        let (tree, deferred) = rayon::join(
            || TreeGraph::read_flat(&mut Reader::new(tree_bytes)),
            || Self::read_deferred(deferred_bytes, &histories),
        );
        st.tree = tree?;
        st.deferred = Some(deferred?);
        Ok(st)
    }

    /// The changes of a state index: its first block, which a reader can
    /// take without the rest.
    pub fn read_index_changes(
        bytes: &[u8],
    ) -> Result<crate::fasthash::FastMap<ChangeHash, Arc<ChangeMeta>>, ObjectError> {
        Ok(Self::read_index_changes_listed(bytes)?
            .into_iter()
            .map(|(h, m)| (h, Arc::new(m)))
            .collect())
    }

    /// The same changes as a list in the order the index keeps them, which
    /// a current index keeps in log order ([`log_order`]): what the log
    /// needs (oxygen#94).
    pub fn read_index_changes_listed(
        bytes: &[u8],
    ) -> Result<Vec<(ChangeHash, ChangeMeta)>, ObjectError> {
        use crate::flat::Reader;
        let mut r = Reader::new(bytes);
        if r.fixed::<18>()?.as_slice() != Self::INDEX_MAGIC {
            return Err(ObjectError::Schema("state index: not a state index".into()));
        }
        let n = r.count()?;
        let mut changes = Vec::with_capacity(n);
        for _ in 0..n {
            let h = r.object()?;
            changes.push((h, ChangeMeta::read_flat(&mut r)?));
        }
        if !r.at_end() {
            return Err(ObjectError::Schema("state index: trailing bytes".into()));
        }
        Ok(changes)
    }

    /// The deferred block: each node with its graph key and its body,
    /// left encoded (see [`Deferred`]).
    fn read_deferred(
        table: &[u8],
        histories: &Histories,
    ) -> Result<crate::fasthash::FastMap<NodeId, Arc<Deferred>>, ObjectError> {
        use crate::flat::Reader;
        let _t = crate::timing::phase("index: deferred");
        let mut r = Reader::new(table);
        let mut deferred = crate::fasthash::FastMap::default();
        let n = r.count()?;
        deferred.reserve(n);
        for _ in 0..n {
            let id = NodeId(r.fixed::<16>()?);
            let key = r.fixed::<32>()?;
            // The length is of bytes in the histories, not in this table:
            // checked against the histories' size below.
            let (at, len, sum) = (r.u64()?, r.u32()? as usize, r.fixed::<32>()?);
            if at
                .checked_add(len as u64)
                .is_none_or(|end| end > histories.len())
            {
                return Err(ObjectError::Schema(
                    "state index: a file's history lies outside it".into(),
                ));
            }
            let d = Deferred {
                raw: Some(RawHistory {
                    from: histories.clone(),
                    at,
                    len,
                    sum,
                }),
                key: std::sync::OnceLock::from(key),
                ..Deferred::default()
            };
            deferred.insert(id, Arc::new(d));
        }
        if !r.at_end() {
            return Err(ObjectError::Schema("state index: trailing bytes".into()));
        }
        Ok(deferred)
    }

    /// One node's deferred history as the index encodes it.
    fn read_deferred_body(bytes: &[u8]) -> Result<DeferredBody, ObjectError> {
        use crate::flat::Reader;
        let bad = |what: &str| ObjectError::Schema(format!("state index: {what}"));
        let node = |r: &mut Reader| -> Result<NodeId, ObjectError> { Ok(NodeId(r.fixed::<16>()?)) };
        let atom = |r: &mut Reader| -> Result<AtomRef, ObjectError> {
            Ok(match r.u8()? {
                0 => AtomRef::Start,
                1 => AtomRef::End,
                2 => AtomRef::Atom {
                    change: r.object()?,
                    index: r.u32()?,
                },
                _ => return Err(bad("atom")),
            })
        };
        let mut r = Reader::new(bytes);
        let base = match r.u8()? {
            0 => None,
            1 => Some(DeferredBase::Blob(r.object()?)),
            _ => Some(DeferredBase::Inline(cbor::decode(r.bytes()?)?)),
        };
        let mut edits = Vec::new();
        for _ in 0..r.count()? {
            let c = r.object()?;
            let mut es = Vec::new();
            for _ in 0..r.count()? {
                es.push(match r.u8()? {
                    0 => Edit::Insert {
                        node: node(&mut r)?,
                        after: atom(&mut r)?,
                        before: atom(&mut r)?,
                        lines: crate::lines::Lines::default(),
                    },
                    1 => {
                        let node = node(&mut r)?;
                        let mut atoms = Vec::new();
                        for _ in 0..r.count()? {
                            atoms.push(crate::objects::AtomRange {
                                change: r.object()?,
                                from: r.u32()?,
                                to: r.u32()?,
                            });
                        }
                        Edit::Delete { node, atoms }
                    }
                    _ => return Err(bad("edit")),
                });
            }
            edits.push((c, es));
        }
        let mut texts = Vec::new();
        for _ in 0..r.count()? {
            texts.push(match r.u8()? {
                0 => TextRef::Blob(
                    r.object()?,
                    r.u32()?,
                    match r.u8()? {
                        0 => crate::objects::TextForm::Cbor,
                        1 => crate::objects::TextForm::Raw,
                        _ => crate::objects::TextForm::CborFallback,
                    },
                ),
                _ => TextRef::Inline {
                    change: r.object()?,
                    start: r.u32()?,
                    end: r.u32()?,
                },
            });
        }
        if !r.at_end() {
            return Err(bad("trailing bytes in a file's history"));
        }
        Ok(DeferredBody { base, edits, texts })
    }
}

impl Clone for State {
    fn clone(&self) -> State {
        State {
            changes: self.changes.clone(),
            tree: self.tree.clone(),
            files: self.files.clone(),
            replaces: self.replaces.clone(),
            // A cache: the clone builds its own.
            closure: Default::default(),
            levels: Default::default(),
            deferred: self.deferred.clone(),
            unified_view: self.unified_view.clone(),
        }
    }
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub enum FileContent {
    Text {
        bytes: Vec<u8>,
        materialized: Materialized,
    },
    /// Live blobs; more than one is a replace conflict (spec §9.3).
    Opaque {
        blobs: Vec<ObjectRef>,
    },
    Dir,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct FileState {
    pub node: NodeId,
    pub kind: NodeKind,
    pub mode: Mode,
    pub content: FileContent,
}

impl FileState {
    pub fn is_conflicted(&self) -> bool {
        match &self.content {
            FileContent::Text { materialized, .. } => materialized.is_conflicted(),
            FileContent::Opaque { blobs } => blobs.len() > 1,
            FileContent::Dir => false,
        }
    }
}

/// See `State::conflict_regions`.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct ConflictRegions {
    pub plain: Vec<(AtomKey, Vec<u8>)>,
    pub region_atoms: Vec<AtomKey>,
    pub contributors: Vec<ChangeHash>,
}

/// The whole tree of a state, materialized.
#[derive(Clone, Debug, Default)]
pub struct Snapshot {
    pub tree: TreeView,
    pub files: BTreeMap<String, FileState>,
    /// Text files left unrendered by [`State::materialize_except`]: their
    /// `bytes` are empty and they are known not to be conflicted.
    pub unrendered: BTreeSet<String>,
}

impl Snapshot {
    pub fn is_conflicted(&self) -> bool {
        self.tree.is_conflicted() || self.files.values().any(|f| f.is_conflicted())
    }
}

fn vertex(r: AtomRef) -> Vertex {
    match r {
        AtomRef::Start => Vertex::Start,
        AtomRef::End => Vertex::End,
        AtomRef::Atom { change, index } => Vertex::Atom((change, index)),
    }
}

/// Replace self references in a tree op with the concrete change hash.
pub fn resolve_tree_op(op: &TreeOp, me: ChangeHash) -> TreeOp {
    match op {
        TreeOp::RemoveNode { node, placement } => TreeOp::RemoveNode {
            node: *node,
            placement: placement.resolve(me),
        },
        TreeOp::Move {
            node,
            old,
            parent,
            name,
        } => TreeOp::Move {
            node: *node,
            old: old.resolve(me),
            parent: *parent,
            name: name.clone(),
        },
        TreeOp::SetMode { node, old, mode } => TreeOp::SetMode {
            node: *node,
            old: old.resolve(me),
            mode: *mode,
        },
        add @ TreeOp::AddNode { .. } => add.clone(),
    }
}

/// Apply one change's edits of one file (inserts and deletes; replaces
/// live beside the graphs) to that file's graph.
fn apply_file_edits(g: &mut FileGraph, h: ChangeHash, edits: Vec<Edit>) {
    // The graph's kept order, if any, follows the edits (see `Order`).
    let order = g.take_order();
    let mut inserts: Vec<(Vertex, Vertex, u32, u32)> = Vec::new();
    let mut deleted: Vec<crate::graph::AtomKey> = Vec::new();
    let mut next = 0u32;
    for e in edits {
        match e {
            Edit::Insert {
                after,
                before,
                lines,
                ..
            } => {
                let first = next;
                let count = lines.len() as u32;
                next += count;
                let (after, before) = (vertex(after.resolve(h)), vertex(before.resolve(h)));
                inserts.push((after, before, first, count));
                g.insert_owned(h, first, after, before, lines);
            }
            Edit::Delete { atoms, .. } => {
                for r in atoms {
                    let r = r.resolve(h);
                    deleted.extend((r.from..r.to).map(|i| (r.change, i)));
                    g.delete(h, &r);
                }
            }
            Edit::Replace { .. } => {}
        }
    }
    if let Some(o) = order {
        g.keep_spliced(o, h, &inserts, &deleted);
    }
}

/// Build the graphs of deferred entries: their checkpoint bases and the
/// texts of inserts read without their lines are loaded in one call, then
/// the graphs are built in parallel (they are independent).
fn build_deferred(
    taken: Vec<(NodeId, Deferred)>,
    load: impl FnOnce(&[ObjectRef]) -> Result<Vec<Vec<u8>>, ObjectError>,
    read_change: impl Fn(&ChangeHash) -> Result<Arc<Vec<u8>>, ObjectError>,
) -> Result<Vec<(NodeId, FileGraph)>, ObjectError> {
    let mut taken: Vec<(NodeId, DeferredBody)> = taken
        .into_iter()
        .map(|(n, d)| Ok((n, d.into_body()?)))
        .collect::<Result<_, ObjectError>>()?;
    let mut refs: Vec<ObjectRef> = taken
        .iter()
        .filter_map(|(_, d)| match &d.base {
            Some(DeferredBase::Blob(r)) => Some(*r),
            _ => None,
        })
        .collect();
    // Text blobs of inserts read without their lines, after the bases.
    let bases = refs.len();
    refs.extend(taken.iter().flat_map(|(_, d)| {
        d.texts.iter().filter_map(|t| match t {
            TextRef::Blob(r, _, _) => Some(*r),
            TextRef::Inline { .. } => None,
        })
    }));
    // And the encodings of changes whose inline inserts are needed,
    // each read once.
    let reading = crate::timing::phase("graphs: read changes and blobs");
    let mut changes: HashMap<ChangeHash, Arc<Vec<u8>>> = HashMap::new();
    for t in taken.iter().flat_map(|(_, d)| d.texts.iter()) {
        if let TextRef::Inline { change, .. } = t {
            if !changes.contains_key(change) {
                changes.insert(*change, read_change(change)?);
            }
        }
    }
    let n_changes = changes.len();
    drop(reading);
    let reading = crate::timing::phase("graphs:   of which blobs");
    let mut blobs = if refs.is_empty() {
        Vec::new()
    } else {
        load(&refs)?
    };
    if crate::timing::enabled() {
        eprintln!(
            "timing graphs: {} graphs, {n_changes} changes, {} blobs",
            taken.len(),
            refs.len()
        );
    }
    if blobs.len() != refs.len() {
        return Err(ObjectError::Schema("graph blobs: wrong count".into()));
    }
    drop(reading);
    let lines = crate::timing::phase("graphs: hand out blobs");
    let mut texts = blobs.split_off(bases).into_iter();
    let mut blobs = blobs.into_iter();
    // Hand each graph its own blobs, in order, then fill in its inserts'
    // lines and build it, all in parallel: graphs are independent.
    type Work = (NodeId, DeferredBody, Option<Vec<u8>>, Vec<Vec<u8>>);
    let mut work: Vec<Work> = Vec::with_capacity(taken.len());
    for (node, d) in taken.drain(..) {
        let wanted = d
            .texts
            .iter()
            .filter(|t| matches!(t, TextRef::Blob(..)))
            .count();
        let mine: Vec<Vec<u8>> = texts.by_ref().take(wanted).collect();
        if mine.len() != wanted {
            return Err(ObjectError::Schema("text blobs: wrong count".into()));
        }
        let bytes = match &d.base {
            Some(DeferredBase::Blob(_)) => Some(
                blobs
                    .next()
                    .ok_or_else(|| ObjectError::Schema("graph blobs: wrong count".into()))?,
            ),
            _ => None,
        };
        work.push((node, d, bytes, mine));
    }
    drop(lines);
    let _t = crate::timing::phase("graphs: lines and build");
    use rayon::prelude::*;
    let changes = &changes;
    let built: Vec<(NodeId, FileGraph)> = work
        .into_par_iter()
        .map(|(node, mut d, bytes, mine)| {
            // Fill in the lines of inserts read without them, in order.
            let mut want = std::mem::take(&mut d.texts).into_iter();
            let mut mine = mine.into_iter();
            for (_, edits) in d.edits.iter_mut() {
                for e in edits.iter_mut() {
                    if let Edit::Insert { lines, .. } = e {
                        if lines.is_empty() {
                            *lines = match want.next() {
                                Some(TextRef::Blob(_, count, form)) => {
                                    let content = mine.next().ok_or_else(|| {
                                        ObjectError::Schema("text blobs: wrong count".into())
                                    })?;
                                    crate::objects::text_lines(&content, count, form)?
                                }
                                Some(TextRef::Inline { change, start, end }) => {
                                    let bytes = &changes[&change];
                                    let range = bytes
                                        .get(start as usize..end as usize)
                                        .ok_or_else(|| {
                                            ObjectError::Schema(
                                                "an inline insert lies outside its change".into(),
                                            )
                                        })?;
                                    crate::objects::inline_lines(range)?
                                }
                                None => {
                                    return Err(ObjectError::Schema(
                                        "an insert's text is missing".into(),
                                    ))
                                }
                            };
                        }
                    }
                }
            }
            let mut graph = match (d.base.take(), bytes) {
                (Some(DeferredBase::Inline(v)), _) => FileGraph::from_value(&v)?,
                (Some(DeferredBase::Blob(_)), Some(bytes)) => {
                    FileGraph::from_value(&cbor::decode(&bytes)?)?
                }
                _ => FileGraph::default(),
            };
            for (h, edits) in std::mem::take(&mut d.edits) {
                apply_file_edits(&mut graph, h, edits);
            }
            Ok((node, graph))
        })
        .collect::<Result<_, ObjectError>>()?;
    Ok(built)
}

impl State {
    pub fn new() -> Self {
        Self::default()
    }

    /// An empty state for history only: changes and tree are kept, file
    /// edits are recorded but no file graph is built until
    /// [`State::load_graphs`] asks for it. For log and other commands that
    /// need no file contents, or only a few files.
    pub fn new_history_only() -> Self {
        State {
            deferred: Some(Default::default()),
            ..Self::default()
        }
    }

    /// Whether `node`'s file graph is built (always, in a full state).
    pub fn is_built(&self, node: &NodeId) -> bool {
        self.deferred.is_none() || self.files.contains_key(node)
    }

    /// Whether this state builds file graphs only when asked
    /// ([`State::new_history_only`]), even if it has none waiting.
    pub fn is_history_only(&self) -> bool {
        self.deferred.is_some()
    }

    /// Whether this state holds every file graph.
    pub fn is_full(&self) -> bool {
        self.deferred.as_ref().is_none_or(|d| d.is_empty())
    }

    /// The graph of `node`, which must be loaded: asking a history-only
    /// state for a graph it has not built is a bug, never an empty file.
    fn graph(&self, node: &NodeId) -> Option<&FileGraph> {
        if let Some(d) = &self.deferred {
            assert!(
                !d.contains_key(node),
                "file graph {} of a history-only state was not loaded (State::load_graphs)",
                node.short(12)
            );
        }
        self.files.get(node).map(|g| g.as_ref())
    }

    /// Operations over every file need every graph.
    fn require_full(&self, what: &str) {
        assert!(
            self.is_full(),
            "{what} needs every file graph; this state was loaded for history only"
        );
    }

    /// Build the graphs of `nodes` in a history-only state: from their
    /// checkpoint base (blobs read in one batch through `load`) and their
    /// deferred edits. Nodes already built, or with no graph, are skipped.
    pub fn load_graphs(
        &mut self,
        nodes: &[NodeId],
        load: impl FnOnce(&[ObjectRef]) -> Result<Vec<Vec<u8>>, ObjectError>,
        read_change: impl Fn(&ChangeHash) -> Result<Arc<Vec<u8>>, ObjectError>,
    ) -> Result<(), ObjectError> {
        let Some(deferred) = self.deferred.as_mut() else {
            return Ok(());
        };
        let taken: Vec<(NodeId, Deferred)> = nodes
            .iter()
            .filter_map(|n| deferred.remove(n).map(|d| (*n, Arc::unwrap_or_clone(d))))
            .collect();
        let built = build_deferred(taken, load, read_change)?;
        self.files
            .extend(built.into_iter().map(|(n, g)| (n, Arc::new(g))));
        Ok(())
    }

    /// The graphs of `nodes`, built from this history-only state's deferred
    /// edits without building them in it: the state stays history only, so
    /// changes read without their lines can still be added to it and every
    /// graph keeps its key. For a caller that renders a few files of many
    /// states in turn (verifying an import).
    pub fn build_graphs_detached(
        &self,
        nodes: &[NodeId],
        load: impl FnOnce(&[ObjectRef]) -> Result<Vec<Vec<u8>>, ObjectError>,
        read_change: impl Fn(&ChangeHash) -> Result<Arc<Vec<u8>>, ObjectError>,
    ) -> Result<Vec<(NodeId, FileGraph)>, ObjectError> {
        let Some(deferred) = self.deferred.as_ref() else {
            return Ok(Vec::new());
        };
        // A file this state has built has no deferred history: its graph
        // is the built one, copied. Built from an empty history instead, it
        // rendered as an empty file, and a merge taking this path recorded
        // the whole file as inserted again, conflicting with itself (#71).
        let mut built: Vec<(NodeId, FileGraph)> = Vec::new();
        let mut copies: Vec<(NodeId, Deferred)> = Vec::new();
        for n in nodes {
            match (self.files.get(n), deferred.get(n)) {
                (Some(g), _) => built.push((*n, (**g).clone())),
                (None, d) => copies.push((*n, d.map(|d| (**d).clone()).unwrap_or_default())),
            }
        }
        let mut out = build_deferred(copies, load, read_change)?;
        out.extend(built);
        Ok(out)
    }

    /// Render a graph built for this state (see
    /// [`State::build_graphs_detached`]), as [`State::render_file`] does.
    pub fn render_graph(&self, g: &FileGraph) -> (Vec<u8>, Materialized) {
        let m = g.materialize(self);
        let bytes = g.render(&m, &|h| self.label(h));
        (bytes, m)
    }

    /// Every file graph a history-only state has not built, deleted files'
    /// included.
    pub fn unbuilt_nodes(&self) -> Vec<NodeId> {
        self.deferred
            .as_ref()
            .map(|d| d.keys().copied().collect())
            .unwrap_or_default()
    }

    /// Drop the built graphs of `nodes`, for a caller rendering a large
    /// tree a batch at a time. The state can no longer render those files,
    /// so it must not be used for them again.
    pub fn drop_graphs(&mut self, nodes: &[NodeId]) {
        for n in nodes {
            self.files.remove(n);
        }
    }

    /// Build every graph a history-only state has not built, making it full.
    pub fn load_all_graphs(
        &mut self,
        load: impl FnOnce(&[ObjectRef]) -> Result<Vec<Vec<u8>>, ObjectError>,
        read_change: impl Fn(&ChangeHash) -> Result<Arc<Vec<u8>>, ObjectError>,
    ) -> Result<(), ObjectError> {
        let nodes: Vec<NodeId> = match &self.deferred {
            Some(d) => d.keys().copied().collect(),
            None => return Ok(()),
        };
        self.load_graphs(&nodes, load, read_change)?;
        self.deferred = None;
        Ok(())
    }

    /// The one text a file's history is, when it is a single insert of
    /// the whole file into nothing (a file added once and never edited,
    /// with no checkpoint base): its rendering is that text, so it needs
    /// no graph. `None` otherwise, or when the graph is built.
    pub fn single_text(&self, node: &NodeId) -> Option<TextRef> {
        if self.files.contains_key(node) {
            return None;
        }
        // A history that cannot be read is no single text: building its
        // graph reports why.
        let d = self.deferred.as_ref()?.get(node)?.body().ok()?;
        let [(_, edits)] = d.edits.as_slice() else {
            return None;
        };
        let [Edit::Insert {
            after: AtomRef::Start,
            before: AtomRef::End,
            lines,
            ..
        }] = edits.as_slice()
        else {
            return None;
        };
        match (&d.base, d.texts.as_slice()) {
            (None, [t]) if lines.is_empty() => Some(*t),
            _ => None,
        }
    }

    /// A key for the graph of `node` in a history-only state that has not
    /// built it: equal keys mean equal graphs, so equal renderings (a
    /// rendering depends only on the graph and on the changes' own
    /// immutable metadata). It names the node, its checkpoint base (a blob,
    /// an inline value, or none) and the set of later changes that edited
    /// the file.
    /// `None` for a full state or a graph already built.
    pub fn graph_key(&self, node: &NodeId) -> Option<[u8; 32]> {
        let deferred = self.deferred.as_ref()?;
        if self.files.contains_key(node) {
            return None;
        }
        let mut h = blake3::Hasher::new();
        h.update(b"oxy graph key 1\0");
        // A change's edits are applied per node: the node is part of it.
        h.update(&node.0);
        let Some(d) = deferred.get(node) else {
            // Never edited: the empty graph.
            h.update(&[0]);
            return Some(*h.finalize().as_bytes());
        };
        // A history that cannot be read has no key: the file is read.
        if let Some(k) = d.key.get() {
            return Some(*k);
        }
        let k = d.graph_key(h).ok()?;
        Some(*d.key.get_or_init(|| k))
    }

    pub fn len(&self) -> usize {
        self.changes.len()
    }

    /// Turn this history-only state's built graphs back into deferred
    /// histories, each starting from its graph encoded as a checkpoint
    /// graph and stored by `store` (it returns the blob's reference): what
    /// they held stays, and memory goes back to what a history-only state
    /// costs. A graph built later again decodes that blob and applies only
    /// the edits that came after. Graph keys change (the base is now a
    /// blob), which only means a cached rendering is not reused.
    pub fn shed_graphs<E>(
        &mut self,
        mut store: impl FnMut(&[u8]) -> Result<ObjectRef, E>,
    ) -> Result<usize, E> {
        use rayon::prelude::*;
        let Some(deferred) = self.deferred.as_mut() else {
            return Ok(0);
        };
        let graphs: Vec<(NodeId, Arc<FileGraph>)> = self.files.drain().collect();
        let encoded: Vec<(NodeId, Vec<u8>)> = graphs
            .par_iter()
            .map(|(n, g)| (*n, cbor::encode(&g.to_value())))
            .collect();
        drop(graphs);
        let shed = encoded.len();
        for (node, bytes) in encoded {
            let blob = store(&bytes)?;
            deferred.insert(
                node,
                Arc::new(Deferred::from_body(DeferredBody {
                    base: Some(DeferredBase::Blob(blob)),
                    ..DeferredBody::default()
                })),
            );
        }
        Ok(shed)
    }

    /// How many file graphs this state has built.
    pub fn built_count(&self) -> usize {
        self.files.len()
    }

    /// Atoms in this state's built graphs, live and deleted: what their
    /// memory follows, since a graph keeps every line its file ever had.
    pub fn built_atoms(&self) -> usize {
        self.files.values().map(|g| g.atom_count()).sum()
    }

    /// Edits held in the deferred histories of files not built, and the
    /// most any one file holds (for measuring, #25).
    pub fn deferred_edit_count(&self) -> (usize, usize) {
        let Some(d) = self.deferred.as_ref() else {
            return (0, 0);
        };
        let mut total = 0;
        let mut most = 0;
        for v in d.values() {
            let n: usize = v
                .body()
                .map_or(0, |b| b.edits.iter().map(|(_, e)| e.len()).sum());
            total += n;
            most = most.max(n);
        }
        (total, most)
    }
}

impl Deferred {
    /// See [`State::graph_key`]: `h` has taken the node already.
    fn graph_key(&self, mut h: blake3::Hasher) -> Result<[u8; 32], ObjectError> {
        let d = self.body()?;
        match &d.base {
            None => h.update(&[0]),
            Some(DeferredBase::Blob(r)) => h.update(&[1]).update(&r.0),
            Some(DeferredBase::Inline(v)) => h.update(&[2]).update(&cbor::encode(v)),
        };
        let mut changes: Vec<&ChangeHash> = d.edits.iter().map(|(c, _)| c).collect();
        changes.sort();
        changes.dedup();
        h.update(&(changes.len() as u64).to_le_bytes());
        for c in changes {
            h.update(&c.0);
        }
        Ok(*h.finalize().as_bytes())
    }
}

impl State {
    pub fn is_empty(&self) -> bool {
        self.changes.is_empty()
    }
    pub fn contains(&self, h: &ChangeHash) -> bool {
        self.changes.contains_key(h)
    }
    pub fn get(&self, h: &ChangeHash) -> Option<&ChangeMeta> {
        self.changes.get(h).map(|m| &**m)
    }
    pub fn changes(&self) -> impl Iterator<Item = (&ChangeHash, &ChangeMeta)> {
        self.changes.iter().map(|(h, m)| (h, &**m))
    }
    pub fn tree(&self) -> &TreeGraph {
        &self.tree
    }

    /// The tree as every reader sees it: the tree graph's view, with the
    /// same path added on several lines of history with identical content
    /// shown as one entry rather than a name conflict (spec §9.2). The
    /// text files among such candidates must be built in a history-only
    /// state ([`State::unify_candidates`]); until they are, they are shown
    /// as the conflict.
    pub fn view(&self) -> TreeView {
        if let Some(v) = self.unified_view.get() {
            return v.clone();
        }
        let raw = self.tree.view();
        if !raw
            .conflicts
            .iter()
            .any(|c| matches!(c, crate::tree::TreeConflict::Name { .. }))
        {
            return raw;
        }
        let (view, decided) = self.tree.unify(raw, |es| self.same_content(es));
        if decided {
            let _ = self.unified_view.set(view.clone());
        }
        view
    }

    /// The text files [`State::view`] must compare and a history-only
    /// state has not built: candidates of the tree's name conflicts.
    pub fn unify_candidates(&self) -> Vec<NodeId> {
        let Some(deferred) = &self.deferred else {
            return Vec::new();
        };
        if self.unified_view.get().is_some() {
            return Vec::new();
        }
        let raw = self.tree.view();
        let mut out = Vec::new();
        for c in &raw.conflicts {
            if let crate::tree::TreeConflict::Name { nodes, .. } = c {
                out.extend(nodes.iter().filter(|n| deferred.contains_key(*n)));
            }
        }
        out
    }

    /// Whether the entries' contents are identical, unconflicted: the same
    /// rendering for text files, the same single blob for opaque ones,
    /// symlinks and subrepos. `None` for a text file not built yet.
    fn same_content(&self, entries: &[&crate::tree::TreeEntry]) -> Option<bool> {
        let text = |e: &crate::tree::TreeEntry| e.kind == NodeKind::File && !e.mode.opaque;
        let mut first: Option<Vec<u8>> = None;
        let mut first_blob: Option<ObjectRef> = None;
        for e in entries {
            if text(e) {
                if self
                    .deferred
                    .as_ref()
                    .is_some_and(|d| d.contains_key(&e.node))
                {
                    return None;
                }
                let (bytes, m) = self.render_file(&e.node);
                if m.is_conflicted() {
                    return Some(false);
                }
                match &first {
                    None => first = Some(bytes),
                    Some(b) if *b == bytes => {}
                    Some(_) => return Some(false),
                }
            } else {
                let [blob] = self.live_blobs(&e.node)[..] else {
                    return Some(false);
                };
                match first_blob {
                    None => first_blob = Some(blob),
                    Some(b) if b == blob => {}
                    Some(_) => return Some(false),
                }
            }
        }
        Some(true)
    }
    pub fn file(&self, node: &NodeId) -> Option<&FileGraph> {
        self.graph(node)
    }

    /// Add a change whose dependencies are already present.
    pub fn add(&mut self, change: Change) -> Result<ChangeHash, StateError> {
        let h = change.hash();
        self.add_hashed(h, change)
    }

    /// [`State::add`] for a change whose hash `h` is already known (it was
    /// just stored), so it is not encoded again.
    pub fn add_hashed(&mut self, h: ChangeHash, change: Change) -> Result<ChangeHash, StateError> {
        if self.changes.contains_key(&h) {
            return Ok(h);
        }
        for d in &change.deps {
            if !self.changes.contains_key(d) {
                return Err(StateError::MissingDep(*d, h));
            }
        }
        self.add_known(h, change, Vec::new());
        Ok(h)
    }

    /// Add a change in any order; call `validate` once all are in.
    pub fn add_unchecked(&mut self, change: Change) -> ChangeHash {
        let h = change.hash();
        self.add_known(h, change, Vec::new());
        h
    }

    /// Add a change just stored under hash `h` to a history-only state that
    /// has built some graphs: an edit of a built graph is applied whole
    /// (from `full`), and one of a file not built waits without its text
    /// (from `skeleton` and its `texts`, as the store reads it back), so the
    /// state holds no text it has not built. A Git import adds each change
    /// it makes this way, keeping memory to what changes.
    pub fn add_known_mixed(
        &mut self,
        h: ChangeHash,
        full: Change,
        skeleton: Change,
        texts: Vec<TextRef>,
    ) {
        if self.deferred.is_none() || self.changes.contains_key(&h) {
            self.add_known(h, full, Vec::new());
            return;
        }
        let Change {
            version,
            change_id,
            deps,
            author,
            hlc,
            message,
            tree_ops,
            edits: full_edits,
            kind,
        } = full;
        let mut texts = texts.into_iter();
        let mut edits = Vec::with_capacity(full_edits.len());
        let mut kept = Vec::new();
        for (f, s) in full_edits.into_iter().zip(skeleton.edits) {
            let text = match &s {
                Edit::Insert { lines, .. } if lines.is_empty() => texts.next(),
                _ => None,
            };
            if self.files.contains_key(&f.node()) {
                edits.push(f);
            } else {
                edits.push(s);
                kept.extend(text);
            }
        }
        let c = Change {
            version,
            change_id,
            deps,
            author,
            hlc,
            message,
            tree_ops,
            edits,
            kind,
        };
        self.add_known(h, c, kept);
    }

    /// Add a change read from the store under hash `h`, without hashing it
    /// again (for a version 2 change that means chunking all its text).
    /// `texts` names, in order, the inserts left without their lines (see
    /// [`crate::store::Store::get_change_lazy`]); only a history-only state
    /// takes them, and only for graphs it has not built.
    pub fn add_known(&mut self, h: ChangeHash, change: Change, texts: Vec<TextRef>) {
        if self.changes.contains_key(&h) {
            return;
        }
        let meta = ChangeMeta::of(&change);
        self.unified_view = Default::default();
        self.apply_with(h, change, texts);
        self.changes.insert(h, Arc::new(meta));
        {
            let mut m = self.closure.write().unwrap();
            m.sets.clear();
            m.held = 0;
        }
    }

    // ------------------------------------------------------------ checkpoints (spec §11)

    /// Serialize the whole state in payload version 1: change metadata,
    /// tree, file graphs, replaces, all in one value (spec §11).
    pub fn checkpoint_payload(&self) -> Vec<u8> {
        self.require_full("a checkpoint");
        let mut nodes: Vec<&NodeId> = self.files.keys().collect();
        nodes.sort();
        let files = Value::Array(
            nodes
                .into_iter()
                .map(|n| Value::Array(vec![Value::Bytes(n.0.to_vec()), self.files[n].to_value()]))
                .collect(),
        );
        cbor::encode(&self.payload_value(o::FORMAT_VERSION, files))
    }

    /// Each file graph encoded on its own, by node, for a version 2 payload
    /// (spec §11): the graphs of large files become blobs of their own.
    pub fn checkpoint_file_graphs(&self) -> Vec<(NodeId, Vec<u8>)> {
        use rayon::prelude::*;
        self.require_full("a checkpoint");
        let mut nodes: Vec<(&NodeId, &FileGraph)> =
            self.files.iter().map(|(n, g)| (n, g.as_ref())).collect();
        nodes.sort_by_key(|(n, _)| **n);
        nodes
            .into_par_iter()
            .map(|(n, g)| (*n, cbor::encode(&g.to_value())))
            .collect()
    }

    /// The payload version 2 index for the graphs of
    /// [`State::checkpoint_file_graphs`]: a graph encoded in fewer than
    /// [`INLINE_GRAPH_LIMIT`] bytes is kept inline, a larger one is named by
    /// `blob_of` (the ObjectRef of the blob holding its encoding).
    pub fn checkpoint_index(
        &self,
        graphs: &[(NodeId, Vec<u8>)],
        mut blob_of: impl FnMut(&[u8]) -> ObjectRef,
    ) -> Vec<u8> {
        let files = Value::Array(
            graphs
                .iter()
                .map(|(n, enc)| {
                    let entry = if enc.len() < INLINE_GRAPH_LIMIT {
                        cbor::decode(enc).expect("an encoded graph decodes")
                    } else {
                        Value::Bytes(blob_of(enc).0.to_vec())
                    };
                    Value::Array(vec![Value::Bytes(n.0.to_vec()), entry])
                })
                .collect(),
        );
        cbor::encode(&self.payload_value(CHECKPOINT_INDEX_VERSION, files))
    }

    /// The payload map around `files`: changes, tree and replaces as in version 1.
    fn payload_value(&self, version: u64, files: Value) -> Value {
        let mut hashes: Vec<&ChangeHash> = self.changes.keys().collect();
        hashes.sort();
        let changes = Value::Array(
            hashes
                .into_iter()
                .map(|h| {
                    let m = &self.changes[h];
                    Value::Array(vec![
                        Value::Bytes(h.0.to_vec()),
                        Value::Bytes(m.change_id.0.to_vec()),
                        Value::Array(m.deps.iter().map(|d| Value::Bytes(d.0.to_vec())).collect()),
                        Value::Text(m.message.clone()),
                        Value::Uint(m.hlc.wall_ms),
                        Value::Uint(m.hlc.logical as u64),
                        Value::Uint(m.kind.tag()),
                        Value::Bytes(m.author.0.to_vec()),
                    ])
                })
                .collect(),
        );
        let mut rnodes: Vec<&NodeId> = self.replaces.keys().collect();
        rnodes.sort();
        let opt =
            |o: &Option<ObjectRef>| o.map(|r| Value::Bytes(r.0.to_vec())).unwrap_or(Value::Null);
        let replaces = Value::Array(
            rnodes
                .into_iter()
                .map(|n| {
                    Value::Array(vec![
                        Value::Bytes(n.0.to_vec()),
                        Value::Array(
                            self.replaces[n]
                                .iter()
                                .map(|(c, o, nw)| {
                                    Value::Array(vec![Value::Bytes(c.0.to_vec()), opt(o), opt(nw)])
                                })
                                .collect(),
                        ),
                    ])
                })
                .collect(),
        );
        Value::map([
            (0, Value::Uint(version)),
            (1, changes),
            (2, self.tree.to_value()),
            (3, files),
            (4, replaces),
        ])
    }

    /// Decode a version 1 payload (the whole state in one value). A
    /// version 2 payload needs its graph blobs: use [`State::from_checkpoint`].
    pub fn from_checkpoint_payload(bytes: &[u8]) -> Result<State, ObjectError> {
        State::from_checkpoint(bytes, |_| {
            Err(ObjectError::Schema(
                "a version 2 checkpoint needs its graph blobs".into(),
            ))
        })
    }

    /// Decode a checkpoint payload of either version (spec §11). For
    /// version 2, `load` returns the contents of the named graph blobs, in
    /// order; the graphs are then decoded in parallel.
    pub fn from_checkpoint(
        bytes: &[u8],
        load: impl FnOnce(&[ObjectRef]) -> Result<Vec<Vec<u8>>, ObjectError>,
    ) -> Result<State, ObjectError> {
        use rayon::prelude::*;
        let v = cbor::decode(bytes)?;
        let m = o::map(&v, "checkpoint payload")?;
        let version = o::uint(o::field(m, 0, "version")?, "version")?;
        if version != o::FORMAT_VERSION && version != CHECKPOINT_INDEX_VERSION {
            return Err(ObjectError::Schema(format!(
                "checkpoint payload version {version} is not supported"
            )));
        }
        let mut st = State::payload_common(m)?;
        let entries = o::array(o::field(m, 3, "files")?, "files")?;
        let mut parsed: Vec<(NodeId, &Value)> = Vec::with_capacity(entries.len());
        for f in entries {
            let e = o::array(f, "file")?;
            if e.len() != 2 {
                return Err(ObjectError::Schema("file: expected [node, graph]".into()));
            }
            let node = NodeId::from_slice(o::bytes(&e[0], "node")?)
                .ok_or_else(|| ObjectError::Schema("node: expected 16 bytes".into()))?;
            parsed.push((node, &e[1]));
        }
        // Version 2 names large graphs by blob: fetch those in one batch.
        let mut fetched: HashMap<NodeId, Vec<u8>> = HashMap::new();
        if version == CHECKPOINT_INDEX_VERSION {
            let wanted: Vec<(NodeId, ObjectRef)> = parsed
                .iter()
                .filter(|(_, e)| matches!(e, Value::Bytes(_)))
                .map(|(n, e)| Ok((*n, o::oref(e, "graph blob")?)))
                .collect::<Result<_, ObjectError>>()?;
            if !wanted.is_empty() {
                let refs: Vec<ObjectRef> = wanted.iter().map(|(_, r)| *r).collect();
                let contents = {
                    let _t = crate::timing::phase("load:   graph blobs read");
                    load(&refs)?
                };
                if contents.len() != refs.len() {
                    return Err(ObjectError::Schema("graph blobs: wrong count".into()));
                }
                fetched.extend(wanted.iter().map(|(n, _)| *n).zip(contents));
            }
        } else if parsed.iter().any(|(_, e)| matches!(e, Value::Bytes(_))) {
            return Err(ObjectError::Schema(
                "a version 1 payload holds graphs inline".into(),
            ));
        }
        // File graphs are independent; decode and build them in parallel.
        let graphs = parsed
            .par_iter()
            .map(|(node, entry)| {
                let graph = match fetched.get(node) {
                    Some(bytes) => FileGraph::from_value(&cbor::decode(bytes)?)?,
                    None => FileGraph::from_value(entry)?,
                };
                Ok((*node, graph))
            })
            .collect::<Result<Vec<_>, ObjectError>>()?;
        st.files
            .extend(graphs.into_iter().map(|(n, g)| (n, Arc::new(g))));
        Ok(st)
    }

    /// The graph blobs a payload names: empty for version 1.
    pub fn checkpoint_graph_refs(bytes: &[u8]) -> Result<Vec<ObjectRef>, ObjectError> {
        let v = cbor::decode(bytes)?;
        let m = o::map(&v, "checkpoint payload")?;
        let mut out = Vec::new();
        for f in o::array(o::field(m, 3, "files")?, "files")? {
            if let Some(entry @ Value::Bytes(_)) = o::array(f, "file")?.get(1) {
                out.push(o::oref(entry, "graph blob")?);
            }
        }
        Ok(out)
    }

    /// Decode a checkpoint payload for history only: a version 2 payload's
    /// graphs are noted, not read or built, and come later through
    /// [`State::load_graphs`]. A version 1 payload holds them inline and is
    /// decoded in full.
    pub fn from_checkpoint_history(bytes: &[u8]) -> Result<State, ObjectError> {
        let v = cbor::decode(bytes)?;
        let m = o::map(&v, "checkpoint payload")?;
        let version = o::uint(o::field(m, 0, "version")?, "version")?;
        if version != CHECKPOINT_INDEX_VERSION {
            return State::from_checkpoint_payload(bytes);
        }
        let mut st = State::payload_common(m)?;
        let mut deferred = crate::fasthash::FastMap::default();
        for f in o::array(o::field(m, 3, "files")?, "files")? {
            let e = o::array(f, "file")?;
            if e.len() != 2 {
                return Err(ObjectError::Schema("file: expected [node, graph]".into()));
            }
            let node = NodeId::from_slice(o::bytes(&e[0], "node")?)
                .ok_or_else(|| ObjectError::Schema("node: expected 16 bytes".into()))?;
            let base = match &e[1] {
                Value::Bytes(_) => DeferredBase::Blob(o::oref(&e[1], "graph blob")?),
                other => DeferredBase::Inline(other.clone()),
            };
            deferred.insert(
                node,
                Arc::new(Deferred::from_body(DeferredBody {
                    base: Some(base),
                    ..DeferredBody::default()
                })),
            );
        }
        st.deferred = Some(deferred);
        Ok(st)
    }

    /// Changes, tree and replaces of a payload of either version.
    fn payload_common(m: &BTreeMap<u64, Value>) -> Result<State, ObjectError> {
        let mut st = State::new();
        for c in o::array(o::field(m, 1, "changes")?, "changes")? {
            let e = o::array(c, "change meta")?;
            if e.len() != 8 {
                return Err(ObjectError::Schema("change meta: expected 8 fields".into()));
            }
            let hash = o::oref(&e[0], "hash")?;
            let meta = ChangeMeta {
                change_id: crate::ids::ChangeId::from_slice(o::bytes(&e[1], "change id")?)
                    .ok_or_else(|| ObjectError::Schema("change id: expected 16 bytes".into()))?,
                deps: o::array(&e[2], "deps")?
                    .iter()
                    .map(|d| o::oref(d, "dep"))
                    .collect::<Result<Vec<_>, _>>()?,
                message: o::text(&e[3], "message")?.to_owned(),
                hlc: Hlc {
                    wall_ms: o::uint(&e[4], "wall")?,
                    logical: o::u32_(&e[5], "logical")?,
                },
                kind: ChangeKind::from_tag(o::uint(&e[6], "kind")?)?,
                author: o::oref(&e[7], "author")?,
            };
            st.changes.insert(hash, Arc::new(meta));
        }
        st.tree = TreeGraph::from_value(o::field(m, 2, "tree")?)?;
        let opt = |v: &Value| -> Result<Option<ObjectRef>, ObjectError> {
            if v.is_null() {
                Ok(None)
            } else {
                o::oref(v, "blob").map(Some)
            }
        };
        for r in o::array(o::field(m, 4, "replaces")?, "replaces")? {
            let e = o::array(r, "replace")?;
            if e.len() != 2 {
                return Err(ObjectError::Schema(
                    "replace: expected [node, records]".into(),
                ));
            }
            let node = NodeId::from_slice(o::bytes(&e[0], "node")?)
                .ok_or_else(|| ObjectError::Schema("node: expected 16 bytes".into()))?;
            let mut recs = Vec::new();
            for rec in o::array(&e[1], "records")? {
                let x = o::array(rec, "record")?;
                if x.len() != 3 {
                    return Err(ObjectError::Schema("record: expected 3 fields".into()));
                }
                recs.push((o::oref(&x[0], "change")?, opt(&x[1])?, opt(&x[2])?));
            }
            st.replaces.insert(node, recs);
        }
        Ok(st)
    }

    pub fn validate(&self) -> Result<(), StateError> {
        for (h, c) in &self.changes {
            for d in &c.deps {
                if !self.changes.contains_key(d) {
                    return Err(StateError::MissingDep(*d, *h));
                }
            }
        }
        Ok(())
    }

    fn apply_with(&mut self, h: ChangeHash, c: Change, texts: Vec<TextRef>) {
        for (i, op) in c.tree_ops.iter().enumerate() {
            self.tree.apply(h, i as u32, &resolve_tree_op(op, h));
        }
        if let Some(deferred) = self.deferred.as_mut() {
            // History only: edits of files not built wait, grouped per file,
            // with the texts of inserts read without their lines.
            let mut texts = texts.into_iter();
            let mut per_node: HashMap<NodeId, (Vec<Edit>, Vec<TextRef>)> = HashMap::new();
            for e in c.edits {
                match e {
                    Edit::Replace { node, old, new } => {
                        self.replaces.entry(node).or_default().push((h, old, new))
                    }
                    e => {
                        let slot = per_node.entry(e.node()).or_default();
                        if matches!(&e, Edit::Insert { lines, .. } if lines.is_empty()) {
                            slot.1.extend(texts.next());
                        }
                        slot.0.push(e);
                    }
                }
            }
            for (node, (edits, node_texts)) in per_node {
                match self.files.get_mut(&node) {
                    Some(g) => {
                        assert!(
                            node_texts.is_empty(),
                            "a change read without its text applied to a built graph"
                        );
                        apply_file_edits(Arc::make_mut(g), h, edits)
                    }
                    None => {
                        let d = Arc::make_mut(deferred.entry(node).or_default());
                        d.add(h, edits, node_texts);
                        d.key = Default::default();
                    }
                }
            }
            return;
        }
        assert!(
            texts.is_empty(),
            "a change read without its text applied to a full state"
        );
        let mut counters: HashMap<NodeId, u32> = HashMap::new();
        for e in c.edits {
            match e {
                Edit::Insert {
                    node,
                    after,
                    before,
                    lines,
                } => {
                    let idx = counters.entry(node).or_default();
                    let first = *idx;
                    *idx += lines.len() as u32;
                    Arc::make_mut(self.files.entry(node).or_default()).insert_owned(
                        h,
                        first,
                        vertex(after.resolve(h)),
                        vertex(before.resolve(h)),
                        lines,
                    );
                }
                Edit::Delete { node, atoms } => {
                    let g = Arc::make_mut(self.files.entry(node).or_default());
                    for r in atoms {
                        g.delete(h, &r.resolve(h));
                    }
                }
                Edit::Replace { node, old, new } => {
                    self.replaces.entry(node).or_default().push((h, old, new));
                }
            }
        }
    }

    /// Transitive dependencies of `h`, excluding `h` itself.
    pub fn closure(&self, h: &ChangeHash) -> Arc<HashSet<ChangeHash>> {
        if let Some(c) = self.closure.read().unwrap().sets.get(h) {
            return c.clone();
        }
        let mut out = HashSet::new();
        let mut stack: Vec<ChangeHash> = self
            .changes
            .get(h)
            .map(|c| c.deps.clone())
            .unwrap_or_default();
        while let Some(d) = stack.pop() {
            if out.insert(d) {
                if let Some(c) = self.changes.get(&d) {
                    stack.extend(c.deps.iter().copied());
                }
            }
        }
        let rc = Arc::new(out);
        let mut memo = self.closure.write().unwrap();
        if memo.held + rc.len() > CLOSURE_MEMO_HASHES {
            memo.sets.clear();
            memo.held = 0;
        }
        memo.held += rc.len();
        memo.sets.insert(*h, rc.clone());
        rc
    }

    /// `h`'s level (see `levels`), worked out for it and its ancestors not
    /// yet known, without recursion: histories are deeper than a stack. An
    /// unknown change has level 0.
    fn level(&self, h: &ChangeHash) -> u32 {
        // Nearly always known: read without excluding the other threads
        // rendering beside this one (oxygen#96).
        if let Some(l) = self.levels.read().unwrap().get(h) {
            return *l;
        }
        let mut levels = self.levels.write().unwrap();
        if let Some(l) = levels.get(h) {
            return *l;
        }
        let mut stack: Vec<(ChangeHash, bool)> = vec![(*h, false)];
        while let Some((c, expanded)) = stack.pop() {
            if levels.contains_key(&c) {
                continue;
            }
            let deps: &[ChangeHash] = self
                .changes
                .get(&c)
                .map(|m| m.deps.as_slice())
                .unwrap_or(&[]);
            if expanded {
                let l = deps
                    .iter()
                    .map(|d| levels.get(d).map_or(0, |l| l + 1))
                    .max()
                    .unwrap_or(0);
                levels.insert(c, l);
            } else {
                stack.push((c, true));
                for d in deps {
                    if !levels.contains_key(d) {
                        stack.push((*d, false));
                    }
                }
            }
        }
        levels[h]
    }

    /// Changes that no other change depends on directly, sorted.
    pub fn heads(&self) -> Vec<ChangeHash> {
        let depended: crate::fasthash::FastSet<&ChangeHash> =
            self.changes.values().flat_map(|c| c.deps.iter()).collect();
        let mut heads: Vec<ChangeHash> = self
            .changes
            .keys()
            .filter(|h| !depended.contains(h))
            .copied()
            .collect();
        heads.sort();
        heads
    }

    /// Maximal elements among `hashes` (those not in the closure of another),
    /// sorted. One walk down from all of them at once: whatever the walk
    /// reaches is below some other element. Linear in the changes reached,
    /// where comparing every pair's closures was quadratic in the number
    /// of heads, which a bookmark of independent changes has many of.
    pub fn maximal(&self, hashes: &[ChangeHash]) -> Vec<ChangeHash> {
        let set: BTreeSet<ChangeHash> = hashes.iter().copied().collect();
        let mut below: crate::fasthash::FastSet<ChangeHash> = Default::default();
        let mut stack: Vec<ChangeHash> = set
            .iter()
            .filter_map(|h| self.changes.get(h))
            .flat_map(|c| c.deps.iter().copied())
            .collect();
        while let Some(d) = stack.pop() {
            if below.insert(d) {
                if let Some(c) = self.changes.get(&d) {
                    stack.extend(c.deps.iter().copied());
                }
            }
        }
        set.into_iter().filter(|h| !below.contains(h)).collect()
    }

    pub fn materialize_file(&self, node: &NodeId) -> Materialized {
        self.graph(node)
            .map(|g| g.materialize(self))
            .unwrap_or_default()
    }

    /// Live lines of a text node in materialized order.
    pub fn live_lines(&self, node: &NodeId) -> Vec<(AtomKey, Vec<u8>)> {
        let Some(g) = self.graph(node) else {
            return vec![];
        };
        let m = g.materialize(self);
        m.live_atoms()
            .into_iter()
            .map(|a| (a, g.content(&a).unwrap_or_default().to_vec()))
            .collect()
    }

    /// The nodes with replace records: opaque files, as their blobs go.
    pub fn opaque_nodes(&self) -> Vec<NodeId> {
        let mut nodes: Vec<NodeId> = self.replaces.keys().copied().collect();
        nodes.sort();
        nodes
    }

    /// Live blobs of an opaque node: every `new` that no later `Replace` retired.
    pub fn live_blobs(&self, node: &NodeId) -> Vec<ObjectRef> {
        let mut live: Vec<ObjectRef> = self
            .live_replaces(node)
            .into_iter()
            .map(|(_, n)| n)
            .collect();
        live.sort();
        live.dedup();
        live
    }

    /// The replace records whose blob is live, with the change that set it
    /// (spec §9.3). A record's blob is superseded by a later replacement of
    /// that blob: one whose change depends on the record's change, as every
    /// replacement depends on the change that made its old blob current
    /// (§8.1). Two replacements that do not know each other both stay live
    /// (a replace conflict), and a file reverted to an earlier blob keeps it.
    /// A replacement without that dependency, from before the rule, is read
    /// as before: it supersedes every record that did not come after it.
    pub fn live_replaces(&self, node: &NodeId) -> Vec<(ChangeHash, ObjectRef)> {
        let Some(rs) = self.replaces.get(node) else {
            return vec![];
        };
        // Dependence by the bounded walk rather than whole closures: on a
        // long history a closure is a hundred thousand changes, and the
        // memo cannot keep them all (#107).
        let dep =
            |a: &ChangeHash, b: &ChangeHash| a != b && <Self as DepOracle>::depends_on(self, a, b);
        let knows_its_old = |s: &ChangeHash, old: &ObjectRef| {
            rs.iter()
                .any(|(q, _, n)| n.as_ref() == Some(old) && q != s && dep(s, q))
        };
        rs.iter()
            .filter_map(|(c, _, n)| n.map(|n| (*c, n)))
            .filter(|(c, n)| {
                !rs.iter().any(|(s, o, _)| {
                    s != c && *o == Some(*n) && (dep(s, c) || (!knows_its_old(s, n) && !dep(c, s)))
                })
            })
            .collect()
    }

    /// Short label for conflict markers: `changeid-prefix "message"`.
    pub fn label(&self, h: &ChangeHash) -> String {
        match self.changes.get(h) {
            Some(c) => format!(
                "{} \"{}\"",
                c.change_id.short(8),
                c.message.lines().next().unwrap_or("")
            ),
            None => format!("#{}", h.short(12)),
        }
    }

    /// For resolving a conflicted text node (spec §9.6): the live atoms that
    /// are not part of any conflict region in order, every live atom that is
    /// part of a region, and every change that contributed to a region.
    /// The changes that introduced the live blobs of an opaque node, for a
    /// replace conflict (spec §9.3).
    pub fn blob_contributors(&self, node: &NodeId, blobs: &[ObjectRef]) -> Vec<ChangeHash> {
        let mut out: Vec<ChangeHash> = Vec::new();
        for (h, _, new) in self.replaces.get(node).map(|v| v.as_slice()).unwrap_or(&[]) {
            if new.is_some_and(|b| blobs.contains(&b)) && !out.contains(h) {
                out.push(*h);
            }
        }
        out.sort();
        out
    }

    pub fn conflict_regions(&self, node: &NodeId) -> ConflictRegions {
        let mut out = ConflictRegions::default();
        let Some(g) = self.graph(node) else {
            return out;
        };
        let m = g.materialize(self);
        let mut contributors = BTreeSet::new();
        for seg in &m.segments {
            match seg {
                crate::graph::Segment::Lines(atoms) => {
                    for a in atoms {
                        out.plain
                            .push((*a, g.content(a).unwrap_or_default().to_vec()));
                    }
                }
                crate::graph::Segment::Conflict(sides) => {
                    for side in sides {
                        for a in side {
                            out.region_atoms.push(*a);
                            contributors.insert(a.0);
                        }
                    }
                }
                crate::graph::Segment::Zombie { deleted, inserted } => {
                    for a in deleted {
                        contributors.insert(a.0);
                        contributors.extend(g.deleters(a).copied());
                    }
                    for a in inserted {
                        out.region_atoms.push(*a);
                        contributors.insert(a.0);
                    }
                }
            }
        }
        out.contributors = contributors.into_iter().collect();
        out
    }

    /// A file graph materialized with each conflict's sides earliest change
    /// first (oxygen#121), by the clock of the oldest change on a side, so
    /// markers read in the order the work was done and `oxy resolve --both`
    /// keeps that order.
    fn materialize_ordered(&self, g: &FileGraph) -> Materialized {
        let mut m = g.materialize(self);
        let when = |h: &ChangeHash| match self.changes.get(h) {
            Some(c) => (false, c.hlc, *h),
            None => (true, Default::default(), *h),
        };
        for seg in &mut m.segments {
            if let crate::graph::Segment::Conflict(sides) = seg {
                sides.sort_by_cached_key(|s| s.iter().map(|a| when(&a.0)).min());
            }
        }
        m
    }

    /// For each conflict of a text node, in marker order (spec §9.5): the
    /// text of its sides, earliest change first, when it is an adjacent
    /// insertion that keeping both sides settles (oxygen#121); `None` for
    /// the others.
    pub fn both_sides(&self, node: &NodeId) -> Vec<Option<Vec<Vec<u8>>>> {
        let Some(g) = self.graph(node) else {
            return Vec::new();
        };
        let m = self.materialize_ordered(g);
        let mut out = Vec::new();
        for (i, seg) in m.segments.iter().enumerate() {
            match seg {
                crate::graph::Segment::Lines(_) => {}
                crate::graph::Segment::Conflict(sides) if g.adjacent_insertion(&m, i, self) => {
                    out.push(Some(
                        sides
                            .iter()
                            .map(|s| {
                                s.iter()
                                    .flat_map(|a| g.content(a).unwrap_or_default().to_vec())
                                    .collect()
                            })
                            .collect(),
                    ));
                }
                _ => out.push(None),
            }
        }
        out
    }

    pub fn render_file(&self, node: &NodeId) -> (Vec<u8>, Materialized) {
        let Some(g) = self.graph(node) else {
            return (Vec::new(), Materialized::default());
        };
        let m = self.materialize_ordered(g);
        let bytes = g.render(&m, &|h| self.label(h));
        (bytes, m)
    }

    /// Every file rendered, in parallel: files are independent, and
    /// rendering them is most of a snapshot's cost in a large tree.
    pub fn materialize(&self) -> Snapshot {
        use rayon::prelude::*;
        self.require_full("materializing a state");
        let tree = self.view();
        let entries: Vec<(&String, &crate::tree::TreeEntry)> = tree.entries.iter().collect();
        let rendered: Vec<(String, FileState)> = entries
            .par_iter()
            .map(|(path, e)| (path.to_string(), self.file_state(e)))
            .collect();
        let files: BTreeMap<String, FileState> = rendered.into_iter().collect();
        Snapshot {
            tree,
            files,
            unrendered: BTreeSet::new(),
        }
    }

    /// Like [`State::materialize`], leaving the text files of nodes in
    /// `skip` unrendered (listed in [`Snapshot::unrendered`]); every other
    /// file's graph must be loaded. For a caller that already knows those
    /// files are unchanged and not conflicted. Its tree is the tree
    /// graph's own view, as a Git import builds against it: same-path
    /// additions are not unified ([`State::view`]), since the import's
    /// merge records Git's tree for them itself.
    pub fn materialize_except(&self, skip: &HashSet<NodeId>) -> Snapshot {
        self.materialize_leaving(self.tree.view(), |_, e| skip.contains(&e.node))
    }

    /// As [`State::materialize_except`], rendering only the text files at
    /// `render` (a few, where the caller knows the rest unchanged), with
    /// the tree as [`State::view`] has it.
    pub fn materialize_only(&self, render: &std::collections::HashSet<&str>) -> Snapshot {
        self.materialize_leaving(self.view(), |p, _| !render.contains(p))
    }

    /// As [`State::materialize_only`], rendering only the text files of
    /// the nodes in `render` (a change's edited files), without a set of
    /// every other file to leave out.
    pub fn materialize_nodes(&self, render: &HashSet<NodeId>) -> Snapshot {
        self.materialize_leaving(self.view(), |_, e| !render.contains(&e.node))
    }

    fn materialize_leaving(
        &self,
        tree: TreeView,
        leave: impl Fn(&str, &crate::tree::TreeEntry) -> bool + Sync,
    ) -> Snapshot {
        use rayon::prelude::*;
        let entries: Vec<(&String, &crate::tree::TreeEntry)> = tree.entries.iter().collect();
        let rendered: Vec<(String, FileState, bool)> = entries
            .par_iter()
            .map(|(path, e)| {
                if e.kind == NodeKind::File && !e.mode.opaque && leave(path, e) {
                    let f = FileState {
                        node: e.node,
                        kind: e.kind,
                        mode: e.mode,
                        content: FileContent::Text {
                            bytes: Vec::new(),
                            materialized: Materialized::default(),
                        },
                    };
                    (path.to_string(), f, true)
                } else {
                    (path.to_string(), self.file_state(e), false)
                }
            })
            .collect();
        // In the view's path order, so the maps are built without searching.
        let unrendered: BTreeSet<String> = rendered
            .iter()
            .filter(|(_, _, skipped)| *skipped)
            .map(|(p, _, _)| p.clone())
            .collect();
        let files: BTreeMap<String, FileState> =
            rendered.into_iter().map(|(p, f, _)| (p, f)).collect();
        Snapshot {
            tree,
            files,
            unrendered,
        }
    }

    /// A snapshot of only `paths` (those `view` has), `view` being this
    /// state's tree view: for a caller that works from what changed. The
    /// graphs of those text files must be loaded.
    pub fn materialize_paths<'p>(
        &self,
        view: crate::tree::TreeView,
        paths: impl IntoIterator<Item = &'p str>,
    ) -> Snapshot {
        use rayon::prelude::*;
        // On every core: a merge may render thousands (oxygen#96).
        let entries: Vec<(&str, &crate::tree::TreeEntry)> = paths
            .into_iter()
            .filter_map(|p| Some((p, view.entries.get(p)?)))
            .collect();
        let files = entries
            .par_iter()
            .map(|(p, e)| (p.to_string(), self.file_state(e)))
            .collect::<Vec<_>>()
            .into_iter()
            .collect();
        Snapshot {
            tree: view,
            files,
            unrendered: BTreeSet::new(),
        }
    }

    /// The state of one tree entry: a directory, an opaque file's live
    /// blobs, or a text file rendered from its graph (which must be built).
    pub fn entry_state(&self, e: &crate::tree::TreeEntry) -> FileState {
        self.file_state(e)
    }

    fn file_state(&self, e: &crate::tree::TreeEntry) -> FileState {
        {
            let content = match e.kind {
                NodeKind::Dir => FileContent::Dir,
                NodeKind::Symlink | NodeKind::Subrepo => FileContent::Opaque {
                    blobs: self.live_blobs(&e.node),
                },
                NodeKind::File if e.mode.opaque => FileContent::Opaque {
                    blobs: self.live_blobs(&e.node),
                },
                NodeKind::File => {
                    let (bytes, materialized) = self.render_file(&e.node);
                    FileContent::Text {
                        bytes,
                        materialized,
                    }
                }
            };
            FileState {
                node: e.node,
                kind: e.kind,
                mode: e.mode,
                content,
            }
        }
    }
}

impl DepOracle for State {
    /// Whether `a` depends on `b`, by a walk down from `a` that stops at
    /// changes no higher than `b`: they cannot have `b` below them. Nothing
    /// of the walk is kept, so asking about many pairs costs no more memory
    /// than one walk (#25); a closure already memoised answers at once.
    fn depends_on(&self, a: &ChangeHash, b: &ChangeHash) -> bool {
        if a == b {
            return true;
        }
        if let Some(c) = self.closure.read().unwrap().sets.get(a) {
            return c.contains(b);
        }
        if !self.changes.contains_key(b) {
            return false;
        }
        let lb = self.level(b);
        if self.level(a) <= lb {
            return false;
        }
        let mut seen: crate::fasthash::FastSet<ChangeHash> = Default::default();
        // A dependency is looked for as it is found, before the walk goes
        // below it: most questions are about a direct one (spec §8.1 makes
        // a deletion depend on the inserter of the line after it, which is
        // what a render asks about), and a depth-first walk that only
        // looked when it came back up went through thousands of changes
        // first (oxygen#137: 4.4 billion steps rendering the VS Code tree).
        let mut stack: Vec<ChangeHash> = self
            .changes
            .get(a)
            .map(|m| m.deps.clone())
            .unwrap_or_default();
        if stack.contains(b) {
            return true;
        }
        while let Some(c) = stack.pop() {
            if c == *b {
                return true;
            }
            if !seen.insert(c) || self.level(&c) <= lb {
                continue;
            }
            if let Some(m) = self.changes.get(&c) {
                if m.deps.contains(b) {
                    return true;
                }
                stack.extend(m.deps.iter().copied());
            }
        }
        false
    }
}

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

    /// A changes block of thousands comes back whole and in order, and a
    /// truncated one is refused (oxygen#94).
    #[test]
    fn a_long_changes_block_decodes_in_order() {
        use crate::flat::Writer;
        let metas: Vec<(ChangeHash, ChangeMeta)> = (0..5000u32)
            .map(|i| {
                let mut h = [0u8; 32];
                h[..4].copy_from_slice(&i.to_le_bytes());
                let meta = ChangeMeta {
                    change_id: crate::ids::ChangeId([i as u8; 16]),
                    deps: (0..i % 3).map(|d| ObjectRef([d as u8; 32])).collect(),
                    author: ObjectRef([7; 32]),
                    hlc: Hlc {
                        wall_ms: 1_000_000 - i as u64,
                        logical: i,
                    },
                    message: "x".repeat((i % 50) as usize),
                    kind: ChangeKind::Normal,
                };
                (ObjectRef(h), meta)
            })
            .collect();
        let mut w = Writer::default();
        w.fixed(State::INDEX_MAGIC);
        w.count(metas.len());
        for (h, m) in &metas {
            w.object(h);
            m.write_flat(&mut w);
        }
        assert_eq!(State::read_index_changes_listed(&w.0).unwrap(), metas);
        let mut cut = w.0.clone();
        cut.pop();
        assert!(
            State::read_index_changes_listed(&cut).is_err(),
            "a truncated block is refused"
        );
    }
    use crate::ids::ChangeId;
    use crate::objects::{AtomRange, ChangeKind, Hlc, OpRef};

    fn change(deps: Vec<ChangeHash>, tree_ops: Vec<TreeOp>, edits: Vec<Edit>) -> Change {
        let mut c = Change {
            version: crate::objects::CHANGE_VERSION,
            change_id: ChangeId::random(),
            deps,
            author: ObjectRef::ZERO,
            hlc: Hlc {
                wall_ms: 1,
                logical: 0,
            },
            message: "m".into(),
            tree_ops,
            edits,
            kind: ChangeKind::Normal,
        };
        c.normalize();
        c
    }

    #[test]
    fn build_edit_and_materialize() {
        let mut st = State::new();
        let f = NodeId::random();
        let c1 = change(
            vec![],
            vec![TreeOp::AddNode {
                node: f,
                kind: NodeKind::File,
                mode: Mode::TEXT,
                parent: NodeId::ROOT,
                name: "a.txt".into(),
            }],
            vec![Edit::Insert {
                node: f,
                after: AtomRef::Start,
                before: AtomRef::End,
                lines: vec![b"one\n".to_vec(), b"two\n".to_vec()].into(),
            }],
        );
        let h1 = st.add(c1).unwrap();
        // Depends on h1: append a line and delete "one".
        let c2 = change(
            vec![h1],
            vec![],
            vec![
                Edit::Delete {
                    node: f,
                    atoms: vec![AtomRange {
                        change: h1,
                        from: 0,
                        to: 1,
                    }],
                },
                Edit::Insert {
                    node: f,
                    after: AtomRef::Atom {
                        change: h1,
                        index: 1,
                    },
                    before: AtomRef::End,
                    lines: vec![b"three\n".to_vec()].into(),
                },
            ],
        );
        let h2 = st.add(c2).unwrap();
        assert_eq!(st.heads(), vec![h2]);
        assert!(st.depends_on(&h2, &h1));
        assert!(!st.depends_on(&h1, &h2));
        let snap = st.materialize();
        assert!(!snap.is_conflicted());
        let a = &snap.files["a.txt"];
        assert!(matches!(&a.content, FileContent::Text { bytes, .. } if bytes == b"two\nthree\n"));
        assert_eq!(st.live_lines(&f).len(), 2);
    }

    #[test]
    fn missing_dependency_is_rejected() {
        let mut st = State::new();
        let c = change(vec![ObjectRef([1; 32])], vec![], vec![]);
        assert!(matches!(st.add(c), Err(StateError::MissingDep(..))));
    }

    #[test]
    fn self_references_and_multi_insert_indexing() {
        let mut st = State::new();
        let f = NodeId::random();
        let me = OpRef {
            change: ObjectRef::ZERO,
            index: 0,
        };
        let c1 = change(
            vec![],
            vec![
                TreeOp::AddNode {
                    node: f,
                    kind: NodeKind::File,
                    mode: Mode::TEXT,
                    parent: NodeId::ROOT,
                    name: "x".into(),
                },
                TreeOp::SetMode {
                    node: f,
                    old: me,
                    mode: Mode {
                        exec: true,
                        opaque: false,
                    },
                },
            ],
            vec![
                Edit::Insert {
                    node: f,
                    after: AtomRef::Start,
                    before: AtomRef::End,
                    lines: vec![b"a\n".to_vec(), b"c\n".to_vec()].into(),
                },
                // Insert between the two lines just inserted, using self references.
                Edit::Insert {
                    node: f,
                    after: AtomRef::Atom {
                        change: ObjectRef::ZERO,
                        index: 0,
                    },
                    before: AtomRef::Atom {
                        change: ObjectRef::ZERO,
                        index: 1,
                    },
                    lines: vec![b"b\n".to_vec()].into(),
                },
            ],
        );
        st.add(c1).unwrap();
        let snap = st.materialize();
        let x = &snap.files["x"];
        assert_eq!(
            x.mode,
            Mode {
                exec: true,
                opaque: false
            }
        );
        assert!(matches!(&x.content, FileContent::Text { bytes, .. } if bytes == b"a\nb\nc\n"));
    }

    #[test]
    fn opaque_replace_and_conflict() {
        let mut st = State::new();
        let f = NodeId::random();
        let b1 = ObjectRef([1; 32]);
        let b2 = ObjectRef([2; 32]);
        let b3 = ObjectRef([3; 32]);
        let h1 = st
            .add(change(
                vec![],
                vec![TreeOp::AddNode {
                    node: f,
                    kind: NodeKind::File,
                    mode: Mode::OPAQUE,
                    parent: NodeId::ROOT,
                    name: "img".into(),
                }],
                vec![Edit::Replace {
                    node: f,
                    old: None,
                    new: Some(b1),
                }],
            ))
            .unwrap();
        assert_eq!(st.live_blobs(&f), vec![b1]);
        st.add(change(
            vec![h1],
            vec![],
            vec![Edit::Replace {
                node: f,
                old: Some(b1),
                new: Some(b2),
            }],
        ))
        .unwrap();
        st.add(change(
            vec![h1],
            vec![],
            vec![Edit::Replace {
                node: f,
                old: Some(b1),
                new: Some(b3),
            }],
        ))
        .unwrap();
        let snap = st.materialize();
        assert!(snap.is_conflicted());
        assert!(
            matches!(&snap.files["img"].content, FileContent::Opaque { blobs } if blobs == &vec![b2, b3])
        );
    }

    #[test]
    fn checkpoint_payload_round_trip_and_replay() {
        let mut st = State::new();
        let f = NodeId::random();
        let h1 = st
            .add(change(
                vec![],
                vec![TreeOp::AddNode {
                    node: f,
                    kind: NodeKind::File,
                    mode: Mode::TEXT,
                    parent: NodeId::ROOT,
                    name: "a".into(),
                }],
                vec![Edit::Insert {
                    node: f,
                    after: AtomRef::Start,
                    before: AtomRef::End,
                    lines: vec![b"1\n".to_vec(), b"2\n".to_vec()].into(),
                }],
            ))
            .unwrap();
        let h2 = st
            .add(change(
                vec![h1],
                vec![],
                vec![Edit::Delete {
                    node: f,
                    atoms: vec![AtomRange {
                        change: h1,
                        from: 0,
                        to: 1,
                    }],
                }],
            ))
            .unwrap();
        let payload = st.checkpoint_payload();
        let back = State::from_checkpoint_payload(&payload).unwrap();
        assert_eq!(back.len(), 2);
        assert_eq!(back.heads(), vec![h2]);
        assert_eq!(back.get(&h1).unwrap().message, "m");
        assert_eq!(back.checkpoint_payload(), payload, "deterministic");
        // Replaying a further change onto the restored state equals a full build.
        let c3 = change(
            vec![h2],
            vec![],
            vec![Edit::Insert {
                node: f,
                after: AtomRef::Atom {
                    change: h1,
                    index: 1,
                },
                before: AtomRef::End,
                lines: vec![b"3\n".to_vec()].into(),
            }],
        );
        let mut full = st.clone();
        full.add(c3.clone()).unwrap();
        let mut restored = back;
        restored.add(c3).unwrap();
        let (a, b) = (full.materialize(), restored.materialize());
        assert_eq!(
            a.files.keys().collect::<Vec<_>>(),
            b.files.keys().collect::<Vec<_>>()
        );
        assert_eq!(a.files["a"].content, b.files["a"].content);
        assert!(
            matches!(&b.files["a"].content, FileContent::Text { bytes, .. } if bytes == b"2\n3\n")
        );
        assert!(State::from_checkpoint_payload(b"nonsense").is_err());
    }

    /// History-only states (O10f): the same state as a full load once its
    /// graphs are built, by replay and from a version 2 checkpoint with
    /// later changes; unloaded graphs are never read as empty.
    #[test]
    fn graph_keys_change_exactly_when_a_graph_does() {
        let (f, g) = (NodeId::random(), NodeId::random());
        let add = |node: NodeId, name: &str| TreeOp::AddNode {
            node,
            kind: NodeKind::File,
            mode: Mode::TEXT,
            parent: NodeId::ROOT,
            name: name.into(),
        };
        let ins = |node: NodeId, line: &[u8]| Edit::Insert {
            node,
            after: AtomRef::Start,
            before: AtomRef::End,
            lines: vec![line.to_vec()].into(),
        };
        let c1 = change(
            vec![],
            vec![add(f, "f"), add(g, "g")],
            vec![ins(f, b"f\n"), ins(g, b"g\n")],
        );
        let h1 = c1.hash();
        let c2 = change(vec![h1], vec![], vec![ins(g, b"g2\n")]);
        let h2 = c2.hash();
        let c3 = change(vec![h2], vec![], vec![ins(f, b"f2\n")]);
        let load = |cs: &[&Change]| {
            let mut st = State::new_history_only();
            for c in cs {
                st.add((*c).clone()).unwrap();
            }
            st
        };
        let s1 = load(&[&c1]);
        let s2 = load(&[&c1, &c2]);
        let s3 = load(&[&c1, &c2, &c3]);
        let key = |s: &State, n: &NodeId| s.graph_key(n).unwrap();
        assert_eq!(key(&s1, &f), key(&load(&[&c1]), &f), "same state, same key");
        assert_eq!(key(&s1, &f), key(&s2, &f), "an edit elsewhere keeps it");
        assert_ne!(key(&s1, &g), key(&s2, &g), "an edit of the file changes it");
        assert_ne!(key(&s2, &f), key(&s3, &f));
        assert_ne!(key(&s1, &f), key(&s1, &g), "different files differ");
        let mut built = load(&[&c1]);
        built
            .load_graphs(&[f], |_| unreachable!(), |_| unreachable!())
            .unwrap();
        assert!(built.graph_key(&f).is_none(), "a built graph has no key");
        let mut full = State::new();
        full.add(c1.clone()).unwrap();
        assert!(full.graph_key(&f).is_none(), "nor has a full state");

        // Through a version 2 checkpoint of c1, keys name its graph blobs
        // or inline values: stable across loads, changed by later edits.
        let graphs = full.checkpoint_file_graphs();
        let index = full.checkpoint_index(&graphs, |enc| {
            crate::hash::hash_object(crate::hash::ObjectType::Blob, enc)
        });
        let cp = |extra: &[&Change]| {
            let mut st = State::from_checkpoint_history(&index).unwrap();
            for c in extra {
                st.add((*c).clone()).unwrap();
            }
            st
        };
        assert_eq!(key(&cp(&[]), &f), key(&cp(&[]), &f));
        assert_eq!(key(&cp(&[]), &f), key(&cp(&[&c2]), &f));
        assert_ne!(key(&cp(&[]), &g), key(&cp(&[&c2]), &g));
        assert_ne!(key(&cp(&[&c2]), &f), key(&cp(&[&c2, &c3]), &f));

        // Partial rendering: skipped files are listed and empty, the others
        // render as in a full state.
        let mut part = load(&[&c1, &c2]);
        part.load_graphs(&[g], |_| unreachable!(), |_| unreachable!())
            .unwrap();
        let snap = part.materialize_except(&[f].into_iter().collect());
        assert_eq!(snap.unrendered, ["f".to_string()].into_iter().collect());
        let mut whole = State::new();
        whole.add(c1.clone()).unwrap();
        whole.add(c2.clone()).unwrap();
        assert_eq!(snap.files["g"], whole.materialize().files["g"]);
        assert!(
            matches!(&snap.files["f"].content, FileContent::Text { bytes, .. } if bytes.is_empty())
        );
    }

    #[test]
    fn history_only_states_build_the_same_graphs_on_demand() {
        let small = NodeId::random();
        let big = NodeId::random();
        let add = |node: NodeId, name: &str| TreeOp::AddNode {
            node,
            kind: NodeKind::File,
            mode: Mode::TEXT,
            parent: NodeId::ROOT,
            name: name.into(),
        };
        let lines: crate::lines::Lines = (0..300)
            .map(|i| format!("line {i}\n").into_bytes())
            .collect();
        let c1 = change(
            vec![],
            vec![add(small, "small"), add(big, "big")],
            vec![
                Edit::Insert {
                    node: small,
                    after: AtomRef::Start,
                    before: AtomRef::End,
                    lines: vec![b"x\n".to_vec()].into(),
                },
                Edit::Insert {
                    node: big,
                    after: AtomRef::Start,
                    before: AtomRef::End,
                    lines,
                },
            ],
        );
        let h1 = c1.hash();
        let c2 = change(
            vec![h1],
            vec![],
            vec![
                Edit::Delete {
                    node: big,
                    atoms: vec![AtomRange {
                        change: h1,
                        from: 2,
                        to: 9,
                    }],
                },
                Edit::Insert {
                    node: big,
                    after: AtomRef::Atom {
                        change: h1,
                        index: 20,
                    },
                    before: AtomRef::Atom {
                        change: h1,
                        index: 21,
                    },
                    lines: vec![b"new a\n".to_vec(), b"new b\n".to_vec()].into(),
                },
                Edit::Insert {
                    node: small,
                    after: AtomRef::Atom {
                        change: h1,
                        index: 0,
                    },
                    before: AtomRef::End,
                    lines: vec![b"y\n".to_vec()].into(),
                },
            ],
        );
        let h2 = c2.hash();
        let c3 = change(
            vec![h2],
            vec![],
            vec![Edit::Delete {
                node: small,
                atoms: vec![AtomRange {
                    change: h1,
                    from: 0,
                    to: 1,
                }],
            }],
        );

        let mut full = State::new();
        for c in [&c1, &c2, &c3] {
            full.add(c.clone()).unwrap();
        }
        // By replay: nothing built until asked, then exactly the full graphs.
        let mut hist = State::new_history_only();
        for c in [&c1, &c2, &c3] {
            hist.add(c.clone()).unwrap();
        }
        assert!(!hist.is_full() && hist.files.is_empty(), "no graph built");
        assert_eq!(hist.len(), 3);
        hist.load_graphs(
            &[big],
            |_| unreachable!("no checkpoint"),
            |_| unreachable!(),
        )
        .unwrap();
        assert_eq!(
            hist.render_file(&big).0,
            full.render_file(&big).0,
            "one file on its own"
        );
        assert!(!hist.is_full());
        hist.load_graphs(&[small, big], |_| unreachable!(), |_| unreachable!())
            .unwrap();
        assert!(hist.is_full());
        assert_eq!(hist.checkpoint_payload(), full.checkpoint_payload());

        // From a version 2 checkpoint of c1+c2, then c3 applied later.
        let mut base = State::new();
        base.add(c1.clone()).unwrap();
        base.add(c2.clone()).unwrap();
        let graphs = base.checkpoint_file_graphs();
        let mut blobs: HashMap<ObjectRef, Vec<u8>> = HashMap::new();
        let index = base.checkpoint_index(&graphs, |enc| {
            let r = crate::hash::hash_object(crate::hash::ObjectType::Blob, enc);
            blobs.insert(r, enc.to_vec());
            r
        });
        let mut from_cp = State::from_checkpoint_history(&index).unwrap();
        from_cp.add(c3.clone()).unwrap();
        assert!(!from_cp.is_full());
        let mut fetched = 0;
        from_cp
            .load_graphs(
                &[small, big],
                |refs| {
                    fetched += refs.len();
                    refs.iter()
                        .map(|r| {
                            blobs
                                .get(r)
                                .cloned()
                                .ok_or_else(|| ObjectError::Schema("missing".into()))
                        })
                        .collect()
                },
                |_| unreachable!(),
            )
            .unwrap();
        assert_eq!(
            fetched, 1,
            "the large graph's blob, in one batch; the small one was inline"
        );
        assert_eq!(from_cp.checkpoint_payload(), full.checkpoint_payload());
    }

    #[test]
    #[should_panic(expected = "was not loaded")]
    fn a_history_only_state_never_reads_an_unloaded_graph_as_empty() {
        let f = NodeId::random();
        let mut st = State::new_history_only();
        st.add(change(
            vec![],
            vec![TreeOp::AddNode {
                node: f,
                kind: NodeKind::File,
                mode: Mode::TEXT,
                parent: NodeId::ROOT,
                name: "a".into(),
            }],
            vec![Edit::Insert {
                node: f,
                after: AtomRef::Start,
                before: AtomRef::End,
                lines: vec![b"a\n".to_vec()].into(),
            }],
        ))
        .unwrap();
        st.render_file(&f);
    }

    #[test]
    #[should_panic(expected = "needs every file graph")]
    fn materializing_a_history_only_state_is_refused() {
        let f = NodeId::random();
        let mut st = State::new_history_only();
        st.add(change(
            vec![],
            vec![TreeOp::AddNode {
                node: f,
                kind: NodeKind::File,
                mode: Mode::TEXT,
                parent: NodeId::ROOT,
                name: "a".into(),
            }],
            vec![Edit::Insert {
                node: f,
                after: AtomRef::Start,
                before: AtomRef::End,
                lines: vec![b"a\n".to_vec()].into(),
            }],
        ))
        .unwrap();
        st.materialize();
    }

    /// Version 2 payloads (spec §11): small graphs inline, large ones by
    /// blob; the state read back is the same as from version 1.
    #[test]
    fn checkpoint_index_round_trip_matches_version_1() {
        let mut st = State::new();
        let small = NodeId::random();
        let big = NodeId::random();
        let add = |node: NodeId, name: &str| TreeOp::AddNode {
            node,
            kind: NodeKind::File,
            mode: Mode::TEXT,
            parent: NodeId::ROOT,
            name: name.into(),
        };
        let big_lines: crate::lines::Lines = (0..400)
            .map(|i| format!("line {i} of a long file\n").into_bytes())
            .collect();
        let h1 = st
            .add(change(
                vec![],
                vec![add(small, "small"), add(big, "big")],
                vec![
                    Edit::Insert {
                        node: small,
                        after: AtomRef::Start,
                        before: AtomRef::End,
                        lines: vec![b"x\n".to_vec()].into(),
                    },
                    Edit::Insert {
                        node: big,
                        after: AtomRef::Start,
                        before: AtomRef::End,
                        lines: big_lines,
                    },
                ],
            ))
            .unwrap();
        st.add(change(
            vec![h1],
            vec![],
            vec![Edit::Delete {
                node: big,
                atoms: vec![AtomRange {
                    change: h1,
                    from: 3,
                    to: 7,
                }],
            }],
        ))
        .unwrap();

        let graphs = st.checkpoint_file_graphs();
        let mut blobs: HashMap<ObjectRef, Vec<u8>> = HashMap::new();
        let index = st.checkpoint_index(&graphs, |enc| {
            let r = crate::hash::hash_object(crate::hash::ObjectType::Blob, enc);
            blobs.insert(r, enc.to_vec());
            r
        });
        assert_eq!(blobs.len(), 1, "only the large graph is a blob of its own");
        assert_eq!(
            State::checkpoint_graph_refs(&index).unwrap(),
            blobs.keys().copied().collect::<Vec<_>>()
        );
        assert!(
            State::checkpoint_graph_refs(&st.checkpoint_payload())
                .unwrap()
                .is_empty(),
            "version 1 names none"
        );
        assert!(index.len() < st.checkpoint_payload().len());

        let back = State::from_checkpoint(&index, |refs| {
            refs.iter()
                .map(|r| {
                    blobs
                        .get(r)
                        .cloned()
                        .ok_or_else(|| ObjectError::Schema("missing".into()))
                })
                .collect()
        })
        .unwrap();
        assert_eq!(
            back.checkpoint_payload(),
            st.checkpoint_payload(),
            "the same state as version 1"
        );
        assert_eq!(
            back.materialize().files["big"].content,
            st.materialize().files["big"].content
        );
        // Deterministic: the same state gives the same index.
        assert_eq!(
            back.checkpoint_index(&back.checkpoint_file_graphs(), |enc| {
                crate::hash::hash_object(crate::hash::ObjectType::Blob, enc)
            }),
            index
        );
        // A missing graph blob is an error, not an empty file.
        assert!(
            State::from_checkpoint(&index, |_| Err(ObjectError::Schema("gone".into()))).is_err()
        );
        assert!(
            State::from_checkpoint_payload(&index).is_err(),
            "version 2 needs its blobs"
        );
        // Version 1 payloads still read, and unknown versions do not.
        assert!(State::from_checkpoint(&st.checkpoint_payload(), |_| unreachable!()).is_ok());
        let mut v3 = cbor::decode(&index).unwrap();
        if let Value::Map(m) = &mut v3 {
            m.insert(0, Value::Uint(3));
        }
        assert!(State::from_checkpoint(&cbor::encode(&v3), |_| unreachable!()).is_err());
    }

    #[test]
    fn maximal_elements() {
        let mut st = State::new();
        let h1 = st.add(change(vec![], vec![], vec![])).unwrap();
        let h2 = st.add(change(vec![h1], vec![], vec![])).unwrap();
        let h3 = st.add(change(vec![], vec![], vec![])).unwrap();
        let mut m = st.maximal(&[h1, h2, h3]);
        m.sort();
        let mut want = vec![h2, h3];
        want.sort();
        assert_eq!(m, want);
        assert_eq!(st.heads(), want);
    }

    /// The bounded walk of `depends_on` agrees with the closure on random
    /// graphs, and the memo stays within its bound.
    #[test]
    fn depends_on_matches_the_closure() {
        use rand::{Rng, SeedableRng};
        let mut rng = rand::rngs::StdRng::seed_from_u64(11);
        for _ in 0..10 {
            let mut st = State::new();
            let mut all: Vec<ChangeHash> = Vec::new();
            for _ in 0..80 {
                let deps: Vec<ChangeHash> =
                    all.iter().copied().filter(|_| rng.gen_bool(0.06)).collect();
                all.push(st.add(change(deps, vec![], vec![])).unwrap());
            }
            let unknown = ObjectRef([0xee; 32]);
            for a in &all {
                let want = st.closure(a).clone();
                st.closure.write().unwrap().sets.clear();
                st.closure.write().unwrap().held = 0;
                for b in all.iter().chain([&unknown]) {
                    assert_eq!(
                        st.depends_on(a, b),
                        a == b || want.contains(b),
                        "{a:?} on {b:?}"
                    );
                }
            }
            assert!(st.closure.write().unwrap().held <= CLOSURE_MEMO_HASHES);
        }
    }

    /// The one-walk `maximal` agrees with the definition (not in the
    /// closure of another element) on random graphs, with repeated and
    /// unknown hashes among the input.
    #[test]
    fn maximal_matches_its_definition() {
        use rand::{Rng, SeedableRng};
        let mut rng = rand::rngs::StdRng::seed_from_u64(7);
        for _ in 0..20 {
            let mut st = State::new();
            let mut all: Vec<ChangeHash> = Vec::new();
            for _ in 0..60 {
                let deps: Vec<ChangeHash> =
                    all.iter().copied().filter(|_| rng.gen_bool(0.05)).collect();
                all.push(st.add(change(deps, vec![], vec![])).unwrap());
            }
            let mut pick: Vec<ChangeHash> =
                all.iter().copied().filter(|_| rng.gen_bool(0.4)).collect();
            pick.extend(pick.clone().into_iter().take(3));
            pick.push(ObjectRef([0xee; 32]));
            let set: BTreeSet<ChangeHash> = pick.iter().copied().collect();
            let want: Vec<ChangeHash> = set
                .iter()
                .filter(|h| !set.iter().any(|o| o != *h && st.closure(o).contains(h)))
                .copied()
                .collect();
            assert_eq!(st.maximal(&pick), want);
            assert_eq!(st.maximal(&all), st.heads());
        }
    }
}