vivac 0.15.6

//! The derived index: a binary snapshot of a folded `Tree`, so a command can
//! skip re-reading and re-folding a log it has already folded once.
//!
//! `LOADING.md` §4 is the specification this file exists to satisfy. Five
//! rules from it are not this module's opinion -- they are what makes an
//! index different from a second copy of the truth:
//!
//! - **Deleting it changes no output, ever.** Every `Tree` this module hands
//!   back -- fresh from the index, or from the index plus a tail of new
//!   events -- is exactly what folding the whole log would have produced.
//!   The round-trip and tail-application tests at the bottom of this file
//!   check that directly; the crate's own byte-for-byte harness checked it
//!   against real command output.
//! - **A write never rewrites it.** `allow_persist` is the one knob `load`
//!   takes, and callers that may append to the log pass `false` -- see
//!   `ops::Ctx::load_for_write`.
//! - **A log that folds with an anomaly gets no index.** A `pending`
//!   forward reference or a repeated `num` (`Tree::has_pending`,
//!   `Tree::repeated_nums`) both depend on the fold that produced them to
//!   ever be fixed up or reported again; loading straight from the index
//!   skips that fold, so `persist` refuses to write while either is
//!   non-empty.
//! - **It never fails.** A missing file, the wrong magic, the wrong
//!   version, an offset that points outside the file, the control ULID not
//!   where the header says, or a permission this process does not have are
//!   all answered the same way: fall back to folding the whole log, never
//!   surface as a command failure.
//! - **It is written to a temporary file and renamed.** `write_atomically`
//!   is the only function that ever touches the real path, and only by
//!   renaming a sibling temporary file that already holds the whole thing.
//!
//! ## Format
//!
//! A fixed-width header, then: a node table in ascending `num` order; the
//! spans arena `Node::refs` and `Node::governs` point into; a flat table of
//! flags; the stack; the roots; the vivacs; and the text blob. The header
//! carries every section's own start, so reading one never has to walk the
//! ones before it.
//!
//! ## Staying current
//!
//! The header carries the log's own byte length and modification time. A
//! match means the index is exactly current. A mismatch means the log
//! either grew -- read from the header's own saved offset, once the last
//! event this index ever folded is confirmed still sitting where the header
//! says, by its ULID and its `seq` -- or changed underneath it, in which
//! case there is no tail to apply and the whole log is folded fresh.

use crate::anchor::AnchorRef;
use crate::event::{Event, Flag, Kind, State, VivacKind};
use crate::failure::Failure;
use crate::model::{
    fold, AgainstSpan, ArmSpan, LaneState, Node, Note, RawParts, Span, Tree, Vivac, Where,
};
use crate::store::Store;
use std::collections::BTreeMap;
use std::fs::{self, File};
use std::io::{BufRead, BufReader, Seek, SeekFrom, Write as IoWrite};
use std::path::Path;

const MAGIC: u64 = u64::from_le_bytes(*b"vivacIDX");
// `t594`: version 6's header carried one stack and four segment counters
// for the whole tree; version 7 carries a lane table instead, one stack
// and six counters per lane, plus whether `main` has ever been claimed.
// Version 8 adds the wheres table -- `Tree.wheres`, one photograph per
// `where.changed` folded. Version 9 widens each vivac record with its own
// `anchors`, one entry per repository the lane had declared when it wrote
// (`t594` task 4). Version 10 adds the two BRANCH MOVED candidate tables,
// `Tree.own_focus` and `Tree.other_focus` (`t594` task 6, §2.7). Version 11
// widens each lane record with `seq_wrote`, the seq of the last event that
// lane wrote of any kind (`t594` tramo 5 task 2). Version 12 widens each
// node record with `born_seq` and `born_lane`, the `seq` and the lane of
// its own `node.created` -- what `why`'s "born in lane" line used to get by
// folding the whole log on every call, `--full` or not (`t594` tramo 7): a
// record this shape read under an earlier version would misparse silently,
// which is exactly what a version bump exists to refuse instead.
// `Header::parse` refuses any version but this one and `try_load_index`
// falls back to folding the log, which is what the index is derived from --
// so bumping this needs no migration and no command. Version 13 changes no
// record's shape but what three spans inside it mean: `opened`, `closed`
// and a decision's `declared` used to carry the ten-character UTC date
// `clock::date_of` sliced at fold time, and now carry the full RFC 3339
// stamp instead, so the local date can be taken in the reader's own zone
// at print time rather than baked in as UTC when the index was written
// (`d797`, `f731`). A version-12 index still parses byte for byte -- the
// spans are still offsets into the same text arena -- but its text holds
// the old ten-character dates, and `date_of` on one of those returns it
// unchanged rather than reading it as an instant, so a reader would keep
// showing UTC dates from a stale index forever without the refusal below.
const FORMAT_VERSION: u32 = 13;
const ULID_LEN: usize = 26;
const SPAN_LEN: usize = 8;
const FLAG_RECORD_LEN: usize = 1 + SPAN_LEN;
/// A note's own moment and text, mirroring `FLAG_RECORD_LEN`'s shape: no tag
/// byte, since a note carries no enum the way a flag carries its kind.
const NOTE_RECORD_LEN: usize = SPAN_LEN * 2;
/// An arm is two spans into the text arena -- the folder it runs in and the
/// command itself (`d441`) -- so the flat arms table is shaped like the
/// notes table above it, one record per arm.
const ARM_RECORD_LEN: usize = SPAN_LEN * 2;
/// A decision's declaration: the pillar or rule's own `num`, the sentence
/// span, a presence byte for `declared` and its span. `t426` §1.3.
const AGAINST_RECORD_LEN: usize = 8 + SPAN_LEN + 1 + SPAN_LEN;
const NODE_RECORD_LEN: usize = ULID_LEN
    + 8
    + 1
    + 1
    + 8
    + 1
    + 1
    + 8 // born_seq
    + SPAN_LEN // born_lane
    + SPAN_LEN * 4
    + 1
    + SPAN_LEN
    + SPAN_LEN
    + SPAN_LEN
    + 4
    + 4
    + 4
    + 4
    + 4
    + 4
    + 4
    + 4
    + 1;

/// `LOADING.md` §4 "El umbral, con su número": a stale index is left alone
/// below this many pending events, because applying them in memory is cheap
/// enough to fit inside the write budget, and only a read is ever allowed to
/// pay for rewriting the file itself.
const TAIL_REFRESH_THRESHOLD: usize = 200;

/// Builds the `Tree` a command needs, using the derived index when it can.
///
/// `allow_persist` is `false` for a command that may append to the log: a
/// write must never pay the cost of rewriting the index (`LOADING.md` §4
/// "Cuándo se reescribe"), even though it is free to read a warm or stale
/// one exactly like a read does. The only error this can return is a
/// genuine failure to read `events` itself -- everything the index's own
/// file touches is caught internally and answered by folding the log --
/// or `t411` §13's refusal, once `events` itself holds a line only a newer
/// vivac could have written.
pub fn load(store: &Store, allow_persist: bool) -> Result<Tree, Failure> {
    if let Some(loaded) = try_load_index(store) {
        return Ok(match loaded {
            Loaded::Fresh(tree) => tree,
            Loaded::Grown {
                mut tree,
                tail_len,
                fold_end_offset,
                last,
                unterminated,
            } => {
                if allow_persist && tail_len > TAIL_REFRESH_THRESHOLD {
                    persist(store, &tree, fold_end_offset, last.as_ref());
                }
                // A trailing chunk with no `\n` yet is re-evaluated on every
                // read, so it counts for whoever reads now and is never
                // written down: what gets persisted above is only what will
                // not change again.
                tree.broken_lines += usize::from(unterminated);
                tree
            }
        });
    }
    let tail = read_tracked(&store.log(), 0)?;
    let mut tree = fold(&tail.events, tail.broken);
    if allow_persist {
        persist(store, &tree, tail.end_offset, tail.last.as_ref());
    }
    // Same reasoning as above: added only now, never persisted.
    tree.broken_lines += usize::from(tail.unterminated);
    Ok(tree)
}

enum Loaded {
    Fresh(Tree),
    Grown {
        tree: Tree,
        tail_len: usize,
        fold_end_offset: u64,
        last: Option<LastEvent>,
        /// Whether the tail read behind this tree stopped on a line with no
        /// `\n` yet. Carried out separately from `tree.broken_lines` so the
        /// caller can persist the committed count first and add this after.
        unterminated: bool,
    },
}

#[derive(Clone)]
pub(crate) struct LastEvent {
    pub(crate) line_offset: u64,
    pub(crate) id: String,
    pub(crate) seq: u64,
}

// ---------------------------------------------------------------------------
// Reading the log, tracking byte offsets. Separate from `Store::read_all`
// on purpose: that function's contract (broken lines counted and skipped,
// invalid UTF-8 propagated) is exercised by the rest of the suite already,
// and this module must reproduce it exactly for a tail applied on top of an
// old index to agree with a fresh fold -- `read_tracked_agrees_with_store_
// read_all` and `a_stale_index_picks_up_the_tail`, below, are what prove it
// does.
// ---------------------------------------------------------------------------

pub(crate) struct Tracked {
    pub(crate) events: Vec<Event>,
    pub(crate) broken: usize,
    /// Whether this read ended in a line with no `\n` yet. Kept apart from
    /// `broken`: it is not consumed, so the next read from the same offset
    /// sees the same bytes and would count them again -- a caller that
    /// keeps its own running total has to add this in fresh each time,
    /// never accumulate it.
    pub(crate) unterminated: bool,
    /// Byte offset at EOF, counted from the very start of the file
    /// regardless of `from_offset`.
    pub(crate) end_offset: u64,
    pub(crate) last: Option<LastEvent>,
}

/// A last line with no `\n` yet still counts as broken, the way a whole
/// read has always counted it -- reported through `unterminated`, apart
/// from `broken`, since it is not consumed: an append that stopped
/// mid-write is not a line until it ends, and `end_offset` never lands
/// inside one -- which is what lets a tail read agree with a whole fold.
pub(crate) fn read_tracked(path: &Path, from_offset: u64) -> Result<Tracked, Failure> {
    let f = match File::open(path) {
        Ok(f) => f,
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
            return Ok(Tracked {
                events: Vec::new(),
                broken: 0,
                unterminated: false,
                end_offset: from_offset,
                last: None,
            })
        }
        Err(e) => return Err(e.into()),
    };
    read_tracked_in(&f, path, from_offset)
}

/// The same read against a log already open, for a caller that keeps it
/// open between refreshes. The open, not the read, is what the system
/// charges for: reopening the log to read what another process appended
/// cost the write path its budget under contention (`f599`).
///
/// `path` still travels alongside `f`: the one line here that meets a
/// line neither valid JSON nor an unreadable-known-event -- `t411` §13's
/// own case -- re-reads the whole file from scratch to name the right
/// line number, and that has always meant its own independent open, not
/// this one's handle.
pub(crate) fn read_tracked_in(f: &File, path: &Path, from_offset: u64) -> Result<Tracked, Failure> {
    let mut reader = BufReader::new(f);
    reader.seek(SeekFrom::Start(from_offset))?;
    let mut cursor = from_offset;
    let mut events = Vec::new();
    let mut broken = 0usize;
    let mut unterminated = false;
    let mut last = None;
    let mut raw = Vec::new();
    loop {
        raw.clear();
        let n = reader.read_until(b'\n', &mut raw)?;
        if n == 0 {
            break;
        }
        if raw.last() != Some(&b'\n') {
            // An append that stopped mid-write: these bytes are not a line
            // yet. They still count as broken, the way a whole read has
            // always counted them, but `cursor` stays behind them, so
            // `end_offset` never lands inside a line and a tail read sees
            // exactly the bytes a whole fold sees. Kept apart from `broken`
            // rather than folded into it: unlike every other broken line,
            // these bytes are not consumed, so the next read from the same
            // offset counts them again, and a caller that accumulates its
            // own total needs to know that this one does not accumulate.
            if !String::from_utf8_lossy(&raw).trim().is_empty() {
                unterminated = true;
            }
            break;
        }
        let line_offset = cursor;
        cursor += n as u64;
        let mut bytes = raw.as_slice();
        if bytes.last() == Some(&b'\n') {
            bytes = &bytes[..bytes.len() - 1];
        }
        if bytes.last() == Some(&b'\r') {
            bytes = &bytes[..bytes.len() - 1];
        }
        let line = String::from_utf8(bytes.to_vec()).map_err(|_| {
            std::io::Error::new(
                std::io::ErrorKind::InvalidData,
                "stream did not contain valid UTF-8",
            )
        })?;
        if line.trim().is_empty() {
            continue;
        }
        match serde_json::from_str::<Event>(&line) {
            Ok(e) => {
                last = Some(LastEvent {
                    line_offset,
                    id: e.id.clone(),
                    seq: e.seq,
                });
                events.push(e);
            }
            Err(_) => match crate::event::unknown_reason_for(&line) {
                // `from_offset` may sit mid-file, so the line number this
                // read would report is only ever right when it starts at
                // byte zero. Rather than reconstruct that count, a tail
                // read that hits this case just re-reads the whole file --
                // `crate::store::read_all_from` -- which starts at zero and
                // so names the correct line.
                Some(_) => match crate::store::read_all_from(path) {
                    Err(e) => return Err(e),
                    // The log changed under the two reads and the full one
                    // no longer sees the problem: treat this line the way
                    // an ordinary broken line has always been treated.
                    Ok(_) => broken += 1,
                },
                None => broken += 1,
            },
        }
    }
    Ok(Tracked {
        events,
        broken,
        unterminated,
        end_offset: cursor,
        last,
    })
}

