Artifact [c22465c8b2]

Artifact c22465c8b2c52c78bafb7d0cbba6355cf5fa7b0e494fb75a1f1bd100f78dde3c:

• File src/pack.zig — part of check-in [e5db38a6df] at 2026-08-05 21:16:30 on branch sub-manifest — citations: re-key 31 verified line ranges, baseline 512 to 481

  Each was located by reading the draft text that supports what the code comment actually claims, then adversarially re-checked against a verbatim quote of those lines. Clamping a range into its section would satisfy the checker while pointing at the wrong paragraph, so the checker alone was never the gate.

  Of 33 proposals, 2 were refuted on exactly that ground -- the quote verified byte-for-byte and still did not state the claim -- and 4 came back unresolved because no draft's §4.1 states what those comments assert at all. Those are mis-sectioned, not line-rotted, and need a section decision rather than a re-key.

  Also restores container.zig's `decode` doc block, which 7d26a31305 left stranded above `decodePrefix` when that function was inserted between the comment and the function it describes.

  test 3495/3495, fmt. Issue #18. (user: dmestas, size: 142377) [annotate] [blame] [check-ins using]

//! The `zit pack` re-deltification pass — the fossil `extra_deltification`
//! analog (storage-delta-base-selection.md §6). A whole-store generational
//! rewrite that reconstructs every content artifact and re-stores it through
//! the `pickBase` best-of-N policy (logstore.zig), so a source frame's stored
//! encoding — including a verbatim wire delta whose sender chose the base — is
//! never trusted at rest. This is the pure store-to-store core, reached both by
//! the CLI wiring in main.zig (`runPack`) and by the DST/fuzz harnesses that
//! exercise the transform in-process; the CLI-layer temp-file/rename plumbing
//! stays in main.zig.
//!
//! In-module file: it reaches `logstore`/`artifact` through relative sibling
//! imports (the same way logstore.zig reaches delta/container), not through the
//! public `zit` module — a relative import here shares the one module-local copy
//! of each dependency rather than pulling a second into a foreign root.

const std = @import("std");
const logstore = @import("logstore.zig");
const artifact = @import("artifact.zig");

/// The binary digest width the pack hex helpers round-trip (sha256/sha3-256).
const digest_len = 32;

/// pack's closed error set: memory, a destination write fault, or a source
/// artifact whose reconstructed content does not hash to its name (§7.7) or
/// whose frame is otherwise corrupt.
pub const PackError = error{ OutOfMemory, Storage, Corrupt };

/// One repack's tally: the artifact count (== the source's content frames), how
/// many landed as deltas in the new log, how many the `Filter` dropped, how many
/// NEW shun rows the new generation gained over the source's table, and the
/// on-disk sizes before/after. `artifacts` counts the source frames WALKED, so
/// `artifacts - dropped` is what the new generation holds.
///
/// `shuns_recorded` is an OBSERVATION, not a restatement of `extra_shuns.len`:
/// it is measured off `dst`'s own table at the row-count gate below. That is the
/// difference that matters to the one caller that reports a takedown to an
/// operator — `zit shun`'s success line names a number the new store actually
/// holds rather than the number the run intended to write, so no future
/// `extra_shuns` caller can over-report a removal it did not record.
pub const PackStats = struct {
    artifacts: usize,
    deltas: usize,
    dropped: usize,
    shuns_recorded: usize,
    bytes_before: u64,
    bytes_after: u64,
};

/// The reclamation seam: a borrowed predicate deciding, per artifact NAME, which
/// of the source's frames the new generation simply does not receive. This is
/// the pack FILTER — the only mechanism by which `zit pack` removes bytes
/// rather than merely relocating them.
///
/// NO DRAFT SPECIFIES THIS. Sync §3.3 L518-525 defines the repository shun table
/// as "a persistent, repository-local set keyed by complete artifact name" and
/// L527-529 makes it mask the phantom scan; nothing in any of the four drafts says
/// whether, when, or how a shunned artifact's BYTES leave the store. Reclamation
/// is therefore a decision this implementation makes on its own, and the rules
/// stated across this module — filter-before-reconstruction, the tombstone
/// outliving the content, the end-state sweep — are ITS rules, referred to below
/// as "the reclaim contract" and deliberately carrying no section number so they
/// are never mistaken for normative text. ERRATUM CANDIDATE against sync §3.3: the
/// draft describes a set that grows without bound and never states the operation
/// that empties it, leaving every implementation to invent both the trigger and
/// the ordering constraints that make it safe.
///
/// Keyed on the DIGEST, not a source rid, because the drop set must be askable of
/// BOTH stores: `packVerifyShunned` re-asks this predicate over `dst`'s own
/// entries, and dst rids are freshly assigned — they do not correspond to src
/// rids — so a rid-keyed predicate could not express the end-state sweep at all.
/// The name is the only key both generations share. (It also leaves room for a
/// caller whose drop set is not in the source's shun table, but the sweep is the
/// reason it is required.) A predicate rather than a
/// `[]const []const u8` drop list because `zit pack`'s own set is "everything in
/// the shun table", which the store already indexes in O(1); materializing that
/// index as a slice would duplicate it for no gain.
///
/// DOMAIN — enforced HERE, in the seam, not delegated to implementations. A
/// `Filter` is asked about an entry name of ANY stored width
/// (`logstore.min_hash_len .. max_hash_len`), because `packVerifyShunned` walks
/// every interned name including abbreviated ones, but a drop set the drafts
/// define as keyed by COMPLETE artifact name (sync §3.3 L518-525) cannot contain
/// a prefix. `drops` therefore answers false for an incomplete width WITHOUT
/// consulting `dropFn` at all, so no implementation can be the reason an
/// artifact nobody named is deleted: a `dropFn` answering true for a short key
/// would reclaim a frame the operator never asked about, and
/// `packVerifyShunned` — which re-asks the SAME predicate — would agree the
/// deletion was correct. One enforcement point rather than an obligation
/// restated in every implementation. This forecloses a hypothetical caller whose
/// drop set is legitimately keyed on abbreviated names; no such caller can exist
/// for a set keyed by complete name, and inventing one would be inventing
/// behavior the drafts do not describe.
pub const Filter = struct {
    ctx: *anyopaque,
    dropFn: *const fn (ctx: *anyopaque, digest: []const u8) bool,

    /// The identity filter: nothing is dropped, so a repack relocates every
    /// frame exactly as it did before the filter existed. Every call site that
    /// is not reclaiming passes this.
    pub const keep_all: Filter = .{ .ctx = @constCast(&keep_all_ctx), .dropFn = keepAll };

    /// Does the new generation omit the artifact named `digest`?
    fn drops(self: Filter, digest: []const u8) bool {
        std.debug.assert(digest.len >= logstore.min_hash_len);
        std.debug.assert(digest.len <= logstore.max_hash_len);
        // The DOMAIN rule (see above), enforced once for every implementation.
        if (!logstore.isCompleteDigest(digest.len)) return false;
        return self.dropFn(self.ctx, digest);
    }
};

/// `keep_all`'s unused context — a real addressable byte, so the identity filter
/// holds a valid pointer rather than `undefined`. Never dereferenced.
const keep_all_ctx: u8 = 0;

fn keepAll(_: *anyopaque, _: []const u8) bool {
    return false;
}

/// The `Filter` `zit pack` runs: drop every artifact the source store's own shun
/// table names. The store IS the context — no wrapper state, and the drop set
/// stays the store's O(1) index rather than a copied list.
///
/// The tombstones themselves are NOT dropped: `packLog` carries the whole set
/// forward, so a reclaimed artifact reads as absent for the same reason it did
/// before the pack, and a later sync will not re-fetch what the operator removed.
pub fn shunFilter(src: *const logstore.LogStore) Filter {
    return .{ .ctx = @constCast(src), .dropFn = dropsIfShunned };
}

fn dropsIfShunned(ctx: *anyopaque, digest: []const u8) bool {
    const src: *const logstore.LogStore = @ptrCast(@alignCast(ctx));
    // `Filter.drops` has already screened the width, so `isShunned`'s own
    // `isCompleteDigest` assert cannot fire here — restated as a precondition
    // because this function is reached only through that seam.
    std.debug.assert(logstore.isCompleteDigest(digest.len));
    return src.isShunned(digest);
}

/// A borrowed-key set of complete artifact names, and the `Filter` over it. The
/// drop set for a reclaim whose members are not (only) the source's shun table:
/// `zit shun --force` must drop the source's already-deferred shuns AND the
/// hashes named in that very invocation, a union `shunFilter` cannot express
/// because the second half is not in the source store at all.
///
/// It exists so the drop predicate and the `packLog` `extra_shuns` list are
/// DERIVED FROM ONE OBJECT rather than assembled twice. Two independently built
/// lists would eventually disagree, and the failure is silent and one-directional:
/// a name tombstoned but not dropped reads as absent while its bytes ship in the
/// next clone, and no integrity gate objects (`packVerifyShunned` re-asks the
/// FILTER, so a name the filter never held is a name it never looks for).
///
/// Keys are BORROWED, never duped: every member comes either from a
/// `logstore.ShunRecord` the source store owns or from a digest the caller
/// decoded, and both outlive the pack. Sized once by the caller; membership is
/// O(1) and the whole set is bounded by the shun-table size plus the argument
/// count.
pub const NameSet = struct {
    names: std.StringHashMapUnmanaged(void) = .empty,

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

    /// Add one COMPLETE artifact name (20 or 32 raw bytes). The width is a
    /// CHECKED error, not an assert: this set is built from operator input and
    /// its members decide which frames a `ReleaseFast` build permanently deletes,
    /// where an assert compiles to nothing. A short key would sit in the set
    /// matching nothing (`Filter.drops` screens incomplete widths before asking),
    /// i.e. fail-OPEN on a removal the operator was told happened.
    pub fn add(self: *NameSet, gpa: std.mem.Allocator, name: []const u8) error{ OutOfMemory, BadKey }!void {
        if (!logstore.isCompleteDigest(name.len)) return error.BadKey;
        try self.names.put(gpa, name, {});
    }

    pub fn count(self: *const NameSet) usize {
        return self.names.count();
    }

    pub fn contains(self: *const NameSet, name: []const u8) bool {
        return self.names.contains(name);
    }

    /// The `Filter` that drops exactly this set's members. `@constCast` because
    /// `Filter` holds an opaque mutable context pointer for the general case;
    /// `dropsIfInSet` only reads through it.
    pub fn filter(self: *const NameSet) Filter {
        return .{ .ctx = @constCast(self), .dropFn = dropsIfInSet };
    }
};

fn dropsIfInSet(ctx: *anyopaque, digest: []const u8) bool {
    const set: *const NameSet = @ptrCast(@alignCast(ctx));
    // `Filter.drops` has already screened the width; restated as a precondition
    // because this function is reached only through that seam.
    std.debug.assert(logstore.isCompleteDigest(digest.len));
    return set.contains(digest);
}

test "pack: NameSet refuses an incomplete name and drops exactly its members" {
    const gpa = std.testing.allocator;
    var set: NameSet = .{};
    defer set.deinit(gpa);

    // A prefix cannot be a member: the shun table is keyed by COMPLETE artifact
    // name (sync §3.3 L518-519), and a short key would match nothing while
    // reading as a recorded removal.
    try std.testing.expectError(error.BadKey, set.add(gpa, "abcd"));
    try std.testing.expectError(error.BadKey, set.add(gpa, "a" ** 21));
    try std.testing.expectEqual(@as(usize, 0), set.count());

    try set.add(gpa, "a" ** 32);
    try set.add(gpa, "b" ** 20); // the other complete width
    try set.add(gpa, "a" ** 32); // idempotent as set membership
    try std.testing.expectEqual(@as(usize, 2), set.count());

    const f = set.filter();
    try std.testing.expect(f.drops("a" ** 32));
    try std.testing.expect(f.drops("b" ** 20));
    try std.testing.expect(!f.drops("c" ** 32));
    // The domain rule, enforced in the seam: an abbreviated key is never asked
    // about, so no set membership can delete a frame nobody named in full.
    try std.testing.expect(!f.drops(("a" ** 32)[0..8]));
}

/// A source artifact's ordered, bounded base-candidate hints for the repack — a
/// check-in manifest's lineage parent (§6.1) or a file blob's same-path
/// predecessors (§6.2, nearest first), fed to `putArtifactPacked` ahead of the
/// chronological window. Inline fixed buffer (≤ `max_base_candidates`), so the
/// hint map holds no per-entry allocation and every fill is bounded.
const PriorityHints = struct {
    buf: [logstore.max_base_candidates]logstore.Rid = undefined,
    len: u8 = 0,

    /// The recorded hints in priority order.
    fn slice(self: *const PriorityHints) []const logstore.Rid {
        return self.buf[0..self.len];
    }

    /// Append `rid` unless the buffer is full or already holds it (bounded dedup
    /// — a base offered by two lanes, or a path listing the same blob twice,
    /// costs no slot). Priority order is preserved: earlier pushes rank higher.
    fn push(self: *PriorityHints, rid: logstore.Rid) void {
        std.debug.assert(rid >= 1); // a valid log rid is 1-based
        if (self.len >= self.buf.len) return;
        for (self.buf[0..self.len]) |r| if (r == rid) return;
        self.buf[self.len] = rid;
        self.len += 1;
    }
};

/// Repack every artifact from `src` into the fresh, empty `dst` (same
/// `hash_kind`), choosing each frame's delta base by OUR size-minimizing policy
/// instead of trusting the source's stored encoding (§6.1, the fossil
/// `extra_deltification` analog). Every content artifact — including a verbatim
/// wire delta (`putDeltaContainer`), decoded to full bytes first — is run
/// through `pickBase` against its priority hints (a check-in manifest's lineage
/// parent §6.1; a file blob's same-path predecessors §6.2, fossil's mlink analog)
/// plus the chronological window of already-packed frames, so the sender's base
/// choice stops being trusted at rest. Iteration is source offset order, which
/// guarantees a chosen base is already present in `dst` before its dependents
/// (candidates are always earlier frames, and `delta_of` is a byte offset in the
/// NEW file, assigned at write time). The live config survives as exactly one
/// frame; superseded config records are dropped (§E). The repository shun table
/// survives whole, in persisted order — dropping it would make the set's
/// durability claim false the first time anyone packs.
///
/// `filter` is the pack FILTER, the one mechanism by which a repack removes
/// artifact bytes rather than relocating them: an artifact whose name it drops is
/// not copied, and `PackStats.dropped` counts them. `Filter.keep_all` reproduces
/// pre-filter behavior exactly. THE FILTER CHECK PRECEDES RECONSTRUCTION — here
/// and in `deriveCandidateHints` — and that ordering is normative, not an
/// optimisation: content is shunned precisely because its bytes may be unlawful
/// or corrupt, and a corrupt frame fails the §7.7 verify below. Reconstruct
/// first and the one operation that could remove such a frame is the one that
/// refuses to run because of it, forever.
///
/// `extra_shuns` are tombstones that exist ONLY in the new generation: rows the
/// caller wants recorded by the same operation that removes the content they
/// name, written here rather than appended to `src` first. That placement is the
/// whole point, and it is a crash-safety rule, not a convenience — a caller that
/// appended the row to the live log and then rewrote would perform TWO durable
/// transitions and expose a state where the tombstone exists and so does the
/// content. Recording them here leaves exactly one: the caller's rename. Because
/// they land in the same loop as the carried-forward rows — before the artifact
/// loop — `pickBase` cannot root a fresh delta on a newly shunned frame either,
/// for the same reason spelled out at that loop.
///
/// PRECONDITION on `extra_shuns`, checked rather than asserted (below): every
/// row's name is ABSENT from `src.shunList()` and the rows are distinct.
/// `putShun` is idempotent as set membership, so a duplicate would silently
/// collapse and the row-count gate would report `error.Corrupt` — the caller
/// filters the already-shunned names out first, which it must do anyway to tell
/// the operator a repeat was a no-op. Nothing here requires `filter` to drop the
/// names in `extra_shuns`; a caller that tombstones a name without dropping it
/// gets a deferred reclaim, which is `--defer`'s meaning, and a caller reclaiming
/// now derives both from one owned set so the two cannot disagree.
///
/// Three integrity gates bracket the writes: each SOURCE frame that is KEPT is
/// §7.7-verified before it is re-encoded; — MANDATORY (§3) — after the final
/// write every PACKED artifact is reconstructed through its NEW chain and
/// re-hashed; and `packVerifyShunned` re-asks the filter over `dst`'s own
/// entries, so a pack that failed to remove what it was told to remove never
/// gets swapped in. Any mismatch aborts before `commit`, leaving `dst` a
/// discarded temp and `src` untouched. `dst` MUST be a just-opened empty log;
/// `src` is never written. Deterministic: fixed offset-order iteration, a
/// deterministic `pickBase` and a filter that is a pure function of the name make
/// the packed bytes a pure function of the source content — so `pack∘pack`
/// reproduces the same log.
pub fn packLog(
    gpa: std.mem.Allocator,
    src: *logstore.LogStore,
    dst: *logstore.LogStore,
    filter: Filter,
    extra_shuns: []const logstore.ShunRecord,
) PackError!PackStats {
    // §6.1 base-availability precondition (load-bearing): `dst` MUST be a
    // just-opened empty log. If it already held frames, the hint translation
    // `dst.lookup(src.hashOf(sb))` below could resolve a pre-existing unrelated
    // frame and silently mis-base a re-delta. A CHECKED error, not an assert:
    // the mis-base happens at the point of DECISION, and an assert compiles to
    // nothing in `ReleaseFast` — exactly the build whose caller renames `dst`
    // over the original store. A bad base is caught 300 lines later by
    // `packVerifyPacked` (a mis-based delta reconstructs wrong content), so the
    // consequence is a failed pack either way; failing here makes it
    // diagnosable at its origin, for one integer comparison run once per pack.
    if (dst.entryCount() != 0) return error.Corrupt;
    // The live config survives as exactly one frame; every superseded config
    // record is dropped (the whole of pack's "dead frame" reclaim, §E).
    if (try readSrcConfig(gpa, src)) |oc_val| {
        var owned = oc_val;
        defer owned.deinit(gpa);
        dst.putConfig(owned.config) catch |e| return storagePutError(e);
    }

    // The repository shun table carries across verbatim, in persisted order.
    // Shun rows are the second never-interned frame kind, so — exactly like the
    // config record above — they are in neither `ridsInPersistedOrder` nor any
    // content path below, and a pack that did not re-emit them would silently
    // DROP the whole set. That is fail-OPEN on the one predicate the set exists
    // to make durable: the tombstones vanish while the artifacts they name may
    // survive the repack. Preserving the rows is not the pack FILTER — that
    // is `filter`, which decides which CONTENT the new generation receives — it
    // is what makes the set durable across the repo's own compaction at all. The
    // two are complementary and both are required: `zit pack` hands
    // `shunFilter(src)` in, so a shunned artifact's bytes are RECLAIMED here
    // while the row that named them carries forward. Dropping the row alongside
    // the content would silently UN-shun the artifact, and the next sync would
    // re-fetch exactly what the operator removed. Under `Filter.keep_all` the
    // artifact loop reads through the maintenance reader instead, so the frame
    // relocates like any other and only the row travels with it.
    //
    // ORDERING IS LOAD-BEARING — this loop MUST precede the artifact loop, and
    // moving it after would silently change packed OUTPUT rather than fail.
    // `dst.putArtifactPacked` → `pickBase` probes candidates through
    // `dst.content`, which is masked, so with the rows already in place a
    // shunned frame is never chosen as a fresh delta basis in the new
    // generation (the right answer: a basis whose bytes a reclaim is about to
    // drop must not acquire new dependents). Copy the rows afterwards and
    // `pickBase` would happily root new deltas on it. Nothing downstream notices
    // — packed OUTPUT changes with every integrity gate still satisfied — which
    // is why the dependency is stated here AND pinned by the test "a shunned
    // frame is never chosen as a FRESH delta basis in the new generation".
    for (src.shunList()) |row| {
        dst.putShun(row.hash, row.reason, row.epoch) catch |e| return storageShunError(e);
    }
    // The caller's new tombstones, recorded in the NEW generation only (see the
    // function doc: one durable transition, and no fresh delta rooted on a name
    // about to be dropped). Same loop, so they are subject to the same ordering
    // rule as the carry-forward above.
    for (extra_shuns) |row| {
        dst.putShun(row.hash, row.reason, row.epoch) catch |e| return storageShunError(e);
    }
    // The whole set is in place before the first artifact lands — the ordering
    // above, checked rather than merely described, and CHECKED in every build:
    // this is the one in-code gate on the carry-forward, and its violation is
    // silent UN-shunning (the tombstones vanish while `zit pack` reports a
    // successful reclaim, so the next sync re-fetches exactly what the operator
    // was ordered to remove). An assert would leave that rule enforced only in
    // safe builds, i.e. only described in the binary that ships. `putShun`
    // dedups by name, so a duplicate-row source cannot trip this today; the
    // check exists for the next carry-forward path, not for this one.
    //
    // Widened by `extra_shuns.len`, which also makes this the enforcement point
    // for that parameter's precondition: `putShun` dedups by name, so a row
    // duplicating a carried-forward name (or another `extra_shuns` row) appends
    // nothing and the sum no longer matches. Fail-CLOSED on a caller that
    // believed it was recording a tombstone it was not — the pack aborts and the
    // original store stays in place, rather than a "successful" reclaim whose
    // row count the operator has no way to check.
    if (dst.shunList().len != src.shunList().len + extra_shuns.len) return error.Corrupt;
    // Measured HERE, off `dst`, immediately after the gate that proves the sum is
    // exact — so `PackStats.shuns_recorded` is what the new generation gained, not
    // a copy of `extra_shuns.len`. Guaranteed non-underflowing by the line above.
    const shuns_recorded = dst.shunList().len - src.shunList().len;

    // Artifacts in offset order: every candidate base a frame picks precedes it
    // in the source, so it is already written into `dst` (its NEW offset known)
    // before the dependent that names it.
    const rids = try src.ridsInPersistedOrder(gpa);
    defer gpa.free(rids);

    // Priority base hints (source rid → ordered source base rids), from a first
    // pass over the manifests: a check-in manifest against its primary parent
    // manifest (§6.1), plus each file blob's same-path predecessors — the §6.2
    // mlink analog. These are `pickBase`'s highest-priority candidates; the
    // chronological window (added by `putArtifactPacked`) fills the rest. Absent ⇒
    // the window alone decides.
    var hints: std.AutoHashMapUnmanaged(logstore.Rid, PriorityHints) = .empty;
    defer hints.deinit(gpa);
    try deriveCandidateHints(gpa, src, rids, filter, &hints);

    var cache = logstore.PlaintextCache.init();
    defer cache.deinit(gpa);
    var deltas: usize = 0;
    var dropped: usize = 0;
    for (rids) |src_rid| {
        const digest = src.hashOf(src_rid);
        // The pack FILTER, BEFORE `srcContent`/`artifact.verify` — the ordering is
        // normative (see the function doc). A dropped artifact is not
        // reconstructed, so it is never hash-verified, so bytes that are corrupt
        // as well as unlawful cannot make the operation that removes them abort.
        if (filter.drops(digest)) {
            dropped += 1;
            continue;
        }
        // Non-phantom by construction (`ridsInPersistedOrder` filters phantoms)
        // and read through the MAINTENANCE reader (`srcContent` — see its note
        // on why a repack must not read through the shun mask), so content is
        // never null and the `orelse` is unreachable defensive code rather than
        // a silent drop. A verbatim wire delta is decoded to full bytes here,
        // then re-deltified like everything else — its stored base is not
        // preserved.
        const content = (try srcContent(gpa, src, src_rid, &cache)) orelse continue;
        defer gpa.free(content);
        // §7.7 L1465/L1469-1471 — the reconstructed bytes MUST hash to the
        // artifact's name, recomputed BEFORE any frame is written. A lying frame
        // aborts the pack with the original store untouched.
        var name_buf: [2 * logstore.max_hash_len]u8 = undefined;
        const name = digestToWireName(digest, &name_buf);
        if (!artifact.verify(name, content)) return error.Corrupt;
        // Priority base hints translated to their NEW dst rids (every hint is an
        // earlier-offset frame, so it is already written into `dst`). An unheld
        // hint is dropped. Absent/all-unheld ⇒ the chronological window alone
        // decides. `putArtifactPacked` owns best-of-N + the accept threshold +
        // chain-depth policy — pack only offers the ordered hints and the §6.1
        // monotonic floor.
        var pri_buf: [logstore.max_base_candidates]logstore.Rid = undefined;
        var pri_len: usize = 0;
        if (hints.get(src_rid)) |ph| {
            for (ph.slice()) |sb| {
                if (pri_len >= pri_buf.len) break;
                const base_digest = src.hashOf(sb);
                if (dst.lookup(base_digest)) |db| {
                    // Defence in depth (the reclaim contract): a dropped artifact never entered
                    // `dst`, so `lookup` has already resolved null for it and the
                    // `if` above dropped the hint. `deriveCandidateHints` filters
                    // dropped rids out too, so this is the SECOND guard on the
                    // same fact — checked rather than trusted to either one
                    // alone, because a dropped base offered here would be
                    // silently mis-resolved if `dst` ever stopped being empty at
                    // entry. An ERROR and not an assert, per this file's own
                    // rule above: an assert compiles to nothing in `ReleaseFast`
                    // — exactly the build whose caller renames `dst` over the
                    // original store — and this arm is now reachable on every
                    // `import-git`, over artifacts derived from an untrusted
                    // repository, so an assert here is also a process abort keyed
                    // on repository content.
                    if (filter.drops(base_digest)) return error.Corrupt;
                    pri_buf[pri_len] = db;
                    pri_len += 1;
                }
            }
        }
        // §6.1 monotonic-shrink guard: the new encoding is written only if it is
        // strictly smaller than the existing source frame; otherwise pack keeps
        // that frame verbatim (a compact wire delta fossil chose well is never
        // replaced by a larger re-encoding). This makes pack non-worsening
        // frame-by-frame with TWO acknowledged exceptions, both in `copySrcFrame`:
        //   * the §3.3 deep-chain fallback re-stores a KEPT delta as a full zlib
        //     frame when its base's dst depth reaches the read cap, which can
        //     exceed the source delta's span — reachable only once chains approach
        //     `max_delta_chain`;
        //   * a RECLAIM of a delta base makes its dependents' source frames
        //     unre-storable, so `copySrcFrame` re-encodes them from scratch
        //     (a null floor) and the result can exceed the source delta it
        //     replaces. That is inherent to deleting a basis, not a defect — but
        //     it is why the floor below is NOT a whole-store shrink guarantee
        //     under a non-identity filter. It stays a per-frame BEST-AVAILABLE
        //     encoding: what must never happen is discarding a profitable delta
        //     for a full frame, which is what a floor read off an unre-storable
        //     frame used to cause (pinned by "reclaiming a SHARED delta base
        //     neither inflates the store nor breaks the fixpoint").
        const floor_span = src.frameSpanOf(src_rid);
        const new_rid = switch (dst.putArtifactPacked(digest, content, pri_buf[0..pri_len], floor_span) catch |e| return storagePutError(e)) {
            .rewrote => |rid| rid,
            .kept => try copySrcFrame(gpa, src, dst, src_rid, content, pri_buf[0..pri_len]),
        };
        if (dst.kindOf(new_rid) == .delta) deltas += 1;
    }

    // "KEPT ⇒ PRESENT" — the one direction neither sweep covers, and the gate
    // that makes `PackStats.dropped` self-verifying instead of bookkeeping
    // nobody checks. `packVerifyPacked` answers only "everything present is
    // honest"; `packVerifyShunned` answers only "everything ordered gone is
    // gone". Neither answers "everything NOT ordered gone actually landed" —
    // and the artifact loop has a SECOND exit besides the filter branch (the
    // `orelse continue` on `srcContent`) which omits an artifact WITHOUT
    // incrementing `dropped`. Before the filter existed, pack's contract was
    // "no artifact ever leaves", so any silent omission was a bug by
    // definition; now an omission is indistinguishable from an intended
    // reclaim, and `commit` plus the caller's atomic rename replaces the
    // repository with the truncated generation. A CHECKED error, not an assert:
    // the rename happens in `ReleaseFast` too.
    //
    // Exact and O(1) with no new bookkeeping, because pack interns exactly one
    // `dst` entry per KEPT artifact (`putArtifactPacked`/`putContainer`/
    // `putDeltaContainer`/`putZlib`/`put`, all idempotent over names `src`
    // already holds uniquely) and creates no `dst` phantoms; config and shun
    // frames are never interned, so they do not count.
    if (dst.entryCount() != rids.len - dropped) return error.Corrupt;
    // MANDATORY invariant sweep (§3): reconstruct EVERY packed artifact through
    // its NEW (possibly re-based) delta chain and re-hash it; a mismatch aborts
    // before the commit, so the source is never replaced by a store that fails
    // round-trip. `dst` is a discarded temp on this path.
    try packVerifyPacked(gpa, dst);
    // ...and the reclaim contract's own end-state gate: nothing the filter names
    // survives in `dst`. Disjoint from the sweep above, which only ever answers
    // "everything present is honest" — and a wrongly-copied frame is honest.
    try packVerifyShunned(dst, filter);

    dst.commit() catch return error.Storage; // fsync the new log before the caller renames it
    return .{
        .artifacts = rids.len,
        .deltas = deltas,
        .dropped = dropped,
        .shuns_recorded = shuns_recorded,
        .bytes_before = src.size(),
        .bytes_after = dst.size(),
    };
}

/// The reclaim contract's end-state gate: re-ask `filter` over every entry `dst` actually holds and
/// fail the pack (`error.Corrupt`) if it names one. A rebuild that did not remove
/// what it was told to remove must never be swapped in — fail-open here means
/// `zit pack` reports "reclaimed" while the bytes ship in the next clone.
///
/// The question is asked of the ARTIFACT, not of the bookkeeping: the filter is
/// re-run against `dst`'s own entry names rather than compared to the `dropped`
/// counter or to a recorded drop list, because a verifier answered from the same
/// bookkeeping it is checking verifies nothing. `error.Corrupt`, never a
/// `std.debug.assert`: this rule must hold in a `ReleaseFast` build, where an
/// assert compiles to nothing.
///
/// Every entry, phantom or not — `entryCount` rather than
/// `ridsInPersistedOrder` — because a dropped NAME lingering as an interned
/// phantom would still make `lookup` and `resolveHexPrefix` answer a rid for
/// content the operator was told is gone. O(dst) name comparisons, no allocation.
fn packVerifyShunned(dst: *const logstore.LogStore, filter: Filter) PackError!void {
    const count = dst.entryCount();
    var rid: logstore.Rid = 1;
    while (rid <= count) : (rid += 1) {
        if (filter.drops(dst.hashOf(rid))) return error.Corrupt;
    }
}

/// The §3 post-write invariant sweep: reconstruct every content artifact in the
/// freshly written `dst` through its (possibly re-based) delta chain and confirm
/// it hashes to its own name. A single mismatch is `error.Corrupt`, aborting the
/// pack before its atomic rename — the source store is never replaced by a pack
/// that fails byte-exact round-trip. A shared `PlaintextCache` bounds the
/// re-reconstruction cost over deep chains.
fn packVerifyPacked(gpa: std.mem.Allocator, dst: *logstore.LogStore) PackError!void {
    var cache = logstore.PlaintextCache.init();
    defer cache.deinit(gpa);
    const rids = try dst.ridsInPersistedOrder(gpa);
    defer gpa.free(rids);
    for (rids) |rid| {
        const content = (try srcContent(gpa, dst, rid, &cache)) orelse continue;
        defer gpa.free(content);
        var name_buf: [2 * logstore.max_hash_len]u8 = undefined;
        const name = digestToWireName(dst.hashOf(rid), &name_buf);
        if (!artifact.verify(name, content)) return error.Corrupt;
    }
}

/// `src.readConfig` narrowed to pack's error set (a config frame that will not
/// decode is `Corrupt`, mirroring §7.7's abort-on-corrupt discipline).
fn readSrcConfig(gpa: std.mem.Allocator, src: *logstore.LogStore) PackError!?logstore.OwnedConfig {
    return src.readConfig(gpa) catch |e| switch (e) {
        error.OutOfMemory => error.OutOfMemory,
        error.Storage => error.Storage,
        error.Corrupt => error.Corrupt,
    };
}

/// `contentUnmasked` (the cached shape) narrowed to pack's error set — the
/// MAINTENANCE reader, deliberately, not `contentCached`.
///
/// A repack is not a read: it relocates frames inside one repository and shows
/// nobody the bytes. The repository shun table masks the rid-keyed readers, so a
/// repack that went through `contentCached` would see null for every
/// shunned-but-held frame and — via `packLog`'s `orelse continue` — DROP it,
/// erasing the artifact and its interned name from the new generation. `lookup`
/// and `resolveHexPrefix` would then answer null where they answered a rid, i.e.
/// `zit pack` would change what an unrelated verb prints as a side effect of a
/// read guard.
///
/// Reclamation is a decision the CALLER makes, explicitly, by handing `packLog` a
/// `Filter` that names the artifact — and when it does, the frame is never
/// read at all: the filter check precedes every call to this function, so a
/// dropped artifact reaches neither reader. A shunned frame the filter KEEPS
/// flows through the repack untouched, exactly like every other frame. That
/// separation is what keeps "shunned" (a serving mask, reversible) distinct from
/// "reclaimed" (bytes gone, not reversible), and it is why this reader is
/// unmasked rather than the mask being consulted here.
fn srcContent(gpa: std.mem.Allocator, src: *logstore.LogStore, rid: logstore.Rid, cache: *logstore.PlaintextCache) PackError!?[]u8 {
    return src.contentUnmasked(gpa, rid, cache) catch |e| switch (e) {
        error.OutOfMemory => error.OutOfMemory,
        error.Storage => error.Storage,
        error.Corrupt => error.Corrupt,
    };
}

/// Re-store `src_rid`'s EXISTING frame into `dst` verbatim — the §6.1 "kept"
/// path, taken when no re-encoding beat the source frame's on-disk size. A full
/// zlib container is copied byte-for-byte; a wire delta is copied byte-for-byte
/// and re-pointed at its base's NEW dst frame (same base content, so the delta
/// still reconstructs the same artifact); a raw frame re-stores its plaintext.
/// Preserves fossil's compact base choice unbeaten where pack's re-deltify could
/// not improve it.
///
/// Offset order guarantees a delta's base is already packed UNLESS the reclaim
/// filter dropped it — the one case where "keep the source frame" is not an
/// option at all. That frame is then re-encoded from scratch against `priority`
/// plus the chronological window (a null floor, because there is no keepable
/// span to compare against), never stored as a bare full zlib frame: the
/// caller's `.kept` verdict was reached against a floor read off an unre-storable
/// frame, so it carries no information here. Storing the full form instead would
/// discard a profitable delta and inflate every dependent of a reclaimed base —
/// on the imported-block shape (many delta manifests against one baseline) by
/// several times the whole block, on the one command an operator runs to free
/// disk. Never a dangling delta either way (§3.3).
fn copySrcFrame(
    gpa: std.mem.Allocator,
    src: *logstore.LogStore,
    dst: *logstore.LogStore,
    src_rid: logstore.Rid,
    content: []const u8,
    priority: []const logstore.Rid,
) PackError!logstore.Rid {
    std.debug.assert(src_rid >= 1); // a valid 1-based source log rid
    std.debug.assert(content.len <= logstore.max_payload); // reconstructed artifact bytes
    const digest = src.hashOf(src_rid);
    // `wireUnmasked` for the same reason `srcContent` reads unmasked: a kept
    // frame is COPIED, not served. Through the masked `wire` a shunned-but-held
    // `.zlib`/`.delta` frame would look container-less and be re-stored as a raw
    // frame — silently inflating the packed log and voiding the §6.1
    // monotonic-shrink guarantee for that frame.
    const maybe_wire = src.wireUnmasked(gpa, src_rid) catch |e| switch (e) {
        error.OutOfMemory => return error.OutOfMemory,
        error.Storage => return error.Storage,
        error.Corrupt => return error.Corrupt,
    };
    // A `.raw` frame carries no stored container: re-store the plaintext as raw.
    var w = maybe_wire orelse return dst.put(digest, content) catch |e| return storagePutError(e);
    defer w.deinit(gpa);
    switch (w) {
        .full => |c| return dst.putContainer(digest, c) catch |e| return storagePutError(e),
        .delta => |d| {
            const base_rid = dst.lookup(d.baseHash()) orelse
                return repackUnkeepable(dst, digest, content, priority);
            // Chain-depth safety: a verbatim copy re-links to the base's NEW rid,
            // whose dst depth can EXCEED the source's — pack may re-deltify a
            // source-full base into a delta (depth up to `delta_chain_write_max`),
            // so a kept delta stacked on it lands deeper than it was at the source.
            // Copy only while the base stays below the READ bound so the packed
            // chain still resolves within `max_delta_chain`; a base at/over the
            // bound falls back to full form, exactly as the clone persist path does
            // for a retained foreign chain (§3.3). Without this the post-write
            // invariant sweep (§3) would reject an over-deep chain as corrupt.
            if (dst.depthOf(base_rid) >= logstore.max_delta_chain) {
                return dst.putZlib(digest, content) catch |e| return storagePutError(e);
            }
            return dst.putDeltaContainer(digest, base_rid, d.container) catch |e| return storagePutError(e);
        },
    }
}

/// Store an artifact whose SOURCE frame cannot be re-stored — a delta whose base
/// the reclaim filter dropped — at the best encoding the new generation can
/// offer: `putArtifactPacked` again, now with NO floor (`null`), so the delta it
/// finds against the surviving siblings wins instead of being measured against a
/// frame that no longer has a basis.
///
/// `.kept` is impossible with a null floor — there is nothing to compare
/// against, so `putArtifactPacked` always writes — but the fallback is a real
/// full-form store rather than `unreachable`: this runs in `ReleaseFast` builds
/// where an `unreachable` is undefined behavior, and a dangling delta or a
/// corrupt frame is a far worse outcome than the pre-fix inflation. The assert
/// is what makes that safety net loud: a regression in the floor contract would
/// otherwise degrade SILENTLY to a full-form store on every dependent of a
/// reclaimed base — the exact inflation "reclaiming a SHARED delta base neither
/// inflates the store nor breaks the fixpoint" forbids, and a test that pins
/// frame KINDS under the current encoder would not notice.
fn repackUnkeepable(
    dst: *logstore.LogStore,
    digest: []const u8,
    content: []const u8,
    priority: []const logstore.Rid,
) PackError!logstore.Rid {
    return switch (dst.putArtifactPacked(digest, content, priority, null) catch |e| return storagePutError(e)) {
        .rewrote => |rid| rid,
        .kept => blk: {
            std.debug.assert(false); // a null floor has nothing to keep against
            break :blk dst.putZlib(digest, content) catch |e| storagePutError(e);
        },
    };
}

/// Map a `putArtifact`/`putConfig` failure into pack's error set.
fn storagePutError(e: logstore.PutError) PackError {
    return switch (e) {
        error.OutOfMemory => error.OutOfMemory,
        error.Storage => error.Storage,
    };
}

/// Map a `putShun` failure into pack's error set. Every row re-emitted here came
/// out of `src.shunList()`, so it has already satisfied both of `putShun`'s
/// input rules — `scan` rejects a shun frame whose key is not a complete digest,
/// and `parseShun` caps the reason at `max_shun_reason` — which makes `BadKey`
/// and `ReasonTooLong` reachable only if the source's set disagrees with the
/// rules its own reader enforces. That is source corruption, and it fails the
/// repack rather than quietly dropping a tombstone.
fn storageShunError(e: logstore.ShunError) PackError {
    return switch (e) {
        error.OutOfMemory => error.OutOfMemory,
        error.Storage => error.Storage,
        error.BadKey, error.ReasonTooLong => error.Corrupt,
    };
}

/// Walk the source manifests once, filling `hints` with each artifact's ordered
/// base candidates:
///   * a check-in manifest → its primary parent manifest (§6.1 lineage);
///   * a file blob → its same-path predecessors, nearest first (§6.2, fossil's
///     mlink-biased `N_NEIGHBOR` analog — rebuild.c:501,508).
/// Same-path knowledge is derived HERE, inside pack (the manifest-parsing layer),
/// and handed to `pickBase` as plain `[]Rid`: no path concept leaks into the
/// logstore policy. Every recorded predecessor precedes its blob in offset order
/// (the two-pass offset sort guarantees it), so a chosen base is always already
/// written into `dst` before its dependent. A non-manifest / unparseable artifact
/// contributes no lineage hint. Only memory/storage/corrupt frame faults escape.
///
/// `filter` is honoured on BOTH sides of the hint relation, and the ordering is
/// the same normative one `packLog` states: pass 1 reconstructs every rid just to
/// try parsing it as a manifest, and `srcContent` can itself return
/// `error.Corrupt`, so a dropped artifact must be skipped BEFORE that read — one
/// function earlier than the artifact loop, and the reason a corrupt-but-dropped
/// frame does not abort the pack here either. A dropped artifact is also never
/// OFFERED as a base hint: it will not be in `dst`, so a hint naming it would be
/// dead weight, and `packLog` asserts the mirror of this when it translates hints.
/// Item 3's pairing step over the two flat partition-key maps one lockstep
/// walk (`diffTreeVersions`) filled. For every partition key the child and
/// parent both carry, offer the parent's node as a delta base for the child's.
/// An equal rid — the SAME subtree in both check-ins — pairs nothing, which is
/// exactly why the walk that feeds this never descends into one.
fn pairAgainstParent(
    gpa: std.mem.Allocator,
    src: *logstore.LogStore,
    hints: *std.AutoHashMapUnmanaged(logstore.Rid, PriorityHints),
    child_keys: *const std.StringHashMapUnmanaged(logstore.Rid),
    parent_keys: *const std.StringHashMapUnmanaged(logstore.Rid),
) PackError!void {
    var kit = child_keys.iterator();
    while (kit.next()) |e| {
        const base = parent_keys.get(e.key_ptr.*) orelse continue;
        const child_rid = e.value_ptr.*;
        // "not an interior node delta-based on its own descendant": an equal rid
        // is the same artifact, and a base must sit strictly earlier in the log
        // to be packed first.
        if (base == child_rid) continue;
        if (src.offsetOf(base) >= src.offsetOf(child_rid)) continue;
        try pushHint(gpa, hints, child_rid, base);
    }
}

/// Whether `prefix` -- the first bytes of an artifact -- could begin a
/// structural artifact under [FossilArtifact] §4.4, where a card is a card-type
/// letter followed by zero or more space-separated tokens.
///
/// One-sided by design: false means the artifact provably is not structural,
/// true means only "not ruled out", and the parser still decides. Fewer than
/// two bytes is not enough to rule anything out, so it answers true. Keeping
/// the error on the permissive side is what lets a caller skip the reject case
/// without changing any answer.
fn couldStartStructural(prefix: []const u8) bool {
    if (prefix.len < 2) return true;
    if (prefix[0] < 'A' or prefix[0] > 'Z') return false;
    // The delimiter after a card's type letter: a space before its first token,
    // or the LF of a card that carries none.
    return prefix[1] == ' ' or prefix[1] == '\n';
}

fn deriveCandidateHints(
    gpa: std.mem.Allocator,
    src: *logstore.LogStore,
    rids: []const logstore.Rid,
    filter: Filter,
    hints: *std.AutoHashMapUnmanaged(logstore.Rid, PriorityHints),
) PackError!void {
    // Fresh map in, filled here: the caller hands a just-initialized `.empty`
    // hint map so every entry below is this walk's own.
    std.debug.assert(hints.count() == 0);
    // path → the distinct file-blob rids ever stored at that path, accumulated in
    // manifest-walk order (re-sorted to offset order in pass 2). Keys are gpa-owned
    // dups of the manifest arena's filename bytes; freed with the map below.
    var paths: std.StringHashMapUnmanaged(std.ArrayListUnmanaged(logstore.Rid)) = .empty;
    defer {
        var it = paths.iterator();
        while (it.next()) |e| {
            e.value_ptr.deinit(gpa);
            gpa.free(e.key_ptr.*);
        }
        paths.deinit(gpa);
    }

    // Pass 1: manifest → lineage parent, and accumulate path → blob rids.
    var cache = logstore.PlaintextCache.init();
    // This sweep REVISITS: even after the lockstep diff prunes unchanged
    // subtrees, a sub-manifest that changed is still resolved by every check-in
    // whose walk reaches it, and sub-manifests are stored `.zlib`. Under the
    // default only resolved deltas are memoised, so each of those visits
    // re-inflates the frame. Opt in to memoising undeltaed plaintext too --
    // see the field's own note for why it is off by default.
    cache.memoize_undeltaed = true;
    defer cache.deinit(gpa);

    for (rids) |m_rid| {
        // The filter FIRST — before the reconstruction below, for the reason in
        // this function's doc: a dropped artifact's bytes may be corrupt, and
        // `srcContent` on a corrupt frame is `error.Corrupt`.
        if (filter.drops(src.hashOf(m_rid))) continue;
        // EARLY REJECTION. This loop visits every artifact in the store, and on
        // any real repository the overwhelming majority are file blobs that
        // reach `parseForImport` below only to be discarded -- after being
        // reconstructed in full. Reconstruction is the expensive part and the
        // discard is free, which is the wrong way round: measured on a real
        // history, artifacts stored whole are 23% of the census but 84% of the
        // payload bytes (`zit stat` full-form-bytes), because an artifact is
        // stored whole exactly when delta did not pay -- the large ones.
        //
        // So look before reconstructing. Two bytes decide it: §4.4 requires a
        // structural artifact's cards to appear in non-decreasing card-type
        // order, and a card is a type letter followed by its tokens, so byte 0
        // of ANY structural artifact is a card letter and byte 1 delimits it.
        // A blob is arbitrary bytes and almost never looks like that.
        //
        // Conservative in both directions, which is what keeps output
        // identical: null means "not cheaply knowable" (a `.delta` frame, whose
        // content does not exist until its base is reconstructed) and falls
        // through to the full read, a prefix error falls through too, and the
        // predicate rejects only a definite mismatch. Nothing that would have
        // parsed as a manifest is skipped, so `hints` is unchanged and so is
        // every byte this pack emits.
        if (src.contentPrefixUnmasked(gpa, m_rid, 2) catch null) |prefix| {
            defer gpa.free(prefix);
            if (!couldStartStructural(prefix)) continue;
        }
        const bytes = (try srcContent(gpa, src, m_rid, &cache)) orelse continue;
        defer gpa.free(bytes);
        // Only a check-in manifest carries F/P lineage; a blob (or a control
        // artifact) fails the parse or is a different kind and contributes none.
        // Read path: pack reads stored (fossil-origin) manifests, so it tolerates
        // the legacy F-card trailing separator.
        var loaded = artifact.parseForImport(gpa, bytes) catch |e| switch (e) {
            error.OutOfMemory => return error.OutOfMemory,
            else => continue,
        };
        defer loaded.deinit(gpa);
        const m = &loaded.manifest;
        if (m.kind != .manifest) continue;
        // manifest → primary parent manifest (§6.1). The parent frame precedes
        // this one in offset order, so it is packed before it.
        // A DROPPED parent is not offered: its frame never reaches `dst`, so the
        // hint would be dead weight, and `packLog`'s translation asserts it.
        // Held for item 3 below: a DROPPED parent offers nothing, here or there.
        var parent_for_keys: ?logstore.Rid = null;
        if (primaryParentRid(src, m)) |parent_rid| {
            if (!filter.drops(src.hashOf(parent_rid))) {
                try pushHint(gpa, hints, m_rid, parent_rid);
                parent_for_keys = parent_rid;
            }
        }
        // Items 2 and 3, fused into one LOCKSTEP walk (`diffTreeVersions`):
        // the child's tree is diffed against the primary parent's, descending
        // only where the two differ. One `seen` pair per check-in: sharing a
        // sub-manifest with a SIBLING check-in is the normal case and must
        // still record that check-in's version. Without a usable parent -- a
        // root, or a dropped/phantom/unparseable one -- the child is walked in
        // full, the fallback DESIGN §6 names. The parent's blobs are never
        // re-recorded (`paths = null` on its side): it contributed them on its
        // own pass, and doing it twice would let a revert reappear as a
        // duplicate version.
        var child_keys: std.StringHashMapUnmanaged(logstore.Rid) = .empty;
        defer deinitPartitionKeys(gpa, &child_keys);
        var child_seen: std.AutoHashMapUnmanaged(logstore.Rid, void) = .empty;
        defer child_seen.deinit(gpa);
        var walked_lockstep = false;
        // A child without S nodes fills no partition-key map, so item 3 has
        // nothing to pair -- skip the parent read and parse outright.
        if (m.s.len != 0) {
            if (parent_for_keys) |prid| lockstep: {
                // `pbytes` must outlive the parse: the manifest's card slices
                // borrow it, and freeing it at block exit left every S-card
                // hash pointing into freed memory -- which surfaced as a
                // "corrupt" store on a perfectly healthy tree repository.
                const pbytes = (try srcContent(gpa, src, prid, &cache)) orelse break :lockstep;
                defer gpa.free(pbytes);
                var pl = artifact.parseForImport(gpa, pbytes) catch |e| switch (e) {
                    error.OutOfMemory => return error.OutOfMemory,
                    else => break :lockstep,
                };
                defer pl.deinit(gpa);
                if (pl.manifest.kind != .manifest) break :lockstep;
                // Item 3 pairs this check-in's S nodes with the PRIMARY
                // parent's under an exactly equal canonical partition key.
                // Walked rather than pooled because the design names the
                // parent specifically -- an offset-nearest node from an
                // unrelated branch is the heuristic §6 forbids.
                var parent_keys: std.StringHashMapUnmanaged(logstore.Rid) = .empty;
                defer deinitPartitionKeys(gpa, &parent_keys);
                var parent_seen: std.AutoHashMapUnmanaged(logstore.Rid, void) = .empty;
                defer parent_seen.deinit(gpa);
                try diffTreeVersions(gpa, src, &cache, filter, &paths, &child_keys, &child_seen, &parent_keys, &parent_seen, m.f, m.s, pl.manifest.s, 0);
                try pairAgainstParent(gpa, src, hints, &child_keys, &parent_keys);
                walked_lockstep = true;
            }
        }
        if (!walked_lockstep) {
            try recordTreeVersions(gpa, src, &cache, filter, &paths, &child_keys, &child_seen, m.f, m.s, 0);
        }
    }

    // Pass 2: per path, sort its versions by offset and give each blob its nearest
    // same-path predecessors (offset below), nearest first — ahead of whatever the
    // chronological window later adds.
    var vit = paths.valueIterator();
    while (vit.next()) |list| {
        std.mem.sort(logstore.Rid, list.items, src, blobOffsetLessThan);
        // A path that reverts to a byte-identical earlier blob (A→B→A) records the
        // SAME content-addressed rid more than once, non-consecutively —
        // `recordPathBlob` dedups only the consecutive run, so a revert slips a
        // duplicate past it. The offset sort makes those duplicates adjacent (an
        // identical rid has an identical offset), so one linear pass compacts
        // `versions` to its DISTINCT rids. Without it the predecessor walk below
        // would offer a blob as its OWN same-path base: equal offsets fail the
        // strict offset-below invariant (:362) and `pushHint`'s `rid != base`
        // assert (:379) — a blob is never its own predecessor.
        const versions = dedupSortedRids(list.items);
        for (versions, 0..) |blob_rid, k| {
            // Nearest predecessor first: k-1, k-2, … (all offset-below `blob_rid`).
            // Bounded to `max_base_candidates`: only that many hints are ever kept
            // (`PriorityHints.push` no-ops past the buffer), so a deep same-path
            // history costs O(N·5), never O(N²).
            var j: usize = k;
            var taken: usize = 0;
            while (j > 0 and taken < logstore.max_base_candidates) {
                j -= 1;
                taken += 1;
                // Load-bearing invariant: an offset-sorted `versions` puts every
                // predecessor (index below `k`) strictly earlier in the log than
                // `blob_rid`, so a chosen base is always already packed into `dst`
                // before its dependent. Assert it at the point of use.
                std.debug.assert(src.offsetOf(versions[j]) < src.offsetOf(blob_rid));
                try pushHint(gpa, hints, blob_rid, versions[j]);
            }
        }
    }
}

/// Append `base` to `rid`'s ordered hint list (creating it on first use). The
/// list is bounded and self-deduping (`PriorityHints.push`), so this is a no-op
/// once the buffer is full — the loops that call it need no separate cap.
fn pushHint(
    gpa: std.mem.Allocator,
    hints: *std.AutoHashMapUnmanaged(logstore.Rid, PriorityHints),
    rid: logstore.Rid,
    base: logstore.Rid,
) PackError!void {
    std.debug.assert(rid >= 1 and base >= 1); // both are valid 1-based log rids
    std.debug.assert(rid != base); // a frame is never its own base hint
    const gop = hints.getOrPut(gpa, rid) catch return error.OutOfMemory;
    if (!gop.found_existing) gop.value_ptr.* = .{};
    gop.value_ptr.push(base);
}

/// Record `blob_rid` under `path` in `paths`, owning a dup of the filename bytes
/// on first sight of a path. A file unchanged across consecutive check-ins
/// references the same blob; that run is deduped here (only a change from the
/// last-recorded version appends), and `PriorityHints.push` dedups any residual
/// repeat. Only OOM escapes.
/// Walk one check-in's complete file enumeration -- F leaves and S children to
/// the full tree height (submanifests §5.1) -- recording two different things for
/// two different hint lanes.
///
/// Blobs go into the shared `paths` map, because DESIGN §6 item 2 wants "earlier
/// same-path versions nearest-first": that is a property of the whole log, not of
/// one lineage, and pass 2's offset sort already computes it.
///
/// S nodes go into `keys`, a map private to THIS check-in, because item 3 is a
/// different rule: a child's S node is based on its PRIMARY PARENT's node under
/// the same canonical partition key. Cross-lineage nearest-offset would be the
/// "nearest-key or overlapping-range heuristic" the design forbids, so the caller
/// pairs two of these maps rather than pooling them.
///
/// Reading `m.f` alone was the bug: an S-bearing check-in keeps its files in
/// sub-manifests, so a tree-shaped repository got no same-path hints at all.
///
/// Recursion is bounded by `artifact.max_tree_depth` (16), the §5.3 bound the
/// spec's own resolver enforces, so depth is a constant rather than
/// repository-controlled. `seen` collapses a diamond and rejects a cycle.
///
/// A referenced S node that will not resolve or parse is CORRUPTION and aborts
/// (design §6: "A missing or corrupt referenced S node is corruption and
/// aborts"). Only a *valid* absence -- a flat tree, an unmatched key, a history
/// root -- is silent and falls back to the chronological window.
fn recordTreeVersions(
    gpa: std.mem.Allocator,
    src: *logstore.LogStore,
    cache: *logstore.PlaintextCache,
    filter: Filter,
    paths: ?*std.StringHashMapUnmanaged(std.ArrayListUnmanaged(logstore.Rid)),
    keys: *std.StringHashMapUnmanaged(logstore.Rid),
    seen: *std.AutoHashMapUnmanaged(logstore.Rid, void),
    files: []const artifact.FCard,
    subs: []const artifact.SCard,
    depth: usize,
) PackError!void {
    std.debug.assert(depth <= artifact.max_tree_depth);

    try recordFileBlobs(gpa, src, filter, paths, files);

    for (subs) |sc| {
        var digest_buf: [digest_len]u8 = undefined;
        const digest = hexToDigest(sc.hash, &digest_buf) orelse return error.Corrupt;
        const sub_rid = src.lookup(digest) orelse return error.Corrupt;
        if (src.isPhantom(sub_rid)) return error.Corrupt;
        // A dropped node is excluded rather than fatal: the filter removing it is
        // a decision this pack already made, not damage in the source.
        if (filter.drops(src.hashOf(sub_rid))) continue;
        if ((try seen.getOrPut(gpa, sub_rid)).found_existing) continue;

        // The S card's filename IS the canonical partition key (§3.6). Exact match
        // only -- the design forbids any nearest-key heuristic. First writer wins
        // so a duplicate key inside one tree cannot silently retarget the pair.
        _ = try putPartitionKey(gpa, keys, sc.filename, sub_rid);

        if (depth == artifact.max_tree_depth) return error.Corrupt; // §5.3 bound
        var ls = try loadSubManifest(gpa, src, cache, sub_rid);
        // The recursion borrows `ls`'s card slices, which this defer keeps
        // alive for exactly this scope (`bytes` backs them; freed last).
        defer ls.deinit(gpa);
        try recordTreeVersions(gpa, src, cache, filter, paths, keys, seen, ls.loaded.manifest.f, ls.loaded.manifest.s, depth + 1);
    }
}

/// Item 2's leaf rule, shared by every walk that visits a tree level: each
/// held, kept, non-phantom F-card blob is recorded under its path. A `null`
/// map is the parent-side walk, which records nothing.
fn recordFileBlobs(
    gpa: std.mem.Allocator,
    src: *logstore.LogStore,
    filter: Filter,
    paths: ?*std.StringHashMapUnmanaged(std.ArrayListUnmanaged(logstore.Rid)),
    files: []const artifact.FCard,
) PackError!void {
    const pmap = paths orelse return;
    for (files) |fc| {
        const bh = fc.hash orelse continue;
        var digest_buf: [digest_len]u8 = undefined;
        const digest = hexToDigest(bh, &digest_buf) orelse continue;
        const blob_rid = src.lookup(digest) orelse continue;
        if (src.isPhantom(blob_rid)) continue;
        // A dropped blob is neither a hint's SUBJECT (never packed, so it
        // needs no base) nor a candidate base for its same-path successors.
        if (filter.drops(src.hashOf(blob_rid))) continue;
        try recordPathBlob(gpa, pmap, fc.filename, blob_rid);
    }
}

/// First-writer-wins insert of `key → rid` into a partition-key map, OWNING a
/// dup of the key bytes. The caller's `key` borrows the manifest bytes of the
/// node being walked, and for every node below depth 0 those are the
/// `bytes`/`loaded` pair the walking frame frees on return -- so a borrowed
/// key would outlive its backing store by exactly one recursion level. The
/// map is read later (iterated after the walk) and grown later (a rehash
/// re-reads every key), so the borrow is a use-after-free on both paths; it
/// stayed latent only while every walked tree was one level deep. Mirrors
/// `recordPathBlob`: on a dupe OOM, drop the half-inserted slot so teardown
/// never frees borrowed bytes. Returns whether this call was the first writer
/// for `key`.
fn putPartitionKey(
    gpa: std.mem.Allocator,
    keys: *std.StringHashMapUnmanaged(logstore.Rid),
    key: []const u8,
    rid: logstore.Rid,
) PackError!bool {
    const gop = keys.getOrPut(gpa, key) catch return error.OutOfMemory;
    if (gop.found_existing) return false;
    const owned = gpa.dupe(u8, key) catch {
        // The removal is the OPERATION, so it is not an assert's argument:
        // under `ReleaseFast` the assert vanishes and would take the cleanup
        // with it, leaving a borrowed key in a map whose teardown frees every
        // key it holds.
        const removed = keys.remove(key);
        std.debug.assert(removed);
        return error.OutOfMemory;
    };
    gop.key_ptr.* = owned;
    gop.value_ptr.* = rid;
    return true;
}

/// One resolved, validated sub-manifest node. `bytes` backs every card slice
/// in `loaded`, so both live and die together -- `deinit` releases the parse
/// first and the backing bytes after it.
const LoadedSub = struct {
    bytes: []u8,
    loaded: artifact.ImportedManifest,

    fn deinit(self: *LoadedSub, gpa: std.mem.Allocator) void {
        self.loaded.deinit(gpa);
        gpa.free(self.bytes);
        self.* = undefined;
    }
};

/// Read and parse the sub-manifest node `rid` under the §6 corruption rule
/// every tree walk shares: a referenced S node that will not reconstruct,
/// parse, or claim the `.submanifest` kind is CORRUPTION and aborts.
fn loadSubManifest(
    gpa: std.mem.Allocator,
    src: *logstore.LogStore,
    cache: *logstore.PlaintextCache,
    rid: logstore.Rid,
) PackError!LoadedSub {
    const bytes = (try srcContent(gpa, src, rid, cache)) orelse return error.Corrupt;
    errdefer gpa.free(bytes);
    var loaded = artifact.parseForImport(gpa, bytes) catch |e| switch (e) {
        error.OutOfMemory => return error.OutOfMemory,
        else => return error.Corrupt,
    };
    if (loaded.manifest.kind != .submanifest) {
        loaded.deinit(gpa);
        return error.Corrupt;
    }
    return .{ .bytes = bytes, .loaded = loaded };
}

/// Items 2 and 3 in ONE pass: walk the child's tree LOCKSTEP against the
/// primary parent's, descending only where the two trees DIFFER.
///
/// The prune rule: an S card whose canonical partition key AND resolved node
/// rid both match the parent's names the SAME subtree in both check-ins, and
/// nothing below it can contribute. Item 3 pairs only DIFFERING same-key
/// nodes (`pairAgainstParent` skips an equal rid), and every blob under a
/// shared subtree was recorded into `paths` by the check-in that INTRODUCED
/// the subtree -- which walked it as a changed one, recursively back to a
/// full root/fallback walk. Pass 2 consumes `paths` as an offset-sorted
/// DISTINCT set per path (`dedupSortedRids`), so re-recording a shared
/// subtree's blobs once per referencing check-in added work, never output.
/// Both flat maps still receive a pruned node's own key, so
/// `pairAgainstParent` over the visited keys yields exactly the pairs the two
/// full walks yielded. This is what turns pass 1 from O(check-ins x tree
/// size) into O(check-ins x changed paths) -- the difference between a pack
/// that finishes and one that does not on a deep imported history (#27).
///
/// Corruption detection is preserved, not relaxed. Every DISTINCT
/// sub-manifest referenced by a kept check-in is still resolved and parsed at
/// least once during pass 1: at the first kept, parseable check-in that
/// references it, its key either has no parent counterpart or a differing one
/// (both walk it), or the same-rid parent -- which is then an earlier such
/// check-in, and offsets strictly decrease until a root or fallback full
/// walk. `deriveCandidateHints` completes before `packLog` writes a single
/// frame, so a malformed, phantom, or unresolvable node aborts the pack with
/// `error.Corrupt` before anything destructive, exactly as the full walks
/// did. The §5.3 depth bound and the `seen` cycle/diamond guards apply to
/// both sides; termination follows from the bounded depth and finite card
/// lists.
fn diffTreeVersions(
    gpa: std.mem.Allocator,
    src: *logstore.LogStore,
    cache: *logstore.PlaintextCache,
    filter: Filter,
    paths: *std.StringHashMapUnmanaged(std.ArrayListUnmanaged(logstore.Rid)),
    child_keys: *std.StringHashMapUnmanaged(logstore.Rid),
    child_seen: *std.AutoHashMapUnmanaged(logstore.Rid, void),
    parent_keys: *std.StringHashMapUnmanaged(logstore.Rid),
    parent_seen: *std.AutoHashMapUnmanaged(logstore.Rid, void),
    c_files: []const artifact.FCard,
    c_subs: []const artifact.SCard,
    p_subs: []const artifact.SCard,
    depth: usize,
) PackError!void {
    std.debug.assert(depth <= artifact.max_tree_depth);

    // The child's F leaves at this (visited) level. The parent's F leaves are
    // never recorded, exactly as the old parent-side walk passed `paths = null`.
    try recordFileBlobs(gpa, src, filter, paths, c_files);

    // Index the parent's S cards at this level by canonical partition key --
    // first writer wins, mirroring the flat maps' own insert rule. Keys
    // borrow the parent node's card bytes, which the caller keeps alive for
    // exactly this frame. `matched` marks cards consumed by the child loop so
    // the sweep below descends only the parent-only remainder.
    var pidx: std.StringHashMapUnmanaged(usize) = .empty;
    defer pidx.deinit(gpa);
    for (p_subs, 0..) |psc, i| {
        const gop = pidx.getOrPut(gpa, psc.filename) catch return error.OutOfMemory;
        if (!gop.found_existing) gop.value_ptr.* = i;
    }
    const matched = gpa.alloc(bool, p_subs.len) catch return error.OutOfMemory;
    defer gpa.free(matched);
    @memset(matched, false);

    for (c_subs) |sc| {
        // The child-side checks, in `recordTreeVersions`'s exact order.
        var digest_buf: [digest_len]u8 = undefined;
        const digest = hexToDigest(sc.hash, &digest_buf) orelse return error.Corrupt;
        const sub_rid = src.lookup(digest) orelse return error.Corrupt;
        if (src.isPhantom(sub_rid)) return error.Corrupt;
        // A dropped node is excluded rather than fatal: the filter removing it
        // is a decision this pack already made, not damage in the source.
        if (filter.drops(src.hashOf(sub_rid))) continue;
        if ((try child_seen.getOrPut(gpa, sub_rid)).found_existing) continue;
        const first_key_writer = try putPartitionKey(gpa, child_keys, sc.filename, sub_rid);

        // The parent's same-key node at this level, resolved with the same
        // strictness the old parent-side walk applied. A duplicate child key
        // gets no counterpart -- the flat map gave a second writer no pairing
        // entry either -- and falls through to the full descent.
        var parent_rid: ?logstore.Rid = null;
        if (first_key_writer) {
            if (pidx.get(sc.filename)) |pi| {
                if (!matched[pi]) {
                    var pdigest_buf: [digest_len]u8 = undefined;
                    const pdigest = hexToDigest(p_subs[pi].hash, &pdigest_buf) orelse return error.Corrupt;
                    const prid = src.lookup(pdigest) orelse return error.Corrupt;
                    if (src.isPhantom(prid)) return error.Corrupt;
                    if (!filter.drops(src.hashOf(prid))) {
                        matched[pi] = true;
                        if (!(try parent_seen.getOrPut(gpa, prid)).found_existing) {
                            _ = try putPartitionKey(gpa, parent_keys, p_subs[pi].filename, prid);
                            parent_rid = prid;
                        }
                    }
                }
            }
        }

        if (parent_rid) |prid| {
            // THE PRUNE: same key, same artifact -- the subtree below is
            // byte-shared with the parent, so nothing under it can pair or
            // add a version. Both maps already carry the node's own key.
            if (prid == sub_rid) continue;
            // Same key, different node: a changed subtree. Descend both
            // sides together.
            if (depth == artifact.max_tree_depth) return error.Corrupt; // §5.3 bound
            var cls = try loadSubManifest(gpa, src, cache, sub_rid);
            defer cls.deinit(gpa);
            var pls = try loadSubManifest(gpa, src, cache, prid);
            defer pls.deinit(gpa);
            try diffTreeVersions(gpa, src, cache, filter, paths, child_keys, child_seen, parent_keys, parent_seen, cls.loaded.manifest.f, cls.loaded.manifest.s, pls.loaded.manifest.s, depth + 1);
        } else {
            // No same-key counterpart: a new, renamed, or split subtree, or a
            // dropped/diamond-collapsed parent side. Walk the child side
            // alone, in full -- this node's own checks and inserts already
            // ran, so recurse with its cards directly.
            if (depth == artifact.max_tree_depth) return error.Corrupt; // §5.3 bound
            var cls = try loadSubManifest(gpa, src, cache, sub_rid);
            defer cls.deinit(gpa);
            try recordTreeVersions(gpa, src, cache, filter, paths, child_keys, child_seen, cls.loaded.manifest.f, cls.loaded.manifest.s, depth + 1);
        }
    }

    // Parent-only cards: subtrees the child dropped, renamed, or merged away
    // at this level. The full parent walk visited them, and their keys can
    // still pair a child key found ELSEWHERE in the tree -- a subtree that
    // moved between levels -- so they keep the full parent-side descent.
    for (p_subs, 0..) |psc, i| {
        if (matched[i]) continue;
        var digest_buf: [digest_len]u8 = undefined;
        const digest = hexToDigest(psc.hash, &digest_buf) orelse return error.Corrupt;
        const prid = src.lookup(digest) orelse return error.Corrupt;
        if (src.isPhantom(prid)) return error.Corrupt;
        if (filter.drops(src.hashOf(prid))) continue;
        if ((try parent_seen.getOrPut(gpa, prid)).found_existing) continue;
        _ = try putPartitionKey(gpa, parent_keys, psc.filename, prid);
        if (depth == artifact.max_tree_depth) return error.Corrupt; // §5.3 bound
        var pls = try loadSubManifest(gpa, src, cache, prid);
        defer pls.deinit(gpa);
        try recordTreeVersions(gpa, src, cache, filter, null, parent_keys, parent_seen, pls.loaded.manifest.f, pls.loaded.manifest.s, depth + 1);
    }
}

/// Tear down a canonical-partition-key map. `recordTreeVersions` OWNS every key
/// it inserts, so releasing the map means releasing them too.
fn deinitPartitionKeys(gpa: std.mem.Allocator, keys: *std.StringHashMapUnmanaged(logstore.Rid)) void {
    var it = keys.keyIterator();
    while (it.next()) |key| gpa.free(key.*);
    keys.deinit(gpa);
}

fn recordPathBlob(
    gpa: std.mem.Allocator,
    paths: *std.StringHashMapUnmanaged(std.ArrayListUnmanaged(logstore.Rid)),
    path: []const u8,
    blob_rid: logstore.Rid,
) PackError!void {
    std.debug.assert(path.len > 0); // an F-card filename is never empty
    std.debug.assert(blob_rid >= 1); // a valid 1-based log rid
    const gop = paths.getOrPut(gpa, path) catch return error.OutOfMemory;
    if (!gop.found_existing) {
        // The slot's key is still the caller's borrowed `path`. Own it before any
        // later free can touch it; on a dupe OOM, drop the half-inserted slot so
        // the map's teardown never frees borrowed bytes or reads an undefined list.
        const owned = gpa.dupe(u8, path) catch {
            // Removal first, assert second: the same side-effect-in-an-assert
            // hazard `recordTreeVersions` documents.
            const removed = paths.remove(path);
            std.debug.assert(removed);
            return error.OutOfMemory;
        };
        gop.key_ptr.* = owned;
        gop.value_ptr.* = .empty;
    }
    const list = gop.value_ptr;
    if (list.items.len > 0 and list.items[list.items.len - 1] == blob_rid) return;
    list.append(gpa, blob_rid) catch return error.OutOfMemory;
}

/// Compact an offset-sorted rid slice to its DISTINCT rids IN PLACE, returning
/// the leading distinct prefix. A same-path history that reverts to a
/// byte-identical earlier blob (A→B→A) records one content-addressed rid more
/// than once; the offset sort makes those duplicates adjacent (an identical rid
/// has an identical offset), so a single linear pass drops them. Keeps the pass-2
/// predecessor walk from ever offering a blob as its own same-path base — a blob
/// is never its own predecessor.
fn dedupSortedRids(sorted: []logstore.Rid) []logstore.Rid {
    if (sorted.len == 0) return sorted;
    var w: usize = 1; // count of distinct rids kept in the leading prefix
    var i: usize = 1;
    while (i < sorted.len) : (i += 1) {
        std.debug.assert(sorted[i] >= 1); // a valid 1-based log rid
        if (sorted[i] != sorted[w - 1]) {
            sorted[w] = sorted[i];
            w += 1;
        }
    }
    return sorted[0..w];
}

/// Offset order over two non-phantom blob rids — pass 2's sort key. Ascending
/// offset means earlier-written first, so a blob's predecessors (positions below
/// it) are exactly the same-path versions already packed when it is reached.
fn blobOffsetLessThan(src: *logstore.LogStore, a: logstore.Rid, b: logstore.Rid) bool {
    // Contract: two non-phantom blob rids. `offsetOf` re-asserts range and
    // non-phantom internally; state the boundary contract here explicitly.
    std.debug.assert(a >= 1 and b >= 1);
    std.debug.assert(!src.isPhantom(a) and !src.isPhantom(b));
    return src.offsetOf(a) < src.offsetOf(b);
}

/// The log rid of a manifest's primary (first) parent, or null when it is a
/// root or the parent is not held. `parents` is the space-separated P-card
/// argument (artifact §4.7.13); the first token is the merge-primary parent.
fn primaryParentRid(src: *logstore.LogStore, m: *const artifact.Manifest) ?logstore.Rid {
    if (m.p.len == 0) return null;
    var it = m.p[0].hashes();
    const first = it.next() orelse return null;
    var digest_buf: [digest_len]u8 = undefined;
    const digest = hexToDigest(first, &digest_buf) orelse return null;
    return src.lookup(digest);
}

/// Decodes an artifact name into its binary digest — the log-frame key — writing
/// into `out` and returning the used prefix. Returns null for anything that is
/// not a name this suite mints. Byte-identical to main.zig's `hexToDigest`; pack
/// keeps its own copy so it needs no import above the store layer, and the two
/// must be changed together.
///
/// Both widths of [FossilArtifact] §7.2 are admitted — 20 bytes (sha1) and 32
/// (sha3-256) — because a repository may hold either; `zit import-git` names
/// every check-in under git's sha1. Requiring `digest_len * 2` here would make
/// `pack` silently drop the primary-parent hint on every imported repository,
/// which is the same assumption that broke `commit` in issue #24.
fn hexToDigest(hex: []const u8, out: *[digest_len]u8) ?[]const u8 {
    if (hex.len != 40 and hex.len != digest_len * 2) return null;
    const n = hex.len / 2;
    _ = std.fmt.hexToBytes(out[0..n], hex) catch return null;
    return out[0..n];
}

/// Lowercase-hex-encodes a raw binary `digest` into `out`, returning the used
/// prefix — the log-frame key's wire NAME (sync §3.3). Byte-identical to
/// main.zig's `digestToWireName`.
fn digestToWireName(digest: []const u8, out: *[2 * digest_len]u8) []const u8 {
    std.debug.assert(digest.len <= digest_len);
    const hexdigits = "0123456789abcdef";
    for (digest, 0..) |b, i| {
        out[2 * i] = hexdigits[b >> 4];
        out[2 * i + 1] = hexdigits[b & 0x0f];
    }
    return out[0 .. 2 * digest.len];
}

// ---------------------------------------------------------------------------
// Tests. The broad pack-transform coverage (content equality, re-deltification,
// config survival) lives with the CLI wiring in cmd/sync.zig; what belongs HERE
// is the property that has no other home — that pack preserves the frame kinds
// it does not rewrite.
// ---------------------------------------------------------------------------

const storage = @import("storage.zig");

test "pack: the repository shun table survives a repack" {
    // Shun rows are never interned, so they are in neither `readSrcConfig` nor
    // `ridsInPersistedOrder` — the two sources `packLog` rebuilds `dst` from.
    // A pack that did not re-emit them would silently DROP the entire durable
    // set: `zit pack` writes a new generation and swaps it in, so the
    // tombstones would be gone for good while the artifacts they name may
    // survive the repack. Fail-open on exactly the predicate the set exists to
    // make durable, with no error and no warning.
    const gpa = std.testing.allocator;
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    const io = std.testing.io;

    const first = "s" ** 32;
    const second = "t" ** 20; // the other complete-name width

    const sfile = try tmp.dir.createFile(io, "src", .{ .read = true });
    defer sfile.close(io);
    var sfs = storage.FileStorage.init(io, sfile, 0);
    var src = try logstore.LogStore.open(gpa, sfs.storageFile(), 0);
    defer src.deinit(gpa);
    // A real content-addressed artifact: `packLog` §7.7-verifies every source
    // frame, so the store must hold a genuinely named one for the shun rows to
    // be preserved ALONGSIDE content rather than in a degenerate empty pack.
    const held = "held content that the pack will reconstruct and re-hash";
    var held_digest: [digest_len]u8 = undefined;
    std.crypto.hash.sha3.Sha3_256.hash(held, &held_digest, .{});
    _ = try src.putArtifact(&held_digest, held, null);
    try src.putShun(first, "takedown notice 4711", 1753000000);
    try src.putShun(second, "", 1753000001);
    try src.commit();

    const dfile = try tmp.dir.createFile(io, "packed", .{ .read = true, .truncate = true });
    defer dfile.close(io);
    var dfs = storage.FileStorage.init(io, dfile, 0);
    var dst = try logstore.LogStore.open(gpa, dfs.storageFile(), src.hash_kind);
    defer dst.deinit(gpa);

    _ = try packLog(gpa, &src, &dst, Filter.keep_all, &.{});

    // Present in the packed generation, in persisted order, fields intact.
    try std.testing.expect(dst.isShunned(first));
    try std.testing.expect(dst.isShunned(second));
    const list = dst.shunList();
    try std.testing.expectEqual(@as(usize, 2), list.len);
    try std.testing.expectEqualStrings(first, list[0].hash);
    try std.testing.expectEqualStrings("takedown notice 4711", list[0].reason);
    try std.testing.expectEqual(@as(i64, 1753000000), list[0].epoch);
    try std.testing.expectEqualStrings(second, list[1].hash);
    try std.testing.expectEqual(@as(i64, 1753000001), list[1].epoch);

    // ...and still there once the packed file is reopened from its bytes.
    var rfs = storage.FileStorage.init(io, dfile, dst.size());
    var reopened = try logstore.LogStore.open(gpa, rfs.storageFile(), 0);
    defer reopened.deinit(gpa);
    try std.testing.expectEqual(@as(usize, 2), reopened.shunList().len);
    try std.testing.expect(reopened.isShunned(first));

    // A tombstone is still not content: it takes no rid in the packed log.
    try std.testing.expect(reopened.lookup(first) == null);
}

test "pack: under keep_all a shunned artifact the store still HOLDS survives the repack" {
    // THE REGRESSION LOCK for the identity filter. Reclamation is now a real
    // outcome (`shunFilter`), so this test's job has changed: it pins that a
    // repack asked to drop NOTHING still relocates a shunned-but-held frame
    // exactly as it did before the filter existed. `zit pack`'s own reclaiming
    // behavior is pinned by the `shunFilter` tests below; a store rewrite that
    // is not a reclaim (a clone persist, a fuzz/DST repack) must keep behaving
    // like this one, and `keep_all` is what makes every new branch dead.
    //
    // The sibling test above shuns two names the store does not hold, so it
    // says nothing about the case that actually has a frame to lose. This one
    // does, and it is the case where the read mask nearly turned `zit pack`
    // into a reclaimer by accident.
    //
    // `ridsInPersistedOrder` filters phantoms, and a shunned artifact is NOT a
    // phantom (its frame exists) — so a shunned-but-held rid is in the list
    // `packLog` walks. Had that walk read through the masked `contentCached`,
    // the `orelse continue` would have skipped it and neither its frame nor its
    // interned NAME would reach the packed generation: `lookup` and
    // `resolveHexPrefix` would start answering null where they answered a rid,
    // i.e. `zit pack` would change what an unrelated verb prints. That is the
    // invariant the mask exists to protect, inverted. Dropping shunned content
    // IS a decision a caller makes explicitly, by handing `packLog` a `Filter`
    // that names it; it is not one a read guard gets to make silently, so
    // pack reads unmasked and this pins it under `keep_all`.
    const gpa = std.testing.allocator;
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    const io = std.testing.io;

    const sfile = try tmp.dir.createFile(io, "src", .{ .read = true });
    defer sfile.close(io);
    var sfs = storage.FileStorage.init(io, sfile, 0);
    var src = try logstore.LogStore.open(gpa, sfs.storageFile(), 0);
    defer src.deinit(gpa);

    // Two artifacts: one shunned, one not, so a pack that dropped EVERYTHING
    // could not pass by accident. Both long enough that `putArtifact` stores a
    // zlib container rather than a raw frame — the masked `wire` would answer
    // null for the shunned one and `copySrcFrame` would silently downgrade it.
    const shunned_body = "content an operator was ordered to stop serving, still physically present in the log " ** 4;
    const kept_body = "ordinary content with no tombstone against it, packed the same way it always was " ** 4;
    var shunned_digest: [digest_len]u8 = undefined;
    var kept_digest: [digest_len]u8 = undefined;
    std.crypto.hash.sha3.Sha3_256.hash(shunned_body, &shunned_digest, .{});
    std.crypto.hash.sha3.Sha3_256.hash(kept_body, &kept_digest, .{});
    const shunned_rid = try src.putArtifact(&shunned_digest, shunned_body, null);
    _ = try src.putArtifact(&kept_digest, kept_body, null);
    try src.putShun(&shunned_digest, "takedown notice 4712", 1753000002);
    try src.commit();

    // The premise: the frame IS there and the mask IS on, or the assertions
    // below would hold for an uninteresting reason.
    try std.testing.expect(!src.isPhantom(shunned_rid));
    try std.testing.expectEqual(@as(?[]u8, null), try src.content(gpa, shunned_rid));

    const dfile = try tmp.dir.createFile(io, "packed", .{ .read = true, .truncate = true });
    defer dfile.close(io);
    var dfs = storage.FileStorage.init(io, dfile, 0);
    var dst = try logstore.LogStore.open(gpa, dfs.storageFile(), src.hash_kind);
    defer dst.deinit(gpa);

    const stats = try packLog(gpa, &src, &dst, Filter.keep_all, &.{});

    var rfs = storage.FileStorage.init(io, dfile, dst.size());
    var reopened = try logstore.LogStore.open(gpa, rfs.storageFile(), 0);
    defer reopened.deinit(gpa);

    // The count pack REPORTS and the count it WROTE are the same number.
    // `stats.artifacts` alone cannot say that: it is `rids.len`, captured from
    // the SOURCE before the loop runs, so it stays 2 even if the loop dropped
    // both. Read the written count back out of the packed generation and compare
    // the two.
    const packed_rids = try reopened.ridsInPersistedOrder(gpa);
    defer gpa.free(packed_rids);
    try std.testing.expectEqual(@as(usize, 2), packed_rids.len);
    try std.testing.expectEqual(packed_rids.len, stats.artifacts);

    // THE PIN: the name still resolves in the packed generation...
    const packed_rid = reopened.lookup(&shunned_digest) orelse
        return error.ShunnedArtifactDroppedByPack;
    try std.testing.expect(!reopened.isPhantom(packed_rid));
    // ...as a CONTAINER, not a raw frame. The bodies above are long enough that
    // `putArtifact` stored zlib; had `copySrcFrame` read the masked `wire` it
    // would have seen a container-less frame and re-stored the plaintext raw —
    // a silent inflation that voids §6.1 monotonic-shrink for this frame and
    // that every other assertion in this test survives.
    try std.testing.expectEqual(logstore.FrameKind.zlib, reopened.kindOf(packed_rid));
    // ...the tombstone travelled with it, so the artifact still reads as
    // absent (reclamation stays invisible — pack changed no verb's answer)...
    try std.testing.expect(reopened.isShunned(&shunned_digest));
    try std.testing.expectEqual(@as(?[]u8, null), try reopened.content(gpa, packed_rid));
    // ...and the BYTES are genuinely in the packed file, not merely a name: the
    // maintenance reader reconstructs them and they still hash to their name.
    const carried = (try reopened.contentUnmasked(gpa, packed_rid, null)).?;
    defer gpa.free(carried);
    try std.testing.expectEqualStrings(shunned_body, carried);
    // The unshunned neighbour is untouched, through the ordinary reader.
    const kept_rid = reopened.lookup(&kept_digest).?;
    const kept_out = (try reopened.content(gpa, kept_rid)).?;
    defer gpa.free(kept_out);
    try std.testing.expectEqualStrings(kept_body, kept_out);
}

test "pack: under keep_all a delta chain ROOTED ON a shunned base survives the repack" {
    // The other half of the identity-filter regression lock: with nothing
    // dropped, a delta whose basis is shunned still relocates as a delta. The
    // reclaiming counterpart — what happens when that basis IS dropped — is
    // "pack: shunFilter dropping a delta BASE leaves every dependent readable"
    // below; the two must not be conflated, and neither substitutes for the
    // other.
    //
    // The sibling test above packs two INDEPENDENT artifacts, so it never
    // exercises the interaction this one is about: what a repack does to a
    // delta whose basis is shunned. Three mechanisms have to cooperate, and
    // none of them is exercised anywhere else.
    //
    //   1. `packLog` copies the shun rows into `dst` BEFORE the artifact loop,
    //      so `dst.putArtifactPacked` → `pickBase` — which probes candidates
    //      through the masked `dst.content` — cannot pick the shunned base for
    //      a FRESH delta. It reads null and is skipped like a phantom.
    //   2. Because no re-encoding beat the source's compact delta frame, the
    //      §6.1 monotonic-shrink guard answers `.kept` and `copySrcFrame` copies
    //      the delta container verbatim, RELINKING it through the unmasked
    //      `dst.lookup(d.baseHash())` — which resolves, because `lookup` is not
    //      masked and the base's frame was packed.
    //   3. The packed delta then reconstructs through `contentAtOffset`, which
    //      reaches its basis by BYTE OFFSET and so never re-enters the rid-keyed
    //      mask — sync §7.7 L1444-1450's "content still held locally MAY serve
    //      as the delta basis", holding across a whole-store rewrite.
    //
    // Break any one of the three and this test fails; the isolated `pickBase`
    // pin in logstore_test.zig catches none of them.
    const gpa = std.testing.allocator;
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    const io = std.testing.io;

    const sfile = try tmp.dir.createFile(io, "src", .{ .read = true });
    defer sfile.close(io);
    var sfs = storage.FileStorage.init(io, sfile, 0);
    var src = try logstore.LogStore.open(gpa, sfs.storageFile(), 0);
    defer src.deinit(gpa);

    // Deliberately incompressible bodies: a delta of the derived body against
    // the base is a handful of bytes, while a zlib re-encoding of it is larger
    // than the whole source delta FRAME. That is what forces the §6.1 guard down
    // the `.kept` → `copySrcFrame` path — the one under test. Compressible text
    // would let the repack re-encode the derived body as a full zlib frame and
    // the delta chain would never reach `copySrcFrame` at all.
    var base_body: [1024]u8 = undefined;
    var seed: u32 = 0x2545f491;
    for (&base_body) |*b| {
        seed = seed *% 1664525 +% 1013904223;
        b.* = @truncate(seed >> 24);
    }
    var derived_body: [1024]u8 = base_body;
    derived_body[1000] ^= 0xff; // one byte apart: a tiny delta, a big zlib

    var base_digest: [digest_len]u8 = undefined;
    var derived_digest: [digest_len]u8 = undefined;
    std.crypto.hash.sha3.Sha3_256.hash(&base_body, &base_digest, .{});
    std.crypto.hash.sha3.Sha3_256.hash(&derived_body, &derived_digest, .{});

    const base_rid = try src.putArtifact(&base_digest, &base_body, null);
    const derived_rid = try src.putArtifact(&derived_digest, &derived_body, base_rid);
    // The shun row goes in LAST, on purpose: shunning first would mask the base
    // from the source store's own `pickBase` and there would be no delta to
    // carry. This is the real-world order anyway — an operator shuns content the
    // repository has already been storing and deltifying for some time.
    try src.putShun(&base_digest, "takedown notice 4713", 1753000003);
    try src.commit();

    // The premises, or the assertions below would hold for uninteresting
    // reasons: the derived frame really is a delta, the base really is held and
    // masked, and the delta really does still reconstruct over a masked basis.
    try std.testing.expectEqual(logstore.FrameKind.delta, src.kindOf(derived_rid));
    try std.testing.expect(!src.isPhantom(base_rid));
    try std.testing.expectEqual(@as(?[]u8, null), try src.content(gpa, base_rid));
    const src_derived = (try src.content(gpa, derived_rid)).?;
    defer gpa.free(src_derived);
    try std.testing.expectEqualSlices(u8, &derived_body, src_derived);

    const dfile = try tmp.dir.createFile(io, "packed", .{ .read = true, .truncate = true });
    defer dfile.close(io);
    var dfs = storage.FileStorage.init(io, dfile, 0);
    var dst = try logstore.LogStore.open(gpa, dfs.storageFile(), src.hash_kind);
    defer dst.deinit(gpa);

    _ = try packLog(gpa, &src, &dst, Filter.keep_all, &.{});

    var rfs = storage.FileStorage.init(io, dfile, dst.size());
    var reopened = try logstore.LogStore.open(gpa, rfs.storageFile(), 0);
    defer reopened.deinit(gpa);

    // The base survived as a frame, still masked, still tombstoned.
    const packed_base = reopened.lookup(&base_digest) orelse
        return error.ShunnedBasisDroppedByPack;
    try std.testing.expect(!reopened.isPhantom(packed_base));
    try std.testing.expect(reopened.isShunned(&base_digest));
    try std.testing.expectEqual(@as(?[]u8, null), try reopened.content(gpa, packed_base));

    // THE PIN: the dependent is STILL A DELTA — the chain was relinked, not
    // flattened into a full frame — and it still reconstructs byte-for-byte
    // through a basis that reads as absent by name.
    const packed_derived = reopened.lookup(&derived_digest) orelse
        return error.DeltaOnShunnedBasisDroppedByPack;
    try std.testing.expectEqual(logstore.FrameKind.delta, reopened.kindOf(packed_derived));
    const out = (try reopened.content(gpa, packed_derived)).?;
    defer gpa.free(out);
    try std.testing.expectEqualSlices(u8, &derived_body, out);
}

// ---------------------------------------------------------------------------
// The pack FILTER: reclamation. Everything above pins what a repack asked to
// drop NOTHING does; everything below pins what one asked to drop something does.
// ---------------------------------------------------------------------------

/// A `Filter` over one named digest, for the tests below. Real reclaim filters
/// key on a store index (`shunFilter`); this one keys on a single literal so a
/// test can drop exactly one artifact without standing up a shun table.
const OneDigestFilter = struct {
    target: []const u8,

    fn drops(ctx: *anyopaque, digest: []const u8) bool {
        const self: *const OneDigestFilter = @ptrCast(@alignCast(ctx));
        return std.mem.eql(u8, self.target, digest);
    }

    fn filter(self: *const OneDigestFilter) Filter {
        return .{ .ctx = @constCast(self), .dropFn = drops };
    }
};

/// Open a fresh empty log on a new file named `name` under `dir`. The tests below
/// each pack into their own destination; this removes the four-line preamble.
const TestLog = struct {
    file: std.Io.File,
    fs: storage.FileStorage,
    log: logstore.LogStore,

    fn create(gpa: std.mem.Allocator, io: std.Io, dir: std.Io.Dir, name: []const u8, hash_kind: u8) !*TestLog {
        const self = try gpa.create(TestLog);
        errdefer gpa.destroy(self);
        self.file = try dir.createFile(io, name, .{ .read = true, .truncate = true });
        errdefer self.file.close(io);
        self.fs = storage.FileStorage.init(io, self.file, 0);
        self.log = try logstore.LogStore.open(gpa, self.fs.storageFile(), hash_kind);
        return self;
    }

    fn destroy(self: *TestLog, gpa: std.mem.Allocator, io: std.Io) void {
        self.log.deinit(gpa);
        self.file.close(io);
        gpa.destroy(self);
    }

    /// Reopen the written bytes as a second store — the packed generation read
    /// back from disk, which is the only oracle that proves what was WRITTEN.
    fn reopen(self: *TestLog, gpa: std.mem.Allocator, io: std.Io, rfs: *storage.FileStorage) !logstore.LogStore {
        rfs.* = storage.FileStorage.init(io, self.file, self.log.size());
        return logstore.LogStore.open(gpa, rfs.storageFile(), 0);
    }
};

test "pack: a Filter dropping one blob removes it from the new generation" {
    // The base case of reclamation, stated on the mechanism rather than on
    // the shun table: a named artifact is not copied, `dropped` counts it, and
    // the new log is smaller than the same pack under `keep_all`. The name is
    // gone from BOTH indexes an artifact can be reached by — `lookup` and
    // `resolveHexPrefix` — which is the mirror of what the keep_all tests above
    // assert; a reclaim that left the name resolvable to a frameless rid would
    // pass every byte-count assertion and still be wrong.
    const gpa = std.testing.allocator;
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    const io = std.testing.io;

    const sfile = try tmp.dir.createFile(io, "src", .{ .read = true });
    defer sfile.close(io);
    var sfs = storage.FileStorage.init(io, sfile, 0);
    var src = try logstore.LogStore.open(gpa, sfs.storageFile(), 0);
    defer src.deinit(gpa);

    const doomed_body = "the artifact this pack was told to reclaim, physically present in the source " ** 4;
    const kept_body = "the neighbour with nothing against it, which must survive untouched " ** 4;
    var doomed_digest: [digest_len]u8 = undefined;
    var kept_digest: [digest_len]u8 = undefined;
    std.crypto.hash.sha3.Sha3_256.hash(doomed_body, &doomed_digest, .{});
    std.crypto.hash.sha3.Sha3_256.hash(kept_body, &kept_digest, .{});
    _ = try src.putArtifact(&doomed_digest, doomed_body, null);
    _ = try src.putArtifact(&kept_digest, kept_body, null);
    try src.commit();

    // The same source packed twice: once keeping everything, once dropping one.
    const keep = try TestLog.create(gpa, io, tmp.dir, "keep", src.hash_kind);
    defer keep.destroy(gpa, io);
    const keep_stats = try packLog(gpa, &src, &keep.log, Filter.keep_all, &.{});
    try std.testing.expectEqual(@as(usize, 0), keep_stats.dropped);

    const f: OneDigestFilter = .{ .target = &doomed_digest };
    const drop = try TestLog.create(gpa, io, tmp.dir, "drop", src.hash_kind);
    defer drop.destroy(gpa, io);
    const stats = try packLog(gpa, &src, &drop.log, f.filter(), &.{});

    // `artifacts` stays the count of source frames WALKED; `dropped` is how
    // many of them the new generation did not receive.
    try std.testing.expectEqual(@as(usize, 2), stats.artifacts);
    try std.testing.expectEqual(@as(usize, 1), stats.dropped);
    // Dropping bytes shrinks the log relative to the keep-all pack of the SAME
    // source — the only comparison that isolates the filter's effect.
    try std.testing.expect(stats.bytes_after < keep_stats.bytes_after);

    var rfs: storage.FileStorage = undefined;
    var reopened = try drop.reopen(gpa, io, &rfs);
    defer reopened.deinit(gpa);

    // Exactly one artifact landed, and it is the neighbour.
    const packed_rids = try reopened.ridsInPersistedOrder(gpa);
    defer gpa.free(packed_rids);
    try std.testing.expectEqual(@as(usize, 1), packed_rids.len);

    // THE MIRROR PIN of the keep_all tests: BOTH name indexes now answer null.
    // `lookup` first — a reclaim that interned the name as a phantom would keep
    // answering a rid here...
    try std.testing.expect(reopened.lookup(&doomed_digest) == null);
    // ...and `resolveHexPrefix`, the other door the same name is reachable
    // through, which no byte-count assertion can speak for.
    var hex: [2 * digest_len]u8 = undefined;
    const doomed_name = digestToWireName(&doomed_digest, &hex);
    try std.testing.expectEqual(@as(?logstore.Rid, null), try reopened.resolveHexPrefix(doomed_name));
    try std.testing.expectEqual(@as(?logstore.Rid, null), try reopened.resolveHexPrefix(doomed_name[0..8]));

    // The neighbour is byte-intact through the ordinary reader.
    const kept_rid = reopened.lookup(&kept_digest) orelse return error.KeptArtifactDroppedByPack;
    const out = (try reopened.content(gpa, kept_rid)).?;
    defer gpa.free(out);
    try std.testing.expectEqualStrings(kept_body, out);
}

test "pack: shunFilter reclaims the shunned artifact and keeps its tombstone" {
    // `zit pack`'s own filter, end to end on the store-to-store core: the
    // artifact the shun table names is gone, and the ROW that named it is not.
    // Dropping both would un-shun the artifact — a later sync would re-fetch
    // exactly the bytes the operator was ordered to remove — so the tombstone
    // outliving the content it names is the whole point of deferred reclaim.
    const gpa = std.testing.allocator;
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    const io = std.testing.io;

    const sfile = try tmp.dir.createFile(io, "src", .{ .read = true });
    defer sfile.close(io);
    var sfs = storage.FileStorage.init(io, sfile, 0);
    var src = try logstore.LogStore.open(gpa, sfs.storageFile(), 0);
    defer src.deinit(gpa);

    const shunned_body = "content an operator was ordered to stop serving, and now to stop storing " ** 4;
    const kept_body = "ordinary content with no tombstone against it, packed the way it always was " ** 4;
    var shunned_digest: [digest_len]u8 = undefined;
    var kept_digest: [digest_len]u8 = undefined;
    std.crypto.hash.sha3.Sha3_256.hash(shunned_body, &shunned_digest, .{});
    std.crypto.hash.sha3.Sha3_256.hash(kept_body, &kept_digest, .{});
    _ = try src.putArtifact(&shunned_digest, shunned_body, null);
    _ = try src.putArtifact(&kept_digest, kept_body, null);
    try src.putShun(&shunned_digest, "takedown notice 4714", 1753000004);
    // A second row naming content the store never held, at the OTHER complete
    // width (20 bytes): it must not confuse the filter, and it must carry over.
    try src.putShun("u" ** 20, "", 1753000005);
    try src.commit();

    const dst_log = try TestLog.create(gpa, io, tmp.dir, "packed", src.hash_kind);
    defer dst_log.destroy(gpa, io);
    const stats = try packLog(gpa, &src, &dst_log.log, shunFilter(&src), &.{});
    try std.testing.expectEqual(@as(usize, 2), stats.artifacts);
    try std.testing.expectEqual(@as(usize, 1), stats.dropped);

    var rfs: storage.FileStorage = undefined;
    var reopened = try dst_log.reopen(gpa, io, &rfs);
    defer reopened.deinit(gpa);

    // The bytes are gone from both name indexes...
    try std.testing.expect(reopened.lookup(&shunned_digest) == null);
    var hex: [2 * digest_len]u8 = undefined;
    const shunned_name = digestToWireName(&shunned_digest, &hex);
    try std.testing.expectEqual(@as(?logstore.Rid, null), try reopened.resolveHexPrefix(shunned_name));
    // ...and the tombstones are not, in persisted order, fields intact. The
    // packed store still REFUSES the hash, which is what keeps a later sync
    // from re-fetching it.
    try std.testing.expect(reopened.isShunned(&shunned_digest));
    const rows = reopened.shunList();
    try std.testing.expectEqual(@as(usize, 2), rows.len);
    try std.testing.expectEqualSlices(u8, &shunned_digest, rows[0].hash);
    try std.testing.expectEqualStrings("takedown notice 4714", rows[0].reason);
    try std.testing.expectEqual(@as(i64, 1753000004), rows[0].epoch);
    try std.testing.expectEqualStrings("u" ** 20, rows[1].hash);

    // The neighbour survived, readable.
    const kept_rid = reopened.lookup(&kept_digest) orelse return error.KeptArtifactDroppedByPack;
    const out = (try reopened.content(gpa, kept_rid)).?;
    defer gpa.free(out);
    try std.testing.expectEqualStrings(kept_body, out);
}

test "pack: extra_shuns land ONLY in the new generation, and a duplicate row fails the pack" {
    // The crash-safety invariant `zit shun --force` rests on, made unit-testable
    // by putting the parameter here rather than in the CLI: the tombstone for a
    // freshly named hash exists in `dst` and NOT in `src`, so there is no on-disk
    // state — at any instant, in any build — where the row is durable and the
    // content it names still is. `src` is opened read-only in production; this
    // test asserts the weaker, checkable form (nothing was appended to it).
    const gpa = std.testing.allocator;
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    const io = std.testing.io;

    const sfile = try tmp.dir.createFile(io, "src", .{ .read = true });
    defer sfile.close(io);
    var sfs = storage.FileStorage.init(io, sfile, 0);
    var src = try logstore.LogStore.open(gpa, sfs.storageFile(), 0);
    defer src.deinit(gpa);

    const doomed_body = "content the operator names in this very invocation " ** 4;
    const kept_body = "content with nothing against it " ** 4;
    var doomed: [digest_len]u8 = undefined;
    var kept: [digest_len]u8 = undefined;
    std.crypto.hash.sha3.Sha3_256.hash(doomed_body, &doomed, .{});
    std.crypto.hash.sha3.Sha3_256.hash(kept_body, &kept, .{});
    _ = try src.putArtifact(&doomed, doomed_body, null);
    _ = try src.putArtifact(&kept, kept_body, null);
    try src.commit();
    const src_size_before = src.size();

    // ONE owned name set drives both the drop predicate and the tombstone list,
    // so a name cannot be recorded without being dropped.
    var set: NameSet = .{};
    defer set.deinit(gpa);
    try set.add(gpa, &doomed);
    const rows = [_]logstore.ShunRecord{.{ .hash = &doomed, .epoch = 1753000042, .reason = "takedown 4417" }};

    {
        const dst_log = try TestLog.create(gpa, io, tmp.dir, "packed", src.hash_kind);
        defer dst_log.destroy(gpa, io);
        const stats = try packLog(gpa, &src, &dst_log.log, set.filter(), &rows);
        try std.testing.expectEqual(@as(usize, 2), stats.artifacts);
        try std.testing.expectEqual(@as(usize, 1), stats.dropped);

        // The source is untouched: no row, no growth. The row exists only in the
        // generation that also lacks the content.
        try std.testing.expectEqual(@as(usize, 0), src.shunList().len);
        try std.testing.expectEqual(src_size_before, src.size());

        var rfs: storage.FileStorage = undefined;
        var reopened = try dst_log.reopen(gpa, io, &rfs);
        defer reopened.deinit(gpa);
        try std.testing.expect(reopened.lookup(&doomed) == null);
        try std.testing.expect(reopened.isShunned(&doomed));
        try std.testing.expectEqual(@as(usize, 1), reopened.shunList().len);
        try std.testing.expectEqualStrings("takedown 4417", reopened.shunList()[0].reason);
        try std.testing.expectEqual(@as(i64, 1753000042), reopened.shunList()[0].epoch);
        // The neighbour survived.
        try std.testing.expect(reopened.lookup(&kept) != null);
    }

    // The precondition, enforced rather than described: a row duplicating one the
    // source already carries appends nothing (`putShun` is idempotent as set
    // membership), so the widened row-count gate fails the pack CLOSED instead of
    // reporting a successful reclaim whose tombstone count nobody checked.
    {
        try src.putShun(&doomed, "recorded earlier", 1753000001);
        try src.commit();
        const dst_log = try TestLog.create(gpa, io, tmp.dir, "packed2", src.hash_kind);
        defer dst_log.destroy(gpa, io);
        try std.testing.expectError(
            error.Corrupt,
            packLog(gpa, &src, &dst_log.log, set.filter(), &rows),
        );
    }
}

test "pack: a dropped artifact is never reconstructed or §7.7-verified" {
    // THE ORDERING REQUIREMENT of the reclaim contract, normative and not an
    // optimisation. The filter check MUST precede `srcContent`/`artifact.verify`
    // at BOTH sites, and each site has its own test because each has its own
    // failure mode. THIS test pins the ARTIFACT LOOP: its fixture is a frame that
    // decodes cleanly and merely lies about its name, so only `artifact.verify`
    // rejects it — and `verify` is called nowhere else. The pass-1 site in
    // `deriveCandidateHints` is pinned by "a dropped artifact whose frame will not
    // DECODE does not abort the pack", whose fixture fails `srcContent` itself.
    //
    // Why it is normative: an operator shuns content PRECISELY BECAUSE its bytes
    // are unlawful or corrupt. A frame whose content does not hash to its name
    // aborts the pack with `error.Corrupt` (the §7.7 gate, pinned in
    // cmd/sync.zig). Verify before filtering and such a frame makes `zit pack`
    // fail forever with no way out: the one operation that could remove it is
    // the one that refuses to run because of it. This store packs CLEANLY.
    const gpa = std.testing.allocator;
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    const io = std.testing.io;

    const sfile = try tmp.dir.createFile(io, "src", .{ .read = true });
    defer sfile.close(io);
    var sfs = storage.FileStorage.init(io, sfile, 0);
    var src = try logstore.LogStore.open(gpa, sfs.storageFile(), 0);
    defer src.deinit(gpa);

    // `put` stores bytes verbatim under ANY key, so this models a lying source
    // frame that survives `openReadOnly` (its CRC and hash label are intact) —
    // the same fixture shape as cmd/sync.zig's §7.7 abort test.
    const lying_name = [_]u8{0xab} ** digest_len;
    _ = try src.put(&lying_name, "content these bytes do not name");
    const honest_body = "an honest neighbour, so a pack that dropped everything cannot pass by accident";
    var honest_digest: [digest_len]u8 = undefined;
    std.crypto.hash.sha3.Sha3_256.hash(honest_body, &honest_digest, .{});
    _ = try src.putArtifact(&honest_digest, honest_body, null);
    try src.putShun(&lying_name, "unlawful and corrupt", 1753000006);
    try src.commit();

    // The premise: keep_all — filter absent — genuinely aborts on this store.
    // Without this line the test below could pass because the frame is benign.
    {
        const keep = try TestLog.create(gpa, io, tmp.dir, "keep", src.hash_kind);
        defer keep.destroy(gpa, io);
        try std.testing.expectError(error.Corrupt, packLog(gpa, &src, &keep.log, Filter.keep_all, &.{}));
    }

    // THE PIN: shunned, so never reconstructed, so never verified, so no abort.
    const dst_log = try TestLog.create(gpa, io, tmp.dir, "packed", src.hash_kind);
    defer dst_log.destroy(gpa, io);
    const stats = try packLog(gpa, &src, &dst_log.log, shunFilter(&src), &.{});
    try std.testing.expectEqual(@as(usize, 1), stats.dropped);

    var rfs: storage.FileStorage = undefined;
    var reopened = try dst_log.reopen(gpa, io, &rfs);
    defer reopened.deinit(gpa);
    try std.testing.expect(reopened.lookup(&lying_name) == null);
    try std.testing.expect(reopened.isShunned(&lying_name)); // the row stays
    try std.testing.expect(reopened.lookup(&honest_digest) != null);
}

test "pack: shunFilter dropping a delta BASE leaves every dependent readable and correctly hashed" {
    // The delta-base correctness property of the reclaim contract. Dropping a
    // frame that other
    // frames are encoded AGAINST is the one way a reclaim could strand data, and
    // the reason it cannot is an ordering fact worth pinning rather than
    // trusting: `packLog` decodes every artifact to full plaintext from the
    // SOURCE — which still holds the base — before re-running `pickBase`, so
    // each surviving dependent is reconstructed while its basis is still there
    // and re-encoded against whatever the new generation actually holds.
    // `copySrcFrame` re-encodes rather than copying when a base is absent from
    // `dst` (never a dangling delta, §3.3), and the mandatory `packVerifyPacked`
    // sweep re-hashes every packed artifact through its NEW chain before the
    // swap. Break any of the three and this test fails.
    //
    // Readability and hashes only, deliberately: this test says nothing about the
    // resulting SIZES. That is the sibling test "reclaiming a SHARED delta base
    // neither inflates the store nor breaks the fixpoint", and the split is why
    // this one passed for a while over a store that had quintupled.
    const gpa = std.testing.allocator;
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    const io = std.testing.io;

    const sfile = try tmp.dir.createFile(io, "src", .{ .read = true });
    defer sfile.close(io);
    var sfs = storage.FileStorage.init(io, sfile, 0);
    var src = try logstore.LogStore.open(gpa, sfs.storageFile(), 0);
    defer src.deinit(gpa);

    // Incompressible bodies one byte apart: the source stores tiny deltas, so
    // the §6.1 monotonic guard takes the `.kept` → `copySrcFrame` path and the
    // dependents genuinely arrive as deltas needing a basis.
    var base_body: [1024]u8 = undefined;
    var seed: u32 = 0x2545f491;
    for (&base_body) |*b| {
        seed = seed *% 1664525 +% 1013904223;
        b.* = @truncate(seed >> 24);
    }
    var d1_body: [1024]u8 = base_body;
    d1_body[1000] ^= 0xff;
    var d2_body: [1024]u8 = base_body;
    d2_body[200] ^= 0x0f;

    var base_digest: [digest_len]u8 = undefined;
    var d1_digest: [digest_len]u8 = undefined;
    var d2_digest: [digest_len]u8 = undefined;
    std.crypto.hash.sha3.Sha3_256.hash(&base_body, &base_digest, .{});
    std.crypto.hash.sha3.Sha3_256.hash(&d1_body, &d1_digest, .{});
    std.crypto.hash.sha3.Sha3_256.hash(&d2_body, &d2_digest, .{});

    const base_rid = try src.putArtifact(&base_digest, &base_body, null);
    const d1_rid = try src.putArtifact(&d1_digest, &d1_body, base_rid);
    const d2_rid = try src.putArtifact(&d2_digest, &d2_body, base_rid);
    // Shun LAST, as an operator would: the repository has been deltifying
    // against this base for some time before anyone objects to it.
    try src.putShun(&base_digest, "takedown notice 4715", 1753000007);
    try src.commit();

    // The premises: both dependents really are deltas rooted on the base.
    try std.testing.expectEqual(logstore.FrameKind.delta, src.kindOf(d1_rid));
    try std.testing.expectEqual(logstore.FrameKind.delta, src.kindOf(d2_rid));

    const dst_log = try TestLog.create(gpa, io, tmp.dir, "packed", src.hash_kind);
    defer dst_log.destroy(gpa, io);
    const stats = try packLog(gpa, &src, &dst_log.log, shunFilter(&src), &.{});
    try std.testing.expectEqual(@as(usize, 3), stats.artifacts);
    try std.testing.expectEqual(@as(usize, 1), stats.dropped);

    var rfs: storage.FileStorage = undefined;
    var reopened = try dst_log.reopen(gpa, io, &rfs);
    defer reopened.deinit(gpa);

    // The basis is genuinely gone — not merely masked. Its bytes are not
    // reachable through the MAINTENANCE door either, which is the difference
    // between a shun and a reclaim.
    try std.testing.expect(reopened.lookup(&base_digest) == null);

    // THE PIN: both dependents are readable and hash to their own names. Neither
    // is stranded on the basis that no longer exists.
    for ([_][]const u8{ &d1_digest, &d2_digest }, [_][]const u8{ &d1_body, &d2_body }) |dg, body| {
        const rid = reopened.lookup(dg) orelse return error.DependentStrandedByReclaim;
        const out = (try reopened.content(gpa, rid)).?;
        defer gpa.free(out);
        try std.testing.expectEqualSlices(u8, body, out);
        var hex: [2 * digest_len]u8 = undefined;
        try std.testing.expect(artifact.verify(digestToWireName(dg, &hex), out));
    }
}

/// A DELIBERATELY INCONSISTENT `Filter`: it answers "keep" while the artifact
/// loop is copying frames (`dst` does not hold the target yet) and "drop" once
/// the target has landed. That models the one failure `packVerifyShunned` exists
/// to catch — a filter bug, or a future `copySrcFrame` path, that copies forward
/// an artifact the pack was told to remove — without needing to mutate pack.
const LandedFilter = struct {
    target: []const u8,
    dst: *const logstore.LogStore,

    fn drops(ctx: *anyopaque, digest: []const u8) bool {
        const self: *const LandedFilter = @ptrCast(@alignCast(ctx));
        if (!std.mem.eql(u8, self.target, digest)) return false;
        return self.dst.lookup(self.target) != null;
    }

    fn filter(self: *const LandedFilter) Filter {
        return .{ .ctx = @constCast(self), .dropFn = drops };
    }
};

test "pack: a filter that lets a dropped digest through aborts the pack" {
    // `packVerifyPacked` cannot catch this and never could: it checks that
    // everything PRESENT in `dst` is honest, and a wrongly-copied frame is
    // perfectly honest. The two predicates are disjoint — "present ⇒ honest" vs
    // "ordered gone ⇒ gone" — and only the second one speaks for an operation
    // whose entire purpose is deletion. Fail-open here means `zit pack` reports
    // "reclaimed" while the bytes ship in the next clone, so the end state is
    // checked on the ARTIFACT (re-asking the filter over `dst`'s own rids)
    // rather than on the bookkeeping that was supposed to produce it.
    const gpa = std.testing.allocator;
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    const io = std.testing.io;

    const sfile = try tmp.dir.createFile(io, "src", .{ .read = true });
    defer sfile.close(io);
    var sfs = storage.FileStorage.init(io, sfile, 0);
    var src = try logstore.LogStore.open(gpa, sfs.storageFile(), 0);
    defer src.deinit(gpa);

    const body = "the artifact a buggy filter copied forward after promising to drop it";
    var dg: [digest_len]u8 = undefined;
    std.crypto.hash.sha3.Sha3_256.hash(body, &dg, .{});
    _ = try src.putArtifact(&dg, body, null);
    try src.commit();

    const dst_log = try TestLog.create(gpa, io, tmp.dir, "packed", src.hash_kind);
    defer dst_log.destroy(gpa, io);
    const f: LandedFilter = .{ .target = &dg, .dst = &dst_log.log };
    // The artifact loop copies the frame (the filter says keep), then the sweep
    // re-asks and gets "drop" for something that is demonstrably present.
    try std.testing.expectError(error.Corrupt, packLog(gpa, &src, &dst_log.log, f.filter(), &.{}));
}

test "pack: pack∘pack is still a fixpoint with shun rows present" {
    // Determinism across the reclaim boundary. The first pack removes the
    // shunned artifact and carries the row; the second finds nothing left to
    // remove and must produce BYTE-IDENTICAL output — a reclaim that was not
    // idempotent (a row re-emitted twice, a dropped name re-interned as a
    // phantom, an offset shifted) would show up here as a differing file.
    const gpa = std.testing.allocator;
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    const io = std.testing.io;

    const sfile = try tmp.dir.createFile(io, "src", .{ .read = true });
    defer sfile.close(io);
    var sfs = storage.FileStorage.init(io, sfile, 0);
    var src = try logstore.LogStore.open(gpa, sfs.storageFile(), 0);
    defer src.deinit(gpa);

    const shunned_body = "reclaimed on the first pass, absent by the second " ** 4;
    const a_body = "one surviving artifact " ** 8;
    const b_body = "one surviving artifact, very nearly " ** 8; // deltifiable against a
    var shunned_digest: [digest_len]u8 = undefined;
    var a_digest: [digest_len]u8 = undefined;
    var b_digest: [digest_len]u8 = undefined;
    std.crypto.hash.sha3.Sha3_256.hash(shunned_body, &shunned_digest, .{});
    std.crypto.hash.sha3.Sha3_256.hash(a_body, &a_digest, .{});
    std.crypto.hash.sha3.Sha3_256.hash(b_body, &b_digest, .{});
    _ = try src.putArtifact(&shunned_digest, shunned_body, null);
    const a_rid = try src.putArtifact(&a_digest, a_body, null);
    _ = try src.putArtifact(&b_digest, b_body, a_rid);
    try src.putShun(&shunned_digest, "takedown notice 4716", 1753000008);
    try src.putShun("v" ** 32, "a row naming content nobody here holds", 1753000009);
    try src.commit();

    const p1 = try TestLog.create(gpa, io, tmp.dir, "p1", src.hash_kind);
    defer p1.destroy(gpa, io);
    const s1 = try packLog(gpa, &src, &p1.log, shunFilter(&src), &.{});
    try std.testing.expectEqual(@as(usize, 1), s1.dropped);

    var r1fs: storage.FileStorage = undefined;
    var l1 = try p1.reopen(gpa, io, &r1fs);
    defer l1.deinit(gpa);
    // Both rows survived, so pass 2's filter is derived from the same set.
    try std.testing.expectEqual(@as(usize, 2), l1.shunList().len);

    const p2 = try TestLog.create(gpa, io, tmp.dir, "p2", l1.hash_kind);
    defer p2.destroy(gpa, io);
    const s2 = try packLog(gpa, &l1, &p2.log, shunFilter(&l1), &.{});
    // Nothing left to reclaim: the artifact is already gone from the source.
    try std.testing.expectEqual(@as(usize, 0), s2.dropped);
    try std.testing.expectEqual(s1.bytes_after, s2.bytes_after);

    // Byte-for-byte, not merely equal in size.
    const b1 = try tmp.dir.readFileAlloc(io, "p1", gpa, .limited(1 << 20));
    defer gpa.free(b1);
    const b2 = try tmp.dir.readFileAlloc(io, "p2", gpa, .limited(1 << 20));
    defer gpa.free(b2);
    try std.testing.expectEqualSlices(u8, b1, b2);
}

test "pack: a shunned frame is never chosen as a FRESH delta basis in the new generation" {
    // THE ORDERING PIN for the shun-row carry-forward. `packLog` copies the rows
    // into `dst` BEFORE the artifact loop, and the reason is stated at that loop:
    // `dst.putArtifactPacked` → `pickBase` probes candidates through the MASKED
    // `dst.content`, so with the rows already in place a shunned frame reads as
    // absent and is skipped like a phantom. Copy them afterwards and `pickBase`
    // happily roots new deltas on it — a basis whose bytes a reclaim is about to
    // drop acquires new dependents, and nothing else in the suite notices.
    //
    // The sibling "delta chain ROOTED ON a shunned base" test cannot catch that:
    // its dependent is ALREADY a delta in the source and stays one either way, so
    // its `FrameKind.delta` assertion holds whichever order the rows are copied
    // in. This test's dependent is stored FULL in the source, so the ordering
    // decides whether pack re-encodes it as a delta at all.
    const gpa = std.testing.allocator;
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    const io = std.testing.io;

    const sfile = try tmp.dir.createFile(io, "src", .{ .read = true });
    defer sfile.close(io);
    var sfs = storage.FileStorage.init(io, sfile, 0);
    var src = try logstore.LogStore.open(gpa, sfs.storageFile(), 0);
    defer src.deinit(gpa);

    // Incompressible and one byte apart: a delta of `dependent` against `basis`
    // is a handful of bytes while its own zlib frame is ~1 KiB, so if `pickBase`
    // can SEE the basis it will certainly choose it — the §6.1 monotonic-shrink
    // guard is nowhere near binding. That makes the frame kind below a clean
    // read-out of whether the basis was visible.
    var basis_body: [1024]u8 = undefined;
    var seed: u32 = 0x9e3779b9;
    for (&basis_body) |*b| {
        seed = seed *% 1664525 +% 1013904223;
        b.* = @truncate(seed >> 24);
    }
    var dependent_body: [1024]u8 = basis_body;
    dependent_body[512] ^= 0xff;

    var basis_digest: [digest_len]u8 = undefined;
    var dependent_digest: [digest_len]u8 = undefined;
    std.crypto.hash.sha3.Sha3_256.hash(&basis_body, &basis_digest, .{});
    std.crypto.hash.sha3.Sha3_256.hash(&dependent_body, &dependent_digest, .{});

    // BOTH stored FULL, via `putZlib` rather than `putArtifact`: `putArtifact`
    // runs the store's OWN base selection and — with no shun row in place yet —
    // would deltify the dependent against the basis at write time, leaving
    // nothing for pack to decide. `putZlib` stores an unconditional full frame,
    // so the source holds no delta at all and any delta in the packed generation
    // is one pack CHOSE to create.
    const basis_rid = try src.putZlib(&basis_digest, &basis_body);
    const dependent_rid = try src.putZlib(&dependent_digest, &dependent_body);
    try src.putShun(&basis_digest, "takedown notice 4718", 1753000011);
    try src.commit();

    // The premises: neither source frame is a delta, and the basis is held.
    try std.testing.expectEqual(logstore.FrameKind.zlib, src.kindOf(basis_rid));
    try std.testing.expectEqual(logstore.FrameKind.zlib, src.kindOf(dependent_rid));
    try std.testing.expect(!src.isPhantom(basis_rid));

    // `keep_all`, deliberately: the basis frame is RETAINED, so it is physically
    // available to be mis-chosen. Under `shunFilter` it would be dropped and the
    // question could not even arise — this is the case where only the row
    // ORDERING stands between a reclaim-bound basis and a new dependent.
    const dst_log = try TestLog.create(gpa, io, tmp.dir, "packed", src.hash_kind);
    defer dst_log.destroy(gpa, io);
    const stats = try packLog(gpa, &src, &dst_log.log, Filter.keep_all, &.{});
    try std.testing.expectEqual(@as(usize, 0), stats.dropped);
    // No frame in the packed generation is a delta: the only candidate basis on
    // offer was masked, so `pickBase` had nothing to choose.
    try std.testing.expectEqual(@as(usize, 0), stats.deltas);

    var rfs: storage.FileStorage = undefined;
    var reopened = try dst_log.reopen(gpa, io, &rfs);
    defer reopened.deinit(gpa);

    // THE PIN: the dependent is still a FULL frame. Copy the shun rows after the
    // artifact loop and this reads `.delta` — packed output changed, silently,
    // with every other assertion in the suite still passing.
    const packed_dependent = reopened.lookup(&dependent_digest) orelse
        return error.DependentDroppedByPack;
    try std.testing.expectEqual(logstore.FrameKind.zlib, reopened.kindOf(packed_dependent));
    // ...and it still reads back byte-exact, so the pin is not passing merely
    // because the frame is broken.
    const out = (try reopened.content(gpa, packed_dependent)).?;
    defer gpa.free(out);
    try std.testing.expectEqualSlices(u8, &dependent_body, out);
}

test "pack: reclaiming a SHARED delta base neither inflates the store nor breaks the fixpoint" {
    // The §6.1 monotonic-shrink floor is the SOURCE frame's span, and that floor
    // is only legitimate while the source frame is actually re-storable. Reclaim a
    // delta BASE and every dependent's source frame becomes unkeepable — its
    // basis is not in `dst` — so a floor read off it makes `putArtifactPacked`
    // answer `.kept` for an encoding that cannot be kept, and `copySrcFrame`
    // throws away the profitable delta it had just computed for a FULL zlib
    // frame. Six dependents of one 4 KiB incompressible base is the ordinary
    // shape of an imported block (every delta manifest against the same baseline
    // manifest), and it inflates the store fivefold on the one command an
    // operator runs to free disk.
    //
    // Two properties, both false under that bug: the reclaim keeps the dependents
    // DELTA-ENCODED against each other (so the reclaimed base's bytes survive in
    // exactly one full frame, not N), and `pack∘pack` is still byte-identical. The
    // existing fixpoint test cannot see either — its dropped artifact is a LEAF,
    // so nothing is ever forced down the absent-base path.
    //
    // NOT asserted, because it is not true and never could be: that the log
    // SHRINKS. Deleting a basis relocates its bytes into whichever dependent
    // becomes the new full root, and each surviving sibling's delta must now span
    // two differing regions instead of one — this fixture measures 5760 → 6248
    // bytes. What the reclaim must never do is pay for the deletion N times over
    // (measured 25448 before the fix, a 4.4x blow-up), which is what the frame-kind
    // and one-full-copy pins below actually forbid.
    const gpa = std.testing.allocator;
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    const io = std.testing.io;

    const sfile = try tmp.dir.createFile(io, "src", .{ .read = true });
    defer sfile.close(io);
    var sfs = storage.FileStorage.init(io, sfile, 0);
    var src = try logstore.LogStore.open(gpa, sfs.storageFile(), 0);
    defer src.deinit(gpa);

    // Incompressible, so a full frame is ~4 KiB and a one-region delta is tiny.
    var base_body: [4096]u8 = undefined;
    var seed: u32 = 0x1d872b41;
    for (&base_body) |*b| {
        seed = seed *% 1664525 +% 1013904223;
        b.* = @truncate(seed >> 24);
    }
    var base_digest: [digest_len]u8 = undefined;
    std.crypto.hash.sha3.Sha3_256.hash(&base_body, &base_digest, .{});
    const base_rid = try src.putArtifact(&base_digest, &base_body, null);

    // Each dependent differs from the BASE in its own distinct region, so a
    // sibling-to-sibling delta must encode two differing regions and is never
    // smaller than the source's sibling-to-base delta. That is what makes the
    // §6.1 floor bind and the `.kept` → `copySrcFrame` path the one taken.
    const dep_count = 6;
    var dep_bodies: [dep_count][4096]u8 = undefined;
    var dep_digests: [dep_count][digest_len]u8 = undefined;
    for (&dep_bodies, &dep_digests, 0..) |*body, *dg, i| {
        body.* = base_body;
        const at = 256 + i * 512;
        for (body[at .. at + 128]) |*b| b.* ^= 0xff;
        std.crypto.hash.sha3.Sha3_256.hash(body, dg, .{});
        const rid = try src.putArtifact(dg, body, base_rid);
        // Premise: the source really did store a delta rooted on the base.
        try std.testing.expectEqual(logstore.FrameKind.delta, src.kindOf(rid));
    }
    try src.putShun(&base_digest, "takedown notice 4719", 1753000012);
    try src.commit();

    const p1 = try TestLog.create(gpa, io, tmp.dir, "p1", src.hash_kind);
    defer p1.destroy(gpa, io);
    const s1 = try packLog(gpa, &src, &p1.log, shunFilter(&src), &.{});
    try std.testing.expectEqual(@as(usize, 1), s1.dropped);
    try std.testing.expectEqual(@as(usize, dep_count + 1), s1.artifacts);

    // THE FAN-OUT PIN: all but one dependent stayed a delta. Under the bug every
    // one of them was flattened (`deltas == 0`) because the discarded floor sent
    // each through a bare full-form store.
    try std.testing.expectEqual(@as(usize, dep_count - 1), s1.deltas);
    // ...and its byte consequence, stated independently of the frame kinds so a
    // future encoder change cannot satisfy the letter of the pin while paying for
    // the deletion N times: the packed log holds ONE full copy of the 4 KiB body,
    // not `dep_count` of them. Nothing else in the suite asserts a byte count
    // under a non-identity filter at all.
    try std.testing.expect(s1.bytes_after < 2 * base_body.len);

    var r1fs: storage.FileStorage = undefined;
    var l1 = try p1.reopen(gpa, io, &r1fs);
    defer l1.deinit(gpa);
    try std.testing.expect(l1.lookup(&base_digest) == null);

    // Every dependent is readable and hashes to its own name through its NEW
    // chain, so the pins above are not passing because something was lost.
    for (&dep_digests, &dep_bodies) |*dg, *body| {
        const rid = l1.lookup(dg) orelse return error.DependentStrandedByReclaim;
        const out = (try l1.content(gpa, rid)).?;
        defer gpa.free(out);
        try std.testing.expectEqualSlices(u8, body, out);
        var hex: [2 * digest_len]u8 = undefined;
        try std.testing.expect(artifact.verify(digestToWireName(dg, &hex), out));
    }

    // THE FIXPOINT PIN, with the drop on a SHARED base: pack the packed log again
    // and the bytes must not move.
    const p2 = try TestLog.create(gpa, io, tmp.dir, "p2", l1.hash_kind);
    defer p2.destroy(gpa, io);
    const s2 = try packLog(gpa, &l1, &p2.log, shunFilter(&l1), &.{});
    try std.testing.expectEqual(@as(usize, 0), s2.dropped);
    const b1 = try tmp.dir.readFileAlloc(io, "p1", gpa, .limited(1 << 20));
    defer gpa.free(b1);
    const b2 = try tmp.dir.readFileAlloc(io, "p2", gpa, .limited(1 << 20));
    defer gpa.free(b2);
    try std.testing.expectEqualSlices(u8, b1, b2);
}

test "pack: a dropped artifact whose frame will not DECODE does not abort the pack" {
    // The ORDERING REQUIREMENT of the reclaim contract at its SECOND site:
    // `deriveCandidateHints`
    // pass 1, which reconstructs every rid one function earlier than the artifact
    // loop just to try parsing it as a manifest.
    //
    // The sibling test's fixture — a frame whose content is intact but does not
    // hash to its name — cannot pin that site: such a frame decodes fine, so only
    // `artifact.verify` rejects it, and `verify` is only ever called in the
    // artifact loop. Move the pass-1 filter check after its `srcContent` and that
    // test stays green. This fixture is the one that bites: an UNDECODABLE `.zlib`
    // container (`putContainer` stores it verbatim and recomputes no hash, `scan`
    // accepts it because its CRC is honest), so `srcContent` ITSELF is
    // `error.Corrupt` — the sole failure mode the pass-1 check guards. Exactly the
    // damage class an operator shunning corrupt content is trying to get rid of.
    const gpa = std.testing.allocator;
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    const io = std.testing.io;

    const sfile = try tmp.dir.createFile(io, "src", .{ .read = true });
    defer sfile.close(io);
    var sfs = storage.FileStorage.init(io, sfile, 0);
    var src = try logstore.LogStore.open(gpa, sfs.storageFile(), 0);
    defer src.deinit(gpa);

    // A well-formed §4.1 length prefix over a body that is not a zlib stream.
    const garbage_container = [_]u8{ 0, 0, 0, 16 } ++ ("not a deflate stream" ** 1);
    const undecodable_name = [_]u8{0x5c} ** digest_len;
    _ = try src.putContainer(&undecodable_name, garbage_container);
    const honest_body = "an honest neighbour, so a pack that dropped everything cannot pass by accident";
    var honest_digest: [digest_len]u8 = undefined;
    std.crypto.hash.sha3.Sha3_256.hash(honest_body, &honest_digest, .{});
    _ = try src.putArtifact(&honest_digest, honest_body, null);
    try src.putShun(&undecodable_name, "corrupt beyond reconstruction", 1753000013);
    try src.commit();

    // The premise: keep_all genuinely aborts, and it aborts because the frame will
    // not DECODE (not because it lies about its name).
    {
        const keep = try TestLog.create(gpa, io, tmp.dir, "keep", src.hash_kind);
        defer keep.destroy(gpa, io);
        try std.testing.expectError(error.Corrupt, packLog(gpa, &src, &keep.log, Filter.keep_all, &.{}));
    }

    // THE PIN: dropped, so never decoded at EITHER site, so no abort.
    const dst_log = try TestLog.create(gpa, io, tmp.dir, "packed", src.hash_kind);
    defer dst_log.destroy(gpa, io);
    const stats = try packLog(gpa, &src, &dst_log.log, shunFilter(&src), &.{});
    try std.testing.expectEqual(@as(usize, 1), stats.dropped);

    var rfs: storage.FileStorage = undefined;
    var reopened = try dst_log.reopen(gpa, io, &rfs);
    defer reopened.deinit(gpa);
    try std.testing.expect(reopened.lookup(&undecodable_name) == null);
    try std.testing.expect(reopened.isShunned(&undecodable_name)); // the row stays
    try std.testing.expect(reopened.lookup(&honest_digest) != null);
}

/// Store one artifact under its SHA3-256 name.
fn putNamed(log: *logstore.LogStore, bytes: []const u8) !void {
    var digest: [digest_len]u8 = undefined;
    std.crypto.hash.sha3.Sha3_256.hash(bytes, &digest, .{});
    _ = try log.putArtifact(&digest, bytes, null);
}

/// A check-in manifest over a sub-manifest tree, with an optional primary
/// parent. Only the cards pack reads are meaningful; the rest satisfy §4.
fn treeCheckinBytes(
    gpa: std.mem.Allocator,
    comment: []const u8,
    timestamp: []const u8,
    s: []const artifact.SCard,
    parent: ?[]const u8,
) ![]u8 {
    var parents: [1]artifact.PCard = undefined;
    var p: []const artifact.PCard = &.{};
    if (parent) |name| {
        parents[0] = .{ .parents = name, .count = 1 };
        p = parents[0..1];
    }
    var m: artifact.Manifest = .{
        .arena_state = std.heap.ArenaAllocator.init(gpa),
        .kind = .manifest,
        .a = null,
        .b = null,
        .c = .{ .comment = comment },
        // §4.3: the D card is what keeps this a MANIFEST rather than being
        // reclassified as a SUBMANIFEST -- pack's check-in scan reads exactly
        // the artifacts that stay MANIFEST.
        .d = .{ .timestamp = timestamp },
        .e = null,
        .g = null,
        .h = null,
        .i = null,
        .k = null,
        .l = null,
        .n = null,
        .r = null,
        .u = .{ .login = "tester" },
        .z = null,
        .w = null,
        .f = &.{},
        .j = &.{},
        .m = &.{},
        .p = p,
        .q = &.{},
        .s = s,
        .t = &.{},
    };
    defer m.deinit();
    return artifact.create(&m, gpa);
}

test "pack: a nested sub-manifest tree survives its own partition-key walk" {
    // REGRESSION. `recordTreeVersions` used to insert `sc.filename` -- which
    // borrows the manifest bytes of the node being walked -- straight into the
    // partition-key map. At depth 0 those bytes live in the caller's frame, but
    // every node BELOW depth 0 is loaded into a `bytes`/`loaded` pair the
    // recursion frame frees on return, so a borrowed key outlived its backing
    // store by exactly one level. The map is iterated after the walk and grown
    // during it, so the borrow was a use-after-free on both paths -- latent for
    // as long as every walked tree was one level deep, which is why a
    // k=5 measurement passed and a k=1 import segfaulted.
    //
    // This test is the direct home for that property: the walk must produce a
    // tree of height >= 2, and the keys it recorded must still be readable and
    // FREEABLE after every nested buffer is gone.
    const gpa = std.testing.allocator;
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    const io = std.testing.io;

    // A file set large enough that the content-defined partition nests. `k = 1`
    // is the shallowest threshold and so the one that groups most eagerly --
    // the same setting `import-git --submanifest-k 1` uses.
    const file_count = 256;
    var names: [file_count][8]u8 = undefined;
    var bodies: [file_count][24]u8 = undefined;
    var hashes: [file_count][64]u8 = undefined;
    var files: [file_count]artifact.FCard = undefined;
    for (0..file_count) |i| {
        _ = std.fmt.bufPrint(&names[i], "f{d:0>5}.c", .{i}) catch unreachable;
        _ = std.fmt.bufPrint(&bodies[i], "body {d:0>18}\n", .{i}) catch unreachable;
        _ = artifact.artifactName(.sha3_256, &bodies[i], &hashes[i]);
        files[i] = .{
            .filename = &names[i],
            .hash = &hashes[i],
            .perm = null,
            .old_name = null,
            .mode = .regular,
        };
    }

    var tree = try artifact.buildTree(gpa, &files, 1, .sha3_256);
    defer tree.deinit();
    try std.testing.expect(!tree.isFlat());

    // The precondition the regression needs, asserted rather than assumed: at
    // least one sub-manifest is INTERIOR (carries S cards), so the walk
    // recurses and records keys from a frame that then returns.
    var interior_nodes: usize = 0;
    for (tree.subs) |sub| {
        var loaded = try artifact.parseForImport(gpa, sub.bytes);
        defer loaded.deinit(gpa);
        if (loaded.manifest.s.len != 0) interior_nodes += 1;
    }
    try std.testing.expect(interior_nodes != 0);

    const sfile = try tmp.dir.createFile(io, "src", .{ .read = true });
    defer sfile.close(io);
    var sfs = storage.FileStorage.init(io, sfile, 0);
    var src = try logstore.LogStore.open(gpa, sfs.storageFile(), 0);
    defer src.deinit(gpa);

    for (0..file_count) |i| try putNamed(&src, &bodies[i]);
    // §6.2 storage obligation: `subs` is already ordered so nothing precedes
    // what it references.
    for (tree.subs) |sub| try putNamed(&src, sub.bytes);

    const first = try treeCheckinBytes(gpa, "root check-in", "2026-08-01T00:00:00", tree.s, null);
    defer gpa.free(first);
    try putNamed(&src, first);
    var first_name: [64]u8 = undefined;
    _ = artifact.artifactName(.sha3_256, first, &first_name);

    // A child sharing the same tree. This is what drives the PARENT arm, where
    // the child's key map is iterated against a second walk's -- the read that
    // the borrowed keys made a use-after-free rather than merely a latent one.
    const second = try treeCheckinBytes(gpa, "child check-in", "2026-08-01T00:00:01", tree.s, &first_name);
    defer gpa.free(second);
    try putNamed(&src, second);
    try src.commit();

    const dst = try TestLog.create(gpa, io, tmp.dir, "packed", src.hash_kind);
    defer dst.destroy(gpa, io);
    const stats = try packLog(gpa, &src, &dst.log, Filter.keep_all, &.{});
    try std.testing.expectEqual(@as(usize, 0), stats.dropped);
    try std.testing.expectEqual(src.entryCount(), stats.artifacts);

    // Every artifact is readable from what was WRITTEN, tree nodes included --
    // a walk that mis-keyed itself can still emit a log, so the oracle is the
    // reopened generation and not the return value.
    var rfs: storage.FileStorage = undefined;
    var reopened = try dst.reopen(gpa, io, &rfs);
    defer reopened.deinit(gpa);
    for (tree.subs) |sub| {
        var digest: [digest_len]u8 = undefined;
        std.crypto.hash.sha3.Sha3_256.hash(sub.bytes, &digest, .{});
        const rid = reopened.lookup(&digest) orelse return error.TreeNodeDroppedByPack;
        const out = (try reopened.content(gpa, rid)).?;
        defer gpa.free(out);
        try std.testing.expectEqualStrings(sub.bytes, out);
    }
}