Artifact [efa1b8949f]

Artifact efa1b8949fe3fcb688ccfa142e19c00162bea19e257745e2e56a41c5e4909498:

• File src/repocache.zig — part of check-in [df99743ca4] at 2026-09-16 22:11:23 on branch trunk — file index: built by the derive, at ingest

  fossil fills mlink as each check-in arrives and answers finfo by lookup. zit's derive computed the same row per check-in and dropped it; the index was rebuilt by a separate pass on the first PATH query, 6.8 min on pkgsrc and not resumable. The derive now hands each change set to an index builder as it links, stores it beside the checkpoint, and extends it over the tail on append; a cache hit computes nothing. The on-query build is the fallback and checkpoints every 8192 entries. Index format 4 records completeness so a partial checkpoint is never adopted whole. crosslink reports a tree's files instead of unknown. pkgsrc, fresh clone to first answer: 11.7 min -> 7.1 min + 1s. (user: dmestas, size: 82368) [annotate] [blame] [check-ins using]

//! zit — the persisted derivation cache (a library-internal `zit` module used
//! by `Repository`; reached only through that facade, not re-exported).
//!
//! A `.zit-cache` sidecar beside `.zit` is a pure MEMO of the derived repo-view:
//! repo-state §1 pins derived state as a function of the immutable log, §7.1
//! blesses precomputing the leaf table, and §7.3 keeps full rebuild as the
//! fallback — this file stores the crosslink INPUT records (in persisted order)
//! at a named log prefix so a read-only verb replays them (`RepoView`-side)
//! instead of re-reading and re-CRC'ing the whole log. Deleting the file must
//! never change any answer: it is disposable by construction.
//!
//! The cache is keyed on the append-only log's identity (`logstore.LogKey`:
//! end offset, frame count, last-frame header CRC). A key match serves the
//! cache; any mismatch, a wrong magic/version, or a failed self-checksum is a
//! MISS — the caller rebuilds from the log and rewrites the cache atomically.
//! A stale, truncated, or corrupt cache is therefore never trusted.
//!
//! This module is a dumb container: it validates magic/version/key/checksum and
//! hands the opaque record body to `RepoView`, which alone knows the record
//! format. All effects go through the injected `std.Io` seam (design-seams rule).

const std = @import("std");
const repoview = @import("repoview.zig");
const logstore = @import("logstore.zig");
const storage = @import("storage.zig");
const crc32 = @import("crc32.zig");
const cone_mod = @import("cone.zig");

/// The advisory tier's one root name, beside `.zit` in the checkout root:
/// a DIRECTORY holding every disposable, regenerable derivation. Members
/// today: `repo` (this module's derived-state cache) and `vstat` (update's
/// materialization plan sidecar); the tree-checkpoint design reserves its
/// member here too. Fossil-ignored and never committed.
pub const cache_dirname: []const u8 = ".zit-cache";
/// This module's member inside `cache_dirname`. The PAYLOAD is byte-identical
/// to what the pre-directory code wrote at the root name -- only the location
/// moved, which is the whole migration story.
pub const cache_member: []const u8 = "repo";
pub const cache_filename: []const u8 = cache_dirname ++ "/" ++ cache_member;

/// The store cone (sparse-cone RFC section 3.1): policy, not memo, but
/// deliberately co-located with the disposable tier -- section 5.2's loss
/// rule makes deletion heal toward dense (elided reclassifies as transit),
/// so policy loss costs bandwidth, never data. Absent = dense.
pub const store_cone_member: []const u8 = "store-cone";
pub const store_cone_filename: []const u8 = cache_dirname ++ "/" ++ store_cone_member;

/// Persist the store cone (atomic write-temp + rename, like every member of
/// this tier). Passing null removes it: return to dense. Best-effort like
/// the cache itself -- a failed write leaves the previous policy standing,
/// and the caller's next open sees whichever policy is on disk.
/// The store cone, or null when dense. Absent, unreadable, or malformed
/// reads as DENSE (the section 5.2 loss rule): classification degrades to
/// fetch-everything, never to damage.
pub fn readStoreCone(gpa: std.mem.Allocator, io: std.Io, dir: std.Io.Dir) ?cone_mod.Cone {
    var file = dir.openFile(io, store_cone_filename, .{}) catch return null;
    defer file.close(io);
    const st = file.stat(io) catch return null;
    const size = std.math.cast(usize, st.size) orelse return null;
    if (size == 0 or size > 1 + cone_mod.max_members * (2 + cone_mod.max_member_len)) return null;
    const buf = gpa.alloc(u8, size) catch return null;
    defer gpa.free(buf);
    const n = file.readPositionalAll(io, buf, 0) catch return null;
    if (n != size) return null;
    return cone_mod.Cone.parse(gpa, buf) catch null;
}

pub fn writeStoreCone(gpa: std.mem.Allocator, io: std.Io, dir: std.Io.Dir, cone: ?*const cone_mod.Cone) void {
    if (cone == null) {
        dir.deleteFile(io, store_cone_filename) catch {};
        return;
    }
    const bytes = cone.?.serialize(gpa) catch return;
    defer gpa.free(bytes);
    if (!ensureCacheDir(io, dir)) return;
    var af = dir.createFileAtomic(io, store_cone_filename, .{ .replace = true }) catch return;
    defer af.deinit(io);
    af.file.writePositionalAll(io, bytes, 0) catch return;
    af.file.sync(io) catch return;
    af.replace(io) catch return;
}
/// The persisted log-scan index (`logstore.serializeIndex`) the lazy content
/// channel adopts instead of re-walking every frame header on each open. The
/// blob is self-validating (its own magic/crc/log-key, checked by
/// `logstore.openReadOnlyIndexed`), so this module stays a dumb byte carrier
/// for it: read whole, write atomic, swallow failures — advisory like every
/// member here.
pub const index_member: []const u8 = "index";
pub const index_filename: []const u8 = cache_dirname ++ "/" ++ index_member;

/// The disposition sidecar (zit#182): one bit per log frame recording whether
/// its artifact is crosslink-relevant, so a full derive skips reconstructing
/// the frames that cannot contribute a record. Encoded and validated by
/// `repoview` (`serializeDispositions`/`parseDispositions`); this module is
/// the same dumb byte carrier it is for the log-scan index.
pub const dispositions_member: []const u8 = "dispositions";
pub const dispositions_filename: []const u8 = cache_dirname ++ "/" ++ dispositions_member;

/// The stored dispositions blob, whole, or null on absence/unreadability (a
/// miss, never an error). Validation is the adopter's job. Caller frees.
pub fn tryLoadDispositions(gpa: std.mem.Allocator, io: std.Io, dir: std.Io.Dir) ?[]u8 {
    return tryLoadMember(gpa, io, dir, dispositions_filename);
}

/// Atomically materialize the dispositions blob (temp + fsync + rename).
/// Every failure is swallowed: advisory, like every member here.
pub fn storeDispositions(io: std.Io, dir: std.Io.Dir, blob: []const u8) void {
    storeMember(io, dir, dispositions_filename, blob);
}

/// zit#184 — the cluster index (`clusterindex.serialize`/`parse`): the names
/// of every held cluster artifact, keyed to the log prefix. Same dumb byte
/// carrier as the dispositions map.
pub const clusters_member: []const u8 = "clusters";
pub const clusters_filename: []const u8 = cache_dirname ++ "/" ++ clusters_member;

/// Unlike the other loaders this one PROPAGATES allocation failure: its
/// callers run under the persist's own OOM contract, where a swallowed
/// failure would let a persist report success it did not fully deliver.
/// Storage trouble is still a miss. Caller frees.
pub fn loadClusterIndex(gpa: std.mem.Allocator, io: std.Io, dir: std.Io.Dir) error{OutOfMemory}!?[]u8 {
    var file = dir.openFile(io, clusters_filename, .{}) catch return null;
    defer file.close(io);
    const st = file.stat(io) catch return null;
    const size = std.math.cast(usize, st.size) orelse return null;
    const buf = try gpa.alloc(u8, size);
    errdefer gpa.free(buf);
    const n = file.readPositionalAll(io, buf, 0) catch return null;
    if (n != size) return null;
    return buf;
}

pub fn storeClusterIndex(io: std.Io, dir: std.Io.Dir, blob: []const u8) void {
    storeMember(io, dir, clusters_filename, blob);
}

/// zit#188 — the file-history index (`fileindex.serialize`/`parse`): which
/// check-ins changed which path, keyed to the log tail and the shun
/// fingerprint. Same dumb byte carrier as its siblings.
pub const finfo_member: []const u8 = "finfo";
pub const finfo_filename: []const u8 = cache_dirname ++ "/" ++ finfo_member;

pub fn tryLoadFileIndex(gpa: std.mem.Allocator, io: std.Io, dir: std.Io.Dir) ?[]u8 {
    return tryLoadMember(gpa, io, dir, finfo_filename);
}

pub fn storeFileIndex(io: std.Io, dir: std.Io.Dir, blob: []const u8) void {
    storeMember(io, dir, finfo_filename, blob);
}

/// The stored log-index blob, whole, or null on absence/unreadability (a miss,
/// never an error). Validation is the adopter's job. Caller frees.
pub fn tryLoadLogIndex(gpa: std.mem.Allocator, io: std.Io, dir: std.Io.Dir) ?[]u8 {
    return tryLoadMember(gpa, io, dir, index_filename);
}

/// Atomically materialize the log-index blob (temp + fsync + rename). Every
/// failure is swallowed: the index is an optimization, and the fallback is
/// the full scan it replaces.
pub fn storeLogIndex(io: std.Io, dir: std.Io.Dir, blob: []const u8) void {
    storeMember(io, dir, index_filename, blob);
}

/// The working-tree stat memo (#118): per tracked file, the mtime, size, and
/// content hash recorded the last time a working-tree scan hashed it, so
/// `status`/`diff` re-hash only files whose stat changed instead of every
/// tracked file. The format is owned by `cmd/tracking.zig` (`TreeStat`),
/// which self-validates the blob; this module stays a dumb byte carrier —
/// advisory like every member here, absent or stale costs a re-hash, never
/// an answer.
pub const treestat_member: []const u8 = "treestat";
pub const treestat_filename: []const u8 = cache_dirname ++ "/" ++ treestat_member;

/// The stored tree-stat blob, whole, or null on absence/unreadability (a
/// miss, never an error). Validation is the adopter's job. Caller frees.
pub fn tryLoadTreeStat(gpa: std.mem.Allocator, io: std.Io, dir: std.Io.Dir) ?[]u8 {
    return tryLoadMember(gpa, io, dir, treestat_filename);
}