// ---------------------------------------------------------------------------
// Deciding whether the index on disk is usable.
// ---------------------------------------------------------------------------

fn fingerprint(path: &Path) -> (u64, i64, u32) {
    match fs::metadata(path) {
        Ok(m) => {
            let (secs, nanos) = m
                .modified()
                .ok()
                .and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok())
                .map(|d| (d.as_secs() as i64, d.subsec_nanos()))
                .unwrap_or((0, 0));
            (m.len(), secs, nanos)
        }
        Err(_) => (0, 0, 0),
    }
}

/// Whether the event `last` names still sits where it was read: the check
/// that tells "the log grew" apart from "the log changed" (`LOADING.md` §4
/// "Vigencia"). Shared by the derived index and the resident server.
pub(crate) fn event_still_at(log_path: &Path, last: &LastEvent) -> bool {
    let Ok(f) = File::open(log_path) else {
        return false;
    };
    event_still_at_in(&f, last)
}

/// The same check against a log already open, for a caller that keeps it
/// open between refreshes rather than reopening it on every one -- the
/// open, not the read, is what the system charges for (`f599`).
pub(crate) fn event_still_at_in(f: &File, last: &LastEvent) -> bool {
    let mut reader = BufReader::new(f);
    if reader.seek(SeekFrom::Start(last.line_offset)).is_err() {
        return false;
    }
    let mut raw = Vec::new();
    let n = match reader.read_until(b'\n', &mut raw) {
        Ok(n) => n,
        Err(_) => return false,
    };
    if n == 0 {
        return false;
    }
    let mut bytes = raw.as_slice();
    if bytes.last() == Some(&b'\n') {
        bytes = &bytes[..bytes.len() - 1];
    }
    if bytes.last() == Some(&b'\r') {
        bytes = &bytes[..bytes.len() - 1];
    }
    let Ok(line) = std::str::from_utf8(bytes) else {
        return false;
    };
    let Ok(e) = serde_json::from_str::<Event>(line) else {
        return false;
    };
    e.id == last.id && e.seq == last.seq
}

/// Confirms the last event this index ever folded is still sitting where
/// the header says it is -- the check that tells "the log grew" apart from
/// "the log changed", per `LOADING.md` §4 "Vigencia".
fn control_still_there(log_path: &Path, header: &Header) -> bool {
    if !header.has_last {
        // Nothing was ever folded, so there is nothing to confirm: growing
        // from byte zero needs no control check.
        return header.fold_end_offset == 0;
    }
    event_still_at(
        log_path,
        &LastEvent {
            line_offset: header.last_line_offset,
            id: header.last_ulid.clone(),
            seq: header.last_seq,
        },
    )
}

fn try_load_index(store: &Store) -> Option<Loaded> {
    let bytes = fs::read(store.index_path()).ok()?;
    let header = Header::parse(&bytes)?;
    let log_path = store.log();
    let (cur_len, cur_secs, cur_nanos) = fingerprint(&log_path);

    if header.log_len == cur_len && header.mtime_secs == cur_secs && header.mtime_nanos == cur_nanos
    {
        return build_tree(&bytes, &header).map(Loaded::Fresh);
    }

    if cur_len < header.fold_end_offset || !control_still_there(&log_path, &header) {
        return None;
    }

    let tail = read_tracked(&log_path, header.fold_end_offset).ok()?;
    let mut tree = build_tree(&bytes, &header)?;
    for e in &tail.events {
        tree.apply(e.seq, &e.ts, &e.lane, &e.payload);
    }
    // The tail carries broken lines that already sit behind `fold_end`, and
    // this tree gets persisted, so they have to be added in. `tail.unterminated`
    // is not: `load` adds that to what this returns only after persisting.
    tree.broken_lines += tail.broken;
    let last = tail.last.clone().or_else(|| {
        header.has_last.then(|| LastEvent {
            line_offset: header.last_line_offset,
            id: header.last_ulid.clone(),
            seq: header.last_seq,
        })
    });
    Some(Loaded::Grown {
        tree,
        tail_len: tail.events.len(),
        fold_end_offset: tail.end_offset,
        last,
        unterminated: tail.unterminated,
    })
}

// ---------------------------------------------------------------------------
// Writing.
// ---------------------------------------------------------------------------

/// Every id this format stores travels in a fixed-width slot: the record
/// layout is what makes the node table directly seekable rather than
/// something a reader has to scan. Every id this binary ever mints
/// (`id::ulid`) is exactly this shape, so the check only ever refuses a
/// hand-edited log -- the same log that a repeated `num` or a pending
/// reference already refuses, and for the same reason: better to keep
/// folding it whole than to store something the format cannot represent
/// without silently truncating it.
fn is_ulid_shaped(s: &str) -> bool {
    s.len() == ULID_LEN && s.is_ascii()
}

/// Best-effort: every failure -- an anomaly in the tree, a read-only
/// directory, a full disk -- is swallowed. `LOADING.md` §4 "Si no se puede
/// escribir, no pasa nada" and "Un log con anomalías no lleva índice".
fn persist(store: &Store, tree: &Tree, fold_end_offset: u64, last: Option<&LastEvent>) {
    if tree.has_pending() || !tree.repeated_nums.is_empty() {
        return;
    }
    let ids_fit = tree.nodes_sorted().iter().all(|n| is_ulid_shaped(&n.id))
        && tree.vivacs.iter().all(|v| is_ulid_shaped(&v.id))
        && last.is_none_or(|l| is_ulid_shaped(&l.id));
    if !ids_fit {
        return;
    }
    let (mtime_secs, mtime_nanos) = mtime_of(&store.log());
    let bytes = encode(tree, fold_end_offset, mtime_secs, mtime_nanos, last);
    let _ = write_atomically(&store.index_path(), &bytes);
}

fn mtime_of(path: &Path) -> (i64, u32) {
    match fs::metadata(path).and_then(|m| m.modified()) {
        Ok(t) => match t.duration_since(std::time::UNIX_EPOCH) {
            Ok(d) => (d.as_secs() as i64, d.subsec_nanos()),
            Err(_) => (0, 0),
        },
        Err(_) => (0, 0),
    }
}

/// Never leaves a half-written index behind: the real path is only ever
/// touched by a `rename` of a sibling temporary file that already holds the
/// whole thing. `LOADING.md` §4, and the same rule `t84`'s own registry
/// uses for the same reason.
fn write_atomically(path: &Path, bytes: &[u8]) -> std::io::Result<()> {
    let dir = path
        .parent()
        .ok_or_else(|| std::io::Error::other("index path has no parent"))?;
    let tmp = dir.join(format!("index.tmp.{}", crate::id::ulid()));
    let result = (|| -> std::io::Result<()> {
        let mut f = File::create(&tmp)?;
        f.write_all(bytes)?;
        drop(f);
        fs::rename(&tmp, path)
    })();
    if result.is_err() {
        let _ = fs::remove_file(&tmp);
    }
    result
}

// ---------------------------------------------------------------------------
// The header.
// ---------------------------------------------------------------------------

struct Header {
    seq: u64,
    fold_end_offset: u64,
    has_last: bool,
    last_line_offset: u64,
    last_ulid: String,
    /// The `seq` of the event at `last_line_offset`, kept apart from `seq`
    /// above: `seq` is the tree's own high-water mark, which a hand-edited
    /// log with out-of-order numbers could in principle push past the
    /// physically last line's own value. The control check has to compare
    /// against the line it is actually looking at, not the tree's summary
    /// of every line.
    last_seq: u64,
    log_len: u64,
    mtime_secs: i64,
    mtime_nanos: u32,
    next_num: u64,
    next_vivac_num: u64,
    /// Whether some lane other than the founding one has ever claimed
    /// `main` (`Tree::main_claimed`, `d597`).
    main_claimed: bool,
    broken_lines: u64,
    node_count: u64,
    spans_count: u64,
    flags_count: u64,
    notes_count: u64,
    arms_count: u64,
    against_count: u64,
    roots_count: u64,
    lanes_count: u64,
    wheres_count: u64,
    vivac_count: u64,
    own_focus_count: u64,
    other_focus_count: u64,
    nodes_offset: u64,
    spans_offset: u64,
    flags_offset: u64,
    notes_offset: u64,
    arms_offset: u64,
    against_offset: u64,
    roots_offset: u64,
    lanes_offset: u64,
    wheres_offset: u64,
    vivacs_offset: u64,
    own_focus_offset: u64,
    other_focus_offset: u64,
    text_offset: u64,
    text_len: u64,
    file_len: u64,
}

impl Header {
    fn parse(bytes: &[u8]) -> Option<Header> {
        let mut c = Cursor::new(bytes);
        if c.u64()? != MAGIC {
            return None;
        }
        if c.u32()? != FORMAT_VERSION {
            return None;
        }
        let h = Header {
            seq: c.u64()?,
            fold_end_offset: c.u64()?,
            has_last: c.bool_()?,
            last_line_offset: c.u64()?,
            last_ulid: c.fixed_str(ULID_LEN)?,
            last_seq: c.u64()?,
            log_len: c.u64()?,
            mtime_secs: c.i64()?,
            mtime_nanos: c.u32()?,
            next_num: c.u64()?,
            next_vivac_num: c.u64()?,
            main_claimed: c.bool_()?,
            broken_lines: c.u64()?,
            node_count: c.u64()?,
            spans_count: c.u64()?,
            flags_count: c.u64()?,
            notes_count: c.u64()?,
            arms_count: c.u64()?,
            against_count: c.u64()?,
            roots_count: c.u64()?,
            lanes_count: c.u64()?,
            wheres_count: c.u64()?,
            vivac_count: c.u64()?,
            own_focus_count: c.u64()?,
            other_focus_count: c.u64()?,
            nodes_offset: c.u64()?,
            spans_offset: c.u64()?,
            flags_offset: c.u64()?,
            notes_offset: c.u64()?,
            arms_offset: c.u64()?,
            against_offset: c.u64()?,
            roots_offset: c.u64()?,
            lanes_offset: c.u64()?,
            wheres_offset: c.u64()?,
            vivacs_offset: c.u64()?,
            own_focus_offset: c.u64()?,
            other_focus_offset: c.u64()?,
            text_offset: c.u64()?,
            text_len: c.u64()?,
            file_len: c.u64()?,
        };
        if h.file_len as usize != bytes.len() {
            return None;
        }
        h.check_bounds(bytes.len())?;
        Some(h)
    }

    /// Every section has to fit inside the file this header came from --
    /// "un desplazamiento apunta fuera" is one of `LOADING.md` §4's own
    /// named reasons to regenerate rather than trust what is on disk.
    fn check_bounds(&self, len: usize) -> Option<()> {
        let fits = |off: u64, count: u64, width: u64| -> Option<bool> {
            let size = count.checked_mul(width)?;
            let end = off.checked_add(size)?;
            Some(end as usize <= len)
        };
        if !fits(self.nodes_offset, self.node_count, NODE_RECORD_LEN as u64)? {
            return None;
        }
        if !fits(self.spans_offset, self.spans_count, SPAN_LEN as u64)? {
            return None;
        }
        if !fits(self.flags_offset, self.flags_count, FLAG_RECORD_LEN as u64)? {
            return None;
        }
        if !fits(self.notes_offset, self.notes_count, NOTE_RECORD_LEN as u64)? {
            return None;
        }
        if !fits(self.arms_offset, self.arms_count, ARM_RECORD_LEN as u64)? {
            return None;
        }
        if !fits(
            self.against_offset,
            self.against_count,
            AGAINST_RECORD_LEN as u64,
        )? {
            return None;
        }
        if !fits(self.roots_offset, self.roots_count, 8)? {
            return None;
        }
        let text_end = self.text_offset.checked_add(self.text_len)?;
        if text_end as usize > len {
            return None;
        }
        // The lanes, wheres, vivacs and BRANCH MOVED candidate tables are
        // self-delimiting, like the flat ones above are not: a lane's own
        // `name` and its repositories, a `where.changed` photograph's own
        // repositories, a vivac's `stack` and `working_set`, and every
        // candidate's own path and branch, are all variable-length. All
        // this can check up front is that the table starts inside the
        // file; a truncated record past that fails to parse on its own.
        if self.lanes_offset as usize > len {
            return None;
        }
        if self.wheres_offset as usize > len {
            return None;
        }
        if self.vivacs_offset as usize > len {
            return None;
        }
        if self.own_focus_offset as usize > len {
            return None;
        }
        if self.other_focus_offset as usize > len {
            return None;
        }
        Some(())
    }
}

#[allow(clippy::too_many_arguments)]
fn write_header(buf: &mut Vec<u8>, h: &Header) {
    write_u64(buf, MAGIC);
    write_u32(buf, FORMAT_VERSION);
    write_u64(buf, h.seq);
    write_u64(buf, h.fold_end_offset);
    write_bool(buf, h.has_last);
    write_u64(buf, h.last_line_offset);
    write_ulid(buf, &h.last_ulid);
    write_u64(buf, h.last_seq);
    write_u64(buf, h.log_len);
    buf.extend_from_slice(&h.mtime_secs.to_le_bytes());
    write_u32(buf, h.mtime_nanos);
    write_u64(buf, h.next_num);
    write_u64(buf, h.next_vivac_num);
    write_bool(buf, h.main_claimed);
    write_u64(buf, h.broken_lines);
    write_u64(buf, h.node_count);
    write_u64(buf, h.spans_count);
    write_u64(buf, h.flags_count);
    write_u64(buf, h.notes_count);
    write_u64(buf, h.arms_count);
    write_u64(buf, h.against_count);
    write_u64(buf, h.roots_count);
    write_u64(buf, h.lanes_count);
    write_u64(buf, h.wheres_count);
    write_u64(buf, h.vivac_count);
    write_u64(buf, h.own_focus_count);
    write_u64(buf, h.other_focus_count);
    write_u64(buf, h.nodes_offset);
    write_u64(buf, h.spans_offset);
    write_u64(buf, h.flags_offset);
    write_u64(buf, h.notes_offset);
    write_u64(buf, h.arms_offset);
    write_u64(buf, h.against_offset);
    write_u64(buf, h.roots_offset);
    write_u64(buf, h.lanes_offset);
    write_u64(buf, h.wheres_offset);
    write_u64(buf, h.vivacs_offset);
    write_u64(buf, h.own_focus_offset);
    write_u64(buf, h.other_focus_offset);
    write_u64(buf, h.text_offset);
    write_u64(buf, h.text_len);
    write_u64(buf, h.file_len);
}

fn header_len() -> usize {
    let placeholder = Header {
        seq: 0,
        fold_end_offset: 0,
        has_last: false,
        last_line_offset: 0,
        last_ulid: "0".repeat(ULID_LEN),
        last_seq: 0,
        log_len: 0,
        mtime_secs: 0,
        mtime_nanos: 0,
        next_num: 0,
        next_vivac_num: 0,
        main_claimed: false,
        broken_lines: 0,
        node_count: 0,
        spans_count: 0,
        flags_count: 0,
        notes_count: 0,
        arms_count: 0,
        against_count: 0,
        roots_count: 0,
        lanes_count: 0,
        wheres_count: 0,
        vivac_count: 0,
        own_focus_count: 0,
        other_focus_count: 0,
        nodes_offset: 0,
        spans_offset: 0,
        flags_offset: 0,
        notes_offset: 0,
        arms_offset: 0,
        against_offset: 0,
        roots_offset: 0,
        lanes_offset: 0,
        wheres_offset: 0,
        vivacs_offset: 0,
        own_focus_offset: 0,
        other_focus_offset: 0,
        text_offset: 0,
        text_len: 0,
        file_len: 0,
    };
    let mut buf = Vec::new();
    write_header(&mut buf, &placeholder);
    buf.len()
}

// ---------------------------------------------------------------------------
// Byte-level helpers.
// ---------------------------------------------------------------------------

struct Cursor<'a> {
    buf: &'a [u8],
    pos: usize,
}

impl<'a> Cursor<'a> {
    fn new(buf: &'a [u8]) -> Cursor<'a> {
        Cursor { buf, pos: 0 }
    }

    fn take(&mut self, n: usize) -> Option<&'a [u8]> {
        let end = self.pos.checked_add(n)?;
        let slice = self.buf.get(self.pos..end)?;
        self.pos = end;
        Some(slice)
    }

    fn u8(&mut self) -> Option<u8> {
        self.take(1).map(|b| b[0])
    }

    fn bool_(&mut self) -> Option<bool> {
        self.u8().map(|b| b != 0)
    }

    fn u32(&mut self) -> Option<u32> {
        self.take(4)
            .map(|b| u32::from_le_bytes(b.try_into().unwrap()))
    }

    fn u64(&mut self) -> Option<u64> {
        self.take(8)
            .map(|b| u64::from_le_bytes(b.try_into().unwrap()))
    }

    fn i64(&mut self) -> Option<i64> {
        self.take(8)
            .map(|b| i64::from_le_bytes(b.try_into().unwrap()))
    }

    fn span(&mut self) -> Option<Span> {
        Some(Span {
            start: self.u32()?,
            len: self.u32()?,
        })
    }

    fn fixed_str(&mut self, n: usize) -> Option<String> {
        String::from_utf8(self.take(n)?.to_vec()).ok()
    }

    fn str(&mut self) -> Option<String> {
        let len = self.u32()? as usize;
        String::from_utf8(self.take(len)?.to_vec()).ok()
    }
}

fn write_u8(buf: &mut Vec<u8>, v: u8) {
    buf.push(v);
}

fn write_bool(buf: &mut Vec<u8>, v: bool) {
    buf.push(v as u8);
}

fn write_u32(buf: &mut Vec<u8>, v: u32) {
    buf.extend_from_slice(&v.to_le_bytes());
}

fn write_u64(buf: &mut Vec<u8>, v: u64) {
    buf.extend_from_slice(&v.to_le_bytes());
}

fn write_span(buf: &mut Vec<u8>, s: Span) {
    write_u32(buf, s.start);
    write_u32(buf, s.len);
}

fn write_str(buf: &mut Vec<u8>, s: &str) {
    write_u32(buf, s.len() as u32);
    buf.extend_from_slice(s.as_bytes());
}

fn write_ulid(buf: &mut Vec<u8>, s: &str) {
    debug_assert_eq!(s.len(), ULID_LEN, "a ulid is always {ULID_LEN} bytes");
    buf.extend_from_slice(s.as_bytes());
}

// ---------------------------------------------------------------------------
// Enum <-> byte, by hand: no `repr(u8)` on `event.rs`'s own public enums,
// and no dependency to derive one.
// ---------------------------------------------------------------------------

fn kind_to_u8(k: Kind) -> u8 {
    match k {
        Kind::Goal => 0,
        Kind::Task => 1,
        Kind::Decision => 2,
        Kind::Question => 3,
        Kind::Constraint => 4,
        Kind::Finding => 5,
        Kind::Assumption => 6,
        Kind::Pillar => 7,
        Kind::Rule => 8,
    }
}

fn u8_to_kind(b: u8) -> Option<Kind> {
    Some(match b {
        0 => Kind::Goal,
        1 => Kind::Task,
        2 => Kind::Decision,
        3 => Kind::Question,
        4 => Kind::Constraint,
        5 => Kind::Finding,
        6 => Kind::Assumption,
        7 => Kind::Pillar,
        8 => Kind::Rule,
        _ => return None,
    })
}

fn state_to_u8(s: State) -> u8 {
    match s {
        State::Active => 0,
        State::Done => 1,
        State::Suspended => 2,
        State::Abandoned => 3,
        State::Superseded => 4,
    }
}

fn u8_to_state(b: u8) -> Option<State> {
    Some(match b {
        0 => State::Active,
        1 => State::Done,
        2 => State::Suspended,
        3 => State::Abandoned,
        4 => State::Superseded,
        _ => return None,
    })
}

fn vivac_kind_to_u8(k: VivacKind) -> u8 {
    match k {
        VivacKind::Push => 0,
        VivacKind::Pop => 1,
        VivacKind::Park => 2,
        VivacKind::Manual => 3,
        VivacKind::Auto => 4,
    }
}

fn u8_to_vivac_kind(b: u8) -> Option<VivacKind> {
    Some(match b {
        0 => VivacKind::Push,
        1 => VivacKind::Pop,
        2 => VivacKind::Park,
        3 => VivacKind::Manual,
        4 => VivacKind::Auto,
        _ => return None,
    })
}

fn flag_to_u8(f: Flag) -> u8 {
    match f {
        Flag::Suspect => 0,
        Flag::Review => 1,
        Flag::Stale => 2,
    }
}

fn u8_to_flag(b: u8) -> Option<Flag> {
    Some(match b {
        0 => Flag::Suspect,
        1 => Flag::Review,
        2 => Flag::Stale,
        _ => return None,
    })
}

// ---------------------------------------------------------------------------
// The node table and the flat flags table it points into.
// ---------------------------------------------------------------------------

struct NodeRaw {
    id: String,
    num: u64,
    kind: Kind,
    state: State,
    parent: Option<u64>,
    blocks: bool,
    forced_close: bool,
    born_seq: u64,
    born_lane: Span,
    title: Span,
    why: Span,
    outcome: Span,
    opened: Span,
    closed: Option<Span>,
    refs: Span,
    governs: Span,
    flags_offset: u32,
    flags_count: u32,
    notes_offset: u32,
    notes_count: u32,
    arms_offset: u32,
    arms_count: u32,
    against_offset: u32,
    against_count: u32,
    against_recorded: bool,
}

#[allow(clippy::too_many_arguments)]
fn write_node_record(
    buf: &mut Vec<u8>,
    n: &Node,
    flags_buf: &mut Vec<u8>,
    flags_cursor: &mut u32,
    notes_buf: &mut Vec<u8>,
    notes_cursor: &mut u32,
    arms_buf: &mut Vec<u8>,
    arms_cursor: &mut u32,
    against_buf: &mut Vec<u8>,
    against_cursor: &mut u32,
) {
    let start = buf.len();
    write_ulid(buf, &n.id);
    write_u64(buf, n.num);
    write_u8(buf, kind_to_u8(n.kind));
    write_u8(buf, state_to_u8(n.state));
    write_u64(buf, n.parent.unwrap_or(u64::MAX));
    write_bool(buf, n.blocks);
    write_bool(buf, n.forced_close);
    write_u64(buf, n.born_seq);
    write_span(buf, n.born_lane);
    write_span(buf, n.title);
    write_span(buf, n.why);
    write_span(buf, n.outcome);
    write_span(buf, n.opened);
    match n.closed {
        Some(s) => {
            write_bool(buf, true);
            write_span(buf, s);
        }
        None => {
            write_bool(buf, false);
            write_span(buf, Span::default());
        }
    }
    write_span(buf, n.refs);
    write_span(buf, n.governs);
    let flags_offset = *flags_cursor;
    for (&flag, &span) in &n.flags {
        write_u8(flags_buf, flag_to_u8(flag));
        write_span(flags_buf, span);
    }
    let flags_count = n.flags.len() as u32;
    *flags_cursor += flags_count;
    write_u32(buf, flags_offset);
    write_u32(buf, flags_count);
    let notes_offset = *notes_cursor;
    for note in &n.notes {
        write_span(notes_buf, note.at);
        write_span(notes_buf, note.text);
    }
    let notes_count = n.notes.len() as u32;
    *notes_cursor += notes_count;
    write_u32(buf, notes_offset);
    write_u32(buf, notes_count);
    let arms_offset = *arms_cursor;
    for arm in &n.arms {
        write_span(arms_buf, arm.dir);
        write_span(arms_buf, arm.command);
    }
    let arms_count = n.arms.len() as u32;
    *arms_cursor += arms_count;
    write_u32(buf, arms_offset);
    write_u32(buf, arms_count);
    let against_offset = *against_cursor;
    for a in &n.against {
        write_u64(against_buf, a.node);
        write_span(against_buf, a.why);
        match a.declared {
            Some(s) => {
                write_bool(against_buf, true);
                write_span(against_buf, s);
            }
            None => {
                write_bool(against_buf, false);
                write_span(against_buf, Span::default());
            }
        }
    }
    let against_count = n.against.len() as u32;
    *against_cursor += against_count;
    write_u32(buf, against_offset);
    write_u32(buf, against_count);
    write_bool(buf, n.against_recorded);
    debug_assert_eq!(buf.len() - start, NODE_RECORD_LEN);
}

fn read_node_record(c: &mut Cursor) -> Option<NodeRaw> {
    let id = c.fixed_str(ULID_LEN)?;
    let num = c.u64()?;
    let kind = u8_to_kind(c.u8()?)?;
    let state = u8_to_state(c.u8()?)?;
    let parent_raw = c.u64()?;
    let parent = (parent_raw != u64::MAX).then_some(parent_raw);
    let blocks = c.bool_()?;
    let forced_close = c.bool_()?;
    let born_seq = c.u64()?;
    let born_lane = c.span()?;
    let title = c.span()?;
    let why = c.span()?;
    let outcome = c.span()?;
    let opened = c.span()?;
    let closed_present = c.bool_()?;
    let closed_span = c.span()?;
    let closed = closed_present.then_some(closed_span);
    let refs = c.span()?;
    let governs = c.span()?;
    let flags_offset = c.u32()?;
    let flags_count = c.u32()?;
    let notes_offset = c.u32()?;
    let notes_count = c.u32()?;
    let arms_offset = c.u32()?;
    let arms_count = c.u32()?;
    let against_offset = c.u32()?;
    let against_count = c.u32()?;
    let against_recorded = c.bool_()?;
    Some(NodeRaw {
        id,
        num,
        kind,
        state,
        parent,
        blocks,
        forced_close,
        born_seq,
        born_lane,
        title,
        why,
        outcome,
        opened,
        closed,
        refs,
        governs,
        flags_offset,
        flags_count,
        notes_offset,
        notes_count,
        arms_offset,
        arms_count,
        against_offset,
        against_count,
        against_recorded,
    })
}

fn assemble_nodes(
    raw_nodes: Vec<NodeRaw>,
    flags_table: &[(Flag, Span)],
    notes_table: &[Note],
    arms_table: &[ArmSpan],
    against_table: &[AgainstSpan],
) -> Option<Vec<Node>> {
    let mut out = Vec::with_capacity(raw_nodes.len());
    for r in raw_nodes {
        let start = r.flags_offset as usize;
        let end = start.checked_add(r.flags_count as usize)?;
        let slice = flags_table.get(start..end)?;
        let mut flags = BTreeMap::new();
        for &(f, s) in slice {
            flags.insert(f, s);
        }
        let notes_start = r.notes_offset as usize;
        let notes_end = notes_start.checked_add(r.notes_count as usize)?;
        let notes = notes_table.get(notes_start..notes_end)?.to_vec();
        let arms_start = r.arms_offset as usize;
        let arms_end = arms_start.checked_add(r.arms_count as usize)?;
        let arms = arms_table.get(arms_start..arms_end)?.to_vec();
        let against_start = r.against_offset as usize;
        let against_end = against_start.checked_add(r.against_count as usize)?;
        let against = against_table.get(against_start..against_end)?.to_vec();
        out.push(Node {
            id: r.id,
            num: r.num,
            kind: r.kind,
            title: r.title,
            why: r.why,
            state: r.state,
            parent: r.parent,
            blocks: r.blocks,
            notes,
            outcome: r.outcome,
            refs: r.refs,
            governs: r.governs,
            opened: r.opened,
            closed: r.closed,
            forced_close: r.forced_close,
            flags,
            arms,
            against,
            against_recorded: r.against_recorded,
            born_seq: r.born_seq,
            born_lane: r.born_lane,
        });
    }
    Some(out)
}

// ---------------------------------------------------------------------------
// The vivacs table: variable-length records, self-delimiting via their own
// length prefixes -- `Vivac` never shared `Tree`'s text arena, so there is
// nothing to intern here, only to write down.
// ---------------------------------------------------------------------------

fn write_repo_anchor(buf: &mut Vec<u8>, r: &crate::event::RepoAnchor) {
    write_str(buf, &r.path);
    match &r.branch {
        Some(branch) => {
            write_bool(buf, true);
            write_str(buf, branch);
        }
        None => {
            write_bool(buf, false);
            write_str(buf, "");
        }
    }
    write_str(buf, &r.sha);
}

fn parse_repo_anchor(c: &mut Cursor) -> Option<crate::event::RepoAnchor> {
    let path = c.str()?;
    let branch_present = c.bool_()?;
    let branch_raw = c.str()?;
    let sha = c.str()?;
    Some(crate::event::RepoAnchor {
        path,
        branch: branch_present.then_some(branch_raw),
        sha,
    })
}

fn write_vivac(buf: &mut Vec<u8>, v: &Vivac) {
    write_ulid(buf, &v.id);
    write_u64(buf, v.num);
    write_u64(buf, v.seq);
    write_str(buf, &v.lane);
    write_u8(buf, vivac_kind_to_u8(v.kind));
    write_str(buf, &v.next_intent);
    write_str(buf, &v.anchor.kind);
    write_str(buf, &v.anchor.id);
    write_u32(buf, v.anchors.len() as u32);
    for r in &v.anchors {
        write_repo_anchor(buf, r);
    }
    match &v.node_ref {
        Some(s) => {
            write_bool(buf, true);
            write_str(buf, s);
        }
        None => {
            write_bool(buf, false);
            write_str(buf, "");
        }
    }
    write_str(buf, &v.label);
    write_str(buf, &v.ts);
    write_u32(buf, v.stack.len() as u32);
    for (a, b) in &v.stack {
        write_str(buf, a);
        write_str(buf, b);
    }
    write_u32(buf, v.working_set.len() as u32);
    for w in &v.working_set {
        write_str(buf, w);
    }
}

fn parse_vivacs(bytes: &[u8], header: &Header) -> Option<Vec<Vivac>> {
    let mut c = Cursor::new(bytes.get(header.vivacs_offset as usize..)?);
    let mut out = Vec::with_capacity(header.vivac_count as usize);
    for _ in 0..header.vivac_count {
        let id = c.fixed_str(ULID_LEN)?;
        let num = c.u64()?;
        let seq = c.u64()?;
        let lane = c.str()?;
        let kind = u8_to_vivac_kind(c.u8()?)?;
        let next_intent = c.str()?;
        let anchor_kind = c.str()?;
        let anchor_id = c.str()?;
        let anchors_count = c.u32()?;
        let mut anchors = Vec::with_capacity(anchors_count as usize);
        for _ in 0..anchors_count {
            anchors.push(parse_repo_anchor(&mut c)?);
        }
        let node_ref_present = c.bool_()?;
        let node_ref_raw = c.str()?;
        let node_ref = node_ref_present.then_some(node_ref_raw);
        let label = c.str()?;
        let ts = c.str()?;
        let stack_count = c.u32()?;
        let mut stack = Vec::with_capacity(stack_count as usize);
        for _ in 0..stack_count {
            let a = c.str()?;
            let b = c.str()?;
            stack.push((a, b));
        }
        let working_set_count = c.u32()?;
        let mut working_set = Vec::with_capacity(working_set_count as usize);
        for _ in 0..working_set_count {
            working_set.push(c.str()?);
        }
        out.push(Vivac {
            id,
            num,
            seq,
            lane,
            kind,
            stack,
            working_set,
            next_intent,
            anchor: AnchorRef {
                kind: anchor_kind,
                id: anchor_id,
            },
            anchors,
            node_ref,
            label,
            ts,
        });
    }
    Some(out)
}

// ---------------------------------------------------------------------------
// The lanes table: one variable-length record per lane, self-delimiting the
// same way the vivacs table above is -- a lane's own `name` and its
// repositories have no fixed width either. `t594`: this replaces the single
// stack and the four segment counters version 6 kept for the whole tree.
// ---------------------------------------------------------------------------

fn write_lane(buf: &mut Vec<u8>, key: &str, s: &LaneState) {
    write_str(buf, key);
    write_str(buf, &s.name);
    write_u32(buf, s.repos.len() as u32);
    for r in &s.repos {
        write_str(buf, &r.path);
        match &r.root {
            Some(root) => {
                write_bool(buf, true);
                write_str(buf, root);
            }
            None => {
                write_bool(buf, false);
                write_str(buf, "");
            }
        }
    }
    write_u32(buf, s.stack.len() as u32);
    for &n in &s.stack {
        write_u64(buf, n);
    }
    write_u64(buf, s.seq_change);
    write_u64(buf, s.seq_vivac);
    write_u64(buf, s.seq_wrote);
    write_u64(buf, s.seg_new);
    write_u64(buf, s.seg_closed);
    write_u64(buf, s.seg_notes);
    write_u64(buf, s.seg_events);
}

fn parse_lanes(bytes: &[u8], header: &Header) -> Option<BTreeMap<String, LaneState>> {
    let mut c = Cursor::new(bytes.get(header.lanes_offset as usize..)?);
    let mut out = BTreeMap::new();
    for _ in 0..header.lanes_count {
        let key = c.str()?;
        let name = c.str()?;
        let repos_count = c.u32()?;
        let mut repos = Vec::with_capacity(repos_count as usize);
        for _ in 0..repos_count {
            let path = c.str()?;
            let root_present = c.bool_()?;
            let root_raw = c.str()?;
            repos.push(crate::event::Repo {
                path,
                root: root_present.then_some(root_raw),
            });
        }
        let stack_count = c.u32()?;
        let mut stack = Vec::with_capacity(stack_count as usize);
        for _ in 0..stack_count {
            stack.push(c.u64()?);
        }
        out.insert(
            key,
            LaneState {
                name,
                repos,
                stack,
                seq_change: c.u64()?,
                seq_vivac: c.u64()?,
                seq_wrote: c.u64()?,
                seg_new: c.u64()?,
                seg_closed: c.u64()?,
                seg_notes: c.u64()?,
                seg_events: c.u64()?,
            },
        );
    }
    Some(out)
}

// ---------------------------------------------------------------------------
// The wheres table: one variable-length record per `where.changed` folded,
// self-delimiting the same way the lanes and vivacs tables above are -- a
// photograph's own repositories have no fixed width either. `t594`: this is
// `Tree.wheres`, kept whole and in log order rather than collapsed to the
// last one, since `why` (§5.4) needs the one in force at a node's own `seq`
// and not only the lane's most recent.
// ---------------------------------------------------------------------------

fn write_where_repo(buf: &mut Vec<u8>, r: &crate::event::WhereRepo) {
    write_str(buf, &r.path);
    match &r.branch {
        Some(branch) => {
            write_bool(buf, true);
            write_str(buf, branch);
        }
        None => {
            write_bool(buf, false);
            write_str(buf, "");
        }
    }
    match &r.sha {
        Some(sha) => {
            write_bool(buf, true);
            write_str(buf, sha);
        }
        None => {
            write_bool(buf, false);
            write_str(buf, "");
        }
    }
    write_bool(buf, r.rebasing);
    write_bool(buf, r.missing);
    write_bool(buf, r.withheld);
}

fn parse_where_repo(c: &mut Cursor) -> Option<crate::event::WhereRepo> {
    let path = c.str()?;
    let branch_present = c.bool_()?;
    let branch_raw = c.str()?;
    let sha_present = c.bool_()?;
    let sha_raw = c.str()?;
    Some(crate::event::WhereRepo {
        path,
        branch: branch_present.then_some(branch_raw),
        sha: sha_present.then_some(sha_raw),
        rebasing: c.bool_()?,
        missing: c.bool_()?,
        withheld: c.bool_()?,
    })
}

fn write_where(buf: &mut Vec<u8>, w: &Where) {
    write_u64(buf, w.seq);
    write_str(buf, &w.lane);
    write_u32(buf, w.repos.len() as u32);
    for r in &w.repos {
        write_where_repo(buf, r);
    }
}

fn parse_wheres(bytes: &[u8], header: &Header) -> Option<Vec<Where>> {
    let mut c = Cursor::new(bytes.get(header.wheres_offset as usize..)?);
    let mut out = Vec::with_capacity(header.wheres_count as usize);
    for _ in 0..header.wheres_count {
        let seq = c.u64()?;
        let lane = c.str()?;
        let repos_count = c.u32()?;
        let mut repos = Vec::with_capacity(repos_count as usize);
        for _ in 0..repos_count {
            repos.push(parse_where_repo(&mut c)?);
        }
        out.push(Where { seq, lane, repos });
    }
    Some(out)
}

// ---------------------------------------------------------------------------
// The two BRANCH MOVED candidate tables (`t594` task 6, §2.7): one
// variable-length record per entry, self-delimiting the same way the
// tables above are -- a lane's id, a repository's path and a branch name
// have no fixed width either.
// ---------------------------------------------------------------------------

fn write_own_focus(buf: &mut Vec<u8>, key: &(String, String, String), value: &(u64, u64)) {
    let (lane, path, branch) = key;
    let (seq, node) = value;
    write_str(buf, lane);
    write_str(buf, path);
    write_str(buf, branch);
    write_u64(buf, *seq);
    write_u64(buf, *node);
}

fn parse_own_focus(bytes: &[u8], header: &Header) -> Option<crate::model::OwnFocus> {
    let mut c = Cursor::new(bytes.get(header.own_focus_offset as usize..)?);
    let mut out = BTreeMap::new();
    for _ in 0..header.own_focus_count {
        let lane = c.str()?;
        let path = c.str()?;
        let branch = c.str()?;
        let seq = c.u64()?;
        let node = c.u64()?;
        out.insert((lane, path, branch), (seq, node));
    }
    Some(out)
}

fn write_other_focus(buf: &mut Vec<u8>, key: &(String, String), value: &(u64, String, u64)) {
    let (root, branch) = key;
    let (seq, lane, node) = value;
    write_str(buf, root);
    write_str(buf, branch);
    write_u64(buf, *seq);
    write_str(buf, lane);
    write_u64(buf, *node);
}

fn parse_other_focus(bytes: &[u8], header: &Header) -> Option<crate::model::OtherFocus> {
    let mut c = Cursor::new(bytes.get(header.other_focus_offset as usize..)?);
    let mut out = BTreeMap::new();
    for _ in 0..header.other_focus_count {
        let root = c.str()?;
        let branch = c.str()?;
        let seq = c.u64()?;
        let lane = c.str()?;
        let node = c.u64()?;
        out.insert((root, branch), (seq, lane, node));
    }
    Some(out)
}

// ---------------------------------------------------------------------------
// The remaining fixed-width sections, and putting it all together.
// ---------------------------------------------------------------------------

fn parse_nodes(bytes: &[u8], header: &Header) -> Option<Vec<NodeRaw>> {
    let mut c = Cursor::new(bytes.get(header.nodes_offset as usize..)?);
    let mut out = Vec::with_capacity(header.node_count as usize);
    for _ in 0..header.node_count {
        out.push(read_node_record(&mut c)?);
    }
    Some(out)
}

fn parse_flags(bytes: &[u8], header: &Header) -> Option<Vec<(Flag, Span)>> {
    let mut c = Cursor::new(bytes.get(header.flags_offset as usize..)?);
    let mut out = Vec::with_capacity(header.flags_count as usize);
    for _ in 0..header.flags_count {
        let tag = u8_to_flag(c.u8()?)?;
        let span = c.span()?;
        out.push((tag, span));
    }
    Some(out)
}

fn parse_notes(bytes: &[u8], header: &Header) -> Option<Vec<Note>> {
    let mut c = Cursor::new(bytes.get(header.notes_offset as usize..)?);
    let mut out = Vec::with_capacity(header.notes_count as usize);
    for _ in 0..header.notes_count {
        let at = c.span()?;
        let text = c.span()?;
        out.push(Note { at, text });
    }
    Some(out)
}

fn parse_spans(bytes: &[u8], header: &Header) -> Option<Vec<Span>> {
    let mut c = Cursor::new(bytes.get(header.spans_offset as usize..)?);
    let mut out = Vec::with_capacity(header.spans_count as usize);
    for _ in 0..header.spans_count {
        out.push(c.span()?);
    }
    Some(out)
}

/// The flat arms table: two spans per arm -- folder, then command -- in the
/// same per-node order `write_node_record` wrote them, so
/// `assemble_nodes`'s `[start..end]` slice lands on the right node's own
/// arms.
fn parse_arms(bytes: &[u8], header: &Header) -> Option<Vec<ArmSpan>> {
    let mut c = Cursor::new(bytes.get(header.arms_offset as usize..)?);
    let mut out = Vec::with_capacity(header.arms_count as usize);
    for _ in 0..header.arms_count {
        let dir = c.span()?;
        let command = c.span()?;
        out.push(ArmSpan { dir, command });
    }
    Some(out)
}

/// The flat declarations table: `node`, `why` and an optional `declared`
/// span per record, in the same per-node order `write_node_record` wrote
/// them. `t426` §1.4.
fn parse_against(bytes: &[u8], header: &Header) -> Option<Vec<AgainstSpan>> {
    let mut c = Cursor::new(bytes.get(header.against_offset as usize..)?);
    let mut out = Vec::with_capacity(header.against_count as usize);
    for _ in 0..header.against_count {
        let node = c.u64()?;
        let why = c.span()?;
        let declared_present = c.bool_()?;
        let declared_span = c.span()?;
        let declared = declared_present.then_some(declared_span);
        out.push(AgainstSpan {
            node,
            why,
            declared,
        });
    }
    Some(out)
}

fn parse_u64_list(bytes: &[u8], offset: u64, count: u64) -> Option<Vec<u64>> {
    let mut c = Cursor::new(bytes.get(offset as usize..)?);
    let mut out = Vec::with_capacity(count as usize);
    for _ in 0..count {
        out.push(c.u64()?);
    }
    Some(out)
}

fn parse_text(bytes: &[u8], header: &Header) -> Option<String> {
    let start = header.text_offset as usize;
    let end = start.checked_add(header.text_len as usize)?;
    String::from_utf8(bytes.get(start..end)?.to_vec()).ok()
}

fn build_tree(bytes: &[u8], header: &Header) -> Option<Tree> {
    let raw_nodes = parse_nodes(bytes, header)?;
    let flags_table = parse_flags(bytes, header)?;
    let notes_table = parse_notes(bytes, header)?;
    let arms_table = parse_arms(bytes, header)?;
    let against_table = parse_against(bytes, header)?;
    let nodes = assemble_nodes(
        raw_nodes,
        &flags_table,
        &notes_table,
        &arms_table,
        &against_table,
    )?;
    let spans = parse_spans(bytes, header)?;
    let roots = parse_u64_list(bytes, header.roots_offset, header.roots_count)?;
    let lanes = parse_lanes(bytes, header)?;
    let wheres = parse_wheres(bytes, header)?;
    let vivacs = parse_vivacs(bytes, header)?;
    let own_focus = parse_own_focus(bytes, header)?;
    let other_focus = parse_other_focus(bytes, header)?;
    let text = parse_text(bytes, header)?;
    Some(Tree::from_parts(RawParts {
        text,
        spans,
        nodes,
        roots,
        lanes,
        vivacs,
        wheres,
        own_focus,
        other_focus,
        next_vivac_num: header.next_vivac_num,
        seq: header.seq,
        next_num: header.next_num,
        broken_lines: header.broken_lines as usize,
        main_claimed: header.main_claimed,
    }))
}

fn encode(
    tree: &Tree,
    fold_end_offset: u64,
    mtime_secs: i64,
    mtime_nanos: u32,
    last: Option<&LastEvent>,
) -> Vec<u8> {
    let nodes = tree.nodes_sorted();
    let mut nodes_buf = Vec::new();
    let mut flags_buf = Vec::new();
    let mut flags_cursor = 0u32;
    let mut notes_buf = Vec::new();
    let mut notes_cursor = 0u32;
    let mut arms_buf = Vec::new();
    let mut arms_cursor = 0u32;
    let mut against_buf = Vec::new();
    let mut against_cursor = 0u32;
    for n in &nodes {
        write_node_record(
            &mut nodes_buf,
            n,
            &mut flags_buf,
            &mut flags_cursor,
            &mut notes_buf,
            &mut notes_cursor,
            &mut arms_buf,
            &mut arms_cursor,
            &mut against_buf,
            &mut against_cursor,
        );
    }
    let mut spans_buf = Vec::new();
    for &s in tree.raw_spans() {
        write_span(&mut spans_buf, s);
    }
    let mut roots_buf = Vec::new();
    for &r in &tree.roots {
        write_u64(&mut roots_buf, r);
    }
    // `BTreeMap` iterates in key order already, so the table on disk comes
    // out sorted for free -- the same determinism `nodes_sorted` gives the
    // node table above.
    let mut lanes_buf = Vec::new();
    for (key, s) in &tree.lanes {
        write_lane(&mut lanes_buf, key, s);
    }
    let mut wheres_buf = Vec::new();
    for w in &tree.wheres {
        write_where(&mut wheres_buf, w);
    }
    let mut vivacs_buf = Vec::new();
    for v in &tree.vivacs {
        write_vivac(&mut vivacs_buf, v);
    }
    let mut own_focus_buf = Vec::new();
    for (key, value) in &tree.own_focus {
        write_own_focus(&mut own_focus_buf, key, value);
    }
    let mut other_focus_buf = Vec::new();
    for (key, value) in &tree.other_focus {
        write_other_focus(&mut other_focus_buf, key, value);
    }
    let text = tree.raw_text();
    let text_bytes = text.as_bytes();

    let header_bytes = header_len() as u64;
    let nodes_offset = header_bytes;
    let spans_offset = nodes_offset + nodes_buf.len() as u64;
    let flags_offset = spans_offset + spans_buf.len() as u64;
    let notes_offset = flags_offset + flags_buf.len() as u64;
    let arms_offset = notes_offset + notes_buf.len() as u64;
    let against_offset = arms_offset + arms_buf.len() as u64;
    let roots_offset = against_offset + against_buf.len() as u64;
    let lanes_offset = roots_offset + roots_buf.len() as u64;
    let wheres_offset = lanes_offset + lanes_buf.len() as u64;
    let vivacs_offset = wheres_offset + wheres_buf.len() as u64;
    let own_focus_offset = vivacs_offset + vivacs_buf.len() as u64;
    let other_focus_offset = own_focus_offset + own_focus_buf.len() as u64;
    let text_offset = other_focus_offset + other_focus_buf.len() as u64;
    let file_len = text_offset + text_bytes.len() as u64;

    let (has_last, last_line_offset, last_ulid, last_seq) = match last {
        Some(l) => (true, l.line_offset, l.id.clone(), l.seq),
        None => (false, 0, "0".repeat(ULID_LEN), 0),
    };

    let header = Header {
        seq: tree.seq,
        fold_end_offset,
        has_last,
        last_line_offset,
        last_ulid,
        last_seq,
        // A persisted index always represents the log up to exactly the
        // byte it finished reading: the two never disagree, or a future
        // freshness match could trust bytes this index never folded.
        log_len: fold_end_offset,
        mtime_secs,
        mtime_nanos,
        next_num: tree.next_num,
        next_vivac_num: tree.next_vivac_num,
        main_claimed: tree.main_claimed,
        broken_lines: tree.broken_lines as u64,
        node_count: nodes.len() as u64,
        spans_count: tree.raw_spans().len() as u64,
        flags_count: (flags_buf.len() / FLAG_RECORD_LEN) as u64,
        notes_count: (notes_buf.len() / NOTE_RECORD_LEN) as u64,
        arms_count: (arms_buf.len() / ARM_RECORD_LEN) as u64,
        against_count: (against_buf.len() / AGAINST_RECORD_LEN) as u64,
        roots_count: tree.roots.len() as u64,
        lanes_count: tree.lanes.len() as u64,
        wheres_count: tree.wheres.len() as u64,
        vivac_count: tree.vivacs.len() as u64,
        own_focus_count: tree.own_focus.len() as u64,
        other_focus_count: tree.other_focus.len() as u64,
        nodes_offset,
        spans_offset,
        flags_offset,
        notes_offset,
        arms_offset,
        against_offset,
        roots_offset,
        lanes_offset,
        wheres_offset,
        vivacs_offset,
        own_focus_offset,
        other_focus_offset,
        text_offset,
        text_len: text_bytes.len() as u64,
        file_len,
    };

    let mut out = Vec::with_capacity(file_len as usize);
    write_header(&mut out, &header);
    debug_assert_eq!(out.len() as u64, header_bytes);
    out.extend_from_slice(&nodes_buf);
    out.extend_from_slice(&spans_buf);
    out.extend_from_slice(&flags_buf);
    out.extend_from_slice(&notes_buf);
    out.extend_from_slice(&arms_buf);
    out.extend_from_slice(&against_buf);
    out.extend_from_slice(&roots_buf);
    out.extend_from_slice(&lanes_buf);
    out.extend_from_slice(&wheres_buf);
    out.extend_from_slice(&vivacs_buf);
    out.extend_from_slice(&own_focus_buf);
    out.extend_from_slice(&other_focus_buf);
    out.extend_from_slice(text_bytes);
    out
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::anchor::AnchorRef;
    use crate::event::Body;

    /// A `Store` whose directory is removed when this value drops, whether
    /// the test that made it passed or panicked -- the same promise
    /// `tests/relocate.rs`'s own `Owned` and `tests/lanes.rs`'s own
    /// `RemoveOnDrop` already make for the folders outside every `Sandbox`.
    /// Two tests below never got past a bare `Store::create` at all, and
    /// left their directory behind on every run, not only a failing one.
    struct TmpStore(Store);

    impl std::ops::Deref for TmpStore {
        type Target = Store;
        fn deref(&self) -> &Store {
            &self.0
        }
    }

    impl Drop for TmpStore {
        fn drop(&mut self) {
            std::fs::remove_dir_all(&self.0.root).ok();
        }
    }

    fn tmp_store(name: &str) -> TmpStore {
        let dir = std::env::temp_dir().join(format!(
            "vivac-index-t-{name}-{}-{}",
            std::process::id(),
            crate::id::ulid()
        ));
        TmpStore(Store::create(&dir).unwrap())
    }

    /// Takes `store`'s own write lock and writes `events` raw under it.
    fn write_raw_locked(store: &Store, events: &[Event]) {
        let lock = store.lock_for_write().unwrap();
        store.write_raw(&lock, events).unwrap();
    }

    /// A deterministic stand-in for `id::ulid()`: every id this format
    /// stores is fixed-width, so a test fixture needs the same shape a real
    /// one has, not a short mnemonic like `"n1"`.
    fn fixed_id(n: u32) -> String {
        format!("{n:0>26}")
    }

    #[allow(clippy::too_many_arguments)]
    fn created(
        seq: u64,
        ulid: &str,
        num: u64,
        kind: Kind,
        parent: Option<&str>,
        title: &str,
        refs: Vec<String>,
        governs: Vec<String>,
    ) -> Event {
        Event {
            seq,
            id: fixed_id(seq as u32),
            ts: "2026-09-05T10:00:00Z".to_string(),
            actor: "a_test".to_string(),
            lane: "main".to_string(),
            payload: Body::NodeCreated {
                node: ulid.to_string(),
                num,
                kind,
                title: title.to_string(),
                why: "because it is needed".to_string(),
                parent: parent.map(str::to_string),
                blocks: false,
                refs,
                governs,
                arms: vec![],
                against: None,
            },
        }
    }

    /// Like `created`, with a rule's arms: the fixture
    /// `a_rule_round_trips_its_arms_through_the_index` needs, since
    /// `created` above stays the minimal shape every other fixture wants.
    /// `d441`: each arm is a folder and a command, not a bare string.
    #[allow(clippy::too_many_arguments)]
    fn created_with_arms(
        seq: u64,
        ulid: &str,
        num: u64,
        kind: Kind,
        parent: Option<&str>,
        title: &str,
        arms: Vec<crate::event::Arm>,
    ) -> Event {
        Event {
            seq,
            id: fixed_id(seq as u32),
            ts: "2026-09-05T10:00:00Z".to_string(),
            actor: "a_test".to_string(),
            lane: "main".to_string(),
            payload: Body::NodeCreated {
                node: ulid.to_string(),
                num,
                kind,
                title: title.to_string(),
                why: "because it is needed".to_string(),
                parent: parent.map(str::to_string),
                blocks: false,
                refs: vec![],
                governs: vec![],
                arms,
                against: None,
            },
        }
    }

    fn a_note(seq: u64, ulid: &str, note: &str) -> Event {
        a_note_at(seq, ulid, "2026-09-05T10:01:00Z", note)
    }

    /// Like `a_note`, but with its own `ts`: the fixture two distinct notes
    /// on the same node need to prove each keeps the moment it was written.
    fn a_note_at(seq: u64, ulid: &str, ts: &str, note: &str) -> Event {
        Event {
            seq,
            id: fixed_id(seq as u32),
            ts: ts.to_string(),
            actor: "a_test".to_string(),
            lane: "main".to_string(),
            payload: Body::NodeNoted {
                node: ulid.to_string(),
                note: note.to_string(),
            },
        }
    }

    fn a_flag(seq: u64, ulid: &str, flag: Flag, reason: &str) -> Event {
        Event {
            seq,
            id: fixed_id(seq as u32),
            ts: "2026-09-05T10:02:00Z".to_string(),
            actor: "a_test".to_string(),
            lane: "main".to_string(),
            payload: Body::FlagRaised {
                node: ulid.to_string(),
                flag,
                reason: reason.to_string(),
            },
        }
    }

    fn a_close(seq: u64, ulid: &str, outcome: &str) -> Event {
        Event {
            seq,
            id: fixed_id(seq as u32),
            ts: "2026-09-05T10:03:00Z".to_string(),
            actor: "a_test".to_string(),
            lane: "main".to_string(),
            payload: Body::StateChanged {
                node: ulid.to_string(),
                state: State::Done,
                outcome: outcome.to_string(),
                forced: false,
            },
        }
    }

    fn a_vivac(seq: u64, num: u64, root_id: &str) -> Event {
        Event {
            seq,
            id: fixed_id(seq as u32),
            ts: "2026-09-05T10:04:00Z".to_string(),
            actor: "a_test".to_string(),
            lane: "main".to_string(),
            payload: Body::VivacCreated {
                vivac: fixed_id(900 + seq as u32),
                num,
                kind: VivacKind::Manual,
                stack: vec![(root_id.to_string(), "Root".to_string())],
                working_set: vec!["src/lib.rs".to_string()],
                next_intent: "keep going".to_string(),
                anchor: AnchorRef {
                    kind: "git".to_string(),
                    id: "abc123".to_string(),
                },
                anchors: vec![],
                node_ref: Some(root_id.to_string()),
                label: "a stop".to_string(),
            },
        }
    }

    /// A dump of everything a command can observe about a `Tree`, so two
    /// trees built two different ways can be compared for equality without
    /// `Tree` itself needing to derive it.
    fn snapshot(tree: &Tree) -> String {
        let mut out = String::new();
        out.push_str(&format!(
            "seq={} next_num={} next_vivac_num={} broken={} main_claimed={} total={}\n",
            tree.seq,
            tree.next_num,
            tree.next_vivac_num,
            tree.broken_lines,
            tree.main_claimed,
            tree.total(),
        ));
        out.push_str(&format!("roots={:?}\n", tree.roots));
        // Every lane, not only the one this tree happens to be looked at
        // from: a round trip that only compared one lane would pass even
        // if the index had swallowed the rest (`t594` ruling F).
        for (key, s) in &tree.lanes {
            out.push_str(&format!(
                "lane key={key:?} name={:?} repos={:?} stack={:?} seq_change={} \
                 seq_vivac={} seg_new={} seg_closed={} seg_notes={} seg_events={}\n",
                s.name,
                s.repos,
                s.stack,
                s.seq_change,
                s.seq_vivac,
                s.seg_new,
                s.seg_closed,
                s.seg_notes,
                s.seg_events,
            ));
        }
        for w in &tree.wheres {
            out.push_str(&format!(
                "where seq={} lane={:?} repos={:?}\n",
                w.seq, w.lane, w.repos,
            ));
        }
        // `t594` task 6: both BRANCH MOVED candidate tables, in key order --
        // a round trip that only compared `wheres` above would pass even if
        // `write_own_focus`/`write_other_focus` had swallowed either whole.
        for (key, value) in &tree.own_focus {
            out.push_str(&format!("own_focus key={key:?} value={value:?}\n"));
        }
        for (key, value) in &tree.other_focus {
            out.push_str(&format!("other_focus key={key:?} value={value:?}\n"));
        }
        out.push_str(&format!("repeated_nums={}\n", tree.repeated_nums.len()));
        for n in tree.nodes_sorted() {
            out.push_str(&format!(
                "node num={} id={} kind={:?} state={:?} parent={:?} blocks={} forced={} \
                 born_seq={} born_lane={:?} \
                 title={:?} why={:?} note={:?} outcome={:?} opened={:?} closed={:?} \
                 refs={:?} governs={:?} flags={:?} arms={:?} against={:?} \
                 against_recorded={}\n",
                n.num,
                n.id,
                n.kind,
                n.state,
                n.parent,
                n.blocks,
                n.forced_close,
                n.born_seq,
                n.born_lane(tree),
                n.title(tree),
                n.why(tree),
                n.note(tree),
                n.outcome(tree),
                n.opened(tree),
                n.closed(tree),
                n.refs(tree),
                n.governs(tree),
                n.flags
                    .iter()
                    .map(|(f, s)| (f.word(), tree.text(*s)))
                    .collect::<Vec<_>>(),
                n.arms(tree),
                n.against(tree),
                n.against_recorded,
            ));
        }
        for v in &tree.vivacs {
            out.push_str(&format!(
                "vivac num={} id={} seq={} lane={:?} kind={:?} stack={:?} working_set={:?} \
                 next_intent={:?} anchor={:?} anchors={:?} node_ref={:?} label={:?} ts={:?}\n",
                v.num,
                v.id,
                v.seq,
                v.lane,
                v.kind,
                v.stack,
                v.working_set,
                v.next_intent,
                v.anchor,
                v.anchors,
                v.node_ref,
                v.label,
                v.ts,
            ));
        }
        out
    }

    fn a_varied_event_set() -> Vec<Event> {
        let root_id = fixed_id(1);
        let child_id = fixed_id(2);
        vec![
            created(
                1,
                &root_id,
                1,
                Kind::Goal,
                None,
                "Root goal",
                vec!["ref-a".to_string()],
                vec!["governs-a".to_string()],
            ),
            created(
                2,
                &child_id,
                2,
                Kind::Task,
                Some(&root_id),
                "Child task",
                vec![],
                vec![],
            ),
            a_note(3, &child_id, "a note on the child"),
            a_flag(4, &child_id, Flag::Suspect, "something fell over"),
            a_flag(5, &child_id, Flag::Review, "worth a second look"),
            a_close(6, &child_id, "done for now"),
            a_vivac(7, 1, &root_id),
        ]
    }

    #[test]
    fn read_tracked_agrees_with_store_read_all() {
        let store = tmp_store("agree");
        write_raw_locked(&store, &a_varied_event_set());
        let (want_events, want_broken) = store.read_all().unwrap();
        let got = read_tracked(&store.log(), 0).unwrap();
        assert_eq!(got.broken, want_broken);
        assert_eq!(got.events.len(), want_events.len());
        for (a, b) in got.events.iter().zip(want_events.iter()) {
            assert_eq!(a.id, b.id);
            assert_eq!(a.seq, b.seq);
        }
        std::fs::remove_dir_all(&store.root).ok();
    }

    /// `f599`: an append that stopped mid-write must not be read as a line,
    /// or a tail read on top of it disagrees with what a whole fold would
    /// see once the log finishes growing.
    #[test]
    fn read_tracked_stops_before_a_line_that_has_no_newline_yet() {
        let store = tmp_store("partial");
        let root_id = fixed_id(1);
        let child_id = fixed_id(2);
        let complete = vec![
            created(1, &root_id, 1, Kind::Goal, None, "Root", vec![], vec![]),
            created(
                2,
                &child_id,
                2,
                Kind::Task,
                Some(&root_id),
                "Child",
                vec![],
                vec![],
            ),
        ];
        write_raw_locked(&store, &complete);
        let complete_end = fs::metadata(store.log()).unwrap().len();

        let grandchild_id = fixed_id(3);
        let partial = serde_json::to_string(&created(
            3,
            &grandchild_id,
            3,
            Kind::Task,
            Some(&child_id),
            "Grandchild",
            vec![],
            vec![],
        ))
        .unwrap();
        {
            let mut f = File::options().append(true).open(store.log()).unwrap();
            f.write_all(partial.as_bytes()).unwrap();
        }

        let got = read_tracked(&store.log(), 0).unwrap();
        assert_eq!(
            got.events.len(),
            2,
            "the unfinished line must not count as an event"
        );
        assert_eq!(
            got.broken, 0,
            "the unfinished line is reported through `unterminated`, not folded into `broken`"
        );
        assert!(
            got.unterminated,
            "an unfinished line still counts as broken, the way a whole read counts it"
        );
        assert_eq!(
            got.end_offset, complete_end,
            "end_offset must not land inside the unfinished line"
        );

        // Completing the line lets a later read from `end_offset` pick it up,
        // exactly as a fresh fold over the whole file would see it.
        {
            let mut f = File::options().append(true).open(store.log()).unwrap();
            f.write_all(b"\n").unwrap();
        }
        let tail = read_tracked(&store.log(), got.end_offset).unwrap();
        assert_eq!(tail.events.len(), 1);
        assert_eq!(tail.events[0].id, grandchild_id);

        std::fs::remove_dir_all(&store.root).ok();
    }

    /// `f599`: `read_all_from` (`why`, `changes`) and `read_tracked`
    /// (`tree`, `check`, the resident server) have to agree on a torn
    /// tail, even one cut off in the middle of a multi-byte character --
    /// otherwise the same log is a hard error for one reading and
    /// silently ignored by the other.
    #[test]
    fn read_all_from_and_read_tracked_agree_on_a_tail_torn_mid_character() {
        let store = tmp_store("torn-char");
        let root_id = fixed_id(1);
        write_raw_locked(
            &store,
            &[created(
                1,
                &root_id,
                1,
                Kind::Goal,
                None,
                "Root",
                vec![],
                vec![],
            )],
        );
        let mut partial =
            format!("{{\"seq\":2,\"id\":\"{}\",\"note\":\"caf", fixed_id(2)).into_bytes();
        // The first byte of "é", with no second byte and no `\n`: a tail
        // torn mid-character, not just mid-line.
        partial.push(0xC3);
        {
            let mut f = File::options().append(true).open(store.log()).unwrap();
            f.write_all(&partial).unwrap();
        }

        let (all_events, all_broken) = crate::store::read_all_from(&store.log()).unwrap();
        let got = read_tracked(&store.log(), 0).unwrap();

        assert_eq!(got.events.len(), all_events.len());
        for (a, b) in got.events.iter().zip(all_events.iter()) {
            assert_eq!(a.id, b.id);
            assert_eq!(a.seq, b.seq);
        }
        assert_eq!(got.broken + usize::from(got.unterminated), all_broken);

        std::fs::remove_dir_all(&store.root).ok();
    }

    #[test]
    fn round_trip_preserves_everything_a_command_can_observe() {
        let events = a_varied_event_set();
        let fresh = fold(&events, 0);

        let store = tmp_store("roundtrip");
        write_raw_locked(&store, &events);

        let loaded = load(&store, true).expect("load should succeed");
        assert_eq!(snapshot(&fresh), snapshot(&loaded));
        assert!(
            store.index_path().is_file(),
            "a clean fold should be indexed"
        );

        // And loading again, now purely from the index (no tail to apply),
        // has to agree too.
        let loaded_again = load(&store, false).expect("load should succeed");
        assert_eq!(snapshot(&fresh), snapshot(&loaded_again));

        std::fs::remove_dir_all(&store.root).ok();
    }

    /// Overrides the lane a fixture's event was signed with, since the
    /// helpers above -- `created`, `push_of` -- are all written for the
    /// single-lane fixtures the rest of this file needs.
    fn on_lane(mut e: Event, lane: &str) -> Event {
        e.lane = lane.to_string();
        e
    }

    fn push_of(seq: u64, ulid: &str) -> Event {
        Event {
            seq,
            id: fixed_id(seq as u32),
            ts: "2026-09-05T10:05:00Z".to_string(),
            actor: "a_test".to_string(),
            lane: "main".to_string(),
            payload: Body::Pushed {
                node: ulid.to_string(),
            },
        }
    }

    /// The round trip that already exists, with three lanes instead of
    /// one. Compared with `snapshot`, which after ruling F prints every
    /// lane: a fix that only reached one of them would still pass a
    /// round-trip test that only had one to compare.
    #[test]
    fn a_tree_with_three_lanes_survives_the_round_trip() {
        let a_node = fixed_id(1);
        let b_node = fixed_id(2);
        let c_node = fixed_id(3);
        let events = vec![
            created(1, &a_node, 1, Kind::Goal, None, "A's root", vec![], vec![]),
            on_lane(
                created(2, &b_node, 2, Kind::Task, None, "B's own", vec![], vec![]),
                "b",
            ),
            on_lane(
                created(3, &c_node, 3, Kind::Task, None, "C's own", vec![], vec![]),
                "c",
            ),
            push_of(4, &a_node),
            on_lane(push_of(5, &b_node), "b"),
            on_lane(push_of(6, &c_node), "c"),
            Event {
                seq: 7,
                id: fixed_id(7),
                ts: "2026-09-05T10:06:00Z".to_string(),
                actor: "a_test".to_string(),
                lane: "b".to_string(),
                payload: Body::LaneDeclared {
                    lane: "b".to_string(),
                    name: "feature".to_string(),
                    repos: vec![crate::event::Repo {
                        path: "webapi".to_string(),
                        root: Some("abc123".to_string()),
                    }],
                },
            },
        ];
        let fresh = fold(&events, 0);
        assert_eq!(
            fresh.lanes.len(),
            3,
            "the fixture itself has to touch three lanes"
        );

        let store = tmp_store("three-lanes");
        write_raw_locked(&store, &events);

        let loaded = load(&store, true).expect("load should succeed");
        assert_eq!(snapshot(&fresh), snapshot(&loaded));
        assert!(
            store.index_path().is_file(),
            "a clean fold should be indexed"
        );

        // Purely from the index this time, with no tail to apply -- the
        // check that this really came off disk and not off the fallback
        // fold, `LOADING.md` §4.
        let loaded_again = load(&store, false).expect("load should succeed");
        assert_eq!(snapshot(&fresh), snapshot(&loaded_again));

        std::fs::remove_dir_all(&store.root).ok();
    }

    /// `t594` tramo 7: `born_seq` and `born_lane` are new to the node
    /// record, and a round trip that only compared the fields that already
    /// existed would pass even if `write_node_record`/`read_node_record`
    /// dropped both -- `f278`'s own shape.
    #[test]
    fn a_nodes_birth_seq_and_lane_survive_the_round_trip() {
        let a_node = fixed_id(1);
        let b_node = fixed_id(2);
        let events = vec![
            created(1, &a_node, 1, Kind::Goal, None, "A's root", vec![], vec![]),
            on_lane(
                created(2, &b_node, 2, Kind::Task, None, "B's own", vec![], vec![]),
                "b",
            ),
        ];
        let fresh = fold(&events, 0);
        assert_eq!(fresh.node(&a_node).unwrap().born_seq, 1);
        assert_eq!(fresh.node(&a_node).unwrap().born_lane(&fresh), "main");
        assert_eq!(fresh.node(&b_node).unwrap().born_seq, 2);
        assert_eq!(fresh.node(&b_node).unwrap().born_lane(&fresh), "b");

        let store = tmp_store("birth-seq-lane-roundtrip");
        write_raw_locked(&store, &events);

        let loaded = load(&store, true).expect("load should succeed");
        assert_eq!(loaded.node(&a_node).unwrap().born_seq, 1);
        assert_eq!(loaded.node(&a_node).unwrap().born_lane(&loaded), "main");
        assert_eq!(loaded.node(&b_node).unwrap().born_seq, 2);
        assert_eq!(loaded.node(&b_node).unwrap().born_lane(&loaded), "b");
        assert_eq!(snapshot(&fresh), snapshot(&loaded));

        // Purely from the index this time, with no tail to apply.
        let loaded_again = load(&store, false).expect("load should succeed");
        assert_eq!(snapshot(&fresh), snapshot(&loaded_again));

        std::fs::remove_dir_all(&store.root).ok();
    }

    /// The index is a derived cache, so what this really checks is that
    /// reading it back gives the same answers a fresh fold would: the
    /// brief and `why` both read `wheres` and neither refolds the log.
    #[test]
    fn the_wheres_survive_the_round_trip_with_their_lane_and_seq() {
        let a_node = fixed_id(1);
        let events = vec![
            created(1, &a_node, 1, Kind::Goal, None, "Root", vec![], vec![]),
            Event {
                seq: 2,
                id: fixed_id(2),
                ts: "2026-09-17T10:00:00Z".to_string(),
                actor: "a_test".to_string(),
                lane: "main".to_string(),
                payload: Body::WhereChanged {
                    repos: vec![crate::event::WhereRepo {
                        path: "webapi".to_string(),
                        branch: Some("develop".to_string()),
                        sha: Some("abc123".to_string()),
                        ..Default::default()
                    }],
                },
            },
        ];
        let fresh = fold(&events, 0);
        assert_eq!(fresh.wheres.len(), 1, "the fixture itself has to write one");

        let store = tmp_store("wheres-roundtrip");
        write_raw_locked(&store, &events);

        let loaded = load(&store, true).expect("load should succeed");
        assert_eq!(snapshot(&fresh), snapshot(&loaded));
        assert!(
            store.index_path().is_file(),
            "a clean fold should be indexed"
        );

        // Purely from the index this time, with no tail to apply.
        let loaded_again = load(&store, false).expect("load should succeed");
        assert_eq!(snapshot(&fresh), snapshot(&loaded_again));

        std::fs::remove_dir_all(&store.root).ok();
    }

    /// `t594` task 6: BRANCH MOVED's own two candidate tables. `own_focus`
    /// and `other_focus` are never printed by anything `snapshot` already
    /// walks, so a round trip that only compared the string above would
    /// pass even if `write_own_focus`/`write_other_focus` had swallowed
    /// either table entirely.
    #[test]
    fn the_branch_moved_candidate_tables_survive_the_round_trip() {
        let n1 = fixed_id(1);
        let n2 = fixed_id(7);
        let events = vec![
            created(1, &n1, 1, Kind::Goal, None, "Root", vec![], vec![]),
            Event {
                seq: 2,
                id: fixed_id(2),
                ts: "2026-09-17T10:00:00Z".to_string(),
                actor: "a_test".to_string(),
                lane: "main".to_string(),
                payload: Body::LaneDeclared {
                    lane: "main".to_string(),
                    name: "main".to_string(),
                    repos: vec![crate::event::Repo {
                        path: "webapi".to_string(),
                        root: Some("root-abc".to_string()),
                    }],
                },
            },
            Event {
                seq: 3,
                id: fixed_id(3),
                ts: "2026-09-17T10:00:01Z".to_string(),
                actor: "a_test".to_string(),
                lane: "main".to_string(),
                payload: Body::WhereChanged {
                    repos: vec![crate::event::WhereRepo {
                        path: "webapi".to_string(),
                        branch: Some("develop".to_string()),
                        ..Default::default()
                    }],
                },
            },
            Event {
                seq: 4,
                id: fixed_id(4),
                ts: "2026-09-17T10:00:02Z".to_string(),
                actor: "a_test".to_string(),
                lane: "main".to_string(),
                payload: Body::Pushed { node: n1.clone() },
            },
            Event {
                seq: 5,
                id: fixed_id(5),
                ts: "2026-09-17T10:00:03Z".to_string(),
                actor: "a_test".to_string(),
                lane: "sonar".to_string(),
                payload: Body::LaneDeclared {
                    lane: "sonar".to_string(),
                    name: "sonar".to_string(),
                    repos: vec![crate::event::Repo {
                        path: "service".to_string(),
                        root: Some("root-abc".to_string()),
                    }],
                },
            },
            Event {
                seq: 6,
                id: fixed_id(6),
                ts: "2026-09-17T10:00:04Z".to_string(),
                actor: "a_test".to_string(),
                lane: "sonar".to_string(),
                payload: Body::WhereChanged {
                    repos: vec![crate::event::WhereRepo {
                        path: "service".to_string(),
                        branch: Some("perf/sp".to_string()),
                        ..Default::default()
                    }],
                },
            },
            Event {
                seq: 7,
                id: n2.clone(),
                ts: "2026-09-17T10:00:05Z".to_string(),
                actor: "a_test".to_string(),
                lane: "sonar".to_string(),
                payload: Body::NodeCreated {
                    node: n2.clone(),
                    num: 2,
                    kind: Kind::Task,
                    title: "Optimize the SP".to_string(),
                    why: "because it is needed".to_string(),
                    parent: None,
                    blocks: false,
                    refs: vec![],
                    governs: vec![],
                    arms: vec![],
                    against: None,
                },
            },
            Event {
                seq: 8,
                id: fixed_id(8),
                ts: "2026-09-17T10:00:06Z".to_string(),
                actor: "a_test".to_string(),
                lane: "sonar".to_string(),
                payload: Body::Pushed { node: n2.clone() },
            },
        ];
        let fresh = fold(&events, 0);
        // Every push writes both of its own tables: one entry in `own_focus`
        // per lane that pushed, and one in `other_focus` per repository root
        // commit and branch either lane was on when it did.
        assert_eq!(fresh.own_focus.len(), 2, "main's and sonar's own candidate");
        assert_eq!(
            fresh.other_focus.len(),
            2,
            "root-abc on develop, and on perf/sp"
        );

        let store = tmp_store("branch-moved-candidates-roundtrip");
        write_raw_locked(&store, &events);

        let loaded = load(&store, true).expect("load should succeed");
        assert_eq!(snapshot(&fresh), snapshot(&loaded));
        assert!(
            store.index_path().is_file(),
            "a clean fold should be indexed"
        );

        // Purely from the index this time, with no tail to apply.
        let loaded_again = load(&store, false).expect("load should succeed");
        assert_eq!(snapshot(&fresh), snapshot(&loaded_again));

        std::fs::remove_dir_all(&store.root).ok();
    }

    /// `t594` task 4: a vivac's own `anchors` -- one entry per repository
    /// the lane had declared when it wrote -- is a table `snapshot` never
    /// used to print, so a round trip that only compared the string above
    /// would pass even if `write_vivac`/`parse_vivacs` had swallowed the
    /// field entirely.
    #[test]
    fn a_vivacs_anchors_survive_the_round_trip_with_their_branch_and_sha() {
        let a_node = fixed_id(1);
        let mut vivac = a_vivac(2, 1, &a_node);
        let Body::VivacCreated { anchors, .. } = &mut vivac.payload else {
            panic!("a_vivac always writes a vivac.created");
        };
        *anchors = vec![
            crate::event::RepoAnchor {
                path: "webapi".to_string(),
                branch: Some("develop".to_string()),
                sha: "abc123".to_string(),
            },
            crate::event::RepoAnchor {
                path: "infra".to_string(),
                branch: None,
                sha: "def456".to_string(),
            },
        ];
        let events = vec![
            created(1, &a_node, 1, Kind::Goal, None, "Root", vec![], vec![]),
            vivac,
        ];
        let fresh = fold(&events, 0);
        assert_eq!(
            fresh.vivacs[0].anchors.len(),
            2,
            "the fixture itself has to write two"
        );

        let store = tmp_store("vivac-anchors-roundtrip");
        write_raw_locked(&store, &events);

        let loaded = load(&store, true).expect("load should succeed");
        assert_eq!(snapshot(&fresh), snapshot(&loaded));
        assert!(
            store.index_path().is_file(),
            "a clean fold should be indexed"
        );

        // Purely from the index this time, with no tail to apply.
        let loaded_again = load(&store, false).expect("load should succeed");
        assert_eq!(snapshot(&fresh), snapshot(&loaded_again));

        std::fs::remove_dir_all(&store.root).ok();
    }

    /// `t594`: every other lane fixture in
    /// this file has `seg_notes == seg_closed == 0`, so transposing the
    /// two in `write_lane` left the whole suite green. This one gives
    /// every lane six counters that are all different from one another
    /// and from zero, one lane a declared name and repositories,
    /// `lane.claimed` so `main_claimed` travels too, and one vivac per
    /// lane so `Vivac.lane` does as well. Goes straight at `encode`/
    /// `build_tree`, the same way `two_notes_on_one_node_round_trip_
    /// through_the_index_alone` does, so there is no fallback fold to
    /// hide behind.
    #[test]
    fn a_lane_with_distinct_nonzero_counters_round_trips_through_the_index_alone() {
        let mut events: Vec<Event> = Vec::new();
        let mut seq = 0u64;
        let mut next_num = 0u64;
        let mut next_raw_id = 0u32;
        let mut fresh_id = || {
            next_raw_id += 1;
            fixed_id(next_raw_id)
        };

        events.push(Event {
            seq: {
                seq += 1;
                seq
            },
            id: fresh_id(),
            ts: "2026-09-16T09:00:00Z".to_string(),
            actor: "a_test".to_string(),
            lane: "a".to_string(),
            payload: Body::LaneDeclared {
                lane: "a".to_string(),
                name: "feature-a".to_string(),
                repos: vec![crate::event::Repo {
                    path: "webapi".to_string(),
                    root: Some("abc123".to_string()),
                }],
            },
        });
        events.push(Event {
            seq: {
                seq += 1;
                seq
            },
            id: fresh_id(),
            ts: "2026-09-16T09:00:01Z".to_string(),
            actor: "a_test".to_string(),
            lane: "c".to_string(),
            payload: Body::LaneClaimed {
                lane: "main".to_string(),
            },
        });

        // (lane, vivac num, new, closed, notes) -- chosen so the six
        // counters end up mutually distinct within each lane: `seg_events`
        // is their sum, `seq_vivac` is the vivac's own seq, and
        // `seq_change` is the seq of the last of the three kinds below.
        let plan = [
            ("a", 1u64, 5usize, 6usize, 7usize),
            ("b", 2u64, 4usize, 3usize, 2usize),
            ("c", 3u64, 2usize, 5usize, 1usize),
        ];

        for (lane, vivac_num, new_count, closed_count, notes_count) in plan {
            let vivac_root = fresh_id();
            events.push(on_lane(
                a_vivac(
                    {
                        seq += 1;
                        seq
                    },
                    vivac_num,
                    &vivac_root,
                ),
                lane,
            ));

            let mut ids = Vec::new();
            for _ in 0..new_count {
                next_num += 1;
                let id = fresh_id();
                events.push(on_lane(
                    created(
                        {
                            seq += 1;
                            seq
                        },
                        &id,
                        next_num,
                        Kind::Task,
                        None,
                        "node",
                        vec![],
                        vec![],
                    ),
                    lane,
                ));
                ids.push(id);
            }
            for i in 0..closed_count {
                let id = ids[i % ids.len()].clone();
                events.push(on_lane(
                    a_close(
                        {
                            seq += 1;
                            seq
                        },
                        &id,
                        "done",
                    ),
                    lane,
                ));
            }
            for i in 0..notes_count {
                let id = ids[i % ids.len()].clone();
                events.push(on_lane(
                    a_note(
                        {
                            seq += 1;
                            seq
                        },
                        &id,
                        "note",
                    ),
                    lane,
                ));
            }
        }

        let tree = fold(&events, 0);
        assert!(tree.main_claimed);
        for (lane, _, new_count, closed_count, notes_count) in plan {
            let s = tree.lanes.get(lane).expect("the lane was written to");
            assert_eq!(s.seg_new as usize, new_count);
            assert_eq!(s.seg_closed as usize, closed_count);
            assert_eq!(s.seg_notes as usize, notes_count);
            let mut six = vec![
                s.seq_change,
                s.seq_vivac,
                s.seg_new,
                s.seg_closed,
                s.seg_notes,
                s.seg_events,
            ];
            six.sort_unstable();
            six.dedup();
            assert_eq!(
                six.len(),
                6,
                "lane {lane} has two equal counters: the fixture is not adversarial enough"
            );
            assert!(
                [
                    s.seq_change,
                    s.seq_vivac,
                    s.seg_new,
                    s.seg_closed,
                    s.seg_notes,
                    s.seg_events
                ]
                .iter()
                .all(|&v| v > 0),
                "lane {lane} has a zero counter"
            );
        }

        let bytes = encode(&tree, 0, 0, 0, None);
        let header = Header::parse(&bytes).expect("the header this test just wrote parses");
        let loaded = build_tree(&bytes, &header).expect("the body this test just wrote parses");

        assert_eq!(snapshot(&tree), snapshot(&loaded));
    }

    /// `t594` tramo 5 task 2: `seq_wrote` moves on every event, context
    /// events included, so a fixture whose last write is a `lane.declared`
    /// rather than a node is what tells `seq_wrote` and `seq_change` apart
    /// -- reading one back into the other's slot would still pass every
    /// other lane fixture in this file, since none of them separates the
    /// two. Goes straight at `encode`/`build_tree`, not through `snapshot`,
    /// which does not print this field.
    #[test]
    fn a_lanes_seq_wrote_survives_the_round_trip_even_when_it_differs_from_seq_change() {
        let b_node = fixed_id(1);
        let events = vec![
            on_lane(
                created(1, &b_node, 1, Kind::Task, None, "B's root", vec![], vec![]),
                "b",
            ),
            Event {
                seq: 2,
                id: fixed_id(2),
                ts: "2026-09-18T00:00:00Z".to_string(),
                actor: "a_test".to_string(),
                lane: "b".to_string(),
                payload: Body::LaneDeclared {
                    lane: "b".to_string(),
                    name: "feature".to_string(),
                    repos: vec![],
                },
            },
        ];
        let tree = fold(&events, 0);
        let before = tree.lanes.get("b").expect("lane b wrote");
        assert_eq!(before.seq_change, 1, "the node write, not the declaration");
        assert_eq!(before.seq_wrote, 2, "the declaration moves it too");

        let bytes = encode(&tree, 0, 0, 0, None);
        let header = Header::parse(&bytes).expect("the header this test just wrote parses");
        let loaded = build_tree(&bytes, &header).expect("the body this test just wrote parses");

        let after = loaded
            .lanes
            .get("b")
            .expect("lane b survived the round trip");
        assert_eq!(after.seq_change, before.seq_change);
        assert_eq!(after.seq_wrote, before.seq_wrote);
    }

    /// A version-6 index -- the shape this crate wrote before lanes existed
    /// -- is discarded rather than misread, and the tree that comes out of
    /// the fallback is exactly what a fresh fold produces.
    #[test]
    fn an_index_of_the_previous_format_is_rebuilt() {
        let store = tmp_store("old-format");
        let events = a_varied_event_set();
        write_raw_locked(&store, &events);
        let want = fold(&events, 0);

        let mut bytes = Vec::new();
        bytes.extend_from_slice(&MAGIC.to_le_bytes());
        bytes.extend_from_slice(&6u32.to_le_bytes());
        bytes.extend_from_slice(&[0u8; 200]);
        fs::write(store.index_path(), &bytes).unwrap();

        let got = load(&store, false).unwrap();
        assert_eq!(snapshot(&want), snapshot(&got));

        std::fs::remove_dir_all(&store.root).ok();
    }

    /// `d797`/`f731`: version 12 is not just *a* previous format, it is the
    /// specific one whose `opened`/`closed`/`declared` spans held the
    /// ten-character UTC date `clock::date_of` used to slice at fold time.
    /// `Header::parse` bails out on the version field alone, before it ever
    /// looks at what those spans point to, so this needs no body shaped
    /// like a real version-12 record to prove the point -- the same
    /// garbage-body trick `an_index_of_the_previous_format_is_rebuilt` uses
    /// already demonstrates the refusal. What this test adds is the other
    /// half: the fold this falls back to is `fold(&events, 0)`, which now
    /// interns the full instant for those three spans, so `got` never
    /// carries the old ten-character dates a version-12 file on disk would
    /// have held -- a reader never gets stuck reading UTC out of it forever.
    #[test]
    fn a_version_12_index_is_rebuilt_with_full_instants_not_utc_dates() {
        let store = tmp_store("version-12");
        let events = a_varied_event_set();
        write_raw_locked(&store, &events);
        let want = fold(&events, 0);

        let mut bytes = Vec::new();
        bytes.extend_from_slice(&MAGIC.to_le_bytes());
        bytes.extend_from_slice(&12u32.to_le_bytes());
        bytes.extend_from_slice(&[0u8; 200]);
        fs::write(store.index_path(), &bytes).unwrap();

        let got = load(&store, false).unwrap();
        assert_eq!(snapshot(&want), snapshot(&got));
        for n in got.nodes_sorted() {
            assert!(
                crate::clock::epoch_seconds(n.opened(&got)).is_some(),
                "opened={:?} is not a full instant",
                n.opened(&got)
            );
        }

        std::fs::remove_dir_all(&store.root).ok();
    }

    /// `d390`: `note` went from one `Span` to a flat table the node table
    /// points into, the same shape change `flags` already went through.
    ///
    /// This calls `encode`/`build_tree` directly rather than through `load`:
    /// `load` falls back to folding the log whenever the index fails to
    /// parse (`LOADING.md` §4 "nunca falla"), and the log it would fall back
    /// to is sitting right there, untouched, with the same two notes in it.
    /// A `parse`/`write_header` field landing out of step could come back
    /// `None` and hide behind that fallback with the surrounding suite still
    /// green. Going straight at the encoded bytes leaves nowhere to hide.
    #[test]
    fn two_notes_on_one_node_round_trip_through_the_index_alone() {
        let root_id = fixed_id(1);
        let events = vec![
            created(1, &root_id, 1, Kind::Task, None, "Root", vec![], vec![]),
            a_note_at(2, &root_id, "2026-09-01T00:00:00Z", "first note"),
            a_note_at(3, &root_id, "2026-09-02T00:00:00Z", "second note"),
        ];
        let tree = fold(&events, 0);

        let bytes = encode(&tree, 0, 0, 0, None);
        let header = Header::parse(&bytes).expect("the header this test just wrote parses");
        let loaded = build_tree(&bytes, &header).expect("the body this test just wrote parses");

        let n = loaded.node(&root_id).expect("the node is in the index");
        assert_eq!(
            n.notes(&loaded),
            vec![
                ("2026-09-01T00:00:00Z", "first note"),
                ("2026-09-02T00:00:00Z", "second note"),
            ],
            "both notes, oldest first and each with its own date, survive \
             reading the encoded bytes back"
        );
        assert_eq!(n.note(&loaded), "second note");
    }

    /// `t411`: a rule's `arms` is a new field on `Node`, stored the same way
    /// `d390` proved for notes -- a flat table the node record points into.
    /// Same direct `encode`/`build_tree` call as the note round trip above,
    /// and for the same reason: `load`'s fallback to folding the log must
    /// not be the thing that hides a misplaced field. A pillar carries no
    /// field of its own (`d436`), so it rides along here only to prove its
    /// presence changes nothing about the rule beside it.
    #[test]
    fn a_pillar_and_a_rule_round_trip_through_the_index_alone() {
        let pillar_id = fixed_id(1);
        let rule_id = fixed_id(2);
        let events = vec![
            created(
                1,
                &pillar_id,
                1,
                Kind::Pillar,
                None,
                "Security",
                vec![],
                vec![],
            ),
            created_with_arms(
                2,
                &rule_id,
                2,
                Kind::Rule,
                Some(&pillar_id),
                "Never store a secret",
                vec![crate::event::Arm {
                    dir: "vivac".to_string(),
                    command: "cargo test --bin vivac redact::tests".to_string(),
                }],
            ),
        ];
        let tree = fold(&events, 0);

        let bytes = encode(&tree, 0, 0, 0, None);
        let header = Header::parse(&bytes).expect("the header this test just wrote parses");
        let loaded = build_tree(&bytes, &header).expect("the body this test just wrote parses");

        let pillar = loaded.node(&pillar_id).expect("the pillar is in the index");
        assert_eq!(pillar.arms(&loaded), Vec::<(&str, &str)>::new());
        let rule = loaded.node(&rule_id).expect("the rule is in the index");
        assert_eq!(
            rule.arms(&loaded),
            vec![("vivac", "cargo test --bin vivac redact::tests")]
        );
    }

    /// Rewrites just the header of an already-persisted index, keeping the
    /// body untouched, so a single field can be corrupted without knowing
    /// its byte position by hand.
    fn rewrite_header(store: &Store, edit: impl FnOnce(&mut Header)) {
        let bytes = fs::read(store.index_path()).unwrap();
        let mut header = Header::parse(&bytes).expect("the index this test just wrote parses");
        edit(&mut header);
        let mut out = Vec::new();
        write_header(&mut out, &header);
        out.extend_from_slice(&bytes[header_len()..]);
        fs::write(store.index_path(), &out).unwrap();
    }

    #[test]
    fn a_corrupt_index_is_regenerated_rather_than_trusted() {
        let store = tmp_store("corrupt");
        let events = a_varied_event_set();
        write_raw_locked(&store, &events);
        let want = fold(&events, 0);

        // Bad magic: the very first byte of every valid index.
        load(&store, true).unwrap();
        let mut bytes = fs::read(store.index_path()).unwrap();
        bytes[0] ^= 0xFF;
        fs::write(store.index_path(), &bytes).unwrap();
        assert_eq!(snapshot(&want), snapshot(&load(&store, false).unwrap()));

        // Truncated mid-header.
        load(&store, true).unwrap();
        let bytes = fs::read(store.index_path()).unwrap();
        fs::write(store.index_path(), &bytes[..bytes.len() / 2]).unwrap();
        assert_eq!(snapshot(&want), snapshot(&load(&store, false).unwrap()));

        // Wrong format version: the four bytes right after the eight-byte
        // magic, in every valid index.
        load(&store, true).unwrap();
        let mut bytes = fs::read(store.index_path()).unwrap();
        bytes[8..12].copy_from_slice(&999u32.to_le_bytes());
        fs::write(store.index_path(), &bytes).unwrap();
        assert_eq!(snapshot(&want), snapshot(&load(&store, false).unwrap()));

        // An offset pointing outside the file.
        load(&store, true).unwrap();
        rewrite_header(&store, |h| h.nodes_offset = h.file_len * 100);
        assert_eq!(snapshot(&want), snapshot(&load(&store, false).unwrap()));

        std::fs::remove_dir_all(&store.root).ok();
    }

    #[test]
    fn a_stale_index_picks_up_the_tail() {
        let store = tmp_store("stale");
        let events = a_varied_event_set();
        write_raw_locked(&store, &events);
        load(&store, true).unwrap();
        assert!(store.index_path().is_file());

        let more = vec![created(
            8,
            &fixed_id(3),
            3,
            Kind::Task,
            Some(&fixed_id(1)),
            "A node born after the index",
            vec![],
            vec![],
        )];
        write_raw_locked(&store, &more);

        let (all_events, broken) = store.read_all().unwrap();
        let want = fold(&all_events, broken);

        let got = load(&store, false).unwrap();
        assert_eq!(snapshot(&want), snapshot(&got));
        assert_eq!(got.total(), 3);

        std::fs::remove_dir_all(&store.root).ok();
    }

    /// `t594`: the `Grown` branch used to drop the tail's own broken lines
    /// on the floor instead of folding them into the tree it persists, so a
    /// tree read incrementally undercounted them next to a fresh fold over
    /// the same bytes.
    #[test]
    fn a_broken_line_in_the_tail_is_not_lost_when_the_index_grows() {
        let store = tmp_store("broken-tail");
        let events = a_varied_event_set();
        write_raw_locked(&store, &events);
        load(&store, true).unwrap();
        assert!(store.index_path().is_file());

        {
            let mut f = File::options().append(true).open(store.log()).unwrap();
            f.write_all(b"not json at all\n").unwrap();
        }

        let (all_events, all_broken) = store.read_all().unwrap();
        let want = fold(&all_events, all_broken);
        assert_eq!(want.broken_lines, 1, "the fixture's own line is not broken");

        let got = load(&store, false).unwrap();
        assert_eq!(
            got.broken_lines, want.broken_lines,
            "a broken line in the tail must count the same as a fresh fold"
        );

        std::fs::remove_dir_all(&store.root).ok();
    }

    #[test]
    fn a_log_with_a_repeated_number_is_never_indexed() {
        let store = tmp_store("repeated");
        let events = vec![
            created(
                1,
                &fixed_id(1),
                1,
                Kind::Task,
                None,
                "First",
                vec![],
                vec![],
            ),
            created(
                2,
                &fixed_id(2),
                1,
                Kind::Finding,
                None,
                "Second claims the same num",
                vec![],
                vec![],
            ),
        ];
        write_raw_locked(&store, &events);

        load(&store, true).unwrap();
        assert!(
            !store.index_path().exists(),
            "a log with a repeated num must not be indexed"
        );
    }

    #[test]
    fn a_log_with_a_pending_reference_is_never_indexed() {
        let store = tmp_store("pending");
        // The child names a parent that never arrives.
        let events = vec![created(
            1,
            &fixed_id(1),
            1,
            Kind::Task,
            Some("ghost-parent-that-never-arrives"),
            "Orphaned child",
            vec![],
            vec![],
        )];
        write_raw_locked(&store, &events);

        load(&store, true).unwrap();
        assert!(
            !store.index_path().exists(),
            "a log with a pending reference must not be indexed"
        );
    }

    #[test]
    fn deleting_the_index_changes_nothing() {
        let store = tmp_store("delete");
        let events = a_varied_event_set();
        write_raw_locked(&store, &events);
        load(&store, true).unwrap();
        assert!(store.index_path().is_file());

        let with_index = load(&store, false).unwrap();
        fs::remove_file(store.index_path()).unwrap();
        let without_index = load(&store, false).unwrap();
        assert_eq!(snapshot(&with_index), snapshot(&without_index));

        std::fs::remove_dir_all(&store.root).ok();
    }

    #[test]
    fn a_write_never_persists_the_index() {
        let store = tmp_store("writeonly");
        let events = a_varied_event_set();
        write_raw_locked(&store, &events);
        load(&store, false).unwrap();
        assert!(
            !store.index_path().exists(),
            "allow_persist=false must never create the index"
        );
        std::fs::remove_dir_all(&store.root).ok();
    }
}