/// Atomically materialize the tree-stat blob. Failures swallowed: the memo
/// is an optimization, and the fallback is the full re-hash it replaces.
pub fn storeTreeStat(io: std.Io, dir: std.Io.Dir, blob: []const u8) void {
    storeMember(io, dir, treestat_filename, blob);
}

/// The packed-generation memo (#122): the `logstore.LogKey` of the store as
/// the last `zit pack` left it, so a repeat pack over an unchanged log can
/// answer "already packed" instead of re-walking and re-encoding the whole
/// store. Any append (a commit, a shun row, a pull) changes the key, and the
/// format is owned by the pack CLI (`cmd/sync.zig`); this module stays a dumb
/// byte carrier. Advisory: absent or mismatching just costs a real pack.
pub const packed_member: []const u8 = "packed";
pub const packed_filename: []const u8 = cache_dirname ++ "/" ++ packed_member;

pub fn tryLoadPackedKey(gpa: std.mem.Allocator, io: std.Io, dir: std.Io.Dir) ?[]u8 {
    return tryLoadMember(gpa, io, dir, packed_filename);
}

pub fn storePackedKey(io: std.Io, dir: std.Io.Dir, blob: []const u8) void {
    storeMember(io, dir, packed_filename, blob);
}

fn tryLoadMember(gpa: std.mem.Allocator, io: std.Io, dir: std.Io.Dir, sub_path: []const u8) ?[]u8 {
    var file = dir.openFile(io, sub_path, .{}) catch return null;
    defer file.close(io);
    const st = file.stat(io) catch return null;
    const size = std.math.cast(usize, st.size) orelse return null;
    const buf = gpa.alloc(u8, size) catch return null;
    const n = file.readPositionalAll(io, buf, 0) catch {
        gpa.free(buf);
        return null;
    };
    if (n != size) {
        gpa.free(buf);
        return null;
    }
    return buf;
}

fn storeMember(io: std.Io, dir: std.Io.Dir, sub_path: []const u8, blob: []const u8) void {
    if (!ensureCacheDir(io, dir)) return;
    var af = dir.createFileAtomic(io, sub_path, .{ .replace = true }) catch return;
    defer af.deinit(io);
    af.file.writePositionalAll(io, blob, 0) catch return;
    af.file.sync(io) catch return;
    af.replace(io) catch return;
}

/// Make `cache_dirname` a directory a member write can land in. A regular
/// FILE at the name is the pre-directory layout, dead since the legacy tier's
/// removal: it is deleted and replaced with the directory (advisory — worth
/// one re-derive, never an error). Returns whether the path is now writable
/// as a directory member.
pub fn ensureCacheDir(io: std.Io, dir: std.Io.Dir) bool {
    const st = dir.statFile(io, cache_dirname, .{ .follow_symlinks = false }) catch |err| switch (err) {
        error.FileNotFound => {
            dir.createDir(io, cache_dirname, .default_dir) catch return false;
            return true;
        },
        else => return false,
    };
    switch (st.kind) {
        .directory => return true,
        .file => {
            // A pre-directory cache FILE at the name. Its read tolerance was
            // removed with the legacy tier, so the bytes are dead: delete and
            // start the directory (advisory — worth one re-derive, never an
            // error).
            dir.deleteFile(io, cache_dirname) catch return false;
            dir.createDir(io, cache_dirname, .default_dir) catch return false;
            return true;
        },
        else => return false,
    }
}

/// File magic; distinguishes a `.zit-cache` from any other file that lands at
/// the name. Eight bytes so the fixed header stays 8-aligned.
const magic = "ZITCACHE";

/// On-disk format version. Any mismatch is a MISS: the cache is disposable, so
/// the tables it serializes stay free to evolve without a migration path.
// Bumped to 2 when the shun fingerprint joined the header: extension across a
// shun change is unsound (the body may carry a record the mask must now hide),
// so the fingerprint is part of the cache's identity. An old cache simply
// misses once and is rebuilt.
// Bumped to 3 when the crosslink record gained the C-card comment: an old
// body would misparse under the new reader, so it must miss and rebuild.
// Bumped to 4 when the record gained the Q cards (zit#177), for the same reason.
// Bumped to 5 when each changed file gained its artifact hash (zit#176).
const format_version: u16 = 5;

/// Materialized-image column schema version. Guards the appended image section
/// independently of the record body: a bump makes a warm `.fast_read` fall back
/// to deriving from the still-valid record body (never a wrong answer, never a
/// migration). A cache written by an older build (or with no image) carries
/// `image_len == 0`, so `.fast_read` degrades to a derive there too.
/// Bumped to 4 when the image gained the resolved §5 tag surface; 3 was the
/// §5.2 overrides it subsumes, 2 was adjacency (#98). A cache written by the
/// previous layout is refused by `stored_layout != layout_version` and falls
/// through to the record-body derive, which is still a validated hit — the
/// column-directory length would in fact reject it anyway, but relying on that
/// would make a format change look like a corrupt cache. That hit rewrites the
/// image at the current layout (zit#180), so the fallback is paid once.
/// Bumped to 6 when the image gained the per-rid comment and login columns,
/// so a listing renders without a manifest read per row. Bumped to 7 when it
/// gained the Q-card columns (zit#177), so `info` renders without one at all.
const layout_version: u16 = 7;

/// Bytes before the variable-length `main_branch`:
/// `magic(8) | format_version(2) | layout_version(2) | end(8) | frame_count(8) |
/// last_frame_offset(8) | last_header_crc(4) | main_branch_len(2) |
/// image_len(4) | shun_fp(4)` -- see the append order in `store`.
/// (Historic comment below described the v1 prefix.)
/// last_frame_offset(8) | last_header_crc(4) | main_branch_len(2) |
/// image_len(4)` = 46.
const fixed_prefix_len: usize = 50;

/// Trailing self-checksum width (CRC32 over every earlier byte).
const checksum_len: usize = 4;

// There is deliberately no maximum-cache-size constant here.
//
// One stood at 2 GiB, justified as bounding "a lying on-disk size before any
// allocation (TigerStyle: never trust a claimed length)". The read path never
// trusted a claimed length: it takes `stat`'s size and allocates exactly that,
// so there was no claim to distrust and nothing the cap defended against. What
// it did do was silently refuse a cache larger than a number nobody chose —
// and because a cache is advisory, that refusal is invisible, degrading to a
// full rebuild from the log with no diagnostic. A performance cliff wearing a
// security justification.
//
// The bounds that are real are stated where they bind: `std.math.cast(usize,
// st.size)` on the read (a size that cannot be addressed cannot be read) and
// the `u32` `image_len` header field on the write.

/// Try to serve `view` from the `.zit-cache` in `dir`, validating it against
/// the live `log_file` (the open `.zit`). Returns true only on a fully
/// validated HIT: correct magic and version, an intact self-checksum, a stored
/// key whose tail still matches the live log (`logstore.validateTail` — O(1), no
/// full scan), and a record body `RepoView` accepts. On a hit `view` is filled
/// and self-contained (the caller `deinit`s it). Returns false on ANY miss — no
/// file, wrong magic/version, size out of bounds, checksum failure, a stale key,
/// or a malformed body — and never errors: a cache is advisory, so a missing or
/// unreadable one just means "rebuild from the log". On a false return `view` is
/// untouched or already torn down; the caller must not `deinit` it.
/// The representation a caller wants served from a validated cache. `.full`
/// always derives the full `RepoState` from the retained record body (today's
/// path — every query answerable). `.fast_read` prefers the zero-parse image
/// section (the hot verbs' path — the freed queries only), transparently
/// falling back to the record-body derive when the cache carries no image or
/// the image is stale/malformed. Both are HITs; the choice never changes an
/// answer, only how much work a warm read pays.
pub const LoadMode = enum { full, fast_read };

pub fn tryLoad(
    gpa: std.mem.Allocator,
    io: std.Io,
    dir: std.Io.Dir,
    log_file: storage.StorageFile,
    view: *repoview.RepoView,
    mode: LoadMode,
) bool {
    var file = dir.openFile(io, cache_filename, .{}) catch return false; // absent/unreadable ⇒ miss
    defer file.close(io);
    const st = file.stat(io) catch return false;
    const size: usize = std.math.cast(usize, st.size) orelse return false;
    // A valid cache is at least the fixed header + checksum; reject anything
    // shorter (truncated). There is no upper bound beyond the `usize` cast
    // above: `size` is what `stat` reports, not what the file claims, so the
    // allocation below is the file itself and a ceiling here could only refuse
    // a cache this code had legitimately written.
    if (size < fixed_prefix_len + checksum_len) return false;

    const buf = gpa.alloc(u8, size) catch return false;
    // On the image path `buf` is handed to the view (its columns view into it);
    // on every other path it is a scratch buffer freed here.
    var owned_by_view = false;
    defer if (!owned_by_view) gpa.free(buf);
    const n = file.readPositionalAll(io, buf, 0) catch return false;
    if (n != size) return false;

    if (!std.mem.eql(u8, buf[0..8], magic)) return false;
    if (std.mem.readInt(u16, buf[8..10], .little) != format_version) return false;
    const stored_layout = std.mem.readInt(u16, buf[10..12], .little);

    // Self-checksum over everything but the trailing CRC — a torn or bitflipped
    // cache fails here and is treated as a miss, never a trusted partial view.
    const stored_crc = std.mem.readInt(u32, buf[size - checksum_len ..][0..4], .little);
    var crc = crc32.Crc32.init();
    crc.update(buf[0 .. size - checksum_len]);
    if (crc.final() != stored_crc) return false;

    // The stored key must still describe the live log's tail — the whole
    // invalidation story. Any append, truncation, or a rewritten tail changes
    // it, so a stale cache cannot validate. O(1): one stat + one header read.
    const stored_key: logstore.LogStore.LogKey = .{
        .end = std.mem.readInt(u64, buf[12..20], .little),
        .frame_count = std.mem.readInt(u64, buf[20..28], .little),
        .last_frame_offset = std.mem.readInt(u64, buf[28..36], .little),
        .last_header_crc = std.mem.readInt(u32, buf[36..40], .little),
    };
    if (!logstore.LogStore.validateTail(log_file, stored_key)) return false;

    const mbl: usize = std.mem.readInt(u16, buf[40..42], .little);
    const image_len: usize = std.mem.readInt(u32, buf[42..46], .little);
    // buf[46..50] is the shun fingerprint; the exact-tail hit path does not
    // need it -- an unchanged tail implies an unchanged shun table.
    // Validate the declared spans against the actual file size before slicing
    // (never trust a claimed-huge length): header + main_branch + record body +
    // image + checksum must fit exactly.
    if (fixed_prefix_len + mbl + image_len + checksum_len > size) return false;
    if (mbl == 0) return false; // main_branch is never empty (§6.1: unset ⇒ "trunk")
    const main_branch = buf[fixed_prefix_len .. fixed_prefix_len + mbl];
    const body_end = size - checksum_len - image_len;
    const body = buf[fixed_prefix_len + mbl .. body_end];
    const image = buf[body_end .. size - checksum_len];

    // Fast path: serve the zero-parse image when the schema matches and one is
    // present. A malformed/OOM image is not an error — fall through to the
    // record-body derive, which is still a validated HIT.
    const image_current = image_len > 0 and stored_layout == layout_version;
    if (mode == .fast_read and image_current) {
        if (view.buildFromImage(gpa, buf, image, main_branch)) |_| {
            owned_by_view = true;
            return true;
        } else |_| {}
    }

    // Replay the records without touching the log. On any framing error the
    // view tears itself down and we fall back to a rebuild.
    view.buildFromRecordBody(gpa, body, main_branch) catch return false;

    // An image this build cannot serve — older layout, absent, or malformed —
    // is rewritten from the view just derived, so the store pays this replay
    // once rather than on every read until something appends to the log
    // (zit#180): a HIT stores nothing otherwise, and a quiet repository's log
    // never grows. Same atomic rewrite as a miss; a failed write leaves the
    // cache as it was.
    if (!image_current or mode == .fast_read) {
        const shun_fp = std.mem.readInt(u32, buf[46..50], .little);
        var fresh: std.ArrayList(u8) = .empty;
        defer fresh.deinit(gpa);
        if (repoimage.serialize(gpa, &view.state, main_branch, &fresh)) |_| {
            store(gpa, io, dir, stored_key, shun_fp, main_branch, body, fresh.items);
        } else |_| {}
    }
    return true;
}

/// Everything cache extension needs from an existing `.zit-cache`, without
/// building a view. `buf` owns the backing bytes; `main_branch` and `body`
/// are windows into it.
pub const RawCache = struct {
    buf: []u8,
    key: logstore.LogStore.LogKey,
    shun_fp: u32,
    main_branch: []const u8,
    body: []const u8,

    pub fn deinit(self: *RawCache, gpa: std.mem.Allocator) void {
        gpa.free(self.buf);
        self.* = undefined;
    }
};

/// Read and structurally validate the `.zit-cache`, returning its key, shun
/// fingerprint and record body WITHOUT checking the log's tail -- the caller
/// is deciding between "still exact", "extendable prefix" and "stale", and
/// that decision needs the stored key first. Magic, version and the
/// self-checksum are still enforced: a torn cache is null here exactly as it
/// is a miss in `tryLoad`.
pub fn loadForExtension(gpa: std.mem.Allocator, io: std.Io, dir: std.Io.Dir) ?RawCache {
    var file = dir.openFile(io, cache_filename, .{}) catch return null;
    defer file.close(io);
    const st = file.stat(io) catch return null;
    const size: usize = std.math.cast(usize, st.size) orelse return null;
    if (size < fixed_prefix_len + checksum_len) return null;
    const buf = gpa.alloc(u8, size) catch return null;
    errdefer gpa.free(buf);
    const n = file.readPositionalAll(io, buf, 0) catch {
        gpa.free(buf);
        return null;
    };
    if (n != size) {
        gpa.free(buf);
        return null;
    }
    if (!std.mem.eql(u8, buf[0..8], magic) or
        std.mem.readInt(u16, buf[8..10], .little) != format_version)
    {
        gpa.free(buf);
        return null;
    }
    const stored_crc = std.mem.readInt(u32, buf[size - checksum_len ..][0..4], .little);
    var crc = crc32.Crc32.init();
    crc.update(buf[0 .. size - checksum_len]);
    if (crc.final() != stored_crc) {
        gpa.free(buf);
        return null;
    }
    const mbl: usize = std.mem.readInt(u16, buf[40..42], .little);
    const image_len: usize = std.mem.readInt(u32, buf[42..46], .little);
    if (fixed_prefix_len + mbl + image_len + checksum_len > size or mbl == 0) {
        gpa.free(buf);
        return null;
    }
    return .{
        .buf = buf,
        .key = .{
            .end = std.mem.readInt(u64, buf[12..20], .little),
            .frame_count = std.mem.readInt(u64, buf[20..28], .little),
            .last_frame_offset = std.mem.readInt(u64, buf[28..36], .little),
            .last_header_crc = std.mem.readInt(u32, buf[36..40], .little),
        },
        .shun_fp = std.mem.readInt(u32, buf[46..50], .little),
        .main_branch = buf[fixed_prefix_len .. fixed_prefix_len + mbl],
        .body = buf[fixed_prefix_len + mbl .. size - checksum_len - image_len],
    };
}

/// Write `body` (a `RepoView.loadCapturing` record body) plus `main_branch` and
/// `key` to the `.zit-cache` in `dir`, atomically (temp file + fsync + rename),
/// so a reader ever sees either the prior valid cache or this one — never a torn
/// file. Best-effort: a read-only directory, a full disk, or any I/O failure is
/// swallowed (the verb has already answered from the just-built view); the cache
/// is an optimization, not a correctness dependency. The temp file is deleted on
/// any early return.
pub fn store(
    gpa: std.mem.Allocator,
    io: std.Io,
    dir: std.Io.Dir,
    key: logstore.LogStore.LogKey,
    shun_fp: u32,
    main_branch: []const u8,
    body: []const u8,
    image: []const u8,
) void {
    if (main_branch.len == 0 or main_branch.len > std.math.maxInt(u16)) return; // §6.1 invariant
    if (image.len > std.math.maxInt(u32)) return; // image_len is a u32 header field
    // No aggregate ceiling: the only reason one stood here was to mirror a
    // read-side cap that was itself unfounded, so writing a cache the reader
    // would refuse is no longer possible by construction.

    // The header is small and fixed; `body` and `image` are the repository's
    // whole derived record stream and image — hundreds of megabytes on a large
    // store. They are written to the file IN PLACE, never concatenated into a
    // second resident copy, with the checksum folded incrementally over the
    // same byte order the reader checks (zit#182: the concatenation buffer was
    // a full extra copy of both at the derive's peak).
    var header: std.ArrayList(u8) = .empty;
    defer header.deinit(gpa);
    header.appendSlice(gpa, magic) catch return;
    appendInt(gpa, &header, u16, format_version) catch return;
    appendInt(gpa, &header, u16, layout_version) catch return;
    appendInt(gpa, &header, u64, key.end) catch return;
    appendInt(gpa, &header, u64, key.frame_count) catch return;
    appendInt(gpa, &header, u64, key.last_frame_offset) catch return;
    appendInt(gpa, &header, u32, key.last_header_crc) catch return;
    appendInt(gpa, &header, u16, @intCast(main_branch.len)) catch return;
    appendInt(gpa, &header, u32, @intCast(image.len)) catch return;
    appendInt(gpa, &header, u32, shun_fp) catch return;
    header.appendSlice(gpa, main_branch) catch return;
    var crc = crc32.Crc32.init();
    crc.update(header.items);
    crc.update(body);
    crc.update(image);
    var trailer: [4]u8 = undefined;
    std.mem.writeInt(u32, &trailer, crc.final(), .little);

    if (!ensureCacheDir(io, dir)) return;
    var af = dir.createFileAtomic(io, cache_filename, .{ .replace = true }) catch return;
    defer af.deinit(io);
    var off: u64 = 0;
    af.file.writePositionalAll(io, header.items, off) catch return;
    off += header.items.len;
    af.file.writePositionalAll(io, body, off) catch return;
    off += body.len;
    af.file.writePositionalAll(io, image, off) catch return;
    off += image.len;
    af.file.writePositionalAll(io, &trailer, off) catch return;
    af.file.sync(io) catch return; // durable before the rename makes it visible
    af.replace(io) catch return; // atomic rename into place
}

fn appendInt(gpa: std.mem.Allocator, s: *std.ArrayList(u8), comptime T: type, v: T) error{OutOfMemory}!void {
    var buf: [@sizeOf(T)]u8 = undefined;
    std.mem.writeInt(T, &buf, v, .little);
    try s.appendSlice(gpa, &buf);
}

// ===========================================================================
// Tests — the load-bearing invariant is HIT ≡ recompute (byte-identical) and
// "a stale/corrupt/wrong-key cache is never served". Fixtures build real logs
// through `LogStore` and derive with the real `RepoView`; expected values come
// from the repo-state RFC (§1 memo, §7.1 precomputed table, §7.3 rebuild).
// ===========================================================================

const artifact = @import("artifact.zig");
const repostate = @import("repostate.zig");
const repoimage = @import("repoimage.zig");
const testing = std.testing;

/// A tmp-backed writable log plus the checkout dir the cache lives in.
const Fixture = struct {
    tmp: std.testing.TmpDir,
    fs: storage.FileStorage,
    log: logstore.LogStore,

    fn init(self: *Fixture, io: std.Io) !void {
        self.tmp = testing.tmpDir(.{});
        const raw = try self.tmp.dir.createFile(io, ".zit", .{ .read = true });
        self.fs = storage.FileStorage.init(io, raw, 0);
        self.log = try logstore.LogStore.open(testing.allocator, self.fs.storageFile(), 0);
    }

    fn deinit(self: *Fixture, io: std.Io) void {
        self.log.deinit(testing.allocator);
        self.fs.file.close(io);
        self.tmp.cleanup();
    }

    /// A freshly-scanned read-only view of the log — the shape production always
    /// derives from (`openReadOnly` populates the frame-count/tail bookkeeping
    /// `logKey` reads). The writable `self.log` used for seeding does not
    /// maintain that bookkeeping, so cache-key work goes through this. Caller
    /// `deinit`s the returned store.
    fn openScan(self: *Fixture) !logstore.LogStore {
        return logstore.LogStore.openReadOnly(testing.allocator, self.fs.storageFile());
    }

    /// Store a manifest/control built by `m` under its own sha3-256 digest.
    fn put(self: *Fixture, m: *const artifact.Manifest) !void {
        const bytes = try artifact.create(m, testing.allocator);
        defer testing.allocator.free(bytes);
        var digest: [32]u8 = undefined;
        std.crypto.hash.sha3.Sha3_256.hash(bytes, &digest, .{});
        _ = try self.log.put(&digest, bytes);
    }
};

/// The 64-hex name of a serialized manifest (so a later child can name it as a
/// parent). Caller frees.
fn nameOf(m: *const artifact.Manifest) ![]u8 {
    const bytes = try artifact.create(m, testing.allocator);
    defer testing.allocator.free(bytes);
    var digest: [32]u8 = undefined;
    std.crypto.hash.sha3.Sha3_256.hash(bytes, &digest, .{});
    var hex: [64]u8 = undefined;
    const lut = "0123456789abcdef";
    for (digest, 0..) |b, i| {
        hex[i * 2] = lut[b >> 4];
        hex[i * 2 + 1] = lut[b & 0x0f];
    }
    return testing.allocator.dupe(u8, &hex);
}

fn checkinManifest(ts: []const u8, parent: ?[]const u8, tags: []const artifact.TCard, pbuf: []artifact.PCard) artifact.Manifest {
    var p_slice: []const artifact.PCard = &.{};
    if (parent) |par| {
        pbuf[0] = .{ .parents = par, .count = 1 };
        p_slice = pbuf[0..1];
    }
    return .{
        .arena_state = std.heap.ArenaAllocator.init(testing.allocator),
        .kind = .manifest,
        .a = null,
        .b = null,
        .c = .{ .comment = "c" },
        .d = .{ .timestamp = ts },
        .e = null,
        .g = null,
        .h = null,
        .i = null,
        .k = null,
        .l = null,
        .n = null,
        .r = null,
        .u = .{ .login = "u" },
        .z = null,
        .w = null,
        .f = &.{},
        .j = &.{},
        .m = &.{},
        .p = p_slice,
        .q = &.{},
        .t = tags,
    };
}

fn controlManifest(ts: []const u8, tags: []const artifact.TCard) artifact.Manifest {
    return .{
        .arena_state = std.heap.ArenaAllocator.init(testing.allocator),
        .kind = .control,
        .a = null,
        .b = null,
        .c = null,
        .d = .{ .timestamp = ts },
        .e = null,
        .g = null,
        .h = null,
        .i = null,
        .k = null,
        .l = null,
        .n = null,
        .r = null,
        .u = .{ .login = "u" },
        .z = null,
        .w = null,
        .f = &.{},
        .j = &.{},
        .m = &.{},
        .p = &.{},
        .q = &.{},
        .t = tags,
    };
}

/// Build a small but structurally rich repo: a trunk root, a same-branch child,
/// a second root that forks trunk, and a control that closes a leaf — exercising
/// leaves, branches, propagation, and a control artifact.
fn seedRepo(fx: *Fixture) !void {
    var pbuf: [1]artifact.PCard = undefined;

    const branch_root = [_]artifact.TCard{.{ .sign = .propagate, .name = "branch", .target = null, .value = "trunk" }};
    var root_m = checkinManifest("2020-01-01T00:00:00", null, &branch_root, &pbuf);
    defer root_m.deinit();
    const root = try nameOf(&root_m);
    defer testing.allocator.free(root);
    try fx.put(&root_m);

    var child_m = checkinManifest("2020-01-02T00:00:00", root, &.{}, &pbuf);
    defer child_m.deinit();
    const child = try nameOf(&child_m);
    defer testing.allocator.free(child);
    try fx.put(&child_m);

    var root2_m = checkinManifest("2020-01-03T00:00:00", null, &branch_root, &pbuf);
    defer root2_m.deinit();
    const root2 = try nameOf(&root2_m);
    defer testing.allocator.free(root2);
    try fx.put(&root2_m);

    const close = [_]artifact.TCard{.{ .sign = .add, .name = "closed", .target = root2, .value = null }};
    var ctl_m = controlManifest("2020-01-04T00:00:00", &close);
    defer ctl_m.deinit();
    try fx.put(&ctl_m);

    // Two siblings of `child` sharing one D-card instant — the exact-mtime tie
    // every ordering tiebreak keys on (zit#173). Distinct sym tags keep the
    // artifacts distinct; both stay open trunk leaves, so branch-name
    // resolution walks the tied-candidate path and `checkinsByMtime` walks the
    // tied-run path, on BOTH backings, in every test seeded from here.
    const tie_a_tags = [_]artifact.TCard{
        .{ .sign = .propagate, .name = "branch", .target = null, .value = "trunk" },
        .{ .sign = .propagate, .name = "sym-tie-a", .target = null, .value = null },
    };
    var tie_a_m = checkinManifest("2020-01-05T00:00:00", child, &tie_a_tags, &pbuf);
    defer tie_a_m.deinit();
    try fx.put(&tie_a_m);
    const tie_b_tags = [_]artifact.TCard{
        .{ .sign = .propagate, .name = "branch", .target = null, .value = "trunk" },
        .{ .sign = .propagate, .name = "sym-tie-b", .target = null, .value = null },
    };
    var tie_b_m = checkinManifest("2020-01-05T00:00:00", child, &tie_b_tags, &pbuf);
    defer tie_b_m.deinit();
    try fx.put(&tie_b_m);

    // A cherry-pick of `root2` onto `child` (zit#177): the one Q card the
    // record, the state and the image must all carry — a target that is
    // NOT a parent.
    var pick_m = checkinManifest("2020-01-06T00:00:00", child, &.{}, &pbuf);
    defer pick_m.deinit();
    const pick_q = [_]artifact.QCard{.{ .sign = .cherrypick, .target = root2, .baseline = null }};
    pick_m.q = &pick_q;
    try fx.put(&pick_m);
}

/// Assert two views are observably identical across every derived query the
/// verbs read. Rids match between the views because the same records intern in
/// the same order, so a per-rid comparison is valid.
/// zit#177: the Q cards of `rid` agree between two backings — count, order,
/// sign and target.
fn expectPicksEqual(a: *const repostate.RepoState, b: *const repostate.RepoState, rid: repostate.Rid) !void {
    const pa = try a.picksOf(testing.allocator, rid);
    defer testing.allocator.free(pa);
    const pb = try b.picksOf(testing.allocator, rid);
    defer testing.allocator.free(pb);
    try testing.expectEqual(pa.len, pb.len);
    for (pa, pb) |x, y| {
        try testing.expectEqual(x.sign, y.sign);
        try testing.expectEqualStrings(x.target, y.target);
    }
}

fn expectViewsEqual(a: *repoview.RepoView, b: *repoview.RepoView) !void {
    try testing.expectEqualStrings(a.main_branch, b.main_branch);
    try testing.expectEqual(a.store.store().count(), b.store.store().count());

    // Leaf table: same membership AND same iteration order (§7.5 SHOULD order).
    var la = a.state.leaves();
    var lb = b.state.leaves();
    while (la.next()) |ra| {
        const rb = lb.next() orelse return error.LeafCountMismatch;
        try testing.expectEqual(ra, rb);
    }
    try testing.expect(lb.next() == null);

    const count = a.store.store().count();
    var rid: repostate.Rid = 1;
    while (rid <= count) : (rid += 1) {
        try testing.expectEqualStrings(a.hashOf(rid), b.hashOf(rid));
        try testing.expectEqual(a.state.mtimeOf(rid), b.state.mtimeOf(rid));
        try testing.expectEqual(a.state.isLeaf(rid), b.state.isLeaf(rid));
        try testing.expectEqual(a.state.isOpenLeaf(rid), b.state.isOpenLeaf(rid));
        try testing.expectEqual(a.state.isAmbiguousLeaf(rid), b.state.isAmbiguousLeaf(rid));
        // branchOf requires an event row (§6.1): skip eventless phantoms and
        // control carriers, whose absence-of-event was already asserted above.
        if (a.state.mtimeOf(rid) != null)
            try testing.expectEqualStrings(a.state.branchOf(rid, a.main_branch), b.state.branchOf(rid, b.main_branch));

        const ta = try a.state.tagsOf(testing.allocator, rid);
        defer testing.allocator.free(ta);
        const tb = try b.state.tagsOf(testing.allocator, rid);
        defer testing.allocator.free(tb);
        try testing.expectEqual(ta.len, tb.len);
        for (ta, tb) |ea, eb| {
            try testing.expectEqualStrings(ea.name, eb.name);
            try testing.expectEqual(ea.tagtype, eb.tagtype);
        }
        try expectPicksEqual(&a.state, &b.state, rid);
    }

    const ba = try a.state.branches(testing.allocator);
    defer testing.allocator.free(ba);
    const bb = try b.state.branches(testing.allocator);
    defer testing.allocator.free(bb);
    try testing.expectEqual(ba.len, bb.len);
    for (ba, bb) |ea, eb| {
        try testing.expectEqualStrings(ea.name, eb.name);
        try testing.expectEqual(ea.tip, eb.tip);
        try testing.expectEqual(ea.closed, eb.closed);
    }

    const na = try a.state.tagNames(testing.allocator);
    defer testing.allocator.free(na);
    const nb = try b.state.tagNames(testing.allocator);
    defer testing.allocator.free(nb);
    try testing.expectEqual(na.len, nb.len);
    for (na, nb) |ea, eb| try testing.expectEqualStrings(ea, eb);
}

/// Capture the record body + key + main_branch for `fx`'s current log (through a
/// freshly-scanned read-only store, as production does), then build a
/// from-scratch reference view. Caller `deinit`s `ref` and frees `body`.
fn captureAndReference(fx: *Fixture, ref: *repoview.RepoView, body: *std.ArrayList(u8)) !logstore.LogStore.LogKey {
    var ro = try fx.openScan();
    defer ro.deinit(testing.allocator);
    const key = try ro.logKey();
    var capture: repoview.RepoView = undefined;
    try capture.loadCapturing(testing.allocator, &ro, body);
    // The reference is a plain from-scratch derivation over the same log.
    try ref.load(testing.allocator, &ro);
    capture.deinit(testing.allocator);
    return key;
}

test "cache §1/§7.1: a hit reconstructs a byte-identical view for a representative repo" {
    const io = testing.io;
    var fx: Fixture = undefined;
    try fx.init(io);
    defer fx.deinit(io);
    try seedRepo(&fx);

    var body: std.ArrayList(u8) = .empty;
    defer body.deinit(testing.allocator);
    var ref: repoview.RepoView = undefined;
    const key = try captureAndReference(&fx, &ref, &body);
    defer ref.deinit(testing.allocator);

    store(testing.allocator, io, fx.tmp.dir, key, 0, ref.main_branch, body.items, "");

    var hit: repoview.RepoView = undefined;
    try testing.expect(tryLoad(testing.allocator, io, fx.tmp.dir, fx.fs.storageFile(), &hit, .full));
    defer hit.deinit(testing.allocator);
    try expectViewsEqual(&ref, &hit);
}

test "cache §1: hit equals from-scratch derivation for random check-in op-sequences" {
    const io = testing.io;
    // The DST derivation oracle in miniature: random DAGs of varied sizes,
    // each proved to round-trip through the on-disk cache byte-identically.
    var prng = std.Random.DefaultPrng.init(0xca6e_5eed);
    const rand = prng.random();

    for ([_]usize{ 1, 2, 5, 12, 30, 64 }) |n| {
        var fx: Fixture = undefined;
        try fx.init(io);
        defer fx.deinit(io);

        var names: std.ArrayList([]u8) = .empty;
        defer {
            for (names.items) |nm| testing.allocator.free(nm);
            names.deinit(testing.allocator);
        }
        var pbuf: [1]artifact.PCard = undefined;
        var day: usize = 0;
        while (day < n) : (day += 1) {
            var tsbuf: [32]u8 = undefined;
            const ts = try std.fmt.bufPrint(&tsbuf, "2021-01-01T00:00:{d:0>2}", .{day % 60});
            // Roughly half the check-ins descend from a random earlier one; the
            // rest are fresh roots — a mix of chains, forks, and independent tips.
            const parent: ?[]const u8 = if (names.items.len != 0 and rand.boolean())
                names.items[rand.intRangeLessThan(usize, 0, names.items.len)]
            else
                null;
            const tags: []const artifact.TCard = if (parent == null)
                &[_]artifact.TCard{.{ .sign = .propagate, .name = "branch", .target = null, .value = "trunk" }}
            else
                &.{};
            var m = checkinManifest(ts, parent, tags, &pbuf);
            defer m.deinit();
            try names.append(testing.allocator, try nameOf(&m));
            try fx.put(&m);
        }

        var body: std.ArrayList(u8) = .empty;
        defer body.deinit(testing.allocator);
        var ref: repoview.RepoView = undefined;
        const key = try captureAndReference(&fx, &ref, &body);
        defer ref.deinit(testing.allocator);

        store(testing.allocator, io, fx.tmp.dir, key, 0, ref.main_branch, body.items, "");
        var hit: repoview.RepoView = undefined;
        try testing.expect(tryLoad(testing.allocator, io, fx.tmp.dir, fx.fs.storageFile(), &hit, .full));
        defer hit.deinit(testing.allocator);
        try expectViewsEqual(&ref, &hit);
    }
}

test "cache §7.3: a grown-log key mismatch is a miss (caller rebuilds, then rewrites)" {
    const io = testing.io;
    var fx: Fixture = undefined;
    try fx.init(io);
    defer fx.deinit(io);
    try seedRepo(&fx);

    var body: std.ArrayList(u8) = .empty;
    defer body.deinit(testing.allocator);
    var ref: repoview.RepoView = undefined;
    const key = try captureAndReference(&fx, &ref, &body);
    ref.deinit(testing.allocator);
    store(testing.allocator, io, fx.tmp.dir, key, 0, "trunk", body.items, "");

    // Grow the log by one control artifact: the live tail advances, so the
    // cache written at the old key must now MISS (validateTail sees the change).
    var ctl = controlManifest("2020-02-01T00:00:00", &[_]artifact.TCard{.{ .sign = .add, .name = "wip", .target = null, .value = null }});
    defer ctl.deinit();
    try fx.put(&ctl);

    var view: repoview.RepoView = undefined;
    try testing.expect(!tryLoad(testing.allocator, io, fx.tmp.dir, fx.fs.storageFile(), &view, .full));

    // The rebuild+rewrite path yields a hit again at the new key.
    var body2: std.ArrayList(u8) = .empty;
    defer body2.deinit(testing.allocator);
    var ref2: repoview.RepoView = undefined;
    const grown_key = try captureAndReference(&fx, &ref2, &body2);
    defer ref2.deinit(testing.allocator);
    try testing.expect(grown_key.end != key.end); // the key genuinely advanced
    store(testing.allocator, io, fx.tmp.dir, grown_key, 0, ref2.main_branch, body2.items, "");
    var hit2: repoview.RepoView = undefined;
    try testing.expect(tryLoad(testing.allocator, io, fx.tmp.dir, fx.fs.storageFile(), &hit2, .full));
    defer hit2.deinit(testing.allocator);
    try expectViewsEqual(&ref2, &hit2);
}

test "cache: a corrupt, truncated, wrong-magic, or wrong-version cache is a miss, never trusted" {
    const io = testing.io;
    var fx: Fixture = undefined;
    try fx.init(io);
    defer fx.deinit(io);
    try seedRepo(&fx);

    var body: std.ArrayList(u8) = .empty;
    defer body.deinit(testing.allocator);
    var ref: repoview.RepoView = undefined;
    const key = try captureAndReference(&fx, &ref, &body);
    ref.deinit(testing.allocator);
    store(testing.allocator, io, fx.tmp.dir, key, 0, "trunk", body.items, "");

    // Read the freshly written cache back so we can mutate copies of it.
    var f = try fx.tmp.dir.openFile(io, cache_filename, .{});
    const sz = (try f.stat(io)).size;
    const original = try testing.allocator.alloc(u8, @intCast(sz));
    defer testing.allocator.free(original);
    _ = try f.readPositionalAll(io, original, 0);
    f.close(io);

    const rewrite = struct {
        fn go(dir: std.Io.Dir, aio: std.Io, bytes: []const u8) !void {
            var af = try dir.createFileAtomic(aio, cache_filename, .{ .replace = true });
            defer af.deinit(aio);
            try af.file.writePositionalAll(aio, bytes, 0);
            try af.replace(aio);
        }
    }.go;

    // (a) A flipped self-checksum byte ⇒ miss.
    {
        const bad = try testing.allocator.dupe(u8, original);
        defer testing.allocator.free(bad);
        bad[bad.len - 1] ^= 0xff;
        try rewrite(fx.tmp.dir, io, bad);
        var v: repoview.RepoView = undefined;
        try testing.expect(!tryLoad(testing.allocator, io, fx.tmp.dir, fx.fs.storageFile(), &v, .full));
    }
    // (b) A flipped body byte (checksum now fails) ⇒ miss.
    {
        const bad = try testing.allocator.dupe(u8, original);
        defer testing.allocator.free(bad);
        bad[bad.len / 2] ^= 0x01;
        try rewrite(fx.tmp.dir, io, bad);
        var v: repoview.RepoView = undefined;
        try testing.expect(!tryLoad(testing.allocator, io, fx.tmp.dir, fx.fs.storageFile(), &v, .full));
    }
    // (c) Truncated to half its length ⇒ miss (short read / bad checksum).
    {
        try rewrite(fx.tmp.dir, io, original[0 .. original.len / 2]);
        var v: repoview.RepoView = undefined;
        try testing.expect(!tryLoad(testing.allocator, io, fx.tmp.dir, fx.fs.storageFile(), &v, .full));
    }
    // (d) Wrong magic ⇒ miss.
    {
        const bad = try testing.allocator.dupe(u8, original);
        defer testing.allocator.free(bad);
        bad[0] = 'X';
        try rewrite(fx.tmp.dir, io, bad);
        var v: repoview.RepoView = undefined;
        try testing.expect(!tryLoad(testing.allocator, io, fx.tmp.dir, fx.fs.storageFile(), &v, .full));
    }
    // (e) Wrong version ⇒ miss.
    {
        const bad = try testing.allocator.dupe(u8, original);
        defer testing.allocator.free(bad);
        std.mem.writeInt(u16, bad[8..10], format_version + 1, .little);
        try rewrite(fx.tmp.dir, io, bad);
        var v: repoview.RepoView = undefined;
        try testing.expect(!tryLoad(testing.allocator, io, fx.tmp.dir, fx.fs.storageFile(), &v, .full));
    }
}

test "cache migration: a legacy root FILE is a miss, and the first write replaces it with the directory" {
    const io = testing.io;
    var fx: Fixture = undefined;
    try fx.init(io);
    defer fx.deinit(io);
    try seedRepo(&fx);

    var body: std.ArrayList(u8) = .empty;
    defer body.deinit(testing.allocator);
    var ref: repoview.RepoView = undefined;
    const key = try captureAndReference(&fx, &ref, &body);
    defer ref.deinit(testing.allocator);

    // Write at the member path, then DEMOTE the layout to the legacy shape:
    // the directory collapses to a single root file holding the same bytes.
    store(testing.allocator, io, fx.tmp.dir, key, 0, ref.main_branch, body.items, "");
    var member = try fx.tmp.dir.openFile(io, cache_filename, .{});
    const sz = (try member.stat(io)).size;
    const bytes = try testing.allocator.alloc(u8, @intCast(sz));
    defer testing.allocator.free(bytes);
    _ = try member.readPositionalAll(io, bytes, 0);
    member.close(io);
    try fx.tmp.dir.deleteTree(io, cache_dirname);
    {
        var legacy = try fx.tmp.dir.createFile(io, cache_dirname, .{});
        defer legacy.close(io);
        try legacy.writePositionalAll(io, bytes, 0);
    }

    // (c) The legacy file's read tolerance left with the legacy tier: even
    // valid old bytes at the root name are a MISS, never an error.
    var miss: repoview.RepoView = undefined;
    try testing.expect(!tryLoad(testing.allocator, io, fx.tmp.dir, fx.fs.storageFile(), &miss, .full));

    // The first write replaces the dead file with the directory, and the
    // fresh derivation inside it serves a validated hit.
    store(testing.allocator, io, fx.tmp.dir, key, 0, ref.main_branch, body.items, "");
    const st = try fx.tmp.dir.statFile(io, cache_dirname, .{ .follow_symlinks = false });
    try testing.expect(st.kind == .directory);
    var rehit: repoview.RepoView = undefined;
    try testing.expect(tryLoad(testing.allocator, io, fx.tmp.dir, fx.fs.storageFile(), &rehit, .full));
    defer rehit.deinit(testing.allocator);
    try expectViewsEqual(&ref, &rehit);

    // (d) Deleting the whole directory changes no answer -- the next reader
    // simply derives cold.
    try fx.tmp.dir.deleteTree(io, cache_dirname);
    var cold: repoview.RepoView = undefined;
    try testing.expect(!tryLoad(testing.allocator, io, fx.tmp.dir, fx.fs.storageFile(), &cold, .full));
}

test "cache: an atomic rewrite over a corrupt cache lands a valid, servable one" {
    const io = testing.io;
    var fx: Fixture = undefined;
    try fx.init(io);
    defer fx.deinit(io);
    try seedRepo(&fx);

    // Plant a garbage cache member first (a prior torn/foreign file). The
    // planting itself needs the directory the production writer would have
    // made.
    {
        try testing.expect(ensureCacheDir(io, fx.tmp.dir));
        var af = try fx.tmp.dir.createFileAtomic(io, cache_filename, .{ .replace = true });
        defer af.deinit(io);
        try af.file.writePositionalAll(io, "garbage not a cache", 0);
        try af.replace(io);
    }

    var body: std.ArrayList(u8) = .empty;
    defer body.deinit(testing.allocator);
    var ref: repoview.RepoView = undefined;
    const key = try captureAndReference(&fx, &ref, &body);
    defer ref.deinit(testing.allocator);

    // The garbage file is a clean miss...
    {
        var v: repoview.RepoView = undefined;
        try testing.expect(!tryLoad(testing.allocator, io, fx.tmp.dir, fx.fs.storageFile(), &v, .full));
    }
    // ...and the atomic store replaces it in place with a servable cache.
    store(testing.allocator, io, fx.tmp.dir, key, 0, ref.main_branch, body.items, "");
    var hit: repoview.RepoView = undefined;
    try testing.expect(tryLoad(testing.allocator, io, fx.tmp.dir, fx.fs.storageFile(), &hit, .full));
    defer hit.deinit(testing.allocator);
    try expectViewsEqual(&ref, &hit);
}

test "cache: tryLoad on an absent cache is a miss, not an error" {
    const io = testing.io;
    var fx: Fixture = undefined;
    try fx.init(io);
    defer fx.deinit(io);
    try seedRepo(&fx);
    var v: repoview.RepoView = undefined;
    try testing.expect(!tryLoad(testing.allocator, io, fx.tmp.dir, fx.fs.storageFile(), &v, .full));
}

/// Assert an image-backed view answers every FREED query identically to the
/// derived reference (§7.1 leaf set/order, §7.4 open predicate, §7.5/§8.3 event
/// times, §6.1 branch-of, §8.2 name resolution, plus hash/lookup). The cold
/// queries (`tagsOf`/`branches`/`tagNames`/adjacency) are intentionally not
/// compared — the image does not serve them; `.full` verbs do.
fn expectFreedQueriesEqual(image_view: *repoview.RepoView, ref: *repoview.RepoView) !void {
    try testing.expect(image_view.state.backing == .image);
    try testing.expectEqualStrings(ref.main_branch, image_view.main_branch);
    const count = ref.store.store().count();
    try testing.expectEqual(count, image_view.state.img.?.count());

    var la = ref.state.leaves();
    var lb = image_view.state.leaves();
    while (la.next()) |ra| {
        const rb = lb.next() orelse return error.LeafCountMismatch;
        try testing.expectEqual(ra, rb);
    }
    try testing.expect(lb.next() == null);

    var edges_seen: usize = 0;
    var rid: repostate.Rid = 1;
    while (rid <= count) : (rid += 1) {
        try testing.expectEqualStrings(ref.hashOf(rid), image_view.hashOf(rid));
        try testing.expectEqual(rid, image_view.lookup(ref.hashOf(rid)).?);
        try testing.expectEqual(ref.state.mtimeOf(rid), image_view.state.mtimeOf(rid));
        try testing.expectEqual(ref.state.isLeaf(rid), image_view.state.isLeaf(rid));
        try testing.expectEqual(ref.state.isOpenLeaf(rid), image_view.state.isOpenLeaf(rid));
        if (ref.state.mtimeOf(rid) != null)
            try testing.expectEqualStrings(ref.state.branchOf(rid, ref.main_branch), image_view.state.branchOf(rid, image_view.main_branch));

        // Adjacency (#98), both directions and the primary flag with them. This
        // is the assertion the whole change rests on: the derive and the image
        // hold this relation in DIFFERENT shapes -- `[]const Edge` of u64 rids
        // against narrowed u32 columns -- so "they agree" is a claim that has
        // to be checked rather than assumed. Order is compared too, not just
        // membership: the primary parent is selected positionally by callers.
        for ([2]repostate.RepoState.EdgeDir{ .parents, .children }) |dir| {
            var ea = ref.state.edgesOf(dir, rid);
            var eb = image_view.state.edgesOf(dir, rid);
            while (ea.next()) |a| {
                const b = eb.next() orelse return error.EdgeCountMismatch;
                try testing.expectEqual(a.rid, b.rid);
                try testing.expectEqual(a.is_primary, b.is_primary);
                edges_seen += 1;
            }
            try testing.expect(eb.next() == null);
        }
        // The derived accessors built on those edges must agree too, so a
        // future reader that keeps using them is covered by this test as well.
        try testing.expectEqual(ref.state.parentCount(rid), image_view.state.parentCount(rid));
        try testing.expectEqual(ref.state.primaryParentOf(rid), image_view.state.primaryParentOf(rid));
        try expectPicksEqual(&ref.state, &image_view.state, rid);

        // The resolved §5 tag surface, compared row for row (#98). The image
        // stores what the derive computed -- cancel filtered, precedence
        // applied -- so this checks that the RESOLUTION survived serialization,
        // not that precedence was re-implemented, which the image deliberately
        // does not do. Order is compared too: both arms sort by name and assert
        // strict ascending, so a divergence here is a real one.
        {
            const at = try ref.state.tagsOf(testing.allocator, rid);
            defer testing.allocator.free(at);
            const bt = try image_view.state.tagsOf(testing.allocator, rid);
            defer testing.allocator.free(bt);
            try testing.expectEqual(at.len, bt.len);
            for (at, bt) |x, y| {
                try testing.expectEqualStrings(x.name, y.name);
                try testing.expectEqual(x.tagtype, y.tagtype);
                if (x.value) |xv| {
                    try testing.expectEqualStrings(xv, y.value orelse return error.TagValueMissingFromImage);
                } else {
                    try testing.expect(y.value == null);
                }
            }
        }
        for (std.enums.values(repoimage.DisplayOverride)) |kind| {
            const a = ref.state.displayOverride(rid, kind);
            const b = image_view.state.displayOverride(rid, kind);
            if (a) |av| {
                try testing.expectEqualStrings(av, b orelse return error.OverrideMissingFromImage);
            } else {
                try testing.expect(b == null);
            }
        }
    }
    // A corpus with no linkage rows would compare two empty relations and pass
    // while proving nothing about adjacency -- the failure this whole change
    // exists to prevent. Fail the fixture instead of the reader's confidence.
    try testing.expect(edges_seen > 0);

    {}

    // §8.2: every live symbolic name resolves identically through both views.
    const syms = try ref.state.resolvedSymbols(testing.allocator);
    defer testing.allocator.free(syms);
    for (syms) |s| {
        try testing.expectEqual(ref.state.resolveName(s.name) catch null, image_view.state.resolveName(s.name) catch null);
    }

    // zit#173: the one general timeline query, compared whole. Its exact-mtime
    // tiebreaks must read hashes through the backing-aware door — under image
    // backing the raw store is the panicking trap, so a regression here dies
    // loudly rather than ordering wrong. The fixture guard below fails if the
    // seed ever loses its tied pair, because then the tie path never ran and
    // this comparison would pass while proving nothing.
    const order_ref = try ref.state.checkinsByMtime(testing.allocator);
    defer testing.allocator.free(order_ref);
    const order_img = try image_view.state.checkinsByMtime(testing.allocator);
    defer testing.allocator.free(order_img);
    try testing.expectEqualSlices(repostate.Rid, order_ref, order_img);
    var tie_seen = false;
    var oi: usize = 1;
    while (oi < order_ref.len) : (oi += 1) {
        if (ref.state.mtimeOf(order_ref[oi - 1]).? == ref.state.mtimeOf(order_ref[oi]).?) tie_seen = true;
    }
    try testing.expect(tie_seen);
}

test "cache §7.1/§7.4/§8.2: a .fast_read image load is query-equivalent to a full derive" {
    const io = testing.io;
    var fx: Fixture = undefined;
    try fx.init(io);
    defer fx.deinit(io);
    try seedRepo(&fx);

    var body: std.ArrayList(u8) = .empty;
    defer body.deinit(testing.allocator);
    var ref: repoview.RepoView = undefined;
    const key = try captureAndReference(&fx, &ref, &body);
    defer ref.deinit(testing.allocator);

    var image: std.ArrayList(u8) = .empty;
    defer image.deinit(testing.allocator);
    try repoimage.serialize(testing.allocator, &ref.state, ref.main_branch, &image);
    store(testing.allocator, io, fx.tmp.dir, key, 0, ref.main_branch, body.items, image.items);

    // .fast_read takes the image path; .full derives from the same cache.
    var fast: repoview.RepoView = undefined;
    try testing.expect(tryLoad(testing.allocator, io, fx.tmp.dir, fx.fs.storageFile(), &fast, .fast_read));
    defer fast.deinit(testing.allocator);
    try expectFreedQueriesEqual(&fast, &ref);

    var full: repoview.RepoView = undefined;
    try testing.expect(tryLoad(testing.allocator, io, fx.tmp.dir, fx.fs.storageFile(), &full, .full));
    defer full.deinit(testing.allocator);
    try testing.expect(full.state.backing == .derived);
    try expectViewsEqual(&ref, &full);
}

test "cache: a bit-flip anywhere in the image section fails the CRC ⇒ miss, never a wrong answer" {
    const io = testing.io;
    var fx: Fixture = undefined;
    try fx.init(io);
    defer fx.deinit(io);
    try seedRepo(&fx);

    var body: std.ArrayList(u8) = .empty;
    defer body.deinit(testing.allocator);
    var ref: repoview.RepoView = undefined;
    const key = try captureAndReference(&fx, &ref, &body);
    defer ref.deinit(testing.allocator);

    var image: std.ArrayList(u8) = .empty;
    defer image.deinit(testing.allocator);
    try repoimage.serialize(testing.allocator, &ref.state, ref.main_branch, &image);
    store(testing.allocator, io, fx.tmp.dir, key, 0, ref.main_branch, body.items, image.items);

    // Flip one byte inside the stored image region; the trailing CRC covers it.
    var f = try fx.tmp.dir.openFile(io, cache_filename, .{ .mode = .read_write });
    defer f.close(io);
    const sz: usize = @intCast((try f.stat(io)).size);
    // The image sits just before the 4-byte checksum; flip a byte there.
    const flip_at = sz - checksum_len - 1;
    var one: [1]u8 = undefined;
    _ = try f.readPositionalAll(io, &one, flip_at);
    one[0] ^= 0xFF;
    try f.writePositionalAll(io, &one, flip_at);
    try f.sync(io);

    var v: repoview.RepoView = undefined;
    try testing.expect(!tryLoad(testing.allocator, io, fx.tmp.dir, fx.fs.storageFile(), &v, .fast_read));
}

test "cache zit#180: a stale-layout or absent image is rewritten by the hit that replays it, once" {
    const io = testing.io;
    var fx: Fixture = undefined;
    try fx.init(io);
    defer fx.deinit(io);
    try seedRepo(&fx);

    var body: std.ArrayList(u8) = .empty;
    defer body.deinit(testing.allocator);
    var ref: repoview.RepoView = undefined;
    const key = try captureAndReference(&fx, &ref, &body);
    defer ref.deinit(testing.allocator);
    var image: std.ArrayList(u8) = .empty;
    defer image.deinit(testing.allocator);
    try repoimage.serialize(testing.allocator, &ref.state, ref.main_branch, &image);

    // Once with an image stamped by the previous layout, once with no image at
    // all — the two shapes an older build leaves behind.
    for ([_]bool{ true, false }) |stale_layout| {
        store(testing.allocator, io, fx.tmp.dir, key, 0, ref.main_branch, body.items, if (stale_layout) image.items else "");
        if (stale_layout) {
            var f = try fx.tmp.dir.openFile(io, cache_filename, .{ .mode = .read_write });
            defer f.close(io);
            const sz: usize = @intCast((try f.stat(io)).size);
            const bytes = try testing.allocator.alloc(u8, sz);
            defer testing.allocator.free(bytes);
            _ = try f.readPositionalAll(io, bytes, 0);
            std.mem.writeInt(u16, bytes[10..12], layout_version - 1, .little);
            var crc = crc32.Crc32.init();
            crc.update(bytes[0 .. sz - checksum_len]);
            std.mem.writeInt(u32, bytes[sz - checksum_len ..][0..4], crc.final(), .little);
            try f.writePositionalAll(io, bytes, 0);
        }

        // First read: a hit that had to replay the records.
        var first: repoview.RepoView = undefined;
        try testing.expect(tryLoad(testing.allocator, io, fx.tmp.dir, fx.fs.storageFile(), &first, .fast_read));
        try testing.expect(first.state.backing == .derived);
        try expectViewsEqual(&ref, &first);
        first.deinit(testing.allocator);

        // Second read: the rewritten image serves, and answers the same.
        var second: repoview.RepoView = undefined;
        try testing.expect(tryLoad(testing.allocator, io, fx.tmp.dir, fx.fs.storageFile(), &second, .fast_read));
        defer second.deinit(testing.allocator);
        try testing.expect(second.state.backing == .image);
        try expectFreedQueriesEqual(&second, &ref);
    }
}

test "cache: a CRC-valid but structurally-invalid image ⇒ .fast_read falls back to the record-body derive" {
    const io = testing.io;
    var fx: Fixture = undefined;
    try fx.init(io);
    defer fx.deinit(io);
    try seedRepo(&fx);

    var body: std.ArrayList(u8) = .empty;
    defer body.deinit(testing.allocator);
    var ref: repoview.RepoView = undefined;
    const key = try captureAndReference(&fx, &ref, &body);
    defer ref.deinit(testing.allocator);

    // A non-empty but structurally invalid image: too short for even the header,
    // so `Image.load` returns null and the fast path degrades to a full derive
    // over the record body (still a validated HIT with correct answers).
    store(testing.allocator, io, fx.tmp.dir, key, 0, ref.main_branch, body.items, "not-a-valid-image");

    var v: repoview.RepoView = undefined;
    try testing.expect(tryLoad(testing.allocator, io, fx.tmp.dir, fx.fs.storageFile(), &v, .fast_read));
    defer v.deinit(testing.allocator);
    try testing.expect(v.state.backing == .derived); // fell back to derive
    try expectViewsEqual(&ref, &v);
}

test "cache resume: resumeFrom at every prefix boundary equals a cold capture, view and body, with zero prefix parses" {
    const io = testing.io;
    var fx: Fixture = undefined;
    try fx.init(io);
    defer fx.deinit(io);
    try seedRepo(&fx);

    // The cold-derive reference: full view + full record body.
    var full_body: std.ArrayList(u8) = .empty;
    defer full_body.deinit(testing.allocator);
    var ref: repoview.RepoView = undefined;
    _ = try captureAndReference(&fx, &ref, &full_body);
    defer ref.deinit(testing.allocator);

    var ro = try fx.openScan();
    defer ro.deinit(testing.allocator);
    const rids = try ro.ridsInPersistedOrder(testing.allocator);
    defer testing.allocator.free(rids);
    try testing.expect(rids.len >= 4); // the fixture's four artifacts

    // Every valid resume boundary: the bare header (empty prefix — resume
    // derives everything) through every artifact frame's end (the last one
    // resumes nothing).
    var boundaries: std.ArrayList(u64) = .empty;
    defer boundaries.deinit(testing.allocator);
    try boundaries.append(testing.allocator, logstore.header_len);
    for (rids) |rid| try boundaries.append(testing.allocator, ro.offsetOf(rid) + ro.frameSpanOf(rid));

    for (boundaries.items, 0..) |boundary, i| {
        // The prefix record body a checkpoint would store for this boundary.
        var prefix_body: std.ArrayList(u8) = .empty;
        defer prefix_body.deinit(testing.allocator);
        var prefix_view: repoview.RepoView = undefined;
        try prefix_view.loadCapturingPrefix(testing.allocator, &ro, &prefix_body, boundary);
        prefix_view.deinit(testing.allocator);

        // Resume: replay the prefix body, walk only the tail.
        var combined: std.ArrayList(u8) = .empty;
        defer combined.deinit(testing.allocator);
        var resumed: repoview.RepoView = undefined;
        repoview.parsed_artifacts = 0;
        try resumed.resumeFrom(testing.allocator, &ro, prefix_body.items, boundary, &combined);
        defer resumed.deinit(testing.allocator);

        // §6.1: indistinguishable from the cold derive — same view, and the
        // SAME body a subsequent `.zit-cache` write would store.
        try expectViewsEqual(&ref, &resumed);
        try testing.expectEqualSlices(u8, full_body.items, combined.items);
        // The evidence, not the inference: exactly the tail's artifacts were
        // parsed, zero from below the boundary (i indexes how many artifact
        // frames the prefix covered).
        try testing.expectEqual(@as(u64, rids.len - i), repoview.parsed_artifacts);
    }
}

test "cache capture: view-free captureRecords emits byte-identical records to the deriving walk, split at any boundary" {
    const io = testing.io;
    var fx: Fixture = undefined;
    try fx.init(io);
    defer fx.deinit(io);
    try seedRepo(&fx);

    var full_body: std.ArrayList(u8) = .empty;
    defer full_body.deinit(testing.allocator);
    var ref: repoview.RepoView = undefined;
    _ = try captureAndReference(&fx, &ref, &full_body);
    defer ref.deinit(testing.allocator);

    var ro = try fx.openScan();
    defer ro.deinit(testing.allocator);
    const rids = try ro.ridsInPersistedOrder(testing.allocator);
    defer testing.allocator.free(rids);

    for (rids) |rid| {
        const boundary = ro.offsetOf(rid) + ro.frameSpanOf(rid);
        // prefix ++ tail, both captured WITHOUT a view, equals the deriving
        // walk's whole-body capture byte for byte.
        var stitched: std.ArrayList(u8) = .empty;
        defer stitched.deinit(testing.allocator);
        try repoview.captureRecords(testing.allocator, &ro, logstore.header_len, boundary, &stitched);
        try repoview.captureRecords(testing.allocator, &ro, boundary, (try ro.logKey()).end, &stitched);
        try testing.expectEqualSlices(u8, full_body.items, stitched.items);
    }
}

/// A manifest carrying its own F cards, and optionally a B card naming the
/// baseline they delta against. `checkinManifest`'s file-free sibling.
fn filesManifest(ts: []const u8, parent: ?[]const u8, baseline: ?[]const u8, files: []const artifact.FCard, tags: []const artifact.TCard, pbuf: []artifact.PCard) artifact.Manifest {
    var m = checkinManifest(ts, parent, tags, pbuf);
    m.b = if (baseline) |b| .{ .hash = b } else null;
    m.f = files;
    return m;
}

test "zit#172: a delta chain longer than the manifest cache derives identically under eviction" {
    // The diff step keeps parsed manifests warm and pins the ones a step is
    // using. A chain longer than the cache is what makes eviction run at all,
    // and a step holds a parent, a child and their shared baseline at once —
    // so this is the shape where a wrongly-evicted entry becomes a walk over
    // freed memory. The seeded fixture is three manifests deep and could never
    // reach it. Verified through the byte-identical invariant: whatever the
    // eviction did, the records must still match the deriving walk exactly.
    const io = testing.io;
    var fx: Fixture = undefined;
    try fx.init(io);
    defer fx.deinit(io);

    var names: std.ArrayList([]u8) = .empty;
    defer {
        for (names.items) |n| testing.allocator.free(n);
        names.deinit(testing.allocator);
    }

    // A baseline of a few files, then a chain of deltas over it — longer than
    // `ListMemo.max_parsed` so entries are evicted while the walk runs.
    const tracked = 6;
    var base_files: [tracked]artifact.FCard = undefined;
    var base_names: [tracked][8]u8 = undefined;
    var base_hashes: [tracked][64]u8 = undefined;
    for (0..tracked) |i| {
        base_names[i] = .{ 'f', 'i', 'l', 'e', '0' + @as(u8, @intCast(i)), 0, 0, 0 };
        @memset(&base_hashes[i], 'a');
        base_hashes[i][0] = "0123456789abcdef"[i];
        base_files[i] = .{ .filename = base_names[i][0..5], .hash = &base_hashes[i], .perm = null, .old_name = null, .mode = .regular };
    }
    var pbuf: [1]artifact.PCard = undefined;
    const branch_root = [_]artifact.TCard{.{ .sign = .propagate, .name = "branch", .target = null, .value = "trunk" }};
    var base_m = filesManifest("2020-01-01T00:00:00", null, null, &base_files, &branch_root, &pbuf);
    defer base_m.deinit();
    const base_name = try nameOf(&base_m);
    try names.append(testing.allocator, base_name);
    try fx.put(&base_m);

    const chain = 80; // > ListMemo.max_parsed
    var parent = base_name;
    var edit_hashes: [chain][64]u8 = undefined;
    var ts_buf: [chain][20]u8 = undefined;
    for (0..chain) |i| {
        @memset(&edit_hashes[i], 'b');
        edit_hashes[i][0] = "0123456789abcdef"[i % 16];
        edit_hashes[i][1] = "0123456789abcdef"[(i / 16) % 16];
        // Each delta edits one baseline file, cycling through them.
        const which = i % tracked;
        const edit = [_]artifact.FCard{.{ .filename = base_names[which][0..5], .hash = &edit_hashes[i], .perm = null, .old_name = null, .mode = .regular }};
        _ = try std.fmt.bufPrint(&ts_buf[i], "2020-02-{d:0>2}T{d:0>2}:00:00", .{ (i % 28) + 1, i % 24 });
        var m = filesManifest(ts_buf[i][0..19], parent, base_name, &edit, &.{}, &pbuf);
        defer m.deinit();
        const name = try nameOf(&m);
        try names.append(testing.allocator, name);
        try fx.put(&m);
        parent = name;
    }

    var full_body: std.ArrayList(u8) = .empty;
    defer full_body.deinit(testing.allocator);
    var ref: repoview.RepoView = undefined;
    _ = try captureAndReference(&fx, &ref, &full_body);
    defer ref.deinit(testing.allocator);
    // The chain really was walked: every check-in contributes a record.
    try testing.expect(full_body.items.len > chain);

    var ro = try fx.openScan();
    defer ro.deinit(testing.allocator);
    var stitched: std.ArrayList(u8) = .empty;
    defer stitched.deinit(testing.allocator);
    try repoview.captureRecords(testing.allocator, &ro, logstore.header_len, (try ro.logKey()).end, &stitched);
    try testing.expectEqualSlices(u8, full_body.items, stitched.items);
}

/// The worktree registry (#70): each checkout that opens or updates against
/// this store stamps its own absolute path here, so the store can answer
/// "which trees do I serve" (`zit worktrees`) and relocation flows can warn
/// before stranding satellites. Same disposable-tier discipline as the
/// store cone: advisory, best-effort, self-healing — a lost or stale file
/// re-fills as trees are used, and readers lazily prune entries whose path
/// died or re-pointed elsewhere. Line-based: one absolute path per line.
pub const worktrees_member: []const u8 = "worktrees";
pub const worktrees_filename: []const u8 = cache_dirname ++ "/" ++ worktrees_member;

/// Bound on registered trees: past this the stamp is silently skipped —
/// the registry is advisory, and pruning usually frees slots long before.
pub const max_worktrees: usize = 1024;

/// Read the registry. Absent, unreadable, or malformed reads as EMPTY (the
/// self-healing posture). Caller frees each path and the slice.
pub fn readWorktrees(gpa: std.mem.Allocator, io: std.Io, dir: std.Io.Dir) [][]const u8 {
    const empty: [][]const u8 = &.{};
    var file = dir.openFile(io, worktrees_filename, .{}) catch return empty;
    defer file.close(io);
    const st = file.stat(io) catch return empty;
    const size = std.math.cast(usize, st.size) orelse return empty;
    if (size == 0 or size > max_worktrees * 4096) return empty;
    const buf = gpa.alloc(u8, size) catch return empty;
    defer gpa.free(buf);
    const n = file.readPositionalAll(io, buf, 0) catch return empty;
    if (n != size) return empty;
    var out: std.ArrayList([]const u8) = .empty;
    var it = std.mem.splitScalar(u8, buf[0..n], '\n');
    while (it.next()) |line| {
        if (line.len == 0) continue;
        if (line[0] != '/') continue; // absolute paths only; anything else is damage
        if (out.items.len >= max_worktrees) break;
        const copy = gpa.dupe(u8, line) catch break;
        out.append(gpa, copy) catch {
            gpa.free(copy);
            break;
        };
    }
    return out.toOwnedSlice(gpa) catch empty;
}

pub fn freeWorktrees(gpa: std.mem.Allocator, list: [][]const u8) void {
    for (list) |p| gpa.free(p);
    if (list.len != 0) gpa.free(list);
}

/// Persist the registry (atomic write-temp + rename, like every member of
/// this tier). Best-effort: a failed write leaves the previous registry
/// standing. An empty list removes the file.
pub fn writeWorktrees(gpa: std.mem.Allocator, io: std.Io, dir: std.Io.Dir, paths: []const []const u8) void {
    if (paths.len == 0) {
        dir.deleteFile(io, worktrees_filename) catch {};
        return;
    }
    var body: std.ArrayList(u8) = .empty;
    defer body.deinit(gpa);
    for (paths[0..@min(paths.len, max_worktrees)]) |p| {
        body.appendSlice(gpa, p) catch return;
        body.append(gpa, '\n') catch return;
    }
    if (!ensureCacheDir(io, dir)) return;
    var af = dir.createFileAtomic(io, worktrees_filename, .{ .replace = true }) catch return;
    defer af.deinit(io);
    af.file.writePositionalAll(io, body.items, 0) catch return;
    af.file.sync(io) catch return;
    af.replace(io) catch return;
}

/// Stamp one checkout path into the registry: read-merge-write, dedupe by
/// exact path, append at the end. Best-effort; a lost race self-heals on
/// that tree's next open or update.
pub fn stampWorktree(gpa: std.mem.Allocator, io: std.Io, dir: std.Io.Dir, abs_path: []const u8) void {
    if (abs_path.len == 0 or abs_path[0] != '/') return;
    const existing = readWorktrees(gpa, io, dir);
    defer freeWorktrees(gpa, existing);
    for (existing) |p| {
        if (std.mem.eql(u8, p, abs_path)) return; // already registered
    }
    if (existing.len >= max_worktrees) return;
    var merged: std.ArrayList([]const u8) = .empty;
    defer merged.deinit(gpa);
    for (existing) |p| merged.append(gpa, p) catch return;
    merged.append(gpa, abs_path) catch return;
    writeWorktrees(gpa, io, dir, merged.items);
}

test "worktree registry: stamp dedupes, round-trips, and malformed reads as empty" {
    // Goal: the advisory-tier contract — stamps accumulate without
    // duplicates, an empty write removes the file, and damage (relative
    // path lines) is skipped rather than propagated.
    const gpa = std.testing.allocator;
    const io = std.testing.io;
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();

    stampWorktree(gpa, io, tmp.dir, "/a/tree-one");
    stampWorktree(gpa, io, tmp.dir, "/b/tree-two");
    stampWorktree(gpa, io, tmp.dir, "/a/tree-one"); // dedupe
    const got = readWorktrees(gpa, io, tmp.dir);
    defer freeWorktrees(gpa, got);
    try std.testing.expectEqual(@as(usize, 2), got.len);
    try std.testing.expectEqualStrings("/a/tree-one", got[0]);
    try std.testing.expectEqualStrings("/b/tree-two", got[1]);

    writeWorktrees(gpa, io, tmp.dir, &.{});
    const gone = readWorktrees(gpa, io, tmp.dir);
    defer freeWorktrees(gpa, gone);
    try std.testing.expectEqual(@as(usize, 0), gone.len);

    // A relative-path line is damage: skipped, the rest read.
    var f = try tmp.dir.createFile(io, worktrees_filename, .{});
    try f.writePositionalAll(io, "relative/junk\n/c/tree-three\n", 0);
    f.close(io);
    const healed = readWorktrees(gpa, io, tmp.dir);
    defer freeWorktrees(gpa, healed);
    try std.testing.expectEqual(@as(usize, 1), healed.len);
    try std.testing.expectEqualStrings("/c/tree-three", healed[0]);
}

test "zit#182: a derive consuming the disposition sidecar is byte-identical and reconstructs only the relevant frames" {
    const io = testing.io;
    var fx: Fixture = undefined;
    try fx.init(io);
    defer fx.deinit(io);

    // Two check-ins, one control, and two file blobs: the blobs are the
    // frames the sidecar exists to skip.
    var pbuf: [1]artifact.PCard = undefined;
    var root = checkinManifest("2021-01-01T00:00:01", null, &.{}, &pbuf);
    defer root.deinit();
    try fx.put(&root);
    const root_name = try nameOf(&root);
    defer testing.allocator.free(root_name);
    var pbuf2: [1]artifact.PCard = undefined;
    var child = checkinManifest("2021-01-01T00:00:02", root_name, &.{}, &pbuf2);
    defer child.deinit();
    try fx.put(&child);
    var ctl = controlManifest("2021-01-01T00:00:03", &.{.{ .sign = .add, .name = "closed", .target = root_name, .value = null }});
    defer ctl.deinit();
    try fx.put(&ctl);
    for ([_][]const u8{ "blob one\n", "blob two, longer\n" }) |blob| {
        var digest: [32]u8 = undefined;
        std.crypto.hash.sha3.Sha3_256.hash(blob, &digest, .{});
        _ = try fx.log.put(&digest, blob);
    }

    var scan = try fx.openScan();
    defer scan.deinit(testing.allocator);

    // Full derive: capture the body, distill the map.
    var v1: repoview.RepoView = undefined;
    var body1: std.ArrayList(u8) = .empty;
    defer body1.deinit(testing.allocator);
    var built: ?repoview.BuiltDispositions = null;
    defer if (built) |*b| b.deinit(testing.allocator);
    const parses_before_full = repoview.parsed_artifacts;
    try v1.loadCapturingDispositions(testing.allocator, &scan, &body1, null, &built, null);
    defer v1.deinit(testing.allocator);
    const full_parses = repoview.parsed_artifacts - parses_before_full;
    try testing.expectEqual(@as(u64, 5), full_parses); // every frame reconstructed

    const blob = (try repoview.serializeDispositions(testing.allocator, &scan, &built.?)).?;
    defer testing.allocator.free(blob);
    var disp = repoview.parseDispositions(blob, &scan) orelse return error.TestUnexpectedResult;
    try testing.expectEqual(@as(u64, 5), disp.n_rids);

    // Consuming derive: byte-identical body, and ONLY the three relevant
    // frames are reconstructed — the two blobs are never touched. The map it
    // distills while consuming round-trips to the same bytes.
    var v2: repoview.RepoView = undefined;
    var body2: std.ArrayList(u8) = .empty;
    defer body2.deinit(testing.allocator);
    var built2: ?repoview.BuiltDispositions = null;
    defer if (built2) |*b| b.deinit(testing.allocator);
    const parses_before = repoview.parsed_artifacts;
    try v2.loadCapturingDispositions(testing.allocator, &scan, &body2, &disp, &built2, null);
    defer v2.deinit(testing.allocator);
    const consuming_parses = repoview.parsed_artifacts - parses_before;
    try testing.expectEqual(@as(u64, 3), consuming_parses);
    try testing.expectEqualSlices(u8, body1.items, body2.items);
    const blob2 = (try repoview.serializeDispositions(testing.allocator, &scan, &built2.?)).?;
    defer testing.allocator.free(blob2);
    try testing.expectEqualSlices(u8, blob, blob2);

    // A flipped bit fails the checksum and reads as absent — never an error,
    // never trusted.
    const mangled = try testing.allocator.dupe(u8, blob);
    defer testing.allocator.free(mangled);
    mangled[mangled.len / 2] ^= 0x01;
    try testing.expect(repoview.parseDispositions(mangled, &scan) == null);

    // A wrong shun fingerprint reads as absent even with a valid checksum:
    // a lifted shun must discard the map, or a revived check-in vanishes.
    const refp = try testing.allocator.dupe(u8, blob);
    defer testing.allocator.free(refp);
    refp[34] ^= 0x01; // the fp byte
    var crc = @import("crc32.zig").Crc32.init();
    crc.update(refp[0 .. refp.len - 4]);
    std.mem.writeInt(u32, refp[refp.len - 4 ..][0..4], crc.final(), .little);
    try testing.expect(repoview.parseDispositions(refp, &scan) == null);
}