Artifact [488551238b]

Artifact 488551238bd0be865a4d9008c19634c3fc0da8e2d27cbd6efc634d79944731bb:

• File src/logstore.zig — part of check-in [0df83f350a] at 2026-08-11 18:17:53 on branch sub-manifest — logstore: O(1) LRU eviction in PlaintextCache

  Replace the pick-the-minimum-stamp linear scan with an intrusive doubly linked recency list; get/put/evict are all constant-time. The scan was invisible at small occupancy and dominant at full: a pkgsrc clone kept the cache pinned at max_entries and spent over half its client CPU sweeping the map once per stored artifact. Measured: clone 12:38 -> 8:39 wall, client CPU 694s -> 463s; cloned store verifies with 0 faults. (user: dmestas, size: 264851) [annotate] [blame] [check-ins using]

//! The framed, append-only, content-addressed log over the `StorageFile`
//! capability. One file: a 64-byte header followed by 8-byte-aligned
//! frames, each self-verifying via a header checksum (over the fixed
//! fields) plus a whole-frame checksum (over header+payload). The header
//! checksum is what makes torn-vs-corrupt discrimination exact and
//! size-independent: without it, a torn write (a clean prefix, safe to
//! truncate) and a corrupted length field (unsafe to trust) both read as
//! "not enough of a frame," and nothing short of decoding payload bytes
//! that may not exist could tell them apart.
//!
//! No RFC section behind this file: native storage-engine design. The
//! oracle is this file's own framing invariants, checked by
//! sim/storage_sweep.zig's DST sweep: every acknowledged put stays
//! readable and hash-verifies; a torn tail truncates to exactly the last
//! complete frame; the index rebuilt on open matches the write-time
//! model; and corruption always surfaces as an error, never as silently
//! wrong bytes.
//!
//! A second, narrower checksum guards the frame's fixed 6-byte prefix
//! (`len|flags|hash_len`) on its own, in a 10-byte window whose size
//! never depends on `hash_len`. Without it, scanning a frame whose
//! `hash_len` byte is itself the bitflipped one has no way to know how
//! many bytes the rest of the header needs before it can check the
//! header checksum -- a corrupted `hash_len` that happens to demand more
//! bytes than the file has left reads exactly like a torn tail. The
//! prefix checksum removes that ambiguity: `hash_len` is proven correct
//! (or rejected outright) from a fixed-size read, before it is ever used
//! to size anything else.

const std = @import("std");
const storage = @import("storage.zig");
const crc32 = @import("crc32.zig");
const delta = @import("delta.zig");
/// The §4.1 zlib container codec — the leaf module shared with `repostate` and
/// `sync`. The store compresses artifacts through it (the hash still names the
/// UNCOMPRESSED bytes, sync §7.7 L1465) and inflates them back via
/// `decodeContainer` below.
const codec = @import("container.zig");

/// Local artifact id: dense, 1-based, assigned in intern/put order. 0 is
/// never issued (mirrors repostate.zig's MemStore).
pub const Rid = u64;

/// FILE header size: magic(4) | version u16(2) | hash_kind u8(1) |
/// reserved(53) | crc32(4).
pub const header_len: u64 = 64;
/// Explicit payload bound: keeps a single frame's read/decode work
/// bounded no matter how large a check-in gets.
pub const max_payload: u64 = 1 << 30;
pub const max_hash_len: u8 = 32;
pub const min_hash_len: u8 = 1;
/// The two COMPLETE artifact-name widths, as raw digests. artifact-format §3.2
/// L347-348: "Exactly two artifact-name forms exist: a 40-hexadecimal-digit
/// name, which denotes a SHA-1 hash [RFC3174], and a 64-hexadecimal-digit"
/// name — 20 and 32 raw bytes, algorithm inferred by length. The 64-hex form is
/// SHA3-256, not SHA-256 (§3.2 L349: "name, which denotes a SHA3-256 hash
/// [FIPS202]"); the two algorithms share a digest width and nothing else, so
/// the constant is named for the one the draft names. `isCompleteDigest` is
/// deliberately NOT applied to `put`/`putArtifact`, whose `min_hash_len ..
/// max_hash_len` bound stays opaque; it exists for the one caller whose key
/// must be a complete name and never a prefix (`putShun` — see `ShunRecord`).
pub const sha1_digest_len: usize = 20;
pub const sha3_256_digest_len: usize = 32;

/// Is `n` the length of a complete (non-abbreviated) artifact digest? A whole
/// digest is bijective with the complete hex name; any other width is a prefix
/// or a non-name, and cannot key a set the drafts describe as keyed by complete
/// artifact name. Rejecting every other width is not local taste: §3.2 L349-350
/// makes the two-forms rule normative — "A candidate name of any other length
/// MUST be rejected as invalid" — and the bijection carries that disposition
/// from the hex name to the digest form.
pub fn isCompleteDigest(n: usize) bool {
    return n == sha1_digest_len or n == sha3_256_digest_len;
}
/// Every frame's on-disk span is a multiple of this.
pub const frame_align: u64 = 8;
/// Frozen on-disk format version — the `.zit` ship-format identity. A reader
/// accepts only this exact stored version and rejects any other, older
/// included (`validateHeader`): the writer emits exactly this value, so a
/// header carrying anything else is not a zit-written file. An unknown frame
/// flag-kind is likewise a clean reject, never UB (`scanOneFrame`). Bumped
/// only on a layout change — a copied `.zit` file is a complete,
/// self-verifying repository, so this field is what absorbs any future layout
/// evolution; widening to a compatibility range belongs to that change, not
/// to a version this reader has never produced.
///
/// Adding `FrameKind.shun` did NOT bump this, deliberately: an older binary
/// meeting a shun frame fails the whole open with `error.Corrupt` (fail-closed
/// — see `FrameKind`), which is exactly what a tombstone needs, whereas a bump
/// would reject every shun-free log that binary itself wrote.
pub const version: u16 = 1;
/// Smallest possible frame span: hash_len=1, payload=0 ->
/// align8(26+1+0) = 32.
const min_frame: u64 = 32;
/// Explicit upper bound on delta-chain resolution: a `content()` on a
/// delta frame walks at most this many `delta_of` hops before rejecting
/// the chain as corrupt. Caps read recursion and read-amplification;
/// generous versus the short chains a real store produces.
pub const max_delta_chain: u8 = 32;
/// The deepest delta chain `putArtifact` will CREATE (write policy), kept
/// strictly below the READ bound `max_delta_chain` so a retained foreign
/// chain (a peer with no write cap) still has headroom to resolve. A base
/// already at this depth is stored full, so a locally-produced chain never
/// exceeds this and `content()`'s read walk stays shallow.
pub const delta_chain_write_max: u8 = 16;
/// fossil `N_NEIGHBOR` (content.c): the most delta bases `pickBase` will ever
/// try for one artifact. Bounds the write-path fan-out to at most this many
/// `delta.create` runs per put — the whole reason the candidate window is a
/// fixed slice, not the unbounded log.
pub const max_base_candidates: usize = 5;
/// The most frames `putArtifact`'s backward window walk will ever step over to
/// gather its chronological candidates, independent of phantom density. The
/// candidate cap (`max_base_candidates`) bounds deltas EXAMINED, not frames
/// SCANNED: a run of trailing phantom frames (a normal sync/clone artifact of
/// interned forward-references) is skipped by `continue` without filling a
/// candidate slot, so without a scan bound one put could step over the entire
/// store to gather < 5 real bases — hidden O(n^2) over N puts. This caps the
/// per-put scan to a fixed window: at most this many entries are inspected,
/// so the write-path cost stays O(1) in store size no matter how dense the
/// phantoms. Generous versus `max_base_candidates` so an ordinary phantom run
/// never starves the window of real bases; a base past the window only costs
/// compactness, never correctness (candidates are hints).
pub const max_base_scan: usize = 64;
/// fossil's per-source floor (content.c:908): a candidate whose reconstructed
/// content is shorter than this is skipped BEFORE `delta.create` — the cheap
/// length early-out. A delta's fixed framing overhead swamps any saving a
/// sub-50-byte source could offer, so running the encoder is wasted work.
const delta_min_source: u64 = 50;
/// The §5 accept ratio (fossil content.c:914): a delta is worth storing only
/// when it is STRICTLY under 75% of the uncompressed artifact — `delta.len * 4
/// < bytes.len * 3`. Integer math, no float (`delta_accept_den`/`num` = 4/3),
/// replacing the former looser 9/10 write-path threshold.
const delta_accept_num: u64 = 3;
const delta_accept_den: u64 = 4;

/// The §5 accept test in one pure predicate (fossil content.c:914): is a delta
/// of `delta_len` bytes worth storing for a `target_len`-byte artifact? True
/// iff strictly under 75% — `delta_len * 4 < target_len * 3`. Integer math on
/// u64, overflow-free by a wide margin: `target_len <= max_payload` (< 2^30, the
/// caller's asserted bound) so `target_len * 3 < 2^32`; `delta_len` is a delta
/// of a `<= max_payload` target, exceeding it only by bounded framing overhead,
/// so it stays far below 2^31 and `delta_len * 4` far below 2^33 — neither
/// product approaches u64's range. Exact at the integer boundary — 75/100
/// rejects, 74/100 accepts.
fn deltaProfitable(delta_len: usize, target_len: usize) bool {
    return @as(u64, delta_len) * delta_accept_den < @as(u64, target_len) * delta_accept_num;
}

/// The winning base and the raw (uncompressed) delta reconstructing the
/// artifact from it — `pickBase`'s result. The caller owns `delta` and stores
/// it (e.g. via `putDelta`), or frees it if it chooses not to.
pub const Pick = struct { base: Rid, delta: []u8 };

const magic = "zitL";

/// Payload encoding. `raw` = uncompressed bytes; `zlib` = a fossil zlib
/// container (4-byte BE uncompressed length + RFC1950 stream, sync §4.1);
/// `delta` = a zlib-contained delta whose `delta_of` is the byte offset of
/// the base frame, so resolving a chain link costs one seek and never a
/// second index lookup; `config` = a distinguished singleton repo-config
/// record, NOT content-addressed and never interned; `shun` = one repository
/// shun-table row, likewise never interned (see `ShunRecord`). Any other flag
/// byte is an unknown kind and rejected on open. The
/// content hash always names the UNCOMPRESSED, fully-reconstructed bytes, so
/// encoding never affects naming (sync §7.7 L1465).
///
/// FORMAT-VERSION NOTE for `shun = 4`: `version` deliberately stays 1. A `.zit`
/// carrying a shun frame, read by a binary that predates this kind, hits
/// `readFrameHeader`'s unknown-kind reject and fails the ENTIRE open with
/// `error.Corrupt` — fail-closed by design, never a skipped frame and never
/// UB. That is the behaviour a tombstone requires: an older binary MUST NOT
/// silently ignore one. Bumping `version` would instead reject every shun-free
/// log the older binary wrote, which is strictly worse.
pub const FrameKind = enum(u8) { raw = 0, zlib = 1, delta = 2, config = 3, shun = 4, _ };

/// One past the highest `FrameKind` value: the bound every NUMERIC flag-byte
/// check tests against, and the width of every per-kind table (`storestat`).
/// DERIVED by folding the enum's own fields, never hand-written — `FrameKind`
/// is non-exhaustive, so `switch` arms over it ARE compile-checked when a kind
/// is added but a numeric comparison against a named kind is NOT, and such a
/// comparison left behind turns every repository holding the new kind into
/// `error.Corrupt` for every consumer. Tied to the enum, a future kind widens
/// these checks by itself.
///
/// The density check is what makes `flags < frame_kind_count` exactly equal to
/// "a kind `FrameKind` names": with every value in `0 .. count-1` named there
/// is no hole for an unnamed kind to slip through, and a sparse future enum is
/// a compile error here rather than a silently-admitted flag byte.
pub const frame_kind_count: usize = blk: {
    const fields = @typeInfo(FrameKind).@"enum".fields;
    var highest: usize = 0;
    for (fields) |f| highest = @max(highest, @as(usize, f.value));
    if (fields.len != highest + 1) @compileError("FrameKind values must be dense from 0: a hole would let a numeric flag check admit an unnamed kind");
    break :blk highest + 1;
};

/// Repo-local config: default-branch name, project id, the sync project
/// code (§9.1 `projectcode`, the repository's shared identity — distinct
/// from `project_id`), and an opaque hash-policy byte the storage layer
/// round-trips but never interprets (exactly like the header's
/// `hash_kind`). It has a home IN the file as one distinguished `config`
/// frame; "last config frame wins" is derived by scan order, reusing the
/// log's own append-prefix property.
pub const max_config_str: usize = 255;
/// Fixed config record bound: `[hash_policy u8][branch_len u16][branch]
/// [pid_len u16][pid][pcode_len u16][pcode][login_len u16][login]
/// [secret_len u16][secret][url_len u16][url][seen_submanifest u8]
/// [enable_submanifests u8][submanifest_k u8][submanifest_auto_retune u8]
/// [shape_tag u8][shape_body]` — six length-prefixed strings after the policy
/// byte, then the four submanifest settings bytes and, when the analysis is
/// settled, a shape-tagged extension carrying its SHA-256 cost-shape digest.
/// The extension is absent from legacy and unsettled records, so readers
/// distinguish "scan once" from a real all-zero digest.
pub const max_config_payload: usize = 1 + 6 * (2 + max_config_str) + 4 + 1 + config_shape_digest_len; // = 1580

/// Shape tag for the config frame's optional trailing extension block, written
/// immediately after the four submanifest settings bytes. Tag 0 is never
/// written, so a zero byte left behind by a torn write cannot read as a known
/// shape.
///
/// **Compatibility direction: the tag protects future readers, not past ones.**
/// A reader that meets a shape it cannot read — a tag it does not know, or a tag
/// it does with a body width it does not expect — drops the extension and reports
/// the digest as absent: one re-scan, never `error.Corrupt`. The four settings
/// bytes, and the login, remote and project fields sharing the frame, still
/// parse. Under the bare width whitelist this replaces, every future trailing
/// width had to be added to every version in lockstep, and a version that
/// missed one read the entire config as corrupt.
///
/// The one-time cost runs the other way: a binary built from this branch before
/// the tag existed sees the 37-byte tagged block, misses its {1, 3, 4, 36}
/// whitelist, and reports `Corrupt`. That break is bounded and already paid in
/// kind — this branch's config frames are unreadable to any pre-branch binary
/// regardless, because the four settings bytes are written unconditionally —
/// and it is the last break the trailing block needs.
///
/// Rule for the next field: extend by defining a new tag, never by widening the
/// untagged settings block. Only data whose absence is safe belongs here, since
/// an older reader silently drops what it cannot name. For this digest, dropping
/// it means "re-run the analysis" — correct, only slower.
const config_shape_cost_shape_sha256: u8 = 1;
/// Body width of the `config_shape_cost_shape_sha256` extension.
const config_shape_digest_len: usize = 32;

/// Total trailing width of the untagged legacy digest block — the four settings
/// bytes plus a bare 32-byte digest, written before the tag existed. `parseConfig`
/// matches it by width, ahead of reading any tag, so the width is reserved forever.
const config_shape_legacy_width: usize = 4 + config_shape_digest_len;

/// Body width of every shape tag defined above, listed so the reserved width can
/// be checked against each one at compile time. A tagged block totalling
/// `config_shape_legacy_width` (a 31-byte body) would be matched as an untagged
/// legacy digest before its tag was ever examined, yielding a wrong digest — which
/// suppresses the scan the digest exists to skip. Registering such a shape is a
/// build failure rather than a runtime surprise. Add each new shape's body width
/// here when you define its tag.
const config_shape_body_widths = [_]usize{config_shape_digest_len};
comptime {
    for (config_shape_body_widths) |body| {
        std.debug.assert(4 + 1 + body != config_shape_legacy_width);
    }
}
/// A config frame is never keyed by hash — it is found only via
/// `config_offset`, never through `index`/`lookup` — so its 1-byte embedded
/// hash is inert bookkeeping (`appendFrame` requires some hash). It is never
/// an index key and never compared.
const config_sentinel: []const u8 = "\x00";

/// One repo config. Both slices borrow a backing buffer (`OwnedConfig`);
/// `hash_policy` is opaque to storage.
pub const Config = struct {
    hash_policy: u8,
    default_branch: []const u8,
    project_id: []const u8,
    /// sync §9.1 — the repository's shared project code (`projectcode`,
    /// an artifact-hash-shaped identifier). Empty on a repo that predates
    /// it; `init` generates one and a clone adopts the origin's. Defaulted
    /// so existing construction sites need not restate it.
    project_code: []const u8 = "",
    /// sync §6.3 — the login name a `zit push` authenticates as, or empty
    /// when no credential is stored. Paired with `login_secret_hex`.
    login_user: []const u8 = "",
    /// sync §6.3/DECISION-3 — the 40-hex derived shared secret
    /// (`SHA1(project-code "/" login "/" password)`), NOT the plaintext
    /// password: it is all a signer needs (§6.4) and is scoped to this
    /// repo+login. Empty when no credential is stored; exactly 40 bytes
    /// otherwise. Persisted only under 0600 file permissions by the caller.
    login_secret_hex: []const u8 = "",
    /// sync §4/§6 — the default peer URL a bare `zit sync`/`pull`/`push`
    /// targets, or empty when no remote is saved. A LOCAL, non-synced value:
    /// it never crosses the wire, it just spares the operator from retyping
    /// the endpoint. Anonymous only — a credential never lives in the URL.
    /// Defaulted so existing construction sites need not restate it.
    remote_url: []const u8 = "",
    /// draft-fossil-submanifests-00 §8.1 — "When the first S-bearing artifact
    /// is stored, an implementation SHOULD record a local seen-submanifest
    /// indicator so operational surfaces can warn before the setting is enabled
    /// or a pre-S client is granted access."
    ///
    /// It is also what makes §7.1's phantoms survive a restart at a cost of
    /// zero for every repository predating this draft. A phantom is an interned
    /// entry with no frame — `intern`'s "Phantoms never reach disk" — so an
    /// S-reference gap is re-derived by rescanning the artifacts that name it,
    /// and this flag is the durable answer to "is there anything to rescan?".
    /// False on every existing repository, so the rescan is unreachable there.
    /// One-way: set when the first S reference is stored, never cleared, since
    /// artifact names are immutable ([FossilArtifact] §7.6) and a stored
    /// S-bearing artifact cannot be retired.
    seen_submanifest: bool = false,
    /// draft-fossil-submanifests-00 §8.1 — the `enable-submanifests` setting,
    /// "a project-scope boolean, default false". The one operator act that lets
    /// this repository's writer emit S cards (§8.1) and the one that decides
    /// whether its server sends the §7.2 `avoid-submanifests` pragma.
    ///
    /// KNOWN GAP (recorded, not papered over): §6.3 makes both this and
    /// `submanifest_k` PROJECT scope — "configuration propagating to clones ...
    /// is REQUIRED: sharing by content address works only while every writer
    /// partitions identically". zit's config frame is repo-LOCAL: nothing in
    /// sync §9 carries it, and `clone` adopts only the origin's project code.
    /// So a clone of an enabled repository comes up disabled and unpinned, and
    /// an operator who enables it must re-pin the same k by hand or the two
    /// repositories' trees stop sharing sub-manifests. §6.4 bounds the damage
    /// to cost, not correctness — "Repositories pinned to different values
    /// exchange artifacts correctly ... so the propagation of Section 6.3 need
    /// be reliable only for cost" — which is why this ships as a stated gap
    /// rather than a blocker. Closing it needs a sync §9 config-card group and
    /// belongs with that work, not here.
    enable_submanifests: bool = false,
    /// draft-fossil-submanifests-00 §6.3 — the pinned partition parameter
    /// (`submanifest-k`), or 0 when this repository has none pinned.
    ///
    /// §6.3 fixes the legal range at 1..12 inclusive, so 0 is outside it and
    /// cannot collide with a pinned value; it is the encoding of "no pinned
    /// parameter", which §8.1 forbids enabling emission on. An out-of-range
    /// value that reaches a writer is NOT repaired here — §6.3 requires the
    /// writer to "fall back to emitting flat manifests rather than substitute a
    /// parameter of its own", which storage cannot do on its behalf, so this
    /// layer round-trips whatever byte it was given (like `hash_policy`) and
    /// `repository.submanifest` makes the decision.
    submanifest_k: u8 = 0,
    /// draft-fossil-submanifests-00 §6.4 — "A writer MAY re-tune
    /// automatically only while the repository's submanifest-auto-retune
    /// setting -- a project-scope boolean, default false -- is true. That
    /// setting MUST NOT be enabled by default or inferred from repository
    /// state; setting it true is the durable project-operator act authorizing
    /// subsequent automatic re-tunes." Manual `retune --k N` is unaffected by
    /// this setting either way — it is the automatic path's gate alone.
    submanifest_auto_retune: bool = false,
    /// Last locally settled cost-analysis input shape. Null forces one real
    /// scan; storage does not decide when that scan is settled and never
    /// substitutes an all-zero sentinel.
    /// Declared in terms of the wire constant, not a repeated literal, so the
    /// encode `@memcpy`, `max_config_payload` and this field cannot disagree.
    submanifest_settled_cost_shape: ?[config_shape_digest_len]u8 = null,
};

/// A `readConfig` result: the owned frame payload plus a `Config` whose
/// slices point into it. Mirrors `OwnedConfig`'s owned-backing shape.
pub const OwnedConfig = struct {
    backing: []u8,
    config: Config,

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

/// The repository shun table's per-row record.
///
/// SPEC PROVENANCE, stated exactly. sync §3.3 L518-525 defines the table:
/// "The repository shun table is a persistent, repository-local set keyed by
/// complete artifact name, distinct from peer-has, peer-lacks, and other
/// round-local state. Local administration MAY add or remove names. If /shun
/// configuration exchange is implemented, accepted rows update the same set;
/// Section 9.3 leaves their row serialization out of scope. Omitting that
/// exchange does not omit the local set, which is empty when no local or
/// exchanged entries supply names." Every property this type fixes is that
/// sentence and not local taste: DURABLE from "persistent", per-repository
/// from "repository-local", and keyed by the whole digest from "keyed by
/// complete artifact name" — which §7.7 L1435-1437 restates on the admission
/// side ("A target is shunned for admission when its complete name is in the
/// repository shun table"). §3.3 L527-530 is where the set is READ: "The
/// ordinary phantom scan excludes peer-lacks hashes, names in the repository
/// shun table, and private hashes unless private sync is enabled; shun-sha1
/// does not itself suppress an otherwise eligible gimme." §8.8 L1777-1779 and
/// §9.5 L1988-1992 re-reference that same filter. What remains out of scope is
/// only a row's WIRE serialization (§1 L185-188, §9.3 L1915-1917), and §1
/// L191-194 draws that line explicitly: "Omitting the optional /shun table
/// group omits its configuration exchange, not the repository-local shun set
/// defined in Section 3.3." The `epoch`/`reason` payload below is on the local
/// side of it — a row's serialization, which no draft fixes.
///
/// HISTORY: durability, repository-locality, and whole-digest keying were
/// derived here from the read-side filter alone, before §3.3 carried the
/// definition; the applied text then stated all three verbatim, and nothing in
/// this type had to change.
///
/// Shape: the FRAME's own `hash` field carries the shunned artifact's binary
/// digest, so a row is self-describing on a raw byte scan with no side table;
/// the payload carries only the operator metadata below. Like `config`, a shun
/// frame is never interned into `entries`/`index` — `lookup(hash)` keeps
/// meaning "the CONTENT frame for hash", never a tombstone.
pub const ShunRecord = struct {
    /// The shunned artifact's digest. Owned by the store (freed in `deinit`);
    /// borrowed by `shunList` callers until the next `putShun` or `deinit`.
    hash: []const u8,
    /// Unix seconds the shun was recorded, as supplied by the operator. Signed
    /// and round-tripped verbatim; storage never interprets or bounds it.
    epoch: i64,
    /// Free operator text, at most `max_shun_reason` bytes. Empty when none was
    /// given — a real value, not a missing field. Owned like `hash`.
    reason: []const u8,
};

/// Bound on a shun row's free-text reason. Fixed so a row's payload has a
/// stack-sized upper bound the open-time scan can read without allocating.
pub const max_shun_reason: usize = 200;
/// Fixed shun record bound: `[epoch i64 LE][reason_len u16 LE][reason]` — one
/// scalar plus one length-prefixed string, mirroring `max_config_payload`'s
/// shape. The scan rejects any `.shun` frame declaring more than this before
/// reading a byte of it.
pub const max_shun_payload: usize = 8 + 2 + max_shun_reason; // = 210

pub const OpenError = error{ OutOfMemory, Storage, Corrupt };
/// No read happens on the write path, so `Corrupt` can't occur here —
/// advertising it would be dead mechanism.
pub const PutError = error{ OutOfMemory, Storage };
/// `putShun`'s error set: `PutError` widened by the two input rules a shun row
/// must satisfy. They are RETURNED, not asserted, because an assert compiles to
/// nothing in `ReleaseFast`/`ReleaseSmall` and each violation is unrecoverable
/// once it lands: a wrong-width key writes a durable frame the scan then
/// rejects (`isCompleteDigest`), permanently un-openable and — being mid-file,
/// not at the tail — not truncatable by torn-tail recovery either; an over-long
/// reason overruns `encodeShun`'s fixed stack buffer. Both inputs reach here
/// from an operator (the shun CLI verb) and later from a peer, so neither is a
/// caller-bug-only class the way `putConfig`'s internally-generated fields are.
pub const ShunError = PutError || error{ BadKey, ReasonTooLong };
/// A phantom is purely in-memory bookkeeping: no I/O, so no `Storage`.
pub const InternError = error{OutOfMemory};
pub const ContentError = error{ OutOfMemory, Storage, Corrupt };
pub const CommitError = error{Storage};

/// The framed, content-addressed log. `open` rebuilds `entries`/`index`
/// from the file each time — there is no persistent side index yet.
/// O(n) on open is correct at this scale; a persistent, regenerable
/// sidecar is a noted future extension, built only if a benchmark shows
/// open time missing its target at large artifact counts.
pub const LogStore = struct {
    gpa: std.mem.Allocator,
    file: storage.StorageFile,
    hash_kind: u8,
    /// Structural end of the log: header + every complete frame written so
    /// far. Always a multiple of `frame_align`.
    end: u64,
    /// Total complete frames (content + config) seen by the scan, and the byte
    /// offset of the last one. Populated by `scan`; the derivation-cache key
    /// (`logKey`) reads them so a cached verb can validate the log's tail in
    /// O(1) instead of re-walking every frame. Zero/`header_len` on an empty log.
    frame_count: u64 = 0,
    last_frame_offset: u64 = header_len,
    entries: std.ArrayList(Entry) = .empty,
    index: std.StringHashMapUnmanaged(Rid) = .empty,
    /// Byte offset of the live (latest in scan order) config frame, or null
    /// when the log carries none. Config is found ONLY through this — never
    /// via `index`/`entries` (plan E).
    config_offset: ?u64 = null,
    /// The repository shun table (see `ShunRecord` for its normative
    /// definition), rebuilt by `scan` from every `.shun` frame. Deliberately
    /// the same two-part shape as `entries`/`index`: the list owns each row's
    /// `hash` and `reason` and holds them in APPEND order — which, since the
    /// log only grows, IS persisted order — so `shunList` is a borrow, not a
    /// sorted copy; the map keys on the same owned digest for an O(1)
    /// `isShunned`. Unlike `index` (whose keys `entries` owns), a duplicate
    /// digest here would leak, so `internShun` goes through `getOrPut` and
    /// frees the loser.
    shun_entries: std.ArrayList(ShunRecord) = .empty,
    shun_index: std.StringHashMapUnmanaged(usize) = .empty,
    /// Set when a rollback truncate fails and `end` can no longer be proven
    /// equal to the file's physical end. Every append refuses to run while it
    /// is set, so a later frame never lands at a diverged offset (the storage
    /// invariant: offset bookkeeping equals the file end). One-way: a store
    /// only ever goes healthy -> poisoned, never back.
    poisoned: bool = false,
    /// `pickBase`'s delta-base memo (§5.2 best-of-N). The candidate window is
    /// the tail of the log, so consecutive writes offer overlapping candidates
    /// whose chains share ancestors; without a memo every write re-walks them
    /// from the frame down. Bounded on bytes AND entries like every
    /// `PlaintextCache`, so it is a fixed ceiling, not growth.
    ///
    /// THE STORE OWNS IT, unlike the derivation memo a sweep caller owns, and
    /// the reason is invalidation. A memo keyed by frame offset goes stale the
    /// instant a rollback truncates the log and later writes reuse those
    /// offsets — and only the store sees that happen. Handing this to callers
    /// would hand them an obligation they cannot discharge; keeping it here
    /// puts the knowledge and the duty in the same place (`clearBaseMemo`).
    base_memo: PlaintextCache = .{},

    const Entry = struct {
        /// Owned duplicate; also the index's key (stable across `entries`
        /// reallocation since it's a separate heap allocation).
        hash: []u8,
        /// Frame start offset. Meaningless while `phantom`.
        offset: u64,
        /// On-disk (container/wire) payload length: raw bytes for `raw`,
        /// the zlib-container length for `zlib`/`delta`.
        payload_len: u32,
        /// Payload encoding of this frame. `.raw` while `phantom` (no frame).
        kind: FrameKind,
        /// A phantom was `intern`ed but never `put`: no frame on disk.
        phantom: bool,
        /// Delta-chain depth: 0 for a phantom or a non-delta frame; one more
        /// than the base's depth for a `.delta` frame. Set at `insert`/`scan`
        /// and read by `putArtifact` to enforce `delta_chain_write_max`, so a
        /// locally-created chain never grows past the write cap.
        depth: u8 = 0,
    };

    /// Number of unique artifact holdings rebuilt from the verified log scan.
    /// Config frames are deliberately excluded because they are not artifacts.
    pub fn artifactCount(self: *const LogStore) u64 {
        return @intCast(self.entries.items.len);
    }

    /// Largest retained delta-chain depth among real artifact holdings.
    pub fn maxArtifactDeltaDepth(self: *const LogStore) u8 {
        var max_depth: u8 = 0;
        for (self.entries.items) |entry| {
            if (!entry.phantom) max_depth = @max(max_depth, entry.depth);
        }
        return max_depth;
    }

    /// Opens for append, creating or repairing the file when necessary, and
    /// rebuilds the in-memory index by scanning every complete frame.
    pub fn open(gpa: std.mem.Allocator, file: storage.StorageFile, hash_kind: u8) OpenError!LogStore {
        const file_size = file.size() catch return error.Storage;
        var self: LogStore = .{ .gpa = gpa, .file = file, .hash_kind = hash_kind, .end = header_len };
        errdefer self.deinit(gpa);

        if (file_size == 0) {
            try writeHeader(file, hash_kind);
            std.debug.assert(self.end % frame_align == 0);
            return self;
        }
        if (file_size < header_len) {
            // Torn header: nothing before byte `header_len` can be trusted
            // as a frame boundary either, so start clean.
            file.truncate(0) catch return error.Storage;
            try writeHeader(file, hash_kind);
            std.debug.assert(self.end % frame_align == 0);
            return self;
        }
        self.hash_kind = try validateHeader(file);
        // Slice 1: the append path scans header-only. `openReadOnly` — the path
        // every CLI verb reaches — still verifies payloads, so no guarantee has
        // moved yet; this is the measurable half of the change.
        try scan(&self, true, .headers_only);
        std.debug.assert(self.end % frame_align == 0);
        return self;
    }

    /// Opens an existing log strictly for reads. Unlike `open`, this path never
    /// creates, rewrites, or truncates storage: an empty file, torn header, or
    /// torn frame tail is corruption. Complete-frame checksum and encoding
    /// faults are rejected by the same scanner used by the append path.
    pub fn openReadOnly(gpa: std.mem.Allocator, file: storage.StorageFile) OpenError!LogStore {
        const file_size = file.size() catch return error.Storage;
        if (file_size < header_len) return error.Corrupt;

        var self: LogStore = .{ .gpa = gpa, .file = file, .hash_kind = 0, .end = header_len };
        errdefer self.deinit(gpa);
        self.hash_kind = try validateHeader(file);
        try scan(&self, false, .headers_only);
        std.debug.assert(self.end == file_size);
        std.debug.assert(self.end % frame_align == 0);
        return self;
    }

    pub fn deinit(self: *LogStore, gpa: std.mem.Allocator) void {
        for (self.entries.items) |e| gpa.free(e.hash);
        self.entries.deinit(gpa);
        self.index.deinit(gpa);
        // The shun list owns both strings of every row; `shun_index` borrows
        // the same `hash` allocation as its key, so it frees nothing itself.
        for (self.shun_entries.items) |s| {
            gpa.free(s.hash);
            gpa.free(s.reason);
        }
        self.shun_entries.deinit(gpa);
        self.shun_index.deinit(gpa);
        self.base_memo.deinit(gpa);
        self.* = undefined;
    }

    /// Forget every memoized delta-base plaintext.
    ///
    /// THE INVARIANT: no `base_memo` key is ever at or above `self.end`. The
    /// memo is keyed by frame offset, and a truncation hands the offsets it
    /// frees to the next write — so an entry left above the new `end` would
    /// answer for the frame that used to live there, the one way this memo
    /// could change an answer instead of only the work.
    ///
    /// Only `rollbackArtifactBatch` can break it, because only it removes
    /// frames that an EARLIER put in the same batch already read as a base
    /// candidate. The other two truncation sites (`insert`'s bookkeeping
    /// failure, `appendFrame`'s torn append) roll back a frame that was
    /// appended microseconds earlier and never read, so nothing names its
    /// offset and the invariant already holds — they call nothing, and pay
    /// nothing. Clearing wholesale rather than dropping just the freed range
    /// keeps this to one line on a path that only runs when a write has
    /// already failed; the cost is a re-walk of chains the next write would
    /// have hit.
    fn clearBaseMemo(self: *LogStore) void {
        self.base_memo.clear(self.gpa);
    }

    /// Check THE INVARIANT, at every truncation site that can reach it.
    ///
    /// Vacuous after `rollbackArtifactBatch`, which has just cleared the memo
    /// wholesale — but NOT at the other two sites, where nothing is cleared and
    /// "no memo names this frame" is load-bearing reasoning rather than an
    /// operation. Those are exactly the places a future edit could invalidate
    /// silently, and the failure mode is a memo answering with the bytes of a
    /// frame that no longer exists, so the invariant is checked and not merely
    /// stated. A walk bounded by the memo's entry cap, on paths that only run
    /// when a write has already failed.
    fn assertBaseMemoBelowEnd(self: *const LogStore) void {
        if (!std.debug.runtime_safety) return;
        var it = self.base_memo.map.keyIterator();
        while (it.next()) |k| std.debug.assert(k.* < self.end);
    }

    /// The append-only-log identity a derivation cache is keyed on: the
    /// structural end offset, the total complete-frame count, the byte offset of
    /// the last complete frame, and that frame's fixed-header CRC. The log only
    /// ever grows by whole frames, so a cache built at key K stays valid exactly
    /// while the live log still matches K: an equal `end` (file size) proves no
    /// append and no truncation changed the length, and the last frame still
    /// sitting at `last_frame_offset` with header CRC `last_header_crc` and
    /// ending at `end` pins the truncate-then-reappend-to-the-same-length corner
    /// that `end` alone would miss. `frame_count` is recorded for the identity
    /// but, under the append-only invariant, is subsumed by `end` — see
    /// `validateTail`, which stays O(1) by checking only the tail. Whole-log
    /// payload integrity remains `openReadOnly`'s job on the miss/rebuild path,
    /// so a cache hit will not detect interior bitrot that preserves the file
    /// size and tail frame; deleting the `.zit-cache` (or any append, which
    /// forces a rebuild) restores full-log detection.
    pub const LogKey = struct {
        end: u64,
        frame_count: u64,
        last_frame_offset: u64,
        last_header_crc: u32,
    };

    pub const KeyError = error{ Storage, Corrupt };

    /// The `LogKey` for this already-scanned store — the value a cache write
    /// records. `end`, `frame_count`, and `last_frame_offset` come from the scan
    /// (`open`/`openReadOnly` already walked every frame); only the last frame's
    /// header CRC needs one extra small read, so this is O(1), never a re-walk.
    /// An empty log (no frames) keys on `end == header_len` alone. Never mutates
    /// storage.
    pub fn logKey(self: *const LogStore) KeyError!LogKey {
        if (self.frame_count == 0) {
            return .{ .end = self.end, .frame_count = 0, .last_frame_offset = self.end, .last_header_crc = 0 };
        }
        const info = try tailHeaderInfo(self.file, self.last_frame_offset);
        // The recorded last frame must end exactly at the structural end.
        std.debug.assert(info.frame_end == self.end);
        return .{
            .end = self.end,
            .frame_count = self.frame_count,
            .last_frame_offset = self.last_frame_offset,
            .last_header_crc = info.header_crc,
        };
    }

    /// The `LogKey` describing the PREFIX of this log ending at the frame that
    /// contains `rid`, inclusive — the value `validatePrefix` later checks.
    /// Unlike `logKey`, which reuses the open scan's tail bookkeeping, this
    /// walks frame headers from the start to COUNT the prefix's frames:
    /// config and shun frames carry no rid, so the count cannot be derived
    /// from the artifact index alone. O(prefix frames) of small header reads
    /// — a builder-time capture, never on the open path.
    pub fn logKeyAt(self: *const LogStore, rid: Rid) KeyError!LogKey {
        const target = self.offsetOf(rid);
        const prefix_end = target + self.frameSpanOf(rid);
        std.debug.assert(prefix_end > header_len and prefix_end <= self.end);
        var offset: u64 = header_len;
        var count: u64 = 0;
        var target_crc: u32 = 0;
        while (offset < prefix_end) {
            const info = try tailHeaderInfo(self.file, offset);
            count += 1;
            if (offset == target) target_crc = info.header_crc;
            offset = info.frame_end;
        }
        // The walk must land exactly on the target frame's end, and the target
        // offset must have been one of the boundaries it visited.
        if (offset != prefix_end or count == 0) return error.Corrupt;
        return .{
            .end = prefix_end,
            .frame_count = count,
            .last_frame_offset = target,
            .last_header_crc = target_crc,
        };
    }

    /// O(1) cache-freshness check: is `file` still the append-only log `key` was
    /// computed over? True iff the file size equals `key.end` and — for a
    /// non-empty log — the last frame still occupies `key.last_frame_offset`
    /// with the same header CRC and end. Reads at most one frame header (plus a
    /// stat); no full scan, so a cached verb stays sub-second even on a
    /// multi-gigabyte log. Any I/O failure, a shrunk/grown file, or a changed
    /// tail is a miss (false), never an error — a cache is advisory.
    pub fn validateTail(file: storage.StorageFile, key: LogKey) bool {
        const file_size = file.size() catch return false;
        if (file_size != key.end) return false; // any append or truncation changes the length
        if (key.frame_count == 0) return key.end == header_len; // empty log: length is the whole identity
        // The recorded tail offset must be a real interior frame boundary.
        if (key.last_frame_offset < header_len or key.last_frame_offset >= file_size) return false;
        const info = tailHeaderInfo(file, key.last_frame_offset) catch return false;
        return info.frame_end == key.end and info.header_crc == key.last_header_crc;
    }

    /// Whether `key` describes an intact PREFIX of the live log: the frame it
    /// records as last still verifies at its offset with the same header CRC
    /// and still ends exactly at `key.end`, while the file itself may have
    /// grown past it. This is `validateTail` with the length equality relaxed
    /// to "at least" -- the discriminator between "the log was appended to"
    /// (extendable) and "the log was rewritten" (not). A rewrite -- pack, shun
    /// --force -- replaces the file wholesale, so the old tail frame no longer
    /// verifies at its recorded offset and this answers false.
    pub fn validatePrefix(file: storage.StorageFile, key: LogKey) bool {
        const file_size = file.size() catch return false;
        if (file_size < key.end) return false;
        if (key.frame_count == 0) return key.end == header_len;
        if (key.last_frame_offset < header_len or key.last_frame_offset >= key.end) return false;
        const info = tailHeaderInfo(file, key.last_frame_offset) catch return false;
        return info.frame_end == key.end and info.header_crc == key.last_header_crc;
    }

    /// A fingerprint over the repository shun table, for cache extension to
    /// detect that shun state moved. The derive masks shunned artifacts at
    /// read time, so a record distilled BEFORE a shun is a record the body
    /// must no longer carry -- extension across a shun change would keep it,
    /// and the only sound answer is a full rebuild. Order-sensitive by
    /// construction: the table is rebuilt from frames in log order, so equal
    /// fingerprints mean equal tables.
    pub fn shunFingerprint(self: *const LogStore) u32 {
        var crc = crc32.Crc32.init();
        for (self.shun_entries.items) |s| {
            crc.update(s.hash);
            crc.update(std.mem.asBytes(&s.epoch));
            crc.update(s.reason);
        }
        return crc.final();
    }

    const TailInfo = struct { frame_end: u64, header_crc: u32 };

    /// Read and CRC-verify the fixed header of the frame at `offset`, returning
    /// its end offset and header CRC. A torn or corrupt header is `error.Corrupt`
    /// (`readFrameHeader` already validates the prefix and header CRCs).
    fn tailHeaderInfo(file: storage.StorageFile, offset: u64) KeyError!TailInfo {
        const fh = switch (readFrameHeader(file, offset) catch |e| switch (e) {
            error.Storage => return error.Storage,
            error.Corrupt => return error.Corrupt,
        }) {
            .torn => return error.Corrupt, // a recorded frame offset is never torn
            .header => |h| h,
        };
        var hcrc = crc32.Crc32.init();
        hcrc.update(fh.header_bytes[0 .. 18 + @as(usize, fh.hash_len)]);
        const frame_len = frameSpan(fh.hash_len, fh.payload_len);
        return .{ .frame_end = offset + frame_len, .header_crc = hcrc.final() };
    }

    /// Clone = copy + derive: byte-copies `source` into `dest`, then opens
    /// `dest` — rebuilding the index and recovering
    /// config with no network (no network capability is even in scope here).
    /// The result is an independent, equal repository: same artifact set, same
    /// config, byte-identical log. A damaged copy needs no clone-specific
    /// error — `open`'s torn-tail truncation drops exactly the incomplete
    /// tail, so the copy degrades gracefully through the shared derive path.
    /// Precondition (documented, not asserted — assertions carry no I/O):
    /// `dest` is empty. `hash_kind` is a don't-care: `dest` is non-empty after
    /// the copy, so `open` reads the copied header's stored kind.
    pub fn cloneInto(
        gpa: std.mem.Allocator,
        source: storage.StorageFile,
        dest: storage.StorageFile,
    ) OpenError!LogStore {
        const src_size = source.size() catch return error.Storage;
        var chunk: [64 * 1024]u8 = undefined;
        var copied: u64 = 0;
        // Bounded: `copied` rises by `n > 0` every pass, so the loop runs at
        // most ceil(src_size / chunk.len) times before `copied == src_size`.
        while (copied < src_size) {
            const n = source.pread(copied, &chunk) catch return error.Storage;
            // A 0-count before EOF on a stable file is a device failure, not a
            // boundary — return rather than spin (no unbounded loop).
            if (n == 0) return error.Storage;
            dest.append(chunk[0..n]) catch return error.Storage;
            copied += n;
        }
        std.debug.assert(copied == src_size);
        return open(gpa, dest, 0);
    }

    /// Resolves `hash` to its rid, creating a phantom (no I/O) when
    /// unknown. Phantoms never reach disk.
    pub fn intern(self: *LogStore, hash: []const u8) InternError!Rid {
        std.debug.assert(hash.len >= min_hash_len and hash.len <= max_hash_len);
        if (self.index.get(hash)) |rid| return rid;
        try self.entries.ensureUnusedCapacity(self.gpa, 1);
        const owned = try self.gpa.dupe(u8, hash);
        errdefer self.gpa.free(owned);
        const rid: Rid = self.entries.items.len + 1;
        try self.index.put(self.gpa, owned, rid);
        self.entries.appendAssumeCapacity(.{ .hash = owned, .offset = 0, .payload_len = 0, .kind = .raw, .phantom = true });
        return rid;
    }

    /// Stores `bytes` verbatim (a `raw` frame) under `hash`. Idempotent for
    /// an already-content-bearing hash (returns the existing rid, appends
    /// nothing); upgrades a phantom in place.
    pub fn put(self: *LogStore, hash: []const u8, bytes: []const u8) PutError!Rid {
        std.debug.assert(bytes.len <= max_payload);
        return self.insert(hash, bytes, .raw, 0, 0);
    }

    /// Store `bytes` under `hash`, delta-encoded against the size-minimizing
    /// base `pickBase` (§5) chooses from a small candidate window, else as a
    /// §4.1 zlib container. The window is the caller's lineage `base` hint
    /// (priority) plus the most recent content frames in log order — a bounded
    /// best-of-N search (fossil `extra_deltification`), accepting a delta only
    /// when it is under 75% of the artifact. Every candidate is a HINT, never a
    /// requirement: a null/phantom/config/unreadable/over-deep/unprofitable
    /// base is silently skipped — there is no error path for an ineligible
    /// base, only a compactness difference. Idempotent + phantom-upgrade like
    /// `put`/`putZlib`: an already content-bearing `hash` keeps its stored
    /// form and no encode work runs. This one method hides the whole storage
    /// policy so callers pass only the base hint they naturally hold.
    /// One immutable artifact destined for an all-or-nothing append group.
    pub const ArtifactBatchItem = struct {
        hash: []const u8,
        bytes: []const u8,
        base: ?Rid = null,
    };

    /// Stores an ordered artifact group as one visibility unit. Any member
    /// failure restores the exact prior log prefix and index before returning.
    pub fn putArtifactBatch(self: *LogStore, items: []const ArtifactBatchItem) PutError!void {
        // An empty group is trivially all-or-nothing: nothing to append, and the
        // store is already byte-identical to its prior state. This was an assert
        // until trunk's release-assert guard named the shape — a panic is the
        // wrong answer to "no artifacts", and under --release=fast the assert
        // vanished and the snapshot/restore ran over an empty set anyway.
        if (items.len == 0) return;
        const snapshot = ArtifactBatchSnapshot{
            .end = self.end,
            .frame_count = self.frame_count,
            .last_frame_offset = self.last_frame_offset,
            .entries_len = self.entries.items.len,
        };
        var phantoms: std.ArrayList(PhantomSnapshot) = .empty;
        defer phantoms.deinit(self.gpa);
        try phantoms.ensureTotalCapacity(self.gpa, items.len);
        for (items) |item| {
            const rid = self.index.get(item.hash) orelse continue;
            const entry = self.entries.items[@intCast(rid - 1)];
            if (!entry.phantom) continue;
            phantoms.appendAssumeCapacity(.{ .rid = rid, .entry = entry });
        }
        for (items) |item| {
            _ = self.putArtifact(item.hash, item.bytes, item.base) catch |err| {
                self.rollbackArtifactBatch(snapshot, phantoms.items) catch return error.Storage;
                return err;
            };
        }
    }

    const ArtifactBatchSnapshot = struct {
        end: u64,
        frame_count: u64,
        last_frame_offset: u64,
        entries_len: usize,
    };

    const PhantomSnapshot = struct {
        rid: Rid,
        entry: Entry,
    };

    fn rollbackArtifactBatch(
        self: *LogStore,
        snapshot: ArtifactBatchSnapshot,
        phantoms: []const PhantomSnapshot,
    ) PutError!void {
        // Before the truncate, so the memo is clean on the poisoned path too:
        // every frame this rollback removes may already have been memoized as
        // a base by a later put in the same batch, and the offsets it frees are
        // the ones the next batch writes over.
        self.clearBaseMemo();
        self.file.truncate(snapshot.end) catch {
            self.poisoned = true;
            return error.Storage;
        };
        while (self.entries.items.len > snapshot.entries_len) {
            const entry = self.entries.pop() orelse unreachable;
            const removed = self.index.remove(entry.hash);
            std.debug.assert(removed);
            self.gpa.free(entry.hash);
        }
        for (phantoms) |prior| self.entries.items[@intCast(prior.rid - 1)] = prior.entry;
        self.end = snapshot.end;
        self.frame_count = snapshot.frame_count;
        self.last_frame_offset = snapshot.last_frame_offset;
        self.assertBaseMemoBelowEnd();
    }

    pub fn putArtifact(self: *LogStore, hash: []const u8, bytes: []const u8, base: ?Rid) PutError!Rid {
        std.debug.assert(hash.len >= min_hash_len and hash.len <= max_hash_len);
        std.debug.assert(bytes.len <= max_payload);
        // Idempotent: return before any encode work when the artifact is held.
        const held: ?Rid = self.index.get(hash);
        if (held) |rid| {
            if (!self.entries.items[@intCast(rid - 1)].phantom) return rid;
        }
        var pri_buf: [1]Rid = undefined;
        const priority: []const Rid = if (base) |b| blk: {
            pri_buf[0] = b;
            break :blk pri_buf[0..1];
        } else pri_buf[0..0];
        var cand_buf: [max_base_candidates]Rid = undefined;
        const n = self.gatherCandidates(priority, cand_buf[0..]);
        if (try self.pickBase(self.gpa, hash, bytes, cand_buf[0..n])) |pick| {
            defer self.gpa.free(pick.delta);
            return try self.putDelta(hash, pick.base, pick.delta);
        }
        return self.putZlib(hash, bytes);
    }

    /// Fill `cand_buf` with the best-of-N base candidate window (§5.2) for a
    /// write: the caller's `priority` hints first (in order — a lineage parent,
    /// or the §6.2 same-path predecessors pack derives), then the most recent
    /// content frames in log order. Insertion order IS log order, so a backward
    /// walk over `entries` yields the chronological window for free — no manifest
    /// parsing on the write path. Only the FIRST `max_base_candidates` priority
    /// elements are observed — trailing hints are not examined, which is the
    /// API behavior and not merely an optimization (see the bound below).
    /// Observed priority candidates are deduped against each other and the
    /// chronological walk skips any rid already listed. Returns the count written
    /// (≤ `max_base_candidates`); `pickBase` re-checks full eligibility, so this
    /// only gathers plausible rids. Shared by `putArtifact` and
    /// `putArtifactPacked` so the two write paths pick from an identical window
    /// shape.
    fn gatherCandidates(self: *LogStore, priority: []const Rid, cand_buf: []Rid) usize {
        std.debug.assert(cand_buf.len >= max_base_candidates);
        // A STRUCTURAL bound, not an assert, and the distinction is the whole
        // point. The loop below caps entries ADDED, not iterations: a duplicate
        // hits `ridListed` and `continue`s without advancing `n`, so the
        // fill-check never fires and a duplicate-heavy slice walks the caller's
        // whole array — with `ridListed` linear in `n`, that is the caller's
        // length times the window. The bound that used to stand here was
        // `assert(priority.len <= max_base_candidates)`, which compiles to
        // NOTHING in ReleaseFast/ReleaseSmall: exactly the build where an
        // unbounded scan matters and exactly the build with no guard. Slicing the
        // prefix holds in every mode. Both callers today (putArtifact's one hint,
        // pack's fixed-size buffer) are inside the window regardless.
        const priority_prefix = priority[0..@min(priority.len, max_base_candidates)];
        var n: usize = 0;
        // Priority hints fill first, in caller order, bounded to the window and
        // deduped so a hint that repeats (or coincides across lanes) costs no slot.
        for (priority_prefix) |b| {
            std.debug.assert(b >= 1 and b <= self.entries.items.len);
            if (ridListed(cand_buf[0..n], b)) continue;
            cand_buf[n] = b;
            n += 1;
        }
        // Newest-first; `idx` strictly decreases to 0. Two independent bounds
        // stop the walk: `n < max_base_candidates` (window full) and
        // `scanned < max_base_scan` (frames inspected). The scan cap is the one
        // that bites under load — a run of trailing phantom frames `continue`s
        // without filling a slot, so without it a single put could step over the
        // whole store to gather < 5 candidates (hidden O(n^2) over N puts). The
        // cap holds the per-put scan to a fixed window regardless of phantom
        // density; a real base past the window only costs compactness. A held
        // artifact that reaches this walk is necessarily a phantom (a
        // content-bearing hit returns early in the caller), so the phantom skip
        // already excludes "self"; the priority-dedup skips any hint already
        // listed.
        var idx: usize = self.entries.items.len;
        var scanned: usize = 0;
        while (idx > 0 and n < max_base_candidates and scanned < max_base_scan) {
            idx -= 1;
            scanned += 1;
            const rid: Rid = @intCast(idx + 1);
            if (self.entries.items[idx].phantom) continue;
            if (ridListed(cand_buf[0..n], rid)) continue; // priority hint already listed
            cand_buf[n] = rid;
            n += 1;
        }
        return n;
    }

    /// Whether `rid` already appears in `listed` — the bounded (≤
    /// `max_base_candidates`) dedup guard `gatherCandidates` runs so a priority
    /// hint and the chronological walk never offer the same base twice.
    fn ridListed(listed: []const Rid, rid: Rid) bool {
        for (listed) |r| if (r == rid) return true;
        return false;
    }

    /// The outcome of `putArtifactPacked`: either a fresh re-encoding won the
    /// §6.1 monotonic-shrink comparison and was written (`rewrote`), or the
    /// existing source frame is smaller-or-equal and NOTHING was written
    /// (`kept`) — the caller then copies that existing frame verbatim.
    pub const PackPut = union(enum) { rewrote: Rid, kept };

    /// `putArtifact` with the §6.1 monotonic-shrink guard for `zit pack`: gather
    /// the same best-of-N window, run `pickBase`, and encode the winning form —
    /// but COMMIT the new frame only when its on-disk span is STRICTLY smaller
    /// than `floor_span` (the artifact's existing stored-frame span in the source
    /// store, from `frameSpanOf`). A tie or a larger re-encoding returns `.kept`
    /// and writes nothing, so the caller re-stores the existing frame — often a
    /// compact wire delta whose base fossil already chose well — verbatim.
    /// `pickBase`'s <75%-of-FULL accept is a FIRST-encoding rule; in pack the
    /// reference size is the EXISTING frame, not the full artifact, so this guard
    /// keeps every REWRITTEN frame non-worsening RELATIVE TO THE FLOOR THE CALLER
    /// DECLARES: this frame is size-neutral or smaller than `floor_span`, never
    /// larger. That is a per-frame guarantee only while the caller's floor is a
    /// frame it could actually re-store instead; a caller with no keepable frame
    /// passes `null` and gets the smallest encoding available, which may still
    /// exceed the source frame it replaces. `null` is a distinct STATE, not an
    /// in-band span: "I have no frame to keep" and "the frame I would keep spans
    /// N bytes" are different claims, and an in-band sentinel in the span's own
    /// type could not tell a caller with no keepable frame from one that computed
    /// a garbage span — the second silently forces a rewrite of a frame that
    /// should have been kept. `zit pack` passes `null` for exactly one shape: a
    /// delta whose base the reclaim filter dropped, whose source frame is
    /// unre-storable. (Outside this function pack's
    /// `copySrcFrame` §3.3 deep-chain fallback can likewise enlarge a KEPT frame;
    /// see pack.zig.) `priority` is the caller's ordered base hints
    /// (a lineage parent, or the §6.2 same-path predecessors), offered ahead of
    /// the chronological window. Only its first `max_base_candidates` elements
    /// are observed; a longer slice is accepted and its tail ignored, so a caller
    /// must put its best hints first. Idempotent + phantom-upgrade like
    /// `putArtifact` (a re-put of a held artifact returns `.rewrote` with its
    /// existing rid).
    pub fn putArtifactPacked(self: *LogStore, hash: []const u8, bytes: []const u8, priority: []const Rid, floor_span: ?u64) PutError!PackPut {
        std.debug.assert(hash.len >= min_hash_len and hash.len <= max_hash_len);
        std.debug.assert(bytes.len <= max_payload);
        if (floor_span) |fs| std.debug.assert(fs >= min_frame);
        // Idempotent: return before any encode work when the artifact is held.
        if (self.index.get(hash)) |rid| {
            if (!self.entries.items[@intCast(rid - 1)].phantom) return .{ .rewrote = rid };
        }
        var cand_buf: [max_base_candidates]Rid = undefined;
        const n = self.gatherCandidates(priority, cand_buf[0..]);
        // The best new encoding, compressed to its on-disk container form so its
        // span is comparable to the existing frame's. A profitable delta compares
        // its zlib-contained delta; otherwise the fallback is a full zlib
        // container (exactly what `putZlib` would store).
        if (try self.pickBase(self.gpa, hash, bytes, cand_buf[0..n])) |pick| {
            defer self.gpa.free(pick.delta);
            const container = codec.encode(self.gpa, pick.delta, .default) catch |e| switch (e) {
                error.OutOfMemory => return error.OutOfMemory,
                error.InputTooLarge => unreachable, // pick.delta <= max_payload + bounded overhead
            };
            defer self.gpa.free(container);
            // Keep the smaller; the existing frame wins ties (strict `<`). No
            // floor ⇒ nothing to keep, so the new encoding always wins.
            if (beatsFloor(frameSpan(hash.len, container.len), floor_span)) {
                return .{ .rewrote = try self.putDeltaContainer(hash, pick.base, container) };
            }
            return .kept;
        }
        const container = codec.encode(self.gpa, bytes, .default) catch |e| switch (e) {
            error.OutOfMemory => return error.OutOfMemory,
            error.InputTooLarge => unreachable, // asserted bytes.len <= max_payload (< 2^32)
        };
        defer self.gpa.free(container);
        if (beatsFloor(frameSpan(hash.len, container.len), floor_span)) {
            return .{ .rewrote = try self.putContainer(hash, container) };
        }
        return .kept;
    }

    /// The §6.1 monotonic-shrink comparison, in one place for both of
    /// `putArtifactPacked`'s encodings: a new frame of `span` bytes is written
    /// iff it is STRICTLY smaller than the frame the caller would otherwise keep.
    /// A caller with no keepable frame (`null`) has nothing to lose by writing,
    /// so it always writes — which is why `.kept` is impossible under a null
    /// floor.
    fn beatsFloor(span: u64, floor_span: ?u64) bool {
        const floor = floor_span orelse return true;
        return span < floor;
    }

    /// The §5 delta-base policy as ONE deep module: choose the size-minimizing
    /// base for storing `bytes` (named `hash`) from `candidates`, or null ⇒
    /// store the full form. `candidates` are RIDs the caller believes are
    /// plausible bases (lineage parent, chronological neighbors, same-path
    /// predecessors), in priority order; at most `max_base_candidates` are
    /// examined. This owns the WHOLE decision — the caller owns only WHICH
    /// candidates to offer:
    ///   * best-of-N: run `delta.create` (delta.zig:294) against every ELIGIBLE
    ///     candidate and keep the SMALLEST delta (ties resolve to the earlier,
    ///     higher-priority candidate via the strict `<`);
    ///   * eligibility: content-bearing (non-phantom), non-config, chain-depth
    ///     headroom (`depth < delta_chain_write_max`), and never the artifact
    ///     itself (a self-delta is degenerate) — all free checks, run before
    ///     any I/O so an ineligible candidate costs nothing;
    ///   * cheap length early-out: a candidate whose content is under
    ///     `delta_min_source` bytes is skipped BEFORE `delta.create` (fossil
    ///     content.c:908);
    ///   * accept iff the smallest delta clears the 75% bar (`deltaProfitable`,
    ///     fossil content.c:914).
    /// On accept, returns the winning base rid and the raw delta bytes (caller
    /// owns and stores them, e.g. via `putDelta`); otherwise null. Ineligible /
    /// unreadable candidates are silently skipped — a base is a hint, never a
    /// requirement, so there is no error path for a bad candidate. Only
    /// `OutOfMemory`/`Storage` escape.
    pub fn pickBase(
        self: *LogStore,
        gpa: std.mem.Allocator,
        hash: []const u8,
        bytes: []const u8,
        candidates: []const Rid,
    ) PutError!?Pick {
        std.debug.assert(hash.len >= min_hash_len and hash.len <= max_hash_len);
        std.debug.assert(bytes.len <= max_payload);

        var best: ?[]u8 = null;
        var best_base: Rid = 0;
        // Free the retained best delta if an allocation/storage fault escapes
        // mid-search; the success and reject paths hand it off / free it
        // explicitly, so this only fires on an error return.
        errdefer if (best) |b| gpa.free(b);

        // Candidates are read through the store's base memo (`base_memo`).
        // The window is the tail of the log, so consecutive writes offer
        // overlapping candidates over shared delta ancestors; reconstructing
        // each from its frame down re-walks the same chains once per candidate
        // per write. Measured on the wapp import: 52,232 frame reads without
        // the memo, 26,289 with it, for identical output.
        //
        // IT THINS NO CHECK. `contentAtOffset` consults a memo only for a
        // `delta_of` BASE, so every candidate's own frame is still read,
        // checksum-verified and hash-verified in full on every call; a
        // memoized base was itself produced by a checksum-verified read of
        // that same frame earlier in this process. This is the shipped
        // `contentCached` contract — it changes the work, never the answer —
        // and the one way it could change an answer, a truncated-away offset
        // being reused, is closed by `clearBaseMemo`.
        //
        // Candidate plaintext is read and freed with the STORE's allocator,
        // not the caller's `gpa`. The memo outlives this call and is freed by
        // `LogStore.deinit`, so its entries have to come from the allocator
        // that will free them; `gpa` still owns what the caller gets back —
        // the returned delta.

        // Bounded fan-out: at most `max_base_candidates` `delta.create` runs,
        // even if the caller over-offers (fossil's N_NEIGHBOR = 5).
        const limit = @min(candidates.len, max_base_candidates);
        var i: usize = 0;
        while (i < limit) : (i += 1) {
            const cand = candidates[i];
            std.debug.assert(cand >= 1 and cand <= self.entries.items.len);
            const entry = &self.entries.items[@intCast(cand - 1)];
            // Eligibility — all free, no I/O: skip a phantom, the config
            // singleton, a base already at the write-chain cap, and the
            // artifact itself.
            if (entry.phantom) continue;
            if (entry.kind == .config) continue;
            if (entry.depth >= delta_chain_write_max) continue;
            if (std.mem.eql(u8, entry.hash, hash)) continue;

            // Routed through the masked reader, so it inherits the shun mask: a shunned
            // candidate reads null here and is skipped like a phantom. That is
            // a WRITE-path consequence of a READ guard, and the right one — a
            // new delta rooted on a frame reclamation is about to drop would be
            // unresolvable the moment those bytes went away. It does not
            // conflict with §7.7, which governs a receiver honouring a basis a
            // peer already chose, not this store choosing a fresh one.
            const base_bytes = self.contentCached(self.gpa, cand, &self.base_memo) catch |e| switch (e) {
                error.OutOfMemory => return error.OutOfMemory,
                error.Storage => return error.Storage,
                error.Corrupt => continue, // an unreadable base is not a basis
            } orelse continue; // a phantom or shunned candidate carries no usable content
            defer self.gpa.free(base_bytes);
            // Cheap length early-out (fossil content.c:908): a source too small
            // to ever pay for a delta's framing is skipped before the encoder.
            if (base_bytes.len < delta_min_source) continue;

            const raw = delta.create(gpa, base_bytes, bytes) catch |e| switch (e) {
                error.OutOfMemory => return error.OutOfMemory,
                error.TargetTooLarge => continue, // unencodable target; skip
            };
            // Keep the strictly smaller delta; a tie keeps the earlier (higher
            // priority) candidate. The loser is freed immediately so at most
            // one delta buffer is ever live.
            if (best == null or raw.len < best.?.len) {
                if (best) |b| gpa.free(b);
                best = raw;
                best_base = cand;
            } else {
                gpa.free(raw);
            }
        }

        const smallest = best orelse return null;
        // Accept iff strictly under 75% of the uncompressed artifact. The
        // globally smallest delta is <= every other, so if any candidate clears
        // 75% the smallest does too — testing only the smallest is exact.
        if (!deltaProfitable(smallest.len, bytes.len)) {
            gpa.free(smallest);
            return null;
        }
        return .{ .base = best_base, .delta = smallest };
    }

    /// Stores `bytes` compressed into a fossil zlib container (§4.1, default
    /// level) under `hash`. The hash still names the UNCOMPRESSED bytes, so
    /// storage compression never changes an artifact's address (§4.1
    /// L1206-1208). Idempotent + phantom-upgrade like `put`.
    pub fn putZlib(self: *LogStore, hash: []const u8, bytes: []const u8) PutError!Rid {
        std.debug.assert(bytes.len <= max_payload);
        const container = codec.encode(self.gpa, bytes, .default) catch |e| switch (e) {
            error.OutOfMemory => return error.OutOfMemory,
            error.InputTooLarge => unreachable, // asserted bytes.len <= max_payload (< 2^32)
        };
        defer self.gpa.free(container);
        return self.insert(hash, container, .zlib, 0, 0);
    }

    /// Store an already-encoded full-form §4.1 zlib `container` verbatim under
    /// `hash` as a `.zlib` frame, so a clone that received this exact container
    /// re-serves it byte-for-byte (sync §7.2/§4.1 L918-919: "retransmitted
    /// verbatim"). Unlike `putZlib` — which compresses raw plaintext — the
    /// wire bytes are preserved, not re-created. PRECONDITION (§7.7 L1465-1470):
    /// the caller MUST have verified `hash` names the bytes `container`
    /// inflates to; this store performs no hash recompute (it lacks the
    /// plaintext to). Idempotent + phantom-upgrade like `put`.
    pub fn putContainer(self: *LogStore, hash: []const u8, container: []const u8) PutError!Rid {
        // §4.1 — a container carries at least its 4-byte length prefix.
        std.debug.assert(container.len >= 4 and container.len <= max_payload);
        return self.insert(hash, container, .zlib, 0, 0);
    }

    /// Stores `raw_delta` (a delta reconstructing this artifact from the frame
    /// at `base_rid`) as a zlib-contained `delta` frame by encoding it into a
    /// §4.1 container, then storing that container verbatim via
    /// `putDeltaContainer`. The ingest path (`putArtifact`) uses this to store
    /// a locally created delta. Idempotent + phantom-upgrade like `put`.
    pub fn putDelta(self: *LogStore, hash: []const u8, base_rid: Rid, raw_delta: []const u8) PutError!Rid {
        std.debug.assert(raw_delta.len >= 1 and raw_delta.len <= max_payload);
        const container = codec.encode(self.gpa, raw_delta, .default) catch |e| switch (e) {
            error.OutOfMemory => return error.OutOfMemory,
            error.InputTooLarge => unreachable, // asserted raw_delta.len <= max_payload (< 2^32)
        };
        defer self.gpa.free(container);
        return self.putDeltaContainer(hash, base_rid, container);
    }

    /// Store an already-§4.1-encoded delta `container` (a compressed
    /// [FossilDelta] reconstructing this artifact from the frame at `base_rid`)
    /// verbatim as a `.delta` frame; `delta_of` records the base frame's byte
    /// offset. The stored container IS the wire payload a `cfile` serves
    /// verbatim (§7.2), so a clone that received this exact delta container
    /// re-serves it byte-for-byte — unlike `putDelta`, the container is NOT
    /// re-encoded. PRECONDITION (§7.7 L1465-1470): the caller MUST have verified
    /// `hash` names the bytes this container reconstructs to (the delta applied
    /// to `base_rid`'s content); this store performs no hash recompute.
    /// Idempotent + phantom-upgrade like `put`.
    pub fn putDeltaContainer(self: *LogStore, hash: []const u8, base_rid: Rid, container: []const u8) PutError!Rid {
        std.debug.assert(base_rid >= 1 and base_rid <= self.entries.items.len);
        const base = &self.entries.items[@intCast(base_rid - 1)];
        std.debug.assert(!base.phantom);
        std.debug.assert(base.offset >= header_len and base.offset < self.end);
        // §4.1 — a delta container carries at least its 4-byte length prefix.
        std.debug.assert(container.len >= 4 and container.len <= max_payload);
        const base_off = base.offset;
        // Bookkeeping only: the WRITE cap is `putArtifact`'s job (it refuses a
        // base at `delta_chain_write_max`); the verbatim caller (`persistStore`)
        // enforces the READ bound `max_delta_chain` for a retained foreign
        // chain. `putDeltaContainer` itself is the low-level primitive and
        // stays permissive. Saturating so depth never wraps on a pathologically
        // deep foreign chain.
        const depth: u8 = base.depth +| 1;
        return self.insert(hash, container, .delta, base_off, depth);
    }

    /// Shared insert core: append one frame carrying `payload` under `hash`
    /// with encoding `kind` (and `delta_of` for `.delta`), then update the
    /// index. Idempotent for an already-content-bearing hash; upgrades a
    /// phantom in place. On a bookkeeping failure after the frame landed, the
    /// file is truncated back so disk and memory stay in lockstep; a rollback
    /// truncate that itself fails poisons the store.
    fn insert(self: *LogStore, hash: []const u8, payload: []const u8, kind: FrameKind, delta_of: u64, depth: u8) PutError!Rid {
        std.debug.assert(hash.len >= min_hash_len and hash.len <= max_hash_len);
        std.debug.assert(payload.len <= max_payload);
        std.debug.assert((depth == 0) == (kind != .delta)); // depth is delta-only

        if (self.index.get(hash)) |rid| {
            const entry = &self.entries.items[@intCast(rid - 1)];
            if (!entry.phantom) return rid;
            const res = try self.appendFrame(hash, payload, kind, delta_of);
            entry.offset = res.offset;
            entry.payload_len = res.len;
            entry.kind = kind;
            entry.phantom = false;
            entry.depth = depth;
            return rid;
        }

        try self.entries.ensureUnusedCapacity(self.gpa, 1);
        const owned = try self.gpa.dupe(u8, hash);
        errdefer self.gpa.free(owned);
        const prior_frame_count = self.frame_count;
        const prior_last_frame_offset = self.last_frame_offset;
        const res = try self.appendFrame(hash, payload, kind, delta_of);
        const rid: Rid = self.entries.items.len + 1;
        self.index.put(self.gpa, owned, rid) catch |e| {
            // Frame already landed on disk but bookkeeping failed: undo the
            // write so disk and every tail-key field return to the exact
            // pre-append state. If truncate itself fails, poison rather than
            // permit any later append against a physically diverged tail.
            // No `clearBaseMemo` here: nothing has read this frame, so no memo
            // names `res.offset` (see the invariant on `clearBaseMemo`).
            self.file.truncate(res.offset) catch {
                self.poisoned = true;
            };
            self.end = res.offset;
            self.frame_count = prior_frame_count;
            self.last_frame_offset = prior_last_frame_offset;
            self.assertBaseMemoBelowEnd();
            return e;
        };
        self.entries.appendAssumeCapacity(.{ .hash = owned, .offset = res.offset, .payload_len = res.len, .kind = kind, .phantom = false, .depth = depth });
        return rid;
    }

    /// The payload encoding of `rid` (`.raw` for a phantom). Lets a serve
    /// loop pick a full-content versus delta cfile without a frame read.
    pub fn kindOf(self: *const LogStore, rid: Rid) FrameKind {
        std.debug.assert(rid >= 1 and rid <= self.entries.items.len);
        return self.entries.items[@intCast(rid - 1)].kind;
    }

    /// The on-disk (container/wire) payload length of `rid`; 0 for a phantom.
    /// Symmetric with `hashOf`; the DST sweep uses it to bound the crc-covered
    /// span of a frame whose uncompressed content differs in length.
    pub fn payloadLenOf(self: *const LogStore, rid: Rid) u32 {
        std.debug.assert(rid >= 1 and rid <= self.entries.items.len);
        return self.entries.items[@intCast(rid - 1)].payload_len;
    }

    /// The total on-disk span (aligned prefix + hash + payload) of `rid`'s
    /// frame — the byte cost `zit pack`'s §6.1 monotonic-shrink guard uses as
    /// the floor when deciding whether a re-encoding is worth writing. Asserts a
    /// real (non-phantom) frame; a phantom carries no frame to measure.
    pub fn frameSpanOf(self: *const LogStore, rid: Rid) u64 {
        std.debug.assert(rid >= 1 and rid <= self.entries.items.len);
        const entry = &self.entries.items[@intCast(rid - 1)];
        std.debug.assert(!entry.phantom);
        return frameSpan(entry.hash.len, entry.payload_len);
    }

    /// The delta-chain depth of `rid`: 0 for a non-delta frame (or a phantom),
    /// one more than its base for a `.delta` frame. Lets the verbatim persist
    /// path (`persistStore`) refuse a retained delta whose base already sits at
    /// the `max_delta_chain` READ bound, so a stored chain never exceeds what
    /// `content()` can walk.
    pub fn depthOf(self: *const LogStore, rid: Rid) u8 {
        std.debug.assert(rid >= 1 and rid <= self.entries.items.len);
        return self.entries.items[@intCast(rid - 1)].depth;
    }

    /// The byte offset of `rid`'s frame in the log file — the stable key a
    /// `PlaintextCache` uses to memoise a frame's reconstructed plaintext across
    /// a derivation sweep. Stable because the log is append-only: a written
    /// frame never moves, so its offset names it for the sweep's lifetime.
    pub fn offsetOf(self: *const LogStore, rid: Rid) u64 {
        std.debug.assert(rid >= 1 and rid <= self.entries.items.len);
        std.debug.assert(!self.entries.items[@intCast(rid - 1)].phantom); // a phantom has no frame
        return self.entries.items[@intCast(rid - 1)].offset;
    }

    /// Whether `rid` is a phantom (interned but carrying no frame). Lets a
    /// caller distinguish "held on disk" from "announced but absent" without
    /// reconstructing content.
    pub fn isPhantom(self: *const LogStore, rid: Rid) bool {
        std.debug.assert(rid >= 1 and rid <= self.entries.items.len);
        return self.entries.items[@intCast(rid - 1)].phantom;
    }

    /// The delta-chain depth of the entry whose frame starts at `offset`, for
    /// recovering a delta's depth during the open-time scan. Entries are
    /// appended in ascending offset order there (real frames only, no
    /// phantoms), so a binary search resolves the base with no auxiliary map.
    /// Returns 0 when no entry starts at `offset` (a dangling `delta_of` that
    /// `content()` rejects). Precondition: `entries` is offset-sorted — true
    /// only mid-`scan`, which is this method's sole caller.
    fn depthForBaseOffset(self: *const LogStore, offset: u64) u8 {
        const items = self.entries.items;
        var lo: usize = 0;
        var hi: usize = items.len;
        while (lo < hi) {
            const mid = lo + (hi - lo) / 2;
            const o = items[mid].offset;
            if (o == offset) return items[mid].depth;
            if (o < offset) lo = mid + 1 else hi = mid;
        }
        return 0;
    }

    /// THE READ MASK: is `entry`'s artifact in the repository shun table?
    /// Shared verbatim by `content`, `contentCached` and `wire` so the domain
    /// argument below is stated once instead of re-derived at three call sites.
    ///
    /// A shunned artifact reads as ABSENT even while its content frame is still
    /// on disk. That is what makes reclamation INVISIBLE — dropping the bytes
    /// later changes no verb's output — and so what makes deferring the reclaim
    /// safe.
    ///
    /// Deliberately skips the public `isShunned`'s `isCompleteDigest` assert
    /// (20 or 32 bytes) — but shares its one probe body (`shunHasName`) so the
    /// two can never drift. An ENTRY's hash is not in that domain: `put`/
    /// `intern` accept `min_hash_len .. max_hash_len` (1..32) and keep a name
    /// opaque, and the fuzz fixtures in this file legitimately drive
    /// `content()` on a 9-byte name. Asserting here would therefore trap on a
    /// perfectly valid store.
    ///
    /// Answering false for an out-of-domain width is EXACT here, not fail-open,
    /// and the argument is different from `isShunned`'s caller contract (which
    /// is about wire names an assert must not let through silently): every key
    /// in this map was PROVEN complete before it was inserted — `putShun`
    /// returns `error.BadKey` otherwise, and the open scan fails the whole open
    /// with `error.Corrupt` on a `.shun` frame whose `hash_len` is not complete.
    /// A non-complete entry hash therefore provably cannot be a key, so "not
    /// found" is the true answer and not merely a tolerated miss.
    ///
    /// THE SYNC-SIDE RECONCILIATION. `content`/`wire` are RID-KEYED, so they
    /// answer null for a shunned artifact whose frame is on disk. Three gates
    /// were written when read-ness and held-ness agreed; reconciling them was
    /// always a sync decision, not a storage one, and all three have since been
    /// decided in the sync layer:
    ///
    ///   * `consumeDelta`'s basis-held test — DECIDED, and deliberately left
    ///     reading through this mask. A locally shunned-but-held DELTASRC reads
    ///     null, so an admitted target falls into the dangling-delta branch: the
    ///     target is queued and a phantom is minted for the basis NAME. sync
    ///     §7.7 L1444-1450 makes serving such a basis permissive ("content still
    ///     held locally MAY serve as the delta basis"), so declining to is
    ///     conformant, and L1461-1463 accepts the resulting dead end explicitly.
    ///     The minted phantom is a real phantom in the client's own store —
    ///     `loadContentStore` snapshots a shunned holding AS a phantom (next
    ///     bullet) — and `Client.shun` excludes it from both the §3.3 gimme scan
    ///     and `unresolvedPhantoms`, so it is neither requested nor reported.
    ///   * The SEND-side enumeration — DECIDED. `cmd/sync.zig`'s
    ///     `LogIndexStore` now applies sync §9.5 L1988-1992's "every eligible
    ///     holding after the Section 3.3 shun … filters" to the enumeration
    ///     itself: a shunned frame keeps its adapter rid (§8.2 L1504-1506
    ///     licenses a GAP, never a renumbering) and is MASKED — `held` false,
    ///     `content`/`wire` null, name absent from the wire-name index. So the
    ///     disagreement this comment used to record at `heldImpl` is gone: the
    ///     igot walk no longer advertises what no round can send, and a `gimme`
    ///     for the name lands on §3.3 L532-535's "If absent, silently skip".
    ///     `loadContentStore` (the pull CLIENT's own holdings snapshot) interns-
    ///     but-does-not-hold a shunned artifact — both reads answer null under
    ///     this mask, so the entry becomes a PHANTOM. That is not an announce-side
    ///     hole: the client's igot walk gates on `held`, so a phantom is never
    ///     advertised; it is gimme-eligible, and `Client.shun` (§3.3 L527-530) is
    ///     what excludes it. What remains is only the oddity of a phantom record
    ///     standing for content this repository physically holds — no wire
    ///     consequence today. Both snapshots carry the SAME masked-basis guard: a
    ///     surviving delta rooted on a masked frame is snapshotted (client) or
    ///     served (serve) in its FULL form, since `contentAtOffset` resolves by
    ///     byte offset and this name-level mask does not reach it.
    ///   * `sync.zig`'s §3.3 phantom scan — DECIDED. Both `Client.shun` and
    ///     `Server.shun` now carry the repository table, so §3.3 L527-530's
    ///     exclusion is wired and absorbs the phantom half of this.
    ///
    /// Storage still must not pick the first bullet's repair unilaterally —
    /// declaring a held-but-shunned basis available (gate on `held`) versus
    /// reading it through an explicitly unmasked door is a §7.7 choice, and the
    /// sync layer has taken the conservative one. Note that the send side
    /// deliberately does NOT depend on this mask — `LogIndexStore` states its own
    /// — so revisiting that choice cannot silently reopen the send-side hole.
    fn shunned(self: *const LogStore, entry: *const Entry) bool {
        return self.shunHasName(entry.hash);
    }

    /// The one place the shun table is probed. `shunned` (the read mask) and
    /// `isShunned` (the public predicate) are two DOORS onto this body, not two
    /// implementations of it: they differ only in the caller contract each
    /// enforces on the way in. Widening the set's notion of membership — name
    /// normalization, an exemption concept, a second index — lands here once and
    /// reaches both, which is the point. Two inlined `contains` calls under
    /// divergent contracts could silently answer from different rules, and the
    /// read path is the half where that fails OPEN.
    fn shunHasName(self: *const LogStore, hash: []const u8) bool {
        return self.shun_index.contains(hash);
    }

    /// WHICH RULE a shared read body applies before it touches the frame. Every
    /// rid-keyed reader names its choice at the call site, so "the masked set is
    /// exactly {`content`, `contentCached`, `wire`}" is a fact the compiler
    /// enforces rather than a claim five doc comments make: the mask is written
    /// ONCE per shared body instead of hand-copied per reader, and a new reader
    /// cannot end up unmasked by forgetting a line — there is no default to
    /// fall through to.
    const ReadMask = enum {
        /// An ARTIFACT read: a name in the repository shun table reads as
        /// ABSENT even while its frame is on disk.
        masked,
        /// In-repository frame MAINTENANCE that shows the bytes to nobody: the
        /// shun table is not consulted. See `contentUnmasked` for the full
        /// argument and for why `pack.zig` is the only legitimate caller.
        maintenance,
    };

    /// Owned uncompressed bytes for a rid, or null while it's a phantom or its
    /// name is in the repository shun table (see `shunned`). Re-reads and
    /// re-verifies the frame (both checksums, embedded hash) on every call,
    /// then decodes by kind: `raw` copied out, `zlib` inflated, `delta`
    /// resolved by walking `delta_of` and applying via `delta.apply`.
    /// Never trusts a stale scan result.
    pub fn content(self: *LogStore, gpa: std.mem.Allocator, rid: Rid) ContentError!?[]u8 {
        std.debug.assert(rid >= 1 and rid <= self.entries.items.len);
        const entry = &self.entries.items[@intCast(rid - 1)];
        if (entry.phantom) return null;
        return try self.contentOfEntry(gpa, entry, .masked, null);
    }

    /// The frame read every rid-keyed content reader shares, INCLUDING the rule
    /// it reads under: apply `mask`, then resolve `entry`'s offset and, for a
    /// `.delta` frame, memoize the result when the caller brought a cache. Both
    /// the decode and the policy live here so the masked pair
    /// (`content`/`contentCached`) and the maintenance reader
    /// (`contentUnmasked`) cannot drift in either.
    fn contentOfEntry(self: *LogStore, gpa: std.mem.Allocator, entry: *const Entry, mask: ReadMask, cache: ?*PlaintextCache) ContentError!?[]u8 {
        // The precondition `offsetOf` states for the same dereference: a phantom
        // has no frame, and its `offset` is not a frame start — `readFrameAt`
        // below would decode arbitrary log bytes under this entry's name. Every
        // caller filters phantoms first; this is what keeps the next one honest.
        std.debug.assert(!entry.phantom);
        // BEFORE the frame read, not after. A masked reader must not reconstruct
        // a shunned artifact's plaintext at all, let alone memoise it where a
        // deeper delta could then be served from it. `contentCached`'s "identical
        // bytes to `content(rid)`" includes identical masking.
        if (mask == .masked and self.shunned(entry)) return null;
        const bytes = try self.contentAtOffset(gpa, entry.offset, max_delta_chain, entry.hash, cache);
        // Keep a resolved delta's plaintext hot for any deeper delta that names
        // this frame as its base; raw/zlib top-level frames are left uncached so
        // a delta-free sweep never dups a byte. Best-effort: a failed insert
        // just means no memo, never a wrong or missing answer.
        if (cache) |c| {
            if (entry.kind == .delta or c.memoize_undeltaed) c.put(gpa, entry.offset, bytes) catch {};
        }
        return bytes;
    }

    /// Like `content`, but resolves delta bases through `cache` — a bounded
    /// plaintext memo the caller owns for the span of one derivation sweep. A
    /// full sweep over delta chains would otherwise re-reconstruct each base
    /// once per referencing delta (the k(k+1)/2 read-amplification the plan
    /// projects at ~8.5x on depth-16 chains); memoizing a base's plaintext on
    /// first use makes every later reference a lookup, so each frame is fully
    /// reconstructed at most about twice regardless of chain depth. A `.delta`
    /// frame's own plaintext is also memoized so a deeper delta built on it
    /// hits; `.raw`/`.zlib` top-level results are never inserted, so an
    /// all-raw repository pays zero cache overhead. Identical bytes to
    /// `content(rid)`; the cache changes only the work, never the answer.
    pub fn contentCached(self: *LogStore, gpa: std.mem.Allocator, rid: Rid, cache: *PlaintextCache) ContentError!?[]u8 {
        std.debug.assert(rid >= 1 and rid <= self.entries.items.len);
        const entry = &self.entries.items[@intCast(rid - 1)];
        if (entry.phantom) return null;
        return try self.contentOfEntry(gpa, entry, .masked, cache);
    }

    /// THE MAINTENANCE READER: `content`/`contentCached` with the shun mask
    /// deliberately absent (phantoms still read null — a phantom has no frame to
    /// read). `cache` may be null for the uncached shape.
    ///
    /// THIS IS NOT AN ARTIFACT READ. The masked readers answer "what does this
    /// repository surface under this name", and a shunned name surfaces nothing.
    /// This answers a different question — "what is in this frame" — and exists
    /// for exactly one kind of caller: in-repository frame MAINTENANCE that
    /// relocates bytes without showing them to anybody. `pack.zig` is that
    /// caller, and the only one.
    ///
    /// Why it has to exist. A repack reads every source frame and rewrites it
    /// into a fresh generation. Read through the mask, `packLog`'s `orelse
    /// continue` would silently DROP every shunned-but-held frame — deleting the
    /// artifact AND its interned name, so `lookup`/`resolveHexPrefix` start
    /// answering null where they answered a rid. That is `zit pack` changing
    /// what an unrelated verb prints, which is exactly the invariant the mask
    /// exists to protect ("reclamation is invisible"), and it is a
    /// data-destroying reclamation policy no draft states.
    ///
    /// Any path that hands bytes to an operator, a peer, a working tree or a
    /// derived view MUST use `content`/`contentCached`/`wire`. Reaching for this
    /// one to "fix a null" means you want the masked reader.
    ///
    /// `rid`'s range is CHECKED here, not asserted as on the masked readers.
    /// This door and `wireUnmasked` exist precisely to bypass a safety rule, and
    /// in `ReleaseFast`/`ReleaseSmall` both the assert AND the slice bounds check
    /// are elided — leaving `rid == 0` to underflow before the `@intCast` and
    /// read off the front of `entries`. One compare, on a path that is about to
    /// do file I/O, holds the bound in every optimize mode.
    /// The first `want` plaintext bytes of `rid`, or null when the answer is
    /// not cheaply available. A CLASSIFIER's read: use it to decide whether an
    /// artifact is worth reconstructing, never as content.
    ///
    /// Null has one meaning -- "ask the full reader" -- and three causes: a
    /// phantom (no frame), a `.delta` frame (its content does not exist until
    /// its base is reconstructed, which is the work this call exists to avoid),
    /// and a shorter-than-`want` artifact is NOT one of them, since a short
    /// read is a real answer. Callers MUST treat null as unknown rather than as
    /// a negative, which is what keeps the classification conservative.
    ///
    /// Unmasked, like `contentUnmasked` and for the same reason: it shows the
    /// bytes to nobody. `pack.zig` is the only legitimate caller.
    pub fn contentPrefixUnmasked(
        self: *LogStore,
        gpa: std.mem.Allocator,
        rid: Rid,
        want: usize,
    ) ContentError!?[]u8 {
        if (rid == 0 or rid > self.entries.items.len) return error.Corrupt;
        const entry = &self.entries.items[@intCast(rid - 1)];
        if (entry.phantom) return null;
        const fv = try self.readFrameAt(gpa, entry.offset);
        switch (fv.kind) {
            .raw => {
                defer gpa.free(fv.payload);
                const n = @min(want, fv.payload.len);
                return gpa.dupe(u8, fv.payload[0..n]) catch return error.OutOfMemory;
            },
            .zlib => {
                defer gpa.free(fv.payload);
                return codec.decodePrefix(gpa, fv.payload, want) catch |e| switch (e) {
                    error.OutOfMemory => error.OutOfMemory,
                    error.TruncatedHeader, error.InvalidStream, error.OutputTooLarge => error.Corrupt,
                };
            },
            .delta => {
                gpa.free(fv.payload);
                return null;
            },
            // Not a content artifact (`config`/`shun`) or a kind this build does
            // not know. Neither is this classifier's business, and "unknown"
            // routes the caller to the full reader, which owns the decision.
            else => {
                gpa.free(fv.payload);
                return null;
            },
        }
    }

    pub fn contentUnmasked(self: *LogStore, gpa: std.mem.Allocator, rid: Rid, cache: ?*PlaintextCache) ContentError!?[]u8 {
        if (rid == 0 or rid > self.entries.items.len) return error.Corrupt;
        const entry = &self.entries.items[@intCast(rid - 1)];
        if (entry.phantom) return null;
        return try self.contentOfEntry(gpa, entry, .maintenance, cache);
    }

    /// Reconstructs the uncompressed content of the frame at `offset`.
    /// `depth_remaining` bounds delta recursion: each `delta` hop consumes
    /// one, and a delta reached at zero is `error.Corrupt` (never unbounded
    /// recursion). `expect_hash`, when non-null, requires the frame's embedded
    /// hash to match (the top-level integrity check); recursion into a base
    /// passes null since a base carries its own name. `cache`, when non-null,
    /// memoizes delta-base plaintext (see `contentCached`).
    ///
    /// CARRIES NO SHUN MASK, AND MUST NOT — this is normative, not stylistic.
    /// sync §7.7 L1444-1450:
    ///
    ///     For an admitted target whose DELTASRC name is in the repository shun
    ///     table, the receiver does not apply admission independently to that
    ///     source: content still held locally MAY serve as the delta basis; if
    ///     the content is unavailable, ordinary dangling-delta handling creates
    ///     or retains the source phantom, but the Section 3.3 scan excludes it,
    ///     so it is not requested.
    ///
    /// zit satisfies that STRUCTURALLY and for free. This function is keyed by
    /// OFFSET, and the delta recursion below reaches a base through `delta_of`
    /// — a byte offset — passing `expect_hash = null` because a base carries
    /// its own name. Base resolution therefore never re-enters the rid-keyed
    /// entry points (`content`/`contentCached`/`wire`) where the hash-keyed
    /// mask lives: the mask and the basis path are physically distinct code
    /// paths, so a held-but-shunned base keeps serving every delta rooted on it
    /// while the base itself reads as absent by name.
    ///
    /// Pushing the mask down here would break §7.7 by making that base
    /// unusable and every dependent delta unresolvable. It would also be a
    /// silent breakage — the deltas simply stop reconstructing — so the §7.7
    /// test in `logstore_test.zig` pins the property rather than leaving this
    /// comment as its only defence.
    fn contentAtOffset(self: *LogStore, gpa: std.mem.Allocator, offset: u64, depth_remaining: u8, expect_hash: ?[]const u8, cache: ?*PlaintextCache) ContentError![]u8 {
        std.debug.assert(depth_remaining <= max_delta_chain);
        const fv = try self.readFrameAt(gpa, offset);
        if (expect_hash) |eh| {
            if (!std.mem.eql(u8, fv.hash[0..fv.hash_len], eh)) {
                gpa.free(fv.payload);
                return error.Corrupt;
            }
        }
        switch (fv.kind) {
            .raw => return fv.payload, // ownership transfers to the caller
            .zlib => {
                defer gpa.free(fv.payload);
                return decodeContainer(gpa, fv.payload, max_payload);
            },
            .delta => {
                defer gpa.free(fv.payload);
                if (depth_remaining == 0) return error.Corrupt;
                // A base must be an earlier, in-range frame: a forward or
                // self reference is corruption, and a mid-frame/garbage
                // offset fails `readFrameAt`'s checksums below.
                if (fv.delta_of < header_len or fv.delta_of >= offset) return error.Corrupt;
                const raw_delta = try decodeContainer(gpa, fv.payload, max_payload);
                defer gpa.free(raw_delta);
                // A memoized base: apply against the borrowed plaintext, never
                // re-reading or re-reconstructing the base frame this sweep.
                if (cache) |c| {
                    if (c.get(fv.delta_of)) |hit| {
                        return delta.apply(gpa, hit, raw_delta) catch |e| switch (e) {
                            error.OutOfMemory => error.OutOfMemory,
                            error.MalformedDelta, error.ChecksumMismatch => error.Corrupt,
                        };
                    }
                }
                const base = try self.contentAtOffset(gpa, fv.delta_of, depth_remaining - 1, null, cache);
                defer gpa.free(base);
                // A base is referenced by at least this delta, so memoize it.
                if (cache) |c| c.put(gpa, fv.delta_of, base) catch {};
                return delta.apply(gpa, base, raw_delta) catch |e| switch (e) {
                    error.OutOfMemory => error.OutOfMemory,
                    error.MalformedDelta, error.ChecksumMismatch => error.Corrupt,
                };
            },
            .config, .shun => {
                // A delta base (or content lookup) resolving to a config or
                // shun frame is corruption: neither is content-addressed and
                // neither is ever a base (plan E; `ShunRecord` for shun).
                gpa.free(fv.payload);
                return error.Corrupt;
            },
            _ => unreachable, // readFrameAt rejects any flag byte outside {raw,zlib,delta,config,shun}
        }
    }

    /// One fully validated frame read from `offset`: kind, `delta_of`, the
    /// embedded hash, and the owned on-disk payload (raw bytes, or the zlib
    /// container for `zlib`/`delta`). Caller owns `payload`.
    const FrameView = struct {
        kind: FrameKind,
        delta_of: u64,
        hash: [max_hash_len]u8,
        hash_len: u8,
        payload: []u8,
    };

    /// Reads and checksum-verifies the frame at `offset`, returning its
    /// decoded fields and owned payload. Shares the fixed-header validation
    /// (`readFrameHeader`) with the open-time scanner; unlike the scanner, a
    /// short read here is `error.Corrupt` — the frame was known to exist, so
    /// missing bytes mean the file lied or lost data, not a torn tail to
    /// recover. Any flag byte outside {raw,zlib,delta,config,shun} is rejected.
    fn readFrameAt(self: *LogStore, gpa: std.mem.Allocator, offset: u64) ContentError!FrameView {
        const fh = switch (try readFrameHeader(self.file, offset)) {
            .torn => return error.Corrupt,
            .header => |h| h,
        };

        const pl: usize = fh.payload_len;
        if (pl > max_payload) return error.Corrupt; // wire-derived bound before allocation
        const payload_offset: u64 = offset + fh.fixed_and_crc_len;

        // ONE read for the payload AND its CRC trailer. The trailer sits
        // immediately after the payload, so reading it separately bought a
        // second syscall for four adjacent bytes — and `FileStorage.pread` is
        // unbuffered, so every such call is a real kernel transition. On a
        // deltified artifact the whole chain is walked frame by frame (up to
        // `max_delta_chain`), which multiplied that waste by the chain depth.
        const buf = try gpa.alloc(u8, pl + 4);
        errdefer gpa.free(buf);
        const n = self.file.pread(payload_offset, buf) catch return error.Storage;
        if (n < pl + 4) return error.Corrupt; // frame known to exist: short means lost bytes

        // Whole-frame CRC, unconditionally. No open path verifies payloads any
        // more -- both scan `.headers_only` -- so this is the ONLY thing standing
        // between a rotted payload and a caller that believes it. It ran on every
        // read from the append path even before that change; what is gone is the
        // read-only path's exemption, which was earned by a scan that no longer
        // happens. Restoring any skip here requires restoring that scan.
        var fcrc = crc32.Crc32.init();
        fcrc.update(fh.header_bytes[0..fh.fixed_and_crc_len]);
        fcrc.update(buf[0..pl]);
        if (fcrc.final() != std.mem.readInt(u32, buf[pl..][0..4], .little)) return error.Corrupt;

        // Hand back an allocation of exactly `pl`: every caller frees
        // `fv.payload` directly, so it must be the whole allocation and not a
        // slice of a longer one. Shrinking in place is the ordinary outcome; the
        // copy is the fallback an allocator that refuses to resize would take.
        const payload = shrink: {
            if (gpa.resize(buf, pl)) break :shrink buf[0..pl];
            const exact = try gpa.alloc(u8, pl);
            @memcpy(exact, buf[0..pl]);
            gpa.free(buf);
            break :shrink exact;
        };

        return .{
            .kind = fh.kind,
            .delta_of = fh.delta_of,
            .hash = fh.hash,
            .hash_len = fh.hash_len,
            .payload = payload,
        };
    }

    /// A stored artifact's cfile wire payload (sync §7.2), served verbatim.
    /// Two forms, matching the two stored container kinds:
    ///   * `full` — a §4.1 zlib container of the whole artifact; USIZE is its
    ///     own 4-byte big-endian prefix (`writeCfileVerbatim`).
    ///   * `delta` — a §4.1 zlib container of a [FossilDelta], plus the base
    ///     artifact's hash (DELTASRC) and the reconstructed-target size read
    ///     from the delta header (delta §3.2). USIZE is that `target_size` —
    ///     NEVER the container's 4-byte prefix, which is the raw delta's length
    ///     (sync §7.2 L1214-1216: "in delta form it is not delta size").
    /// `deinit` frees the owned `container` in either arm.
    pub const Wire = union(enum) {
        full: []u8,
        delta: Delta,

        pub const Delta = struct {
            container: []u8,
            base_hash_buf: [max_hash_len]u8,
            base_hash_len: u8,
            /// delta §3.2 — the reconstructed target length; the cfile USIZE.
            target_size: u64,

            pub fn baseHash(self: *const Delta) []const u8 {
                return self.base_hash_buf[0..self.base_hash_len];
            }
        };

        pub fn container(self: *const Wire) []const u8 {
            return switch (self.*) {
                .full => |c| c,
                .delta => |d| d.container,
            };
        }

        pub fn deinit(self: *Wire, gpa: std.mem.Allocator) void {
            switch (self.*) {
                .full => |c| gpa.free(c),
                .delta => |d| gpa.free(d.container),
            }
            self.* = undefined;
        }
    };

    /// The stored cfile payload for `rid` served verbatim (sync §4.1/§7.2), or
    /// null when the frame holds no stored container — a `.raw` frame or a
    /// phantom (the serve loop then falls back to compressing `content()`) — or
    /// when `rid`'s name is in the repository shun table (see `shunned`; the
    /// `content()` fallback masks too, so a shunned `.raw` artifact has no path
    /// out either). A `.zlib` frame yields the `full` arm; a `.delta` frame
    /// yields the `delta` arm, carrying the base hash (read from the base frame)
    /// and the target size (read from the delta header — one small inflate). The
    /// stored container IS the wire payload (normative §4.1/§7.2), so this
    /// exposes no representation the RFC does not already fix. Caller owns the
    /// `Wire`.
    pub fn wire(self: *LogStore, gpa: std.mem.Allocator, rid: Rid) ContentError!?Wire {
        std.debug.assert(rid >= 1 and rid <= self.entries.items.len);
        const entry = &self.entries.items[@intCast(rid - 1)];
        if (entry.phantom) return null;
        return self.wireOfEntry(gpa, entry, .masked);
    }

    /// `wire` with the shun mask deliberately absent — the maintenance door, on
    /// exactly the terms `contentUnmasked` documents at length: `pack.zig`'s
    /// verbatim frame copy needs the stored container of a shunned-but-held
    /// frame to re-emit it into the packed generation. NOT a serve path; a `Wire`
    /// obtained here MUST NOT reach a peer. `rid`'s range is CHECKED for the
    /// reason `contentUnmasked` gives.
    pub fn wireUnmasked(self: *LogStore, gpa: std.mem.Allocator, rid: Rid) ContentError!?Wire {
        if (rid == 0 or rid > self.entries.items.len) return error.Corrupt;
        const entry = &self.entries.items[@intCast(rid - 1)];
        if (entry.phantom) return null;
        return self.wireOfEntry(gpa, entry, .maintenance);
    }

    /// The kind-directed container read both `wire` doors share, INCLUDING the
    /// rule it reads under, so they cannot drift in either.
    fn wireOfEntry(self: *LogStore, gpa: std.mem.Allocator, entry: *const Entry, mask: ReadMask) ContentError!?Wire {
        // `offsetOf`'s precondition, for the same dereference: a phantom's
        // `offset` is not a frame start, so `readFrameAt` would decode arbitrary
        // log bytes and could hand back a checksum-passing neighbour frame under
        // this entry's name. Callers filter phantoms; this keeps the next honest.
        std.debug.assert(!entry.phantom);
        // The stored container IS the wire payload. Masking `content` but not
        // this would leave a shunned artifact's exact bytes reachable by the one
        // path whose whole job is handing them to a peer. Before the kind switch,
        // not inside an arm: every kind that can yield a container must mask, and
        // a guard that only covers the arm the fixtures happen to exercise is a
        // leak wearing a test's clothes.
        if (mask == .masked and self.shunned(entry)) return null;
        switch (entry.kind) {
            .raw => return null,
            .zlib => {
                const fv = try self.readFrameAt(gpa, entry.offset);
                std.debug.assert(fv.kind == .zlib);
                return .{ .full = fv.payload };
            },
            .delta => {
                const fv = try self.readFrameAt(gpa, entry.offset);
                errdefer gpa.free(fv.payload);
                std.debug.assert(fv.kind == .delta);
                // Structural phantom safety (D3): a stored delta's base ALWAYS
                // precedes it in the log; a forward/self reference is corruption.
                if (fv.delta_of < header_len or fv.delta_of >= entry.offset) return error.Corrupt;
                // The reconstructed-target size (the §7.2 USIZE) comes from the
                // delta's own header, not the container prefix (that is the raw
                // delta length). One small inflate reads it without applying a
                // single segment.
                const raw_delta = try decodeContainer(gpa, fv.payload, max_payload);
                defer gpa.free(raw_delta);
                const target_size = delta.targetSize(raw_delta) catch return error.Corrupt;
                // The base frame's own header carries its hash; reading it also
                // checksum-validates that `delta_of` still points at a real frame.
                const base = try self.readFrameAt(gpa, fv.delta_of);
                gpa.free(base.payload);
                // A non-content frame is never a basis. `content()` already
                // rejects this exact log, so without the check `wire()` would
                // hand a peer a delta cfile naming a base that is a config
                // record or a tombstone — a chain no receiver can resolve, and
                // disagreement between the two read paths over one file.
                if (base.kind == .config or base.kind == .shun) return error.Corrupt;
                var out: Wire = .{ .delta = .{
                    .container = fv.payload,
                    .base_hash_buf = undefined,
                    .base_hash_len = base.hash_len,
                    .target_size = target_size,
                } };
                @memcpy(out.delta.base_hash_buf[0..base.hash_len], base.hash[0..base.hash_len]);
                return out;
            },
            // Neither a config nor a shun frame is ever an artifact rid (both
            // are never interned), so neither reaches here; any other kind is
            // rejected on open.
            .config, .shun, _ => unreachable,
        }
    }

    /// Borrows the hash naming `rid`. Every rid, phantom or not, has one.
    pub fn hashOf(self: *const LogStore, rid: Rid) []const u8 {
        std.debug.assert(rid >= 1 and rid <= self.entries.items.len);
        return self.entries.items[@intCast(rid - 1)].hash;
    }
    /// Returns caller-owned non-phantom artifact RIDs in the order their
    /// frames were persisted. Dense RID order is deliberately not used:
    /// `intern` may create a phantom before a later `put` writes its frame.
    pub fn ridsInPersistedOrder(self: *const LogStore, gpa: std.mem.Allocator) error{OutOfMemory}![]Rid {
        var count: usize = 0;
        for (self.entries.items) |entry| {
            if (!entry.phantom) count += 1;
        }
        const rids = try gpa.alloc(Rid, count);
        count = 0;
        for (self.entries.items, 0..) |entry, i| {
            if (entry.phantom) continue;
            rids[count] = @intCast(i + 1);
            count += 1;
        }
        std.mem.sort(Rid, rids, self, ridOffsetLessThan);
        return rids;
    }

    fn ridOffsetLessThan(self: *const LogStore, a: Rid, b: Rid) bool {
        return self.entries.items[@intCast(a - 1)].offset < self.entries.items[@intCast(b - 1)].offset;
    }

    pub fn lookup(self: *const LogStore, hash: []const u8) ?Rid {
        return self.index.get(hash);
    }

    /// Resolve a lowercase-hex artifact-name prefix to the single non-phantom
    /// rid whose full name starts with it, or null when nothing matches
    /// (artifact §3.6 prefix lookup). `error.Ambiguous` when more than one
    /// stored artifact shares the prefix — a prefix matching several artifacts
    /// is REFUSED, never silently narrowed to one (repo-state §8.3 item 8).
    /// General-purpose: this queries EVERY held artifact, file blobs included,
    /// since a blob is a non-phantom frame here just like a manifest. No
    /// allocation — each entry's raw digest is hex-encoded into a fixed stack
    /// buffer. Bound: `self.entries.items.len`. Preconditions (asserted): the
    /// prefix is 4..2*max_hash_len bytes and every byte is a lowercase hex
    /// digit — the caller canonicalizes; a non-hex or out-of-band prefix is the
    /// caller's not-found family, never reaches here.
    pub fn resolveHexPrefix(self: *const LogStore, prefix: []const u8) error{Ambiguous}!?Rid {
        std.debug.assert(prefix.len >= 4 and prefix.len <= 2 * max_hash_len);
        for (prefix) |c| std.debug.assert((c >= '0' and c <= '9') or (c >= 'a' and c <= 'f'));

        const lower = "0123456789abcdef";
        var found: ?Rid = null;
        for (self.entries.items, 0..) |entry, i| {
            if (entry.phantom) continue; // announced but no frame — carries no content
            var hexbuf: [2 * max_hash_len]u8 = undefined;
            const hex_len = entry.hash.len * 2;
            std.debug.assert(hex_len <= hexbuf.len);
            for (entry.hash, 0..) |b, j| {
                hexbuf[j * 2] = lower[b >> 4];
                hexbuf[j * 2 + 1] = lower[b & 0x0f];
            }
            const hex = hexbuf[0..hex_len];
            if (hex.len < prefix.len) continue; // a longer prefix cannot name a shorter hash
            if (std.mem.startsWith(u8, hex, prefix)) {
                if (found != null) return error.Ambiguous; // §8.3 item 8: >1 match refuses
                found = @intCast(i + 1); // rids are 1-based (entries index + 1)
            }
        }
        if (found) |rid| std.debug.assert(!self.entries.items[@intCast(rid - 1)].phantom);
        return found;
    }

    pub fn commit(self: *LogStore) CommitError!void {
        self.file.sync() catch return error.Storage;
    }

    /// Appends a new config frame and makes it the live config (plan E). Latest
    /// wins by scan order, so an edit is one append — no rewrite, no mutable
    /// pointer. On a torn append `appendFrame` rolls the file back and
    /// `config_offset` is left untouched, so failure leaves the prior config
    /// live.
    pub fn putConfig(self: *LogStore, cfg: Config) PutError!void {
        std.debug.assert(cfg.default_branch.len <= max_config_str);
        std.debug.assert(cfg.project_id.len <= max_config_str);
        std.debug.assert(cfg.project_code.len <= max_config_str);
        std.debug.assert(cfg.login_user.len <= max_config_str);
        std.debug.assert(cfg.login_secret_hex.len <= max_config_str);
        // sync §6.3 — a stored secret is either absent or the full 40-hex
        // derived value; no other width is ever persisted.
        std.debug.assert(cfg.login_secret_hex.len == 0 or cfg.login_secret_hex.len == 40);
        std.debug.assert(cfg.remote_url.len <= max_config_str);
        var buf: [max_config_payload]u8 = undefined;
        const payload = encodeConfig(&buf, cfg);
        const res = try self.appendFrame(config_sentinel, payload, .config, 0);
        std.debug.assert(res.offset + payload.len < self.end); // frame landed before the new end
        self.config_offset = res.offset;
    }

    /// The live config, or null when the log carries none (plan E: absent config
    /// means the caller — CLI/repostate — owns the default; storage bakes
    /// none). Re-reads and re-verifies the frame on every call. Caller owns
    /// the returned `OwnedConfig`.
    pub fn readConfig(self: *LogStore, gpa: std.mem.Allocator) ContentError!?OwnedConfig {
        const off = self.config_offset orelse return null;
        const fv = try self.readFrameAt(gpa, off);
        errdefer gpa.free(fv.payload);
        if (fv.kind != .config) return error.Corrupt;
        const parsed = try parseConfig(fv.payload);
        return .{ .backing = fv.payload, .config = parsed };
    }

    /// Record `hash` in this repository's shun table, durably: one appended
    /// `.shun` frame carrying the digest in its own `hash` field and
    /// `epoch`/`reason` in its payload. Never interned, so no rid is issued and
    /// `lookup(hash)` is unaffected. Never removes anything — reclaiming the
    /// shunned artifact's bytes is a separate, later operation.
    ///
    /// IDEMPOTENT AS SET MEMBERSHIP: a digest already in the set appends NO
    /// frame and leaves the existing row's `epoch`/`reason` alone. The durable
    /// record is the one that was actually written; this is a set, not a
    /// last-write-wins cell.
    ///
    /// INPUT RULES, both CHECKED and returned as `ShunError` rather than
    /// asserted — see `ShunError` for why an assert is the wrong instrument for
    /// either (both survive into `ReleaseFast`, both are unrecoverable once the
    /// frame lands, and both take operator text):
    ///
    ///   * `hash` is a COMPLETE digest, 20 or 32 bytes (`isCompleteDigest`),
    ///     else `error.BadKey`. This is the one put path that does NOT take the
    ///     opaque `min_hash_len .. max_hash_len` bound. The shun table is keyed
    ///     by complete artifact name, so a whole-digest key makes an abbreviated
    ///     prefix structurally unrepresentable rather than merely discouraged —
    ///     a prefix admitted here would sit in the set never matching the full
    ///     digest anyone queries, i.e. fail-OPEN, the one failure a tombstone
    ///     must not have. It would also be durable: `scan` rejects such a frame
    ///     outright, so admitting one here bricks the repository for every later
    ///     open. The write end therefore states the rule exactly as strongly as
    ///     the read end does. `hash_kind` is still never read: the algorithm is
    ///     inferred from the width, exactly as artifact-format §3.2 L347-348
    ///     defines the two name forms, so this checks a length and still
    ///     interprets no digest.
    ///   * `reason` is at most `max_shun_reason` bytes, else
    ///     `error.ReasonTooLong`. Unlike `putConfig`'s internally-generated
    ///     fields this is free operator text — the first payload field in this
    ///     file taking any — and `encodeShun` copies it into a fixed
    ///     `max_shun_payload` stack buffer, so the bound is a memory-safety
    ///     bound, not a policy one. An operator-facing caller (the CLI verb) may
    ///     still truncate for ergonomics; it is not required to, because this
    ///     rejects rather than trusts.
    ///
    /// Every fallible step runs BEFORE the append, so nothing that can fail
    /// follows the frame landing: a partially-recorded tombstone (durable on
    /// disk, absent in memory) would read as fail-OPEN until the next reopen,
    /// which is exactly the failure a tombstone must not have.
    pub fn putShun(self: *LogStore, hash: []const u8, reason: []const u8, epoch: i64) ShunError!void {
        if (!isCompleteDigest(hash.len)) return error.BadKey;
        if (reason.len > max_shun_reason) return error.ReasonTooLong;
        if (self.shunHasName(hash)) return;

        const owned_hash = try self.gpa.dupe(u8, hash);
        errdefer self.gpa.free(owned_hash);
        const owned_reason = try self.gpa.dupe(u8, reason);
        errdefer self.gpa.free(owned_reason);
        try self.shun_entries.ensureUnusedCapacity(self.gpa, 1);
        try self.shun_index.ensureUnusedCapacity(self.gpa, 1);

        // The slice's central invariant — a shun frame is NEVER interned —
        // stated here, where the append that could violate it happens, and not
        // only in the tests. Frame-level, deliberately: asserting the digest is
        // absent from `index` would be WRONG (shunning an artifact this repo
        // holds is the ordinary case); what must not change is the intern
        // count. A refactor that routed this append through the content path
        // would make `lookup(hash)` resolve a tombstone as content.
        const entries_before = self.entries.items.len;
        var buf: [max_shun_payload]u8 = undefined;
        const payload = encodeShun(&buf, epoch, reason);
        const res = try self.appendFrame(hash, payload, .shun, 0);
        std.debug.assert(res.offset >= header_len and res.offset + payload.len < self.end); // frame landed before the new end
        std.debug.assert(self.entries.items.len == entries_before);
        self.shunAssumeCapacity(owned_hash, epoch, owned_reason);
    }

    /// THE ONE DOOR onto the in-memory set — the only place either half is
    /// written. Takes ownership of two already-duplicated strings, and both
    /// containers' capacity is already reserved by the caller, so the two writes
    /// below are infallible and nothing can run between them. That is what lets
    /// `putShun` place its only irreversible step (the append) last, and it is
    /// also what makes a HALF-WRITTEN row — listed but not queryable, or
    /// queryable but not listed — unrepresentable rather than merely asserted
    /// against afterwards.
    ///
    /// Which matters more since the read mask than it did before it. `shunned`
    /// answers from `shun_index` while `shunList` reads `shun_entries`, so a
    /// desync no longer just misreports a row to an operator: it leaves an
    /// artifact displayed as suppressed and still SERVED. A `std.debug.assert`
    /// pairing check is the wrong instrument for a rule with that consequence —
    /// it compiles to nothing in `ReleaseFast`/`ReleaseSmall`, i.e. it is absent
    /// from every build that ships. Funnelling both writers through here removes
    /// the failure mode instead of watching for it.
    ///
    /// The two halves cannot disagree about the BYTES either: the map borrows
    /// the row's own allocation as its key, and its value is that row's position.
    /// PRECONDITION: `owned_hash` is not already a key.
    fn shunAssumeCapacity(self: *LogStore, owned_hash: []u8, epoch: i64, owned_reason: []u8) void {
        std.debug.assert(!self.shunHasName(owned_hash));
        const row = self.shun_entries.items.len;
        self.shun_entries.appendAssumeCapacity(.{ .hash = owned_hash, .epoch = epoch, .reason = owned_reason });
        self.shun_index.putAssumeCapacity(owned_hash, row);
    }

    /// Add one scanned shun row to the set, first-writer-wins. A log CAN carry
    /// two frames for one digest (a concurrent writer, a hand-built fixture);
    /// the repeat is dropped BEFORE anything is allocated, so — unlike `index`,
    /// whose keys `entries` owns — this map never takes a duplicate copy of the
    /// row's own allocation that it would then have to free again. Same shape as
    /// `putShun`'s idempotence check, and lands in `shunAssumeCapacity` for the
    /// same reason: one door onto the set.
    fn internShun(self: *LogStore, hash: []const u8, epoch: i64, reason: []const u8) error{OutOfMemory}!void {
        if (self.shunHasName(hash)) return;
        const owned_hash = try self.gpa.dupe(u8, hash);
        errdefer self.gpa.free(owned_hash);
        const owned_reason = try self.gpa.dupe(u8, reason);
        errdefer self.gpa.free(owned_reason);
        try self.shun_entries.ensureUnusedCapacity(self.gpa, 1);
        try self.shun_index.ensureUnusedCapacity(self.gpa, 1);
        self.shunAssumeCapacity(owned_hash, epoch, owned_reason);
    }

    /// Is `hash` in this repository's shun table? O(1), no I/O — the check a
    /// per-round sync filter (sync §3.3 L527-530) can afford per hash.
    ///
    /// Asserts the SAME complete-digest domain `putShun` checks, and for a
    /// sharper reason than symmetry: the two ends of one predicate must accept
    /// one domain or the read end fails OPEN. Sync and CLI callers handle
    /// 40/64-char lowercase-hex WIRE names throughout; passing one here would
    /// otherwise compile, miss silently, and report a shunned artifact as
    /// servable. `lookup` is unasserted because its bad-key miss is fail-SAFE
    /// (you lose content); a false from here suppresses a suppression.
    ///
    /// CALLER CONTRACT — binding on the sync layer that will wire this in, and
    /// stated here because this slice is where the domain is fixed: an assert
    /// is the right instrument only while every argument is locally derived. A
    /// name off the wire is not. sync §3.3 L527-530 puts this predicate in the
    /// per-round phantom scan, so once a peer's announced names reach it, a
    /// name of any other width would trap the process in a safe build — a
    /// one-packet remote denial of service. The sync boundary MUST therefore
    /// reject a name that is not one of artifact-format §3.2's two forms (a
    /// protocol error on that peer's round) BEFORE decoding it to a digest and
    /// calling this. Widening this predicate to answer false instead is the
    /// wrong repair: an out-of-domain query silently answered "not shunned" is
    /// the fail-open case the assert exists to make impossible.
    pub fn isShunned(self: *const LogStore, hash: []const u8) bool {
        std.debug.assert(isCompleteDigest(hash.len));
        return self.shunHasName(hash);
    }

    /// The whole shun table in persisted order, BORROWED — no allocator, no
    /// error, O(1). Append order is persisted order (the log only grows), so
    /// there is nothing to sort and nothing to copy. Every row's `hash` and
    /// `reason` belong to the store and stay valid until the next `putShun` or
    /// `deinit`.
    pub fn shunList(self: *const LogStore) []const ShunRecord {
        return self.shun_entries.items;
    }

    /// Structural end of the log (header + every complete frame written so
    /// far), for callers that track write progress without duplicating
    /// frame-layout arithmetic (e.g. the DST sweep's crash-recovery model).
    pub fn size(self: *const LogStore) u64 {
        return self.end;
    }

    /// The number of interned frames (phantom or real) held so far — lets
    /// `pack.packLog` assert its destination is a just-opened empty log before
    /// it re-bases anything into it (§6.1 base-availability precondition: a
    /// hint translated via `dst.lookup` must only ever resolve a frame THIS
    /// pack wrote, never a pre-existing unrelated one).
    pub fn entryCount(self: *const LogStore) usize {
        return self.entries.items.len;
    }

    const AppendResult = struct { offset: u64, len: u32 };

    /// Encodes one frame and appends it in a single `StorageFile.append`
    /// call — one call is one atomic torn-write draw point, matching the
    /// crash simulator's fault granularity. On failure, truncates back to
    /// the prior structural end so a partial write never lingers; if that
    /// rollback truncate itself fails the store is poisoned and refuses every
    /// later append.
    fn appendFrame(self: *LogStore, hash: []const u8, bytes: []const u8, kind: FrameKind, delta_of: u64) PutError!AppendResult {
        if (self.poisoned) return error.Storage;
        const hl: u64 = hash.len;
        const pl: u64 = bytes.len;
        const frame_len = frameSpan(hl, pl);
        std.debug.assert(frame_len <= max_payload + 64);

        const frame = try self.gpa.alloc(u8, @intCast(frame_len));
        defer self.gpa.free(frame);
        _ = encodeFrame(frame, hash, bytes, @intFromEnum(kind), delta_of);

        const offset = self.end;
        if (self.file.append(frame)) |_| {
            self.end += frame_len;
            self.frame_count += 1;
            self.last_frame_offset = offset;
            std.debug.assert(self.end % frame_align == 0);
            return .{ .offset = offset, .len = @intCast(pl) };
        } else |_| {
            // A torn append may have left a prefix; if the rollback truncate
            // can't remove it, `end` no longer matches the file end — poison.
            // No `clearBaseMemo` here: the partial frame was never read, so no
            // memo names it (see the invariant on `clearBaseMemo`).
            self.file.truncate(self.end) catch {
                self.poisoned = true;
            };
            self.assertBaseMemoBelowEnd();
            return error.Storage;
        }
    }
};

/// A bounded plaintext memo: base frame offset to its reconstructed plaintext.
/// Two holders, with different lifetimes and for the same reason — whoever can
/// prove the keys still mean what they meant owns it. A derivation sweep
/// (repoview.load) owns one for the span of the sweep; the store owns
/// `LogStore.base_memo` for the write path, because only the store sees the
/// truncation that invalidates a key. See `LogStore.contentCached` and
/// `LogStore.clearBaseMemo`. Explicitly bounded on BOTH a
/// byte budget and an entry count; an insert that would exceed either evicts
/// least-recently-used entries first, and a single item larger than the whole
/// byte budget is simply never cached (the budget never grows). No unbounded
/// allocation and no unbounded memoization — deleting the cache can only change
/// speed, never any answer.
pub const PlaintextCache = struct {
    /// Default resident-byte budget: comfortably holds a delta family's hot
    /// bases without letting the memo grow without limit.
    pub const default_max_bytes: usize = 64 << 20;
    /// The smallest entry this cap plans for. Not a limit on anything — purely
    /// the divisor that turns a BYTE budget into an ENTRY budget, chosen well
    /// below a typical stored plaintext so the entry cap sits above where the
    /// byte cap starts binding for any realistic mix.
    const planning_entry_bytes: usize = 1 << 10;

    /// Default hard cap on resident entries, DERIVED from the byte budget so
    /// the two cannot silently disagree about which one is doing the bounding.
    /// It exists to keep per-entry metadata bounded under many tiny items, not
    /// to bound the cache — that is `max_bytes`'s job.
    ///
    /// It was a flat 4096, and on a real 53,773-commit import it was the ONLY
    /// thing evicting: 99,097 evictions, every one forced by the entry cap and
    /// not one by the byte cap, with the resident set sitting at 23,551 KiB —
    /// roughly a third of a 64 MiB budget that never bound at all. The cache
    /// was held to a third of its own stated size by a second, unrelated number,
    /// and the hit rate that came out of it (51%) looked like a working-set
    /// problem rather than the arithmetic accident it was.
    pub const default_max_entries: usize = default_max_bytes / planning_entry_bytes;

    /// One resident plaintext, threaded on the intrusive recency list. Nodes
    /// are individually heap-allocated so the list pointers survive map
    /// resizes; the map holds only the pointer.
    const Node = struct {
        bytes: []u8,
        offset: u64,
        prev: ?*Node = null,
        next: ?*Node = null,
    };

    map: std.AutoHashMapUnmanaged(u64, *Node) = .empty,
    /// The intrusive recency list: `mru` is the most recently used end, `lru`
    /// the eviction end. Every operation — touch on `get`, insert on `put`,
    /// victim selection on eviction — is O(1) by construction. It replaced a
    /// pick-the-minimum-stamp linear scan whose cost was invisible at small
    /// occupancy and dominant at full: a pkgsrc clone (1M artifacts, cache
    /// pinned at max_entries throughout) spent ~55% of its client CPU inside
    /// that scan, one full-map sweep per stored artifact.
    mru: ?*Node = null,
    lru: ?*Node = null,
    bytes_used: usize = 0,
    /// Resident plaintext never exceeds this many bytes (fixed for a cache's
    /// lifetime; the field form lets a test drive eviction with a small budget).
    max_bytes: usize = default_max_bytes,
    /// Resident entry count never exceeds this.
    max_entries: usize = default_max_entries,
    /// Also memoise top-level `.raw`/`.zlib` plaintext, not just resolved
    /// deltas. Off by default because a LINEAR sweep -- `packLog`'s artifact
    /// loop, `packVerifyPacked` -- reads each artifact exactly once, so keeping
    /// its plaintext would duplicate bytes nothing will ask for again.
    ///
    /// A REVISITING sweep is the opposite case, and `deriveCandidateHints` is
    /// one: consecutive check-ins share nearly all of their sub-manifest tree,
    /// so the same node is resolved once per referencing check-in. Sub-manifests
    /// are stored `.zlib`, so under the default every one of those visits
    /// re-inflates the frame -- which is why a pack profile is dominated by
    /// `flate.Decompress.streamInner` and the LZ77 match-copy beneath it rather
    /// than by any tree-walk arithmetic.
    memoize_undeltaed: bool = false,
    /// Observability only (never gates logic): the number of `get` calls that
    /// returned a memoised plaintext. A persist/derivation sweep over delta
    /// chains proves basis reuse by this rising above zero — a re-reconstruction
    /// that the memo replaced with a lookup.
    hits: u64 = 0,
    /// `get` calls that found nothing. Present because a hit COUNT cannot be
    /// read as a hit RATE: a large `hits` on a much larger number of lookups is
    /// a cache that is barely working, and the two are indistinguishable without
    /// this. `SubManifestMemo` had exactly this blind spot, and its budget
    /// turned out to be costing 23 points of hit rate while looking healthy.
    misses: u64 = 0,
    /// Entries dropped by `evictOne` to stay inside the budgets. Distinguishes
    /// "the working set fits" from "the budgets are thrashing" — a distinction
    /// the hit rate alone leaves ambiguous, because a cache can hold a decent
    /// rate while paying to evict constantly.
    evicted: u64 = 0,
    /// `evicted`, split by WHICH budget forced it. There are two caps and they
    /// bind under different data: many small artifacts exhaust `max_entries`
    /// while `max_bytes` sits idle, and a few large ones do the reverse. Without
    /// the split, a thrashing cache says only "raise something" — with it, the
    /// counter names the budget to raise. The two always sum to `evicted`;
    /// when both caps are over, the byte cap is credited, since raising the
    /// entry cap alone would not have avoided the eviction.
    evicted_for_bytes: u64 = 0,
    evicted_for_entries: u64 = 0,
    /// Items never cached because one alone exceeded the whole byte budget.
    /// Silent otherwise, and it means the budget is too small for the data
    /// rather than merely full.
    oversized: u64 = 0,

    pub fn init() PlaintextCache {
        return .{};
    }

    pub fn deinit(self: *PlaintextCache, gpa: std.mem.Allocator) void {
        var it = self.map.valueIterator();
        while (it.next()) |n| {
            gpa.free(n.*.bytes);
            gpa.destroy(n.*);
        }
        self.map.deinit(gpa);
        self.* = undefined;
    }

    /// Drop every memoized plaintext, leaving the cache usable and its bounds
    /// unchanged. For a holder whose keys have stopped meaning what they meant
    /// — see `LogStore.clearBaseMemo`, where a truncated log can hand an old
    /// offset to a new frame. Retains capacity: the next sweep refills it.
    pub fn clear(self: *PlaintextCache, gpa: std.mem.Allocator) void {
        var it = self.map.valueIterator();
        while (it.next()) |n| {
            gpa.free(n.*.bytes);
            gpa.destroy(n.*);
        }
        self.map.clearRetainingCapacity();
        self.mru = null;
        self.lru = null;
        self.bytes_used = 0;
    }

    /// Unlink `n` from the recency list. `n` must be resident.
    fn detach(self: *PlaintextCache, n: *Node) void {
        if (n.prev) |p| p.next = n.next else self.mru = n.next;
        if (n.next) |x| x.prev = n.prev else self.lru = n.prev;
        n.prev = null;
        n.next = null;
    }

    /// Push `n` onto the MRU end. `n` must be detached.
    fn pushFront(self: *PlaintextCache, n: *Node) void {
        std.debug.assert(n.prev == null and n.next == null);
        n.next = self.mru;
        if (self.mru) |m| m.prev = n else self.lru = n;
        self.mru = n;
    }

    /// The memoized plaintext for `offset`, or null; bumps its LRU recency.
    /// The returned slice is borrowed and valid until the next `put`/`deinit`.
    pub fn get(self: *PlaintextCache, offset: u64) ?[]const u8 {
        const n = self.map.get(offset) orelse {
            self.misses += 1;
            return null;
        };
        self.hits += 1;
        if (self.mru != n) {
            self.detach(n);
            self.pushFront(n);
        }
        return n.bytes;
    }

    /// Memoize a private copy of `src` under `offset`, evicting LRU entries to
    /// stay within both bounds. A no-op when `offset` is already resident (base
    /// offsets are unique keys) or when `src` alone exceeds the byte budget.
    /// Only `error.OutOfMemory` escapes.
    pub fn put(self: *PlaintextCache, gpa: std.mem.Allocator, offset: u64, src: []const u8) error{OutOfMemory}!void {
        if (self.map.contains(offset)) return;
        if (src.len > self.max_bytes) {
            self.oversized += 1; // budget too small for the data, not merely full
            return; // never grow the budget for one item
        }
        // Evict until the newcomer fits under both caps (bounded by the
        // resident count — each pass removes exactly one entry).
        while (true) {
            const over_bytes = self.bytes_used + src.len > self.max_bytes;
            const over_entries = @as(usize, self.map.count()) >= self.max_entries;
            if (!over_bytes and !over_entries) break;
            if (!self.evictOne(gpa)) break; // nothing left to evict
            if (over_bytes) self.evicted_for_bytes += 1 else self.evicted_for_entries += 1;
        }
        if (self.bytes_used + src.len > self.max_bytes) return; // still over: skip
        const owned = try gpa.dupe(u8, src);
        errdefer gpa.free(owned);
        const node = try gpa.create(Node);
        errdefer gpa.destroy(node);
        node.* = .{ .bytes = owned, .offset = offset };
        try self.map.put(gpa, offset, node);
        self.pushFront(node);
        self.bytes_used += owned.len;
    }

    /// Evict the least-recently-used entry — the `lru` end of the recency
    /// list, O(1) — false when the cache is empty.
    fn evictOne(self: *PlaintextCache, gpa: std.mem.Allocator) bool {
        const victim = self.lru orelse return false;
        self.detach(victim);
        _ = self.map.remove(victim.offset);
        std.debug.assert(self.bytes_used >= victim.bytes.len);
        self.bytes_used -= victim.bytes.len;
        gpa.free(victim.bytes);
        gpa.destroy(victim);
        self.evicted += 1; // counted here, not on entry: an empty cache evicts nothing
        return true;
    }
};

fn writeHeader(file: storage.StorageFile, hash_kind: u8) OpenError!void {
    var buf: [header_len]u8 = [_]u8{0} ** header_len;
    @memcpy(buf[0..4], magic);
    std.mem.writeInt(u16, buf[4..6], version, .little);
    buf[6] = hash_kind;
    var crc = crc32.Crc32.init();
    crc.update(buf[0..60]);
    std.mem.writeInt(u32, buf[60..64], crc.final(), .little);
    file.append(&buf) catch return error.Storage;
}

/// Returns the stored `hash_kind` on success. Rejects: bad magic, a stored
/// version other than this reader's, or a header crc mismatch.
fn validateHeader(file: storage.StorageFile) OpenError!u8 {
    var buf: [header_len]u8 = undefined;
    const n = file.pread(0, &buf) catch return error.Storage;
    if (n < header_len) return error.Corrupt;
    if (!std.mem.eql(u8, buf[0..4], magic)) return error.Corrupt;
    const stored_version = std.mem.readInt(u16, buf[4..6], .little);
    if (stored_version != version) return error.Corrupt;
    const stored_crc = std.mem.readInt(u32, buf[60..64], .little);
    var crc = crc32.Crc32.init();
    crc.update(buf[0..60]);
    if (crc.final() != stored_crc) return error.Corrupt;
    return buf[6];
}

/// Outcome of reading one frame at a scan offset: either a torn tail (the
/// file ends mid-frame — truncate and stop) or a validated frame.
/// Whether the open-time scan checksums frame payloads.
///
/// `.verify_payloads` streams every payload byte through the crc, which is what
/// both open paths did historically. That makes open cost Θ(repository bytes):
/// a 410 MB store spends ~440 ms of a ~440 ms open on it, and the cost grows
/// with the whole repository rather than with what the caller asked for.
///
/// `.headers_only` reads and validates each frame's header window and then
/// steps over the payload by `frameSpan` arithmetic, leaving open cost
/// Θ(frame count). Every fault that lives in a header window — a bad prefix
/// crc, a bad header crc, an out-of-range `hash_len`, an unknown flag kind —
/// is still caught, because `readFrameHeader` runs unchanged. What is no
/// longer caught at open is a bitflip in already-landed payload bytes; that
/// becomes `content()`'s job, which crc-checks the frame it reads anyway.
///
/// Torn-tail detection survives the mode change intact. A torn frame is a
/// clean prefix — `append()` writes one frame per call and never leaves
/// damaged bytes — so `offset + frame_len > total` catches every truncation
/// that the payload and trailer short-read checks used to catch, using the
/// length the header window already proved trustworthy.
const ScanMode = enum { verify_payloads, headers_only };

/// One frame's disposition, as the scanner sees it.
///
/// THE INVARIANT THAT MAKES `FrameVerifier`'s FATAL/NON-FATAL SPLIT WORK, and
/// the reason a damaged frame is an outcome rather than an error: everything
/// below `readFrameHeader` runs on a checksum-clean header window, so
/// `frame_len` is proven and the next frame boundary is known. The split is
/// therefore drawn by WHERE the fault was found, not by what kind it was:
///
///   * Any error out of `scanOneFrame` — `Corrupt` or `Storage` — is ALWAYS a
///     HEADER-WINDOW fault, because `readFrameHeader` is the only thing that
///     can still fail that way. `frame_len` is untrustworthy, so a walk must
///     stop where it stands.
///   * `.corrupt` (payload crc) and `.storage_fault` (the device would not
///     yield the payload bytes) are ALWAYS post-header faults on a frame of
///     known extent, so a walk may step over the damaged frame and keep
///     inventorying.
///
/// Raising an error below `readFrameHeader` — instead of adding an outcome arm
/// — would silently break that reading and turn a recoverable finding into a
/// truncated walk. `scanOneFrame`'s error set is narrowed to
/// `error{Corrupt,Storage}` to keep the switch on it total.
const ScanOutcome = union(enum) {
    torn,
    /// A complete frame whose header window verified but whose whole-frame crc
    /// did not match its trailer. `frame_len` is carried so the caller can step
    /// past it; the hash and kind are carried because they are what NAMES the
    /// damage to an operator, and they live in the header window
    /// `readFrameHeader` already proved clean.
    corrupt: struct { hash: [max_hash_len]u8, hash_len: u8, kind: FrameKind, frame_len: u64 },
    /// A complete frame whose header window verified but whose payload or
    /// trailer bytes the device refused to yield. Carries the same fields as
    /// `.corrupt` and for the same reason: one bad sector inside one payload
    /// must not end the inventory of a store an operator is fsck-ing precisely
    /// because the disk is failing.
    storage_fault: struct { hash: [max_hash_len]u8, hash_len: u8, kind: FrameKind, frame_len: u64 },
    frame: struct { hash: [max_hash_len]u8, hash_len: u8, frame_len: u64, payload_len: u32, kind: FrameKind, delta_of: u64 },
};

/// Fixed prefix window: len(4)|flags(1)|hash_len(1)|prefix_crc32(4). Its
/// size does not depend on `hash_len` — that is what lets `hash_len` be
/// proven correct (or rejected) before it is used to size anything else.
const prefix_len: usize = 10;

/// A frame's fully validated fixed header — everything before the payload.
/// The payload and its whole-frame crc are the caller's concern: the
/// content reader allocates and returns the payload, the open-time scanner
/// streams it for the crc without allocating.
const FrameHeader = struct {
    payload_len: u32,
    kind: FrameKind,
    delta_of: u64,
    hash: [max_hash_len]u8,
    hash_len: u8,
    /// Span of prefix+hash+delta_of+header_crc, i.e. the payload's offset
    /// relative to the frame start; the whole-frame crc is seeded over
    /// exactly this many bytes of `header_bytes`.
    fixed_and_crc_len: usize,
    /// The verified header bytes, so the caller seeds the whole-frame crc
    /// without re-reading them.
    header_bytes: [22 + max_hash_len]u8,
};

/// Either a torn tail (a short read on the prefix or header window) or a
/// validated fixed header.
const HeaderScan = union(enum) {
    torn,
    header: FrameHeader,
};

/// Reads and checksum-verifies a frame's fixed header at `offset`, shared by
/// the content reader (`readFrameAt`) and the open-time scanner
/// (`scanOneFrame`). Two-stage: first the fixed-size prefix window
/// (hash_len-independent, so a short read there is unambiguously a torn write
/// — nothing about its size can be thrown off by corruption), which proves
/// `hash_len` trustworthy via its own crc; only then is `hash_len` used to
/// size the rest of the header read, at which point the same "short read =
/// torn, checksum fails = corrupt" reasoning applies validly (append() only
/// ever writes one frame per call and a torn write is a clean prefix — never
/// damaged bytes; a full window whose checksum fails can only be a bitflip on
/// already-landed bytes). A short read on either window is `.torn`; the
/// callers decide its disposition (the scanner recovers, the content reader
/// treats it as `error.Corrupt`). Any flag byte outside
/// {raw,zlib,delta,config,shun} is rejected.
fn readFrameHeader(file: storage.StorageFile, offset: u64) error{ Corrupt, Storage }!HeaderScan {
    var prefix: [prefix_len]u8 = undefined;
    const n0 = file.pread(offset, &prefix) catch return error.Storage;
    if (n0 < prefix_len) return .torn;
    const payload_len = std.mem.readInt(u32, prefix[0..4], .little);
    const flags = prefix[4];
    const hash_len = prefix[5];
    if (hash_len < min_hash_len or hash_len > max_hash_len) return error.Corrupt;

    var pcrc = crc32.Crc32.init();
    pcrc.update(prefix[0..6]);
    if (pcrc.final() != std.mem.readInt(u32, prefix[6..][0..4], .little)) return error.Corrupt;
    // A prefix-verified flag byte is trustworthy (a torn write is a clean
    // truncation, never damaged bytes), so any kind `FrameKind` does not name is
    // a genuine unknown-kind frame, rejected — not a torn tail. The bound is
    // `frame_kind_count`, folded from the enum, so adding a kind admits it here
    // without a second edit and without a stale numeric comparison rejecting
    // every repository that holds one.
    if (flags >= frame_kind_count) return error.Corrupt;

    const hl: usize = hash_len;
    const fixed_and_crc_len: usize = frameHeaderSpan(hash_len);
    var hbuf: [22 + max_hash_len]u8 = undefined;
    const n1 = file.pread(offset, hbuf[0..fixed_and_crc_len]) catch return error.Storage;
    if (n1 < fixed_and_crc_len) return .torn;
    var hcrc = crc32.Crc32.init();
    hcrc.update(hbuf[0 .. 18 + hl]);
    if (hcrc.final() != std.mem.readInt(u32, hbuf[18 + hl ..][0..4], .little)) return error.Corrupt;

    var fh: FrameHeader = .{
        .payload_len = payload_len,
        .kind = @enumFromInt(flags),
        .delta_of = std.mem.readInt(u64, hbuf[10 + hl ..][0..8], .little),
        .hash = undefined,
        .hash_len = hash_len,
        .fixed_and_crc_len = fixed_and_crc_len,
        .header_bytes = undefined,
    };
    @memcpy(fh.hash[0..hl], hbuf[10..][0..hl]);
    @memcpy(fh.header_bytes[0..fixed_and_crc_len], hbuf[0..fixed_and_crc_len]);
    return .{ .header = fh };
}

/// Reads and validates exactly one frame at `offset`, for the open-time scan.
/// Delegates the fixed-header validation to `readFrameHeader`; a torn header,
/// a frame that would run past `total`, or a torn payload/trailer all report
/// `.torn` (recoverable tail), while a failed whole-frame crc is
/// `error.Corrupt`. The payload is streamed through the crc in bounded chunks
/// so a large frame never forces a full-payload allocation.
/// The error set is deliberately narrower than `OpenError`: this function takes
/// no allocator and cannot run out of memory, so stating `error{Corrupt,Storage}`
/// makes that structural rather than something a caller has to assume — and
/// keeps `FrameVerifier`'s switch on the failure total, so an `OutOfMemory`
/// could never be silently relabelled as damage.
fn scanOneFrame(file: storage.StorageFile, offset: u64, total: u64, mode: ScanMode) error{ Corrupt, Storage }!ScanOutcome {
    const fh = switch (try readFrameHeader(file, offset)) {
        .torn => return .torn,
        .header => |h| h,
    };

    const frame_len = frameSpan(fh.hash_len, fh.payload_len);
    if (offset + frame_len > total) return .torn;

    // Under `.headers_only` the frame is accepted here: its header window is
    // crc-clean and `frame_len` — derived from that verified window — proves
    // the payload and trailer are both present in the file. The chunk loop
    // below is the entire Θ(bytes) term of an open, and this is where it is
    // skipped.
    if (mode == .headers_only) {
        return .{ .frame = .{ .hash = fh.hash, .hash_len = fh.hash_len, .frame_len = frame_len, .payload_len = fh.payload_len, .kind = fh.kind, .delta_of = fh.delta_of } };
    }

    var crc = crc32.Crc32.init();
    crc.update(fh.header_bytes[0..fh.fixed_and_crc_len]);
    var remaining: u64 = fh.payload_len;
    var pos = offset + @as(u64, fh.fixed_and_crc_len);
    // 64 KiB, matching `cloneInto`'s existing whole-store streaming buffer
    // (`chunk: [64 * 1024]u8`) so the two byte-streaming loops in this file
    // agree on one read size. Not claimed as a measured win: post-Slice-1+2
    // this loop is dead on every open path and runs only under `zit verify`,
    // where the operator is fsck-ing a whole store and fewer, larger preads is
    // simply the right shape. STATED LIMIT: the slot is reserved for the whole
    // call frame, so every `scanOneFrame` — header-only opens included — now
    // carries 64 KiB of stack. Everything in this program runs on the main
    // stack today; a future worker thread has to size for it.
    var chunk: [64 * 1024]u8 = undefined;
    while (remaining > 0) {
        const take: usize = @intCast(@min(remaining, @as(u64, chunk.len)));
        // A device error HERE is not the same fault as one in the header
        // window: `frame_len` is already proven, so the next boundary is known
        // and a walker can step over this frame. Reported, not raised — see
        // `ScanOutcome`'s invariant.
        const got = file.pread(pos, chunk[0..take]) catch {
            return .{ .storage_fault = .{ .hash = fh.hash, .hash_len = fh.hash_len, .kind = fh.kind, .frame_len = frame_len } };
        };
        if (got < take) return .torn;
        crc.update(chunk[0..take]);
        pos += take;
        remaining -= take;
    }
    var trailer: [4]u8 = undefined;
    const nt = file.pread(pos, &trailer) catch {
        return .{ .storage_fault = .{ .hash = fh.hash, .hash_len = fh.hash_len, .kind = fh.kind, .frame_len = frame_len } };
    };
    if (nt < trailer.len) return .torn;
    // A payload-crc fault is REPORTED, not raised: the header window that
    // produced `frame_len` was itself checksum-verified, so this frame's extent
    // is known and a walker can step over it. `scan` turns this straight back
    // into `error.Corrupt` (open-path behaviour is byte-for-byte unchanged);
    // `FrameVerifier` is the caller that wants to keep going. The hash rides
    // along because it is what lets `zit verify` NAME the damaged artifact —
    // `readFrameHeader` crc-verifies the prefix and header windows separately
    // from the payload, so the name survives a payload bitflip intact.
    if (crc.final() != std.mem.readInt(u32, &trailer, .little)) {
        return .{ .corrupt = .{ .hash = fh.hash, .hash_len = fh.hash_len, .kind = fh.kind, .frame_len = frame_len } };
    }

    return .{ .frame = .{ .hash = fh.hash, .hash_len = fh.hash_len, .frame_len = frame_len, .payload_len = fh.payload_len, .kind = fh.kind, .delta_of = fh.delta_of } };
}

/// Rebuilds `self.entries`/`self.index` by scanning frames from
/// `header_len`. With `recover_torn`, a torn tail truncates to the last valid
/// boundary; otherwise it is reported as corruption without mutating storage.
/// A corrupt complete frame always propagates `error.Corrupt`.
fn scan(self: *LogStore, recover_torn: bool, mode: ScanMode) OpenError!void {
    const total = self.file.size() catch return error.Storage;
    const max_frames = (total - header_len) / min_frame + 1;
    var offset: u64 = header_len;
    var frames: u64 = 0;
    while (offset < total) {
        frames += 1;
        std.debug.assert(frames <= max_frames);
        switch (try scanOneFrame(self.file, offset, total, mode)) {
            // Open aborts on a post-header fault exactly as it did when
            // `scanOneFrame` raised it — the outcome arms exist for
            // `FrameVerifier`, and these two lines are what keep that refactor
            // from touching either open path.
            .corrupt => return error.Corrupt,
            .storage_fault => return error.Storage,
            .torn => {
                if (!recover_torn) return error.Corrupt;
                self.file.truncate(offset) catch return error.Storage;
                self.end = offset;
                return;
            },
            .frame => |f| {
                // Every complete frame — content or config — advances the
                // cache-key tail bookkeeping.
                self.frame_count += 1;
                self.last_frame_offset = offset;
                if (f.kind == .config) {
                    // Config is not interned: it never enters entries/index,
                    // and "latest wins" is just the last one scan reaches (plan E).
                    if (self.config_offset) |prev| std.debug.assert(offset > prev);
                    self.config_offset = offset;
                } else if (f.kind == .shun) {
                    // A shun row is not interned either (`ShunRecord`): it
                    // never enters entries/index, so `lookup` never resolves a
                    // tombstone. Unlike config there is no "latest wins" — the
                    // set accumulates every row in scan order, which IS
                    // persisted order.
                    //
                    // `scanOneFrame` streams the payload through the crc
                    // without allocating, so the row's metadata is re-read here
                    // directly. That read is stack-bounded and needs no
                    // allocation because the declared length is checked against
                    // `max_shun_payload` first — which is also where the
                    // `max_shun_reason` bound is enforced on the READ side.
                    //
                    // The complete-digest key rule is enforced on BOTH sides:
                    // `putShun` asserts it, and a frame arriving from the file
                    // (hand-built, foreign, or written by a future kind) is
                    // rejected outright here. Asserting only on the write path
                    // would leave `open` able to rebuild a set containing an
                    // abbreviated key — the same fail-OPEN hole, reached
                    // through the file instead of the API. Fail-closed: a shun
                    // frame we cannot key correctly voids the whole open.
                    if (!isCompleteDigest(f.hash_len)) return error.Corrupt;
                    if (f.payload_len > max_shun_payload) return error.Corrupt;
                    const pl: usize = f.payload_len;
                    var pbuf: [max_shun_payload]u8 = undefined;
                    const payload_offset = offset + frameHeaderSpan(f.hash_len);
                    const got = self.file.pread(payload_offset, pbuf[0..pl]) catch return error.Storage;
                    // `scanOneFrame` already proved the whole frame present —
                    // under `.verify_payloads` crc-clean as well, under
                    // `.headers_only` by the verified `frame_len` — so a short
                    // read here is the file lying, not a torn tail to recover.
                    if (got < pl) return error.Corrupt;
                    const meta = try parseShun(pbuf[0..pl]);
                    // Not-interned, asserted rather than only tested. The whole
                    // purpose of this arm is to bypass the `else` block below,
                    // and nothing but its shape says so; a refactor folding the
                    // two back together would make `lookup(hash)` resolve a
                    // tombstone as content. Frame-level on purpose — the digest
                    // may legitimately be in `index` already (shunning an
                    // artifact this repo holds is the ordinary case), so what
                    // must not move is the intern count.
                    const entries_before = self.entries.items.len;
                    try self.internShun(f.hash[0..f.hash_len], meta.epoch, meta.reason);
                    std.debug.assert(self.entries.items.len == entries_before);
                } else {
                    // Delta depth is recovered from the base frame, which — by
                    // the append-only, strictly-backward `delta_of` invariant —
                    // was scanned earlier and is already in `entries` (offset
                    // order). A base whose offset is not found is a dangling
                    // delta `content()` will reject; depth 0 is a safe placeholder.
                    const depth: u8 = if (f.kind == .delta)
                        self.depthForBaseOffset(f.delta_of) +| 1
                    else
                        0;
                    const owned = try self.gpa.dupe(u8, f.hash[0..f.hash_len]);
                    errdefer self.gpa.free(owned);
                    try self.entries.append(self.gpa, .{ .hash = owned, .offset = offset, .payload_len = f.payload_len, .kind = f.kind, .phantom = false, .depth = depth });
                    errdefer _ = self.entries.pop();
                    const rid: Rid = self.entries.items.len;
                    try self.index.put(self.gpa, owned, rid);
                }
                const next = offset + f.frame_len;
                std.debug.assert(next > offset);
                offset = next;
            },
        }
    }
    self.end = offset;
}

// ---------------------------------------------------------------------------
// On-demand whole-log verification — the store-level half of `zit verify`.
//
// `open` and `openReadOnly` are ABORT-ON-FIRST-FAULT by construction: their job
// is to hand back a store that can be trusted, so the first damaged frame voids
// the whole open. An fsck has the opposite job — it must INVENTORY damage,
// including on a store no open would accept — so it needs a walker that reports
// findings instead of raising them. That walker is `FrameVerifier`; it is
// deliberately built on `storage.StorageFile` rather than on a `LogStore`, so
// it works on exactly the file an operator is worried about.
// ---------------------------------------------------------------------------

/// One damaged frame, named as concretely as the file allows.
pub const FrameFault = struct {
    /// Byte offset of the frame's first byte.
    offset: u64,
    /// The frame's own hash field, valid for `hash_len` bytes. Meaningful only
    /// when `kind` is non-null: a header-window fault is exactly the case where
    /// the name did not survive.
    hash: [max_hash_len]u8,
    /// Zero on a header-window fault.
    hash_len: u8,
    /// The frame's kind, or null for a header-window fault — see `fatal`.
    kind: ?FrameKind,
    /// Which check failed. `Corrupt` is a checksum that did not match —
    /// the payload's when `kind` is non-null, the header window's own when it
    /// is null. `Storage` is the device refusing to yield the bytes, again on
    /// either side of that line.
    cause: error{ Corrupt, Storage },

    /// Whether this fault ended the walk. A header-window fault leaves
    /// `frame_len` unproven, so there is no trustworthy way to find the next
    /// frame boundary and the verifier stops; a post-header fault does not, so
    /// the verifier steps over it and keeps going.
    pub fn fatal(self: FrameFault) bool {
        return self.kind == null;
    }

    /// The frame's hash, borrowed. Empty on a header-window fault.
    pub fn name(self: *const FrameFault) []const u8 {
        return self.hash[0..self.hash_len];
    }
};

/// What a completed `FrameVerifier` walk covered. `bytes_checked` versus
/// `total` is not decoration: a verb that reports "1 fault" after stopping at
/// 8% of the store has lied about the other 92%.
pub const VerifyTotals = struct {
    /// Frames whose header window verified — including those whose payload
    /// then failed, since the header is what proved they were frames.
    frames: u64,
    /// The offset the walk reached, counted from byte 0 (so the file header is
    /// included).
    bytes_checked: u64,
    /// The size snapshot taken at `init`. Frames a concurrent writer appends
    /// during the walk are outside it and are not examined.
    total: u64,
    /// Whether the walk reached `total` (a torn tail counts as reaching it —
    /// the tail is not part of the log).
    complete: bool,
    /// Offset of an incomplete final frame, when one was found. An interrupted
    /// write, not damage: `open` recovers it silently by truncating.
    torn_tail_at: ?u64,
};

/// Walks every frame of a log file, whole-frame-checksumming each one, and
/// yields each fault instead of raising it.
///
/// `next` returns `?FrameFault` rather than an error union on purpose: for an
/// fsck every failure is a reportable finding, and an error union would force
/// the caller back into abort-on-first-fault, which is the exact behaviour this
/// type exists to replace.
pub const FrameVerifier = struct {
    file: storage.StorageFile,
    /// The size snapshot; see `VerifyTotals.total`.
    total: u64,
    offset: u64,
    frames: u64 = 0,
    complete: bool = false,
    torn_tail_at: ?u64 = null,
    /// Latched once the walk has ended, for any reason. `next` answers null
    /// forever after.
    done: bool = false,
    /// Mirrors `scan`'s bound so a pathological advance cannot spin.
    max_frames: u64,

    /// Validates the file header, then positions at the first frame. A file
    /// header that will not verify is `error.Corrupt` from here: nothing past
    /// it can be read as frames at all, so there is no inventory to take.
    pub fn init(file: storage.StorageFile) OpenError!FrameVerifier {
        const total = file.size() catch return error.Storage;
        if (total < header_len) return error.Corrupt;
        _ = try validateHeader(file);
        return .{
            .file = file,
            .total = total,
            .offset = header_len,
            .max_frames = (total - header_len) / min_frame + 1,
        };
    }

    /// The next fault, or null when the walk is over. After a fatal fault
    /// (`FrameFault.fatal`) the walk is over too.
    pub fn next(self: *FrameVerifier) ?FrameFault {
        while (!self.done and self.offset < self.total) {
            // Room for at least one more frame, which is the same bound `scan`
            // states as `frames <= max_frames` after its own increment. Every
            // frame advances by at least `min_frame`, so with `offset < total`
            // this cannot be driven over by file contents.
            std.debug.assert(self.frames < self.max_frames);
            // THE MUTATION-PROOF LEVER: this `.verify_payloads` literal is the
            // only one in the file outside `openReadOnly`. Point it at
            // `.headers_only` and `zit verify` stops checking payload bytes.
            const outcome = scanOneFrame(self.file, self.offset, self.total, .verify_payloads) catch |e| {
                // Either the header window failed its own checksum or the
                // device would not yield it. Both leave `frame_len` unproven,
                // so there is no next boundary to step to. The switch is total
                // over `scanOneFrame`'s narrowed error set: there is no `else`
                // arm here to quietly relabel some future error as damage.
                self.done = true;
                self.complete = false;
                return .{
                    .offset = self.offset,
                    .hash = undefined,
                    .hash_len = 0,
                    .kind = null,
                    .cause = switch (e) {
                        error.Storage => error.Storage,
                        error.Corrupt => error.Corrupt,
                    },
                };
            };
            switch (outcome) {
                .torn => {
                    // Not a fault. `open` recovers a torn tail silently
                    // (`recover_torn = true`) because it means an interrupted
                    // write, not rot; the next write truncates it away.
                    self.torn_tail_at = self.offset;
                    self.done = true;
                    self.complete = true;
                    return null;
                },
                .corrupt => |c| {
                    const at = self.offset;
                    self.frames += 1;
                    const advanced = self.offset + c.frame_len;
                    std.debug.assert(advanced > self.offset);
                    self.offset = advanced;
                    return .{ .offset = at, .hash = c.hash, .hash_len = c.hash_len, .kind = c.kind, .cause = error.Corrupt };
                },
                .storage_fault => |c| {
                    // One bad sector inside one payload is a FINDING, not the
                    // end of the inventory. The header window proved this
                    // frame's extent before the read failed, so the next
                    // boundary is known and the walk goes on — which is the
                    // whole point on the failing disk this verb exists for.
                    const at = self.offset;
                    self.frames += 1;
                    const advanced = self.offset + c.frame_len;
                    std.debug.assert(advanced > self.offset);
                    self.offset = advanced;
                    return .{ .offset = at, .hash = c.hash, .hash_len = c.hash_len, .kind = c.kind, .cause = error.Storage };
                },
                .frame => |f| {
                    self.frames += 1;
                    const advanced = self.offset + f.frame_len;
                    std.debug.assert(advanced > self.offset);
                    self.offset = advanced;
                },
            }
        }
        if (!self.done) {
            // THE CATASTROPHE THIS GUARDS: a walk that ended early without
            // reporting why reads as a clean store. `scanOneFrame` answers
            // `.torn` whenever `offset + frame_len > total`, so every arm that
            // advances lands inside the file and the loop can only fall out
            // with `offset` exactly on `total`. Anything else is a bug in the
            // advance arithmetic, and it must not be reported as "0 faults".
            std.debug.assert(self.offset == self.total);
            self.done = true;
            self.complete = true;
        }
        return null;
    }

    pub fn totals(self: *const FrameVerifier) VerifyTotals {
        return .{
            .frames = self.frames,
            .bytes_checked = self.offset,
            .total = self.total,
            .complete = self.complete,
            .torn_tail_at = self.torn_tail_at,
        };
    }
};

/// Encodes one frame into `buf` (which must be at least the padded frame
/// span) and returns the written slice. Shared by the production write
/// path and by tests that hand-craft torn/corrupt fixtures.
///
/// Layout (all little-endian): len(4) | flags(1) | hash_len(1) |
/// prefix_crc32(4) = crc32(bytes[0..6)) | hash(hash_len) | delta_of(8) |
/// header_crc32(4) = crc32(bytes[0..18+hash_len)) | payload(len) |
/// frame_crc32(4) = crc32(bytes[0..22+hash_len+len)). `delta_of` is the
/// base frame's byte offset for a `delta` frame, and 0 otherwise.
fn encodeFrame(buf: []u8, hash: []const u8, payload: []const u8, flags: u8, delta_of: u64) []const u8 {
    std.debug.assert(hash.len >= min_hash_len and hash.len <= max_hash_len);
    std.debug.assert(payload.len <= max_payload);
    const hl: usize = hash.len;
    const pl: usize = payload.len;
    const hash_len: u8 = @intCast(hl);
    const payload_len: u32 = @intCast(pl);
    const unpadded: usize = frame_fixed_overhead + hl + pl;
    const frame_len: usize = @intCast(frameSpan(hl, pl));
    std.debug.assert(buf.len >= frame_len);

    std.mem.writeInt(u32, buf[0..4], payload_len, .little);
    buf[4] = flags;
    buf[5] = hash_len;
    var pcrc = crc32.Crc32.init();
    pcrc.update(buf[0..6]);
    std.mem.writeInt(u32, buf[6..][0..4], pcrc.final(), .little);

    @memcpy(buf[10..][0..hl], hash);
    std.mem.writeInt(u64, buf[10 + hl ..][0..8], delta_of, .little);

    var hcrc = crc32.Crc32.init();
    hcrc.update(buf[0 .. 18 + hl]);
    std.mem.writeInt(u32, buf[18 + hl ..][0..4], hcrc.final(), .little);

    @memcpy(buf[22 + hl ..][0..pl], payload);

    var fcrc = crc32.Crc32.init();
    fcrc.update(buf[0 .. 22 + hl + pl]);
    std.mem.writeInt(u32, buf[22 + hl + pl ..][0..4], fcrc.final(), .little);

    if (frame_len > unpadded) @memset(buf[unpadded..frame_len], 0);
    return buf[0..frame_len];
}

fn alignUp8(comptime T: type, n: T) T {
    const a: T = frame_align;
    return (n + (a - 1)) / a * a;
}

/// The fixed per-frame overhead (bytes) that brackets the variable hash+payload
/// on disk: len(4) + flags(1) + hash_len(1) + prefix_crc32(4) + delta_of(8) +
/// header_crc32(4) + frame_crc32(4) = 26 (see `encodeFrame`'s layout). Named
/// once so every on-disk span — `frameSpan`, `appendFrame`, and the decode/scan
/// paths — derives the same number and pack's §6.1 monotonic floor can never
/// silently drift from the span `appendFrame` commits.
///
/// Public because the storage simulation needs the same number to say which
/// bytes of a frame are crc-relevant. It had a hand-copied `22` — the header
/// alone — which silently excluded the trailing `frame_crc32` from the span it
/// called meaningful, so an injected bitflip landing in a frame's own checksum
/// read as inert. Deriving it here instead of re-spelling it there is what
/// keeps the simulator's idea of a frame and the encoder's the same shape.
pub const frame_fixed_overhead = 26;

/// The aligned on-disk span of a frame carrying `hash_len`-byte hash and
/// `payload_len`-byte payload — the same `alignUp8(frame_fixed_overhead + hl +
/// pl)` `appendFrame` commits. Used by `frameSpanOf` and pack's §6.1
/// monotonic-shrink guard to compare a re-encoding's byte cost against the frame
/// it would replace, and by the scan/decode paths to advance frame-to-frame.
fn frameSpan(hash_len: u64, payload_len: u64) u64 {
    return alignUp8(u64, frame_fixed_overhead + hash_len + payload_len);
}

/// The payload's offset relative to a frame's start: the fixed header window
/// `encodeFrame` lays out ahead of it — len(4) + flags(1) + hash_len(1) +
/// prefix_crc32(4) + hash(hash_len) + delta_of(8) + header_crc32(4). Named once
/// so `readFrameHeader`'s window and the scan's direct payload read of a shun
/// row derive the same number from the same layout.
///
/// Carries its own domain check rather than inheriting each caller's: the result
/// slices a `[22 + max_hash_len]u8` stack buffer and derives a file offset, so a
/// future third call site reaching here before validating `hash_len` would slice
/// out of bounds. Both current callers pass an already-validated width, which is
/// exactly why the check belongs to the helper and not to them.
fn frameHeaderSpan(hash_len: u8) usize {
    std.debug.assert(hash_len >= min_hash_len and hash_len <= max_hash_len);
    return 22 + @as(usize, hash_len);
}

/// Encodes `cfg` into `buf` as `[hash_policy u8][branch_len u16 LE][branch]
/// [pid_len u16 LE][pid][pcode_len u16 LE][pcode][login_len u16 LE][login]
/// [secret_len u16 LE][secret][url_len u16 LE][url]` and returns the written
/// slice (plan E fixed field order — deterministic, not a hash-map dump). Every
/// string field is always written (length zero when empty), so decode
/// consumption stays exact. Fixed bounds, no unbounded copy.
fn encodeConfig(buf: []u8, cfg: Config) []const u8 {
    std.debug.assert(buf.len >= max_config_payload);
    std.debug.assert(cfg.default_branch.len <= max_config_str);
    std.debug.assert(cfg.project_id.len <= max_config_str);
    std.debug.assert(cfg.project_code.len <= max_config_str);
    std.debug.assert(cfg.login_user.len <= max_config_str);
    std.debug.assert(cfg.login_secret_hex.len <= max_config_str);
    std.debug.assert(cfg.remote_url.len <= max_config_str);
    const bl: usize = cfg.default_branch.len;
    const pl: usize = cfg.project_id.len;
    const cl: usize = cfg.project_code.len;
    const ul: usize = cfg.login_user.len;
    const sl: usize = cfg.login_secret_hex.len;
    const rl: usize = cfg.remote_url.len;

    buf[0] = cfg.hash_policy;
    std.mem.writeInt(u16, buf[1..3], @intCast(bl), .little);
    @memcpy(buf[3..][0..bl], cfg.default_branch);
    std.mem.writeInt(u16, buf[3 + bl ..][0..2], @intCast(pl), .little);
    @memcpy(buf[5 + bl ..][0..pl], cfg.project_id);
    std.mem.writeInt(u16, buf[5 + bl + pl ..][0..2], @intCast(cl), .little);
    @memcpy(buf[7 + bl + pl ..][0..cl], cfg.project_code);
    std.mem.writeInt(u16, buf[7 + bl + pl + cl ..][0..2], @intCast(ul), .little);
    @memcpy(buf[9 + bl + pl + cl ..][0..ul], cfg.login_user);
    std.mem.writeInt(u16, buf[9 + bl + pl + cl + ul ..][0..2], @intCast(sl), .little);
    @memcpy(buf[11 + bl + pl + cl + ul ..][0..sl], cfg.login_secret_hex);
    std.mem.writeInt(u16, buf[11 + bl + pl + cl + ul + sl ..][0..2], @intCast(rl), .little);
    @memcpy(buf[13 + bl + pl + cl + ul + sl ..][0..rl], cfg.remote_url);
    // The four existing submanifest settings bytes are always written as one
    // block. A pre-settings frame ends earlier and parses with defaults. The
    // optional digest is local persistence, present only after a settled scan,
    // and it is written as a shape-tagged extension so a later field can change
    // the block without making this frame unreadable — see
    // `config_shape_cost_shape_sha256` for the direction that buys.
    const submanifest_pos = 13 + bl + pl + cl + ul + sl + rl;
    buf[submanifest_pos] = @intFromBool(cfg.seen_submanifest);
    buf[submanifest_pos + 1] = @intFromBool(cfg.enable_submanifests);
    buf[submanifest_pos + 2] = cfg.submanifest_k;
    buf[submanifest_pos + 3] = @intFromBool(cfg.submanifest_auto_retune);
    if (cfg.submanifest_settled_cost_shape) |digest| {
        buf[submanifest_pos + 4] = config_shape_cost_shape_sha256;
        @memcpy(buf[submanifest_pos + 5 ..][0..digest.len], &digest);
        return buf[0 .. submanifest_pos + 5 + digest.len];
    }
    return buf[0 .. submanifest_pos + 4];
}

/// Parses an `encodeConfig` payload. Every LE length read is bounds-checked
/// against the remaining slice before use, each length is capped at
/// `max_config_str`, and total consumption must be exact (trailing bytes are
/// corruption). Borrows `payload`. All wire-derived, validated before use.
fn parseConfig(payload: []const u8) error{Corrupt}!Config {
    // The writer asserts this same bound (`encodeConfig`), so the reader states
    // it too rather than trusting the frame: the string fields are each capped
    // at `max_config_str`, but the trailing shape block is open-ended by design
    // (an unknown tag is dropped, not rejected on width), so without this a
    // structurally intact frame could carry a trailing run up to `readFrameAt`'s
    // 1 GiB `max_payload` and still parse. Both ends now enforce one limit.
    if (payload.len > max_config_payload) return error.Corrupt;
    if (payload.len < 1) return error.Corrupt; // no hash_policy byte
    const hash_policy = payload[0];
    var pos: usize = 1;

    if (payload.len - pos < 2) return error.Corrupt;
    const blen: usize = std.mem.readInt(u16, payload[pos..][0..2], .little);
    pos += 2;
    if (blen > max_config_str) return error.Corrupt;
    if (payload.len - pos < blen) return error.Corrupt;
    const branch = payload[pos..][0..blen];
    pos += blen;

    if (payload.len - pos < 2) return error.Corrupt;
    const plen: usize = std.mem.readInt(u16, payload[pos..][0..2], .little);
    pos += 2;
    if (plen > max_config_str) return error.Corrupt;
    if (payload.len - pos < plen) return error.Corrupt;
    const pid = payload[pos..][0..plen];
    pos += plen;

    if (payload.len - pos < 2) return error.Corrupt;
    const clen: usize = std.mem.readInt(u16, payload[pos..][0..2], .little);
    pos += 2;
    if (clen > max_config_str) return error.Corrupt;
    if (payload.len - pos < clen) return error.Corrupt;
    const pcode = payload[pos..][0..clen];
    pos += clen;

    // sync §6.3 — the login credential fields are an optional trailing block:
    // a frame written before they existed ends exactly here, and parses with
    // an empty credential (forward-compatible), never as corruption.
    if (pos == payload.len) {
        return .{ .hash_policy = hash_policy, .default_branch = branch, .project_id = pid, .project_code = pcode };
    }

    if (payload.len - pos < 2) return error.Corrupt;
    const ulen: usize = std.mem.readInt(u16, payload[pos..][0..2], .little);
    pos += 2;
    if (ulen > max_config_str) return error.Corrupt;
    if (payload.len - pos < ulen) return error.Corrupt;
    const login = payload[pos..][0..ulen];
    pos += ulen;

    if (payload.len - pos < 2) return error.Corrupt;
    const slen: usize = std.mem.readInt(u16, payload[pos..][0..2], .little);
    pos += 2;
    if (slen > max_config_str) return error.Corrupt;
    if (payload.len - pos < slen) return error.Corrupt;
    const secret = payload[pos..][0..slen];
    pos += slen;

    // sync §4/§6 — the saved remote URL is another optional trailing block:
    // a frame written before it existed ends exactly here and parses with an
    // empty remote (forward-compatible), never as corruption.
    if (pos == payload.len) {
        return .{ .hash_policy = hash_policy, .default_branch = branch, .project_id = pid, .project_code = pcode, .login_user = login, .login_secret_hex = secret };
    }

    if (payload.len - pos < 2) return error.Corrupt;
    const rlen: usize = std.mem.readInt(u16, payload[pos..][0..2], .little);
    pos += 2;
    if (rlen > max_config_str) return error.Corrupt;
    if (payload.len - pos < rlen) return error.Corrupt;
    const url = payload[pos..][0..rlen];
    pos += rlen;

    // Submanifest settings are the final optional trailing block. A frame
    // written before them ends here and parses with defaults. Past the four
    // settings bytes the block is `[shape_tag u8][body]`, the body width fixed
    // by the tag. A tag this reader knows, carrying the body width it knows, is
    // the only case that yields a digest; everything else — an unknown tag, or a
    // known tag whose body width disagrees — is dropped, leaving the digest
    // absent so the next commit re-scans.
    //
    // Dropping rather than rejecting is deliberate in BOTH cases, and it is not
    // the same as accepting a malformed shape: a dropped digest still forces the
    // scan it exists to skip. `readFrameAt` has already verified this frame's CRC
    // before `parseConfig` sees a byte, so a width disagreement here cannot be
    // media corruption — it can only be a differently-versioned or buggy writer,
    // exactly the population the tag exists to tolerate. Returning `error.Corrupt`
    // would take `default_branch`, `project_id`, `project_code`, `login_user`,
    // `login_secret_hex` and `remote_url` down with one regenerable cache field,
    // rendering the repository unopenable to save a scan. See
    // `config_shape_cost_shape_sha256` for the compatibility direction this buys.
    var seen_submanifest = false;
    var enable_submanifests = false;
    var submanifest_k: u8 = 0;
    var submanifest_auto_retune = false;
    var submanifest_settled_cost_shape: ?[config_shape_digest_len]u8 = null;
    if (pos != payload.len) {
        const trailing = payload.len - pos;
        // Widths written before the tag existed: a partial settings block (1 or
        // 3), the full one (4), and the untagged digest (4 + 32). Those four are
        // recognized by width forever, so repositories already carrying them
        // keep parsing; everything new carries a tag instead. The reservation
        // costs one future shape the right to have a 31-byte body.
        if (trailing < 4 and trailing != 1 and trailing != 3) return error.Corrupt;
        seen_submanifest = payload[pos] != 0;
        if (trailing >= 3) {
            enable_submanifests = payload[pos + 1] != 0;
            submanifest_k = payload[pos + 2];
        }
        if (trailing >= 4) {
            submanifest_auto_retune = payload[pos + 3] != 0;
        }
        if (trailing == config_shape_legacy_width) {
            submanifest_settled_cost_shape = payload[pos + 4 ..][0..config_shape_digest_len].*;
        } else if (trailing > 4) {
            const tag = payload[pos + 4];
            const body = payload[pos + 5 ..];
            if (tag == config_shape_cost_shape_sha256 and body.len == config_shape_digest_len) {
                submanifest_settled_cost_shape = body[0..config_shape_digest_len].*;
            }
        }
        pos += trailing;
    }

    if (pos != payload.len) return error.Corrupt; // no trailing bytes
    return .{
        .hash_policy = hash_policy,
        .default_branch = branch,
        .project_id = pid,
        .project_code = pcode,
        .login_user = login,
        .login_secret_hex = secret,
        .remote_url = url,
        .seen_submanifest = seen_submanifest,
        .enable_submanifests = enable_submanifests,
        .submanifest_k = submanifest_k,
        .submanifest_auto_retune = submanifest_auto_retune,
        .submanifest_settled_cost_shape = submanifest_settled_cost_shape,
    };
}

/// One shun row's payload fields, borrowing the buffer they were parsed from.
/// The digest is NOT here: it lives in the frame's own `hash` field, which is
/// what makes a row self-describing on a raw byte scan.
const ShunMeta = struct { epoch: i64, reason: []const u8 };

/// Encodes a shun row's payload into `buf` as `[epoch i64 LE][reason_len u16
/// LE][reason]` and returns the written slice — the same fixed-field-order,
/// deterministic shape as `encodeConfig`. `epoch` is written signed and read
/// back signed; storage never interprets it.
///
/// The `reason` bound stays an assert here — and only here — because its one
/// production caller, `putShun`, rejects an over-long reason UNCONDITIONALLY
/// before reaching this (`error.ReasonTooLong`). That is deliberate: the copy
/// below is into a fixed `max_shun_payload` buffer, so an unchecked over-long
/// reason is a stack overrun, not a formatting defect, and a bound that has to
/// survive `ReleaseFast` cannot live in an assert. Truncating here instead
/// would be worse than rejecting: the durable row and the in-memory row would
/// then disagree about the operator's own text.
fn encodeShun(buf: []u8, epoch: i64, reason: []const u8) []const u8 {
    std.debug.assert(buf.len >= max_shun_payload);
    std.debug.assert(reason.len <= max_shun_reason);
    const rl: usize = reason.len;
    std.mem.writeInt(i64, buf[0..8], epoch, .little);
    std.mem.writeInt(u16, buf[8..10], @intCast(rl), .little);
    @memcpy(buf[10..][0..rl], reason);
    return buf[0 .. 10 + rl];
}

/// Parses an `encodeShun` payload. The length word is bounds-checked against
/// the remaining slice before use, capped at `max_shun_reason`, and total
/// consumption must be exact (trailing bytes are corruption) — the same
/// discipline `parseConfig` applies field by field. Borrows `payload`.
fn parseShun(payload: []const u8) error{Corrupt}!ShunMeta {
    if (payload.len < 8) return error.Corrupt; // no epoch
    const epoch = std.mem.readInt(i64, payload[0..8], .little);
    var pos: usize = 8;

    if (payload.len - pos < 2) return error.Corrupt;
    const rlen: usize = std.mem.readInt(u16, payload[pos..][0..2], .little);
    pos += 2;
    if (rlen > max_shun_reason) return error.Corrupt;
    if (payload.len - pos < rlen) return error.Corrupt;
    const reason = payload[pos..][0..rlen];
    pos += rlen;

    if (pos != payload.len) return error.Corrupt; // no trailing bytes
    return .{ .epoch = epoch, .reason = reason };
}

/// Map the canonical §4.1 decode error set (`codec.DecodeError`) into this
/// layer's `ContentError`: a malformed container — short header, a declared
/// capacity over `max_output`, or a corrupt RFC1950 stream — is `Corrupt`;
/// OOM propagates. The §4.1 codec itself is the leaf `codec` module
/// (`container.zig`), shared with `repostate` and `sync`.
fn decodeContainer(gpa: std.mem.Allocator, wire: []const u8, max_output: usize) ContentError![]u8 {
    return codec.decode(gpa, wire, max_output) catch |e| switch (e) {
        error.OutOfMemory => error.OutOfMemory,
        error.TruncatedHeader, error.InvalidStream, error.OutputTooLarge => error.Corrupt,
    };
}

// ---------------------------------------------------------------------------
// Framing tests (no RFC section behind this file; the oracle is the module
// doc's crash-recovery invariants above).
// ---------------------------------------------------------------------------

fn openTmp(tmp: anytype, io: std.Io, name: []const u8) !storage.FileStorage {
    const raw = try tmp.dir.createFile(io, name, .{ .read = true });
    return storage.FileStorage.init(io, raw, 0);
}

fn writeRawHeader(file: storage.StorageFile, magic_bytes: []const u8, ver: u16, hk: u8, bad_crc: bool) !void {
    std.debug.assert(magic_bytes.len == 4);
    var buf: [header_len]u8 = [_]u8{0} ** header_len;
    @memcpy(buf[0..4], magic_bytes);
    std.mem.writeInt(u16, buf[4..6], ver, .little);
    buf[6] = hk;
    var crc = crc32.Crc32.init();
    crc.update(buf[0..60]);
    var crc_val = crc.final();
    if (bad_crc) crc_val ^= 1;
    std.mem.writeInt(u32, buf[60..64], crc_val, .little);
    try file.append(&buf);
}

test "logstore §C: header round-trips magic/version/hash_kind through open" {
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    const io = std.testing.io;
    var fs = try openTmp(&tmp, io, "log.bin");
    defer fs.file.close(io);
    const file = fs.storageFile();

    var store1 = try LogStore.open(std.testing.allocator, file, 7);
    store1.deinit(std.testing.allocator);

    var raw: [header_len]u8 = undefined;
    const n = try file.pread(0, &raw);
    try std.testing.expectEqual(@as(usize, header_len), n);
    try std.testing.expectEqualStrings(magic, raw[0..4]);
    try std.testing.expectEqual(@as(u16, version), std.mem.readInt(u16, raw[4..6], .little));
    try std.testing.expectEqual(@as(u8, 7), raw[6]);
    for (raw[7..60]) |b| try std.testing.expectEqual(@as(u8, 0), b);
    var crc = crc32.Crc32.init();
    crc.update(raw[0..60]);
    try std.testing.expectEqual(crc.final(), std.mem.readInt(u32, raw[60..64], .little));

    var store2 = try LogStore.open(std.testing.allocator, file, 0); // hint ignored on reopen
    defer store2.deinit(std.testing.allocator);
    try std.testing.expectEqual(@as(u8, 7), store2.hash_kind);
}

test "logstore: logKey + validateTail track the append-only tail in O(1)" {
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    const io = std.testing.io;
    var fs = try openTmp(&tmp, io, "log.bin");
    defer fs.file.close(io);
    const file = fs.storageFile();

    // Empty log: no complete frames, end at the header boundary; validateTail
    // accepts the empty key exactly at that length. logKey is defined over an
    // already-scanned store, so `open` (which scans, and for an empty file just
    // writes the header) is the store the key is read from.
    const empty = blk: {
        var store = try LogStore.open(std.testing.allocator, file, 0);
        defer store.deinit(std.testing.allocator);
        const k = try store.logKey();
        break :blk k;
    };
    try std.testing.expectEqual(@as(u64, 0), empty.frame_count);
    try std.testing.expectEqual(@as(u64, header_len), empty.end);
    try std.testing.expect(LogStore.validateTail(file, empty));

    // Append two frames through a writable store, then read the key from a
    // freshly-scanned store — the exact shape production keys on (`repocache`
    // computes the LogKey over an `openReadOnly` scan, never a mid-append
    // writable store, whose tail bookkeeping is only established by a scan).
    {
        var store = try LogStore.open(std.testing.allocator, file, 0);
        defer store.deinit(std.testing.allocator);
        _ = try store.put(&([_]u8{1} ** 32), "payload one");
        _ = try store.put(&([_]u8{2} ** 32), "payload two two");
        try store.commit();
    }
    var scan2 = try LogStore.openReadOnly(std.testing.allocator, file);
    const k2 = try scan2.logKey();
    scan2.deinit(std.testing.allocator);
    try std.testing.expectEqual(@as(u64, 2), k2.frame_count);
    try std.testing.expect(LogStore.validateTail(file, k2)); // still fresh
    try std.testing.expect(!LogStore.validateTail(file, empty)); // stale (shorter) key is a miss

    // Appending one frame advances the key: the old key no longer validates.
    {
        var store = try LogStore.open(std.testing.allocator, file, 0);
        defer store.deinit(std.testing.allocator);
        _ = try store.put(&([_]u8{3} ** 32), "third");
        try store.commit();
    }
    try std.testing.expect(!LogStore.validateTail(file, k2)); // grown log ⇒ miss
    var scan3 = try LogStore.openReadOnly(std.testing.allocator, file);
    const k3 = try scan3.logKey();
    scan3.deinit(std.testing.allocator);
    try std.testing.expectEqual(@as(u64, 3), k3.frame_count);
    try std.testing.expect(k3.end > k2.end);
    try std.testing.expect(LogStore.validateTail(file, k3));
}

test "logstore §3.6/§8.3-8: resolveHexPrefix resolves a unique prefix, refuses an ambiguous one, misses an unknown" {
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    const io = std.testing.io;
    var fs = try openTmp(&tmp, io, "log.bin");
    defer fs.file.close(io);
    const file = fs.storageFile();

    // Two artifacts sharing the "abcd" nibble prefix and one distinct, so
    // "abcd" is ambiguous while "abcd00"/"ef" each name exactly one.
    const h_shared_a = [_]u8{ 0xab, 0xcd, 0x00 } ++ [_]u8{0} ** 29; // abcd00...
    const h_shared_b = [_]u8{ 0xab, 0xcd, 0x11 } ++ [_]u8{0} ** 29; // abcd11...
    const h_other = [_]u8{ 0xef, 0x01 } ++ [_]u8{0} ** 30; // ef01...
    {
        var store = try LogStore.open(std.testing.allocator, file, 0);
        defer store.deinit(std.testing.allocator);
        _ = try store.put(&h_shared_a, "alpha");
        _ = try store.put(&h_shared_b, "bravo");
        _ = try store.put(&h_other, "other");
        try store.commit();
    }
    var store = try LogStore.openReadOnly(std.testing.allocator, file);
    defer store.deinit(std.testing.allocator);

    // Unique 6-char prefix → the exact rid; the full name round-trips.
    const rid_a = (try store.resolveHexPrefix("abcd00")).?;
    try std.testing.expectEqualSlices(u8, &h_shared_a, store.hashOf(rid_a));
    // A distinct 4-char prefix names the third artifact uniquely.
    const rid_e = (try store.resolveHexPrefix("ef01")).?;
    try std.testing.expectEqualSlices(u8, &h_other, store.hashOf(rid_e));

    // Ambiguous 4-char prefix → refused, never one silently picked.
    try std.testing.expectError(error.Ambiguous, store.resolveHexPrefix("abcd"));

    // A prefix no artifact carries → null (the not-found family, not an error).
    try std.testing.expectEqual(@as(?Rid, null), try store.resolveHexPrefix("dead"));
}

test "logstore §C: open rejects a corrupt file-header crc" {
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    const io = std.testing.io;
    var fs = try openTmp(&tmp, io, "log.bin");
    defer fs.file.close(io);
    const file = fs.storageFile();
    try writeRawHeader(file, magic, version, 0, true);
    try std.testing.expectError(error.Corrupt, LogStore.open(std.testing.allocator, file, 0));
}

test "logstore §C: open rejects a version newer than this reader" {
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    const io = std.testing.io;
    var fs = try openTmp(&tmp, io, "log.bin");
    defer fs.file.close(io);
    const file = fs.storageFile();
    try writeRawHeader(file, magic, version + 1, 0, false);
    try std.testing.expectError(error.Corrupt, LogStore.open(std.testing.allocator, file, 0));
}

test "logstore §C: the frame header crc covers len/flags/hash_len/hash/delta_of" {
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    const io = std.testing.io;
    var fs = try openTmp(&tmp, io, "log.bin");
    defer fs.file.close(io);
    const file = fs.storageFile();

    var seed = try LogStore.open(std.testing.allocator, file, 0);
    seed.deinit(std.testing.allocator); // header only

    var buf: [64]u8 = undefined;
    const frame = encodeFrame(&buf, "aaaa", "payload", @intFromEnum(FrameKind.raw), 0);
    var corrupted: [64]u8 = undefined;
    @memcpy(corrupted[0..frame.len], frame);
    corrupted[10] ^= 0xff; // inside the hash bytes, part of the fixed header
    try file.append(corrupted[0..frame.len]);

    try std.testing.expectError(error.Corrupt, LogStore.open(std.testing.allocator, file, 0));
}

test "logstore §C(inv2): a genuinely torn (prefix-truncated) final frame truncates to exactly the last valid boundary" {
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    const io = std.testing.io;
    var fs = try openTmp(&tmp, io, "log.bin");
    defer fs.file.close(io);
    const file = fs.storageFile();

    var store = try LogStore.open(std.testing.allocator, file, 0);
    _ = try store.put("frame-one-hash", "payload-one");
    try store.commit();
    const boundary = store.size();
    store.deinit(std.testing.allocator);

    var buf: [96]u8 = undefined;
    const frame2 = encodeFrame(&buf, "frame-two-hash", "payload-two", @intFromEnum(FrameKind.raw), 0);
    try file.append(frame2[0..10]); // short of even the fixed header+crc window

    var reopened = try LogStore.open(std.testing.allocator, file, 0);
    defer reopened.deinit(std.testing.allocator);
    try std.testing.expectEqual(boundary, reopened.size());
    try std.testing.expect(reopened.lookup("frame-two-hash") == null);
}

test "logstore §C(inv2): a torn payload/crc tail truncates exactly" {
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    const io = std.testing.io;
    var fs = try openTmp(&tmp, io, "log.bin");
    defer fs.file.close(io);
    const file = fs.storageFile();

    var store = try LogStore.open(std.testing.allocator, file, 0);
    _ = try store.put("frame-a-hash-here", "hello-payload");
    try store.commit();
    const boundary = store.size();
    store.deinit(std.testing.allocator);

    var buf: [128]u8 = undefined;
    const frame2 = encodeFrame(&buf, "frame-b-hash-value", "a longer second payload for this test", @intFromEnum(FrameKind.raw), 0);
    const header_window = 22 + "frame-b-hash-value".len;
    try file.append(frame2[0 .. header_window + 3]); // full header+crc landed; payload/crc badly short

    var reopened = try LogStore.open(std.testing.allocator, file, 0);
    defer reopened.deinit(std.testing.allocator);
    try std.testing.expectEqual(boundary, reopened.size());
}

test "logstore §C(inv4): a bitflipped header on a durable non-final frame surfaces error.Corrupt" {
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    const io = std.testing.io;
    var fs = try openTmp(&tmp, io, "log.bin");
    defer fs.file.close(io);
    const file = fs.storageFile();

    var seed = try LogStore.open(std.testing.allocator, file, 0);
    seed.deinit(std.testing.allocator);

    var buf1: [64]u8 = undefined;
    const f1 = encodeFrame(&buf1, "frame-one-abc", "payload-one", @intFromEnum(FrameKind.raw), 0);
    var corrupted: [64]u8 = undefined;
    @memcpy(corrupted[0..f1.len], f1);
    corrupted[10] ^= 0xff;
    try file.append(corrupted[0..f1.len]);

    var buf2: [64]u8 = undefined;
    const f2 = encodeFrame(&buf2, "frame-two-def", "payload-two", @intFromEnum(FrameKind.raw), 0);
    try file.append(f2);

    try std.testing.expectError(error.Corrupt, LogStore.open(std.testing.allocator, file, 0));
}

test "logstore §C(inv4): a bitflipped complete final frame is error.Corrupt" {
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    const io = std.testing.io;
    var fs = try openTmp(&tmp, io, "log.bin");
    defer fs.file.close(io);
    const file = fs.storageFile();

    var seed = try LogStore.open(std.testing.allocator, file, 0);
    seed.deinit(std.testing.allocator);

    var buf: [64]u8 = undefined;
    const f = encodeFrame(&buf, "final-frame-hash", "payload-for-final", @intFromEnum(FrameKind.raw), 0);
    var corrupted: [64]u8 = undefined;
    @memcpy(corrupted[0..f.len], f);
    corrupted[6] ^= 0xff;
    try file.append(corrupted[0..f.len]); // complete, fully landed, nothing follows

    try std.testing.expectError(error.Corrupt, LogStore.open(std.testing.allocator, file, 0));
}

test "logstore §C(inv4): a corrupted payload on a complete frame surfaces error.Corrupt on read, on both open paths" {
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    const io = std.testing.io;
    var fs = try openTmp(&tmp, io, "log.bin");
    defer fs.file.close(io);
    const file = fs.storageFile();

    var seed = try LogStore.open(std.testing.allocator, file, 0);
    seed.deinit(std.testing.allocator);

    var buf: [96]u8 = undefined;
    const f = encodeFrame(&buf, "payload-hash-value", "some payload bytes here", @intFromEnum(FrameKind.raw), 0);
    var corrupted: [96]u8 = undefined;
    @memcpy(corrupted[0..f.len], f);
    const header_window = 22 + "payload-hash-value".len;
    corrupted[header_window] ^= 0xff; // inside the payload, not the fixed header
    try file.append(corrupted[0..f.len]);

    // BOTH open paths scan `.headers_only`, so a payload bitflip no longer
    // voids either open — the header window is intact and the frame is
    // complete. The fault is not lost, only relocated: `content()` crc-checks
    // the frame it reads, so the first read of the damaged artifact reports it.
    //
    // This assertion was written in Slice 1 with its second half inverted --
    // `openReadOnly` still rejected eagerly then, and the assertion existed as
    // a tripwire against deleting `verified_on_open` without an fsck verb to
    // replace it. `zit verify` shipped, the flag went, and the tripwire fired
    // on exactly the commit it was aimed at. Both halves now read the same way
    // because both paths behave the same way.
    var store = try LogStore.open(std.testing.allocator, file, 0);
    defer store.deinit(std.testing.allocator);
    const rid = store.lookup("payload-hash-value") orelse return error.TestUnexpectedResult;
    try std.testing.expectError(error.Corrupt, store.content(std.testing.allocator, rid));

    var ro = try LogStore.openReadOnly(std.testing.allocator, file);
    defer ro.deinit(std.testing.allocator);
    const ro_rid = ro.lookup("payload-hash-value") orelse return error.TestUnexpectedResult;
    try std.testing.expectError(error.Corrupt, ro.content(std.testing.allocator, ro_rid));
}

// ---------------------------------------------------------------------------
// FrameVerifier / openForVerify — the store half of `zit verify`.
// ---------------------------------------------------------------------------

/// Appends a frame whose header window is valid and whose payload has been
/// bitflipped after encoding, so the whole-frame crc fails while the name,
/// kind, and length stay readable. The exact fixture shape a payload fault
/// takes on disk.
fn appendPayloadCorruptFrame(file: storage.StorageFile, hash: []const u8, payload: []const u8) !void {
    var buf: [256]u8 = undefined;
    const f = encodeFrame(&buf, hash, payload, @intFromEnum(FrameKind.raw), 0);
    var corrupted: [256]u8 = undefined;
    @memcpy(corrupted[0..f.len], f);
    corrupted[frameHeaderSpan(@intCast(hash.len))] ^= 0xff; // first payload byte
    try file.append(corrupted[0..f.len]);
}

test "logstore verify: FrameVerifier names a payload-corrupt frame by offset, hash, and kind" {
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    const io = std.testing.io;
    var fs = try openTmp(&tmp, io, "log.bin");
    defer fs.file.close(io);
    const file = fs.storageFile();

    var seed = try LogStore.open(std.testing.allocator, file, 0);
    seed.deinit(std.testing.allocator);

    const at = file.size() catch unreachable;
    try appendPayloadCorruptFrame(file, "payload-hash-value", "some payload bytes here");

    var v = try FrameVerifier.init(file);
    const fault = v.next() orelse return error.TestExpectedFault;
    // The name is the whole point: a fault an operator cannot paste into
    // `zit artifact` is a fault they cannot act on.
    try std.testing.expectEqual(at, fault.offset);
    try std.testing.expectEqualStrings("payload-hash-value", fault.name());
    try std.testing.expectEqual(FrameKind.raw, fault.kind.?);
    try std.testing.expect(!fault.fatal());

    try std.testing.expect(v.next() == null);
    const t = v.totals();
    try std.testing.expect(t.complete);
    try std.testing.expectEqual(@as(u64, 1), t.frames);
    try std.testing.expectEqual(t.total, t.bytes_checked);
    try std.testing.expect(t.torn_tail_at == null);
}

test "logstore verify: FrameVerifier steps over a payload fault and finds the next one" {
    // INVENTORY, NOT FAIL-FAST — the property that separates an fsck from
    // pack's abort-on-first-fault pre-flight. A verb that stopped here would
    // send an operator round the loop once per damaged frame.
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    const io = std.testing.io;
    var fs = try openTmp(&tmp, io, "log.bin");
    defer fs.file.close(io);
    const file = fs.storageFile();

    var seed = try LogStore.open(std.testing.allocator, file, 0);
    seed.deinit(std.testing.allocator);

    const first_at = file.size() catch unreachable;
    try appendPayloadCorruptFrame(file, "first-damaged-hash", "the first payload");
    // A clean frame between them, so "found the second" cannot be an artifact
    // of the two being adjacent.
    var clean_buf: [256]u8 = undefined;
    try file.append(encodeFrame(&clean_buf, "an-intact-hash-here", "intact payload", @intFromEnum(FrameKind.raw), 0));
    const second_at = file.size() catch unreachable;
    try appendPayloadCorruptFrame(file, "second-damaged-hash", "the second payload");

    var v = try FrameVerifier.init(file);
    const a = v.next() orelse return error.TestExpectedFault;
    try std.testing.expectEqual(first_at, a.offset);
    try std.testing.expectEqualStrings("first-damaged-hash", a.name());
    const b = v.next() orelse return error.TestExpectedFault;
    try std.testing.expectEqual(second_at, b.offset);
    try std.testing.expectEqualStrings("second-damaged-hash", b.name());
    try std.testing.expect(v.next() == null);

    const t = v.totals();
    try std.testing.expect(t.complete);
    try std.testing.expectEqual(@as(u64, 3), t.frames);
    try std.testing.expectEqual(t.total, t.bytes_checked);
}

/// A `StorageFile` that forwards everything to a real one, except that any
/// `pread` overlapping `[bad_from, bad_to)` fails. The one bad sector this
/// program cannot otherwise produce on demand.
const BadSectorFile = struct {
    inner: storage.StorageFile,
    bad_from: u64,
    bad_to: u64,

    fn storageFile(self: *BadSectorFile) storage.StorageFile {
        return .{ .context = self, .vtable = &vtable };
    }

    const vtable: storage.StorageFile.VTable = .{
        .append = appendImpl,
        .pread = preadImpl,
        .sync = syncImpl,
        .size = sizeImpl,
        .truncate = truncateImpl,
    };

    fn appendImpl(ctx: *anyopaque, bytes: []const u8) storage.StorageError!void {
        const self: *BadSectorFile = @ptrCast(@alignCast(ctx));
        return self.inner.append(bytes);
    }
    fn preadImpl(ctx: *anyopaque, offset: u64, buf: []u8) storage.StorageError!usize {
        const self: *BadSectorFile = @ptrCast(@alignCast(ctx));
        if (offset < self.bad_to and offset + buf.len > self.bad_from) return error.Io;
        return self.inner.pread(offset, buf);
    }
    fn syncImpl(ctx: *anyopaque) storage.StorageError!void {
        const self: *BadSectorFile = @ptrCast(@alignCast(ctx));
        return self.inner.sync();
    }
    fn sizeImpl(ctx: *anyopaque) storage.StorageError!u64 {
        const self: *BadSectorFile = @ptrCast(@alignCast(ctx));
        return self.inner.size();
    }
    fn truncateImpl(ctx: *anyopaque, len: u64) storage.StorageError!void {
        const self: *BadSectorFile = @ptrCast(@alignCast(ctx));
        return self.inner.truncate(len);
    }
};

test "logstore verify: an unreadable payload is a finding, not the end of the walk" {
    // ONE BAD SECTOR MUST NOT END THE INVENTORY. A failing disk is the primary
    // reason an operator reaches for an fsck; a walker that stopped at the
    // first I/O error would report one line about a 40 GB store and call it a
    // day. The header window proved this frame's extent before the read
    // failed, so the next boundary is known.
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    const io = std.testing.io;
    var fs = try openTmp(&tmp, io, "log.bin");
    defer fs.file.close(io);
    const file = fs.storageFile();

    var seed = try LogStore.open(std.testing.allocator, file, 0);
    seed.deinit(std.testing.allocator);

    const first_at = file.size() catch unreachable;
    var a_buf: [256]u8 = undefined;
    try file.append(encodeFrame(&a_buf, "unreadable-hash-here", "the unreadable payload", @intFromEnum(FrameKind.raw), 0));
    const second_at = file.size() catch unreachable;
    try appendPayloadCorruptFrame(file, "second-damaged-hash", "the second payload");
    const total = file.size() catch unreachable;

    // The bad sector sits in the FIRST frame's payload, past its header window.
    var bad = BadSectorFile{
        .inner = file,
        .bad_from = first_at + @as(u64, frameHeaderSpan(@intCast("unreadable-hash-here".len))),
        .bad_to = second_at,
    };

    var v = try FrameVerifier.init(bad.storageFile());
    const a = v.next() orelse return error.TestExpectedFault;
    try std.testing.expectEqual(first_at, a.offset);
    try std.testing.expect(a.cause == error.Storage);
    try std.testing.expectEqualStrings("unreadable-hash-here", a.name());
    // NOT fatal: the walk goes on, which is the entire property under test.
    try std.testing.expect(!a.fatal());

    const b = v.next() orelse return error.TestExpectedFault;
    try std.testing.expectEqual(second_at, b.offset);
    try std.testing.expect(b.cause == error.Corrupt);
    try std.testing.expect(v.next() == null);

    const t = v.totals();
    try std.testing.expect(t.complete);
    try std.testing.expectEqual(total, t.bytes_checked);
    try std.testing.expectEqual(@as(u64, 2), t.frames);
}

test "logstore verify: scan still returns error.Storage for an unreadable payload under .verify_payloads" {
    // The `.storage_fault` outcome arm must not have moved open-path behaviour:
    // an open still refuses a store whose bytes the device will not yield.
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    const io = std.testing.io;
    var fs = try openTmp(&tmp, io, "log.bin");
    defer fs.file.close(io);
    const file = fs.storageFile();

    var seed = try LogStore.open(std.testing.allocator, file, 0);
    seed.deinit(std.testing.allocator);
    const at = file.size() catch unreachable;
    var buf: [256]u8 = undefined;
    try file.append(encodeFrame(&buf, "unreadable-hash-here", "the unreadable payload", @intFromEnum(FrameKind.raw), 0));

    var bad = BadSectorFile{
        .inner = file,
        .bad_from = at + @as(u64, frameHeaderSpan(@intCast("unreadable-hash-here".len))),
        .bad_to = file.size() catch unreachable,
    };
    var probe: LogStore = .{ .gpa = std.testing.allocator, .file = bad.storageFile(), .hash_kind = 0, .end = header_len };
    defer probe.deinit(std.testing.allocator);
    try std.testing.expectError(error.Storage, scan(&probe, false, .verify_payloads));
}

test "logstore verify: a header fault stops the walk and totals say the walk was short" {
    // A SHORT WALK IS NEVER SILENTLY REPORTED AS A CLEAN ONE. Past a damaged
    // header window there is no trustworthy frame boundary, so the verifier
    // stops — and must say so, or "1 fault" would imply the rest is fine.
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    const io = std.testing.io;
    var fs = try openTmp(&tmp, io, "log.bin");
    defer fs.file.close(io);
    const file = fs.storageFile();

    var seed = try LogStore.open(std.testing.allocator, file, 0);
    seed.deinit(std.testing.allocator);

    const at = file.size() catch unreachable;
    var buf: [128]u8 = undefined;
    const f = encodeFrame(&buf, "header-damaged-hash", "payload bytes", @intFromEnum(FrameKind.raw), 0);
    var corrupted: [128]u8 = undefined;
    @memcpy(corrupted[0..f.len], f);
    corrupted[6] ^= 0xff; // the prefix crc itself
    try file.append(corrupted[0..f.len]);
    var tail_buf: [128]u8 = undefined;
    try file.append(encodeFrame(&tail_buf, "unreached-frame-hash", "never examined", @intFromEnum(FrameKind.raw), 0));

    var v = try FrameVerifier.init(file);
    const fault = v.next() orelse return error.TestExpectedFault;
    try std.testing.expectEqual(at, fault.offset);
    try std.testing.expect(fault.fatal());
    try std.testing.expect(fault.kind == null);
    try std.testing.expectEqual(@as(usize, 0), fault.name().len);
    try std.testing.expect(v.next() == null);

    const t = v.totals();
    try std.testing.expect(!t.complete);
    try std.testing.expectEqual(at, t.bytes_checked);
    try std.testing.expect(t.bytes_checked < t.total);
}

test "logstore verify: openReadOnly admits a payload-corrupt store and the read, not the open, reports it" {
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    const io = std.testing.io;
    var fs = try openTmp(&tmp, io, "log.bin");
    defer fs.file.close(io);
    const file = fs.storageFile();

    var seed = try LogStore.open(std.testing.allocator, file, 0);
    seed.deinit(std.testing.allocator);
    try appendPayloadCorruptFrame(file, "payload-hash-value", "some payload bytes here");

    // THE SLICE 2 CONTRACT, both halves. This store used to be refused at the
    // door by `openReadOnly`; that eager whole-log check is gone, so the open
    // now succeeds and every verb can operate on the undamaged rest of the
    // repository.
    var store = try LogStore.openReadOnly(std.testing.allocator, file);
    defer store.deinit(std.testing.allocator);

    // The fault is relocated, NOT dropped. `readFrameAt` crc-checks every frame
    // it reads with no exemption left to grant, so the damage surfaces the
    // first time anyone asks for those bytes. If this assertion ever goes
    // green-by-silence, a rotted payload is being handed to callers as truth.
    const rid = store.lookup("payload-hash-value") orelse return error.TestUnexpectedResult;
    try std.testing.expectError(error.Corrupt, store.content(std.testing.allocator, rid));
}

test "logstore verify: scan still returns error.Corrupt for a payload-corrupt frame under .verify_payloads" {
    // The `ScanOutcome.corrupt` refactor must not have moved open-path
    // behaviour by one byte; this is the pin that says so.
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    const io = std.testing.io;
    var fs = try openTmp(&tmp, io, "log.bin");
    defer fs.file.close(io);
    const file = fs.storageFile();

    var seed = try LogStore.open(std.testing.allocator, file, 0);
    seed.deinit(std.testing.allocator);
    try appendPayloadCorruptFrame(file, "payload-hash-value", "some payload bytes here");

    var probe: LogStore = .{ .gpa = std.testing.allocator, .file = file, .hash_kind = 0, .end = header_len };
    defer probe.deinit(std.testing.allocator);
    try std.testing.expectError(error.Corrupt, scan(&probe, false, .verify_payloads));
}

test "logstore §C(inv4): a bitflipped hash_len that would need more bytes than the file has surfaces error.Corrupt, not a torn tail" {
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    const io = std.testing.io;
    var fs = try openTmp(&tmp, io, "log.bin");
    defer fs.file.close(io);
    const file = fs.storageFile();

    var seed = try LogStore.open(std.testing.allocator, file, 0);
    seed.deinit(std.testing.allocator);

    // hash_len=1, payload="" -> frame_len = align8(26+1+0) = 32, fully
    // landed, nothing follows. Corrupting hash_len to an in-range value
    // (20) makes the naive "read 18+hash_len bytes, short = torn" reading
    // demand 38 bytes from a 32-byte frame -- exactly the false-torn gap
    // the prefix checksum closes: it fails on this corrupted byte before
    // any window-size decision is made from it.
    var buf: [64]u8 = undefined;
    const f = encodeFrame(&buf, "h", "", @intFromEnum(FrameKind.raw), 0);
    var corrupted: [64]u8 = undefined;
    @memcpy(corrupted[0..f.len], f);
    corrupted[5] = 20;
    try file.append(corrupted[0..f.len]);

    try std.testing.expectError(error.Corrupt, LogStore.open(std.testing.allocator, file, 0));
}

test "logstore §C: a crc-valid frame with an unknown flag kind (not raw/zlib/delta/config/shun) is rejected on open" {
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    const io = std.testing.io;
    var fs = try openTmp(&tmp, io, "log.bin");
    defer fs.file.close(io);
    const file = fs.storageFile();

    var seed = try LogStore.open(std.testing.allocator, file, 0);
    seed.deinit(std.testing.allocator);

    // The first flag byte `FrameKind` does not name: both crcs are computed
    // correctly over it, so the frame is byte-intact -- it is simply an
    // encoding this reader does not know, which must fail closed rather than be
    // interpreted. Taken from `frame_kind_count` rather than written as 5, so
    // adding a kind moves this fixture to the next unknown byte instead of
    // asserting that a now-KNOWN kind is rejected.
    try std.testing.expectEqual(@as(usize, 5), frame_kind_count); // raw|zlib|delta|config|shun
    var buf: [64]u8 = undefined;
    const f = encodeFrame(&buf, "unknown-flag-hash", "some payload bytes", @intCast(frame_kind_count), 0);
    try file.append(f);

    try std.testing.expectError(error.Corrupt, LogStore.open(std.testing.allocator, file, 0));
}

// ---------------------------------------------------------------------------
// zlib and delta frames (container sync §4.1; verbatim serve sync §7.2;
// delta apply delta §4).
// ---------------------------------------------------------------------------

/// Appends one hand-crafted frame (valid crcs) to `file`, for fixtures that
/// need a payload/flag/delta_of a normal put would never produce.
fn appendCrafted(gpa: std.mem.Allocator, file: storage.StorageFile, hash: []const u8, payload: []const u8, kind: FrameKind, delta_of: u64) !void {
    const frame_len: usize = @intCast(frameSpan(hash.len, payload.len));
    const buf = try gpa.alloc(u8, frame_len);
    defer gpa.free(buf);
    const f = encodeFrame(buf, hash, payload, @intFromEnum(kind), delta_of);
    try file.append(f);
}

test "logstore §C: content on a malformed zlib container is error.Corrupt" {
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    const io = std.testing.io;
    var fs = try openTmp(&tmp, io, "log.bin");
    defer fs.file.close(io);
    const file = fs.storageFile();

    var seed = try LogStore.open(std.testing.allocator, file, 0);
    seed.deinit(std.testing.allocator);

    // A crc-valid zlib frame whose payload is not a valid container: two bytes
    // is short of even the 4-byte length header, so decodeContainer rejects it.
    const hash = "malformed-zlib-container-hash-01";
    try appendCrafted(std.testing.allocator, file, hash, "\x00\x00", .zlib, 0);

    var store = try LogStore.open(std.testing.allocator, file, 0);
    defer store.deinit(std.testing.allocator);
    const rid = store.lookup(hash).?;
    try std.testing.expectError(error.Corrupt, store.content(std.testing.allocator, rid));
}

test "logstore §C/sync §4.1: a zlib container whose declared length exceeds max_payload is rejected without a huge allocation" {
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    const io = std.testing.io;
    var fs = try openTmp(&tmp, io, "log.bin");
    defer fs.file.close(io);
    const file = fs.storageFile();

    var seed = try LogStore.open(std.testing.allocator, file, 0);
    seed.deinit(std.testing.allocator);

    // Declared length 0xFFFFFFFF is far past max_payload; decodeContainer must
    // reject on the declared value before inflating anything.
    const hash = "oversize-declared-length-hash-01";
    try appendCrafted(std.testing.allocator, file, hash, "\xff\xff\xff\xff\x78\x9c", .zlib, 0);

    var store = try LogStore.open(std.testing.allocator, file, 0);
    defer store.deinit(std.testing.allocator);
    const rid = store.lookup(hash).?;
    try std.testing.expectError(error.Corrupt, store.content(std.testing.allocator, rid));
}

test "logstore §C/delta §4: content on a delta whose inflated payload is not a valid delta is error.Corrupt" {
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    const io = std.testing.io;
    var fs = try openTmp(&tmp, io, "log.bin");
    defer fs.file.close(io);
    const file = fs.storageFile();

    var seed = try LogStore.open(std.testing.allocator, file, 0);
    seed.deinit(std.testing.allocator);

    // A valid raw base, then a crc-valid delta frame whose container inflates
    // to bytes that are not a conforming delta -> delta.apply rejects -> Corrupt.
    try appendCrafted(std.testing.allocator, file, "garbage-delta-base-hash-00000001", "base content here", .raw, 0);
    const container = try codec.encode(std.testing.allocator, "this is not a delta at all", .default);
    defer std.testing.allocator.free(container);
    try appendCrafted(std.testing.allocator, file, "garbage-delta-hash-value-0000001", container, .delta, header_len);

    var store = try LogStore.open(std.testing.allocator, file, 0);
    defer store.deinit(std.testing.allocator);
    const rid = store.lookup("garbage-delta-hash-value-0000001").?;
    try std.testing.expectError(error.Corrupt, store.content(std.testing.allocator, rid));
}

test "logstore §C: content on a delta whose delta_of is out of range or mid-frame is error.Corrupt" {
    const cases = [_]struct { name: []const u8, delta_of: u64 }{
        .{ .name = "below-header", .delta_of = 0 },
        .{ .name = "forward-ref", .delta_of = 1 << 20 },
        .{ .name = "mid-base-frame", .delta_of = header_len + 3 },
    };
    for (cases) |c| {
        var tmp = std.testing.tmpDir(.{});
        defer tmp.cleanup();
        const io = std.testing.io;
        var fs = try openTmp(&tmp, io, "log.bin");
        defer fs.file.close(io);
        const file = fs.storageFile();

        var seed = try LogStore.open(std.testing.allocator, file, 0);
        seed.deinit(std.testing.allocator);

        try appendCrafted(std.testing.allocator, file, "bad-delta-of-base-hash-000000001", "base content bytes", .raw, 0);
        const container = try codec.encode(std.testing.allocator, "1\ndelta-ish payload; only delta_of matters here", .default);
        defer std.testing.allocator.free(container);
        try appendCrafted(std.testing.allocator, file, "bad-delta-of-hash-value-00000001", container, .delta, c.delta_of);

        var store = try LogStore.open(std.testing.allocator, file, 0);
        defer store.deinit(std.testing.allocator);
        const rid = store.lookup("bad-delta-of-hash-value-00000001").?;
        try std.testing.expectError(error.Corrupt, store.content(std.testing.allocator, rid));
    }
}

// Seeded deterministic fuzz: decode/apply on adversarial wire bytes must
// yield a value or a declared error, never a crash or a hang.

test "logstore §C/sync §4.1 fuzz: a garbage zlib container never crashes content" {
    var prng = std.Random.DefaultPrng.init(0x2172B);
    const random = prng.random();
    var iter: usize = 0;
    while (iter < 150) : (iter += 1) {
        var tmp = std.testing.tmpDir(.{});
        defer tmp.cleanup();
        const io = std.testing.io;
        var fs = try openTmp(&tmp, io, "log.bin");
        defer fs.file.close(io);
        const file = fs.storageFile();

        var store = try LogStore.open(std.testing.allocator, file, 0);
        store.deinit(std.testing.allocator); // header only; craft the frame directly

        // A crc-valid zlib frame with a random payload must never crash content.
        var garbage: [64]u8 = undefined;
        const gn = random.uintLessThan(usize, garbage.len + 1);
        random.bytes(garbage[0..gn]);
        var ghash: [9]u8 = undefined;
        ghash[0] = 'g';
        std.mem.writeInt(u64, ghash[1..9], iter, .little);
        try appendCrafted(std.testing.allocator, file, &ghash, garbage[0..gn], .zlib, 0);
        var reopened = try LogStore.open(std.testing.allocator, file, 0);
        defer reopened.deinit(std.testing.allocator);
        const grid = reopened.lookup(&ghash).?;
        if (reopened.content(std.testing.allocator, grid)) |maybe| {
            if (maybe) |b| std.testing.allocator.free(b);
        } else |err| switch (err) {
            error.OutOfMemory, error.Storage, error.Corrupt => {},
        }
    }
}

test "logstore §C/delta §4 fuzz: a garbage delta payload or delta_of never crashes content" {
    var prng = std.Random.DefaultPrng.init(0xD317A);
    const random = prng.random();
    var iter: usize = 0;
    while (iter < 150) : (iter += 1) {
        var tmp = std.testing.tmpDir(.{});
        defer tmp.cleanup();
        const io = std.testing.io;
        var fs = try openTmp(&tmp, io, "log.bin");
        defer fs.file.close(io);
        const file = fs.storageFile();

        var store = try LogStore.open(std.testing.allocator, file, 0);
        store.deinit(std.testing.allocator); // header only; craft the frame directly

        // A crc-valid delta frame with a random container payload and a random
        // delta_of must never crash content (Corrupt/OOM/value only).
        var garbage: [64]u8 = undefined;
        const gn = random.uintLessThan(usize, garbage.len + 1);
        random.bytes(garbage[0..gn]);
        var ghash: [9]u8 = undefined;
        ghash[0] = 'x';
        std.mem.writeInt(u64, ghash[1..9], iter, .little);
        try appendCrafted(std.testing.allocator, file, &ghash, garbage[0..gn], .delta, random.int(u64));
        var reopened = try LogStore.open(std.testing.allocator, file, 0);
        defer reopened.deinit(std.testing.allocator);
        const grid = reopened.lookup(&ghash).?;
        if (reopened.content(std.testing.allocator, grid)) |maybe| {
            if (maybe) |b| std.testing.allocator.free(b);
        } else |err| switch (err) {
            error.OutOfMemory, error.Storage, error.Corrupt => {},
        }
    }
}

test "logstore §C: a config frame is not interned — rids stay content-only and contiguous" {
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    const io = std.testing.io;
    var fs = try openTmp(&tmp, io, "log.bin");
    defer fs.file.close(io);
    const file = fs.storageFile();

    var store = try LogStore.open(std.testing.allocator, file, 0);
    defer store.deinit(std.testing.allocator);

    const r1 = try store.put("config-interleave-hash-one-0001", "one");
    try store.putConfig(.{ .hash_policy = 0, .default_branch = "trunk", .project_id = "pid" });
    const r2 = try store.put("config-interleave-hash-two-0001", "two");
    // The config frame consumed a byte range but no rid: content rids stay
    // dense and contiguous, and the sentinel hash is never an index key.
    try std.testing.expectEqual(r1 + 1, r2);
    try std.testing.expect(store.lookup(config_sentinel) == null);
}

test "logstore §C: a delta base offset pointing at a config frame is rejected as corrupt" {
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    const io = std.testing.io;
    var fs = try openTmp(&tmp, io, "log.bin");
    defer fs.file.close(io);
    const file = fs.storageFile();

    var seed = try LogStore.open(std.testing.allocator, file, 0);
    seed.deinit(std.testing.allocator); // header only

    // A config frame at header_len, then a delta whose (valid) container base
    // offset points back at it: resolving the base hits the .config arm.
    var cbuf: [max_config_payload]u8 = undefined;
    const cfg_payload = encodeConfig(&cbuf, .{ .hash_policy = 4, .default_branch = "trunk", .project_id = "pid" });
    try appendCrafted(std.testing.allocator, file, config_sentinel, cfg_payload, .config, 0);
    const container = try codec.encode(std.testing.allocator, "delta-ish bytes; only the base offset matters here", .default);
    defer std.testing.allocator.free(container);
    try appendCrafted(std.testing.allocator, file, "delta-over-config-base-hash-0001", container, .delta, header_len);

    var store = try LogStore.open(std.testing.allocator, file, 0);
    defer store.deinit(std.testing.allocator);
    const rid = store.lookup("delta-over-config-base-hash-0001").?;
    try std.testing.expectError(error.Corrupt, store.content(std.testing.allocator, rid));
}

test "logstore plan E: parseConfig rejects malformed config payloads" {
    var over: [3]u8 = undefined;
    over[0] = 0; // hash_policy
    std.mem.writeInt(u16, over[1..3], @intCast(max_config_str + 1), .little); // branch_len > cap
    const cases = [_][]const u8{
        &[_]u8{}, // (a) empty: no hash_policy byte
        &[_]u8{ 0x07, 0x05 }, // (b) truncated branch_len (1 of 2 bytes)
        &over, // (c) branch_len exceeds max_config_str
        &[_]u8{ 0x00, 0x05, 0x00, 'a', 'b' }, // (d) branch_len 5 exceeds remaining
        &[_]u8{ 0x00, 0x01, 0x00, 'b', 0x02 }, // (e) truncated pid_len after branch
        &[_]u8{ 0x00, 0x01, 0x00, 'b', 0x00, 0x00, 0x09 }, // (f) trailing byte after pid
    };
    for (cases) |c| try std.testing.expectError(error.Corrupt, parseConfig(c));
}

test "logstore plan E/§6.3: parseConfig forward-compat-decodes a legacy pre-credential frame" {
    // A frame written before the login credential block existed carries only
    // the three string fields (branch, project_id, project_code) and ends
    // exactly after project_code — encodeConfig can no longer emit this shape
    // (it always writes the two trailing length words), so the bytes are hand-
    // built. parseConfig's forward-compat arm must accept it with the credential
    // fields defaulting empty, never as corruption.
    const legacy = [_]u8{
        0x04, // hash_policy
        0x05, 0x00, 't', 'r', 'u', 'n', 'k', // branch_len=5, "trunk"
        0x03, 0x00, 'p', 'i', 'd', // pid_len=3, "pid"
        0x05, 0x00, 'c', 'o', 'd', 'e', '9', // pcode_len=5, "code9"
    };
    const cfg = try parseConfig(&legacy);
    try std.testing.expectEqual(@as(u8, 4), cfg.hash_policy);
    try std.testing.expectEqualStrings("trunk", cfg.default_branch);
    try std.testing.expectEqualStrings("pid", cfg.project_id);
    try std.testing.expectEqualStrings("code9", cfg.project_code);
    try std.testing.expectEqual(@as(usize, 0), cfg.login_user.len);
    try std.testing.expectEqual(@as(usize, 0), cfg.login_secret_hex.len);
    try std.testing.expectEqual(@as(usize, 0), cfg.remote_url.len);
}

test "config logstore plan E: encodeConfig/parseConfig round-trips remote_url" {
    // sync §4/§6 — the saved remote is the sixth trailing block; encoding a
    // full config and decoding it must recover the URL verbatim with every
    // other field intact (the fixed plan E field order is a lossless round-trip).
    var buf: [max_config_payload]u8 = undefined;
    const wire = encodeConfig(&buf, .{
        .hash_policy = 7,
        .default_branch = "trunk",
        .project_id = "pid7",
        .project_code = "code7",
        .login_user = "alice",
        .login_secret_hex = "0123456789abcdef0123456789abcdef01234567", // 40 hex
        .remote_url = "http://h.example/xfer",
    });
    const cfg = try parseConfig(wire);
    try std.testing.expectEqual(@as(u8, 7), cfg.hash_policy);
    try std.testing.expectEqualStrings("trunk", cfg.default_branch);
    try std.testing.expectEqualStrings("pid7", cfg.project_id);
    try std.testing.expectEqualStrings("code7", cfg.project_code);
    try std.testing.expectEqualStrings("alice", cfg.login_user);
    try std.testing.expectEqualStrings("0123456789abcdef0123456789abcdef01234567", cfg.login_secret_hex);
    try std.testing.expectEqualStrings("http://h.example/xfer", cfg.remote_url);
}

test "config logstore plan E: a pre-remote frame decodes with empty remote_url" {
    // A frame written before the remote-URL block existed carries the five
    // pre-URL fields (branch, pid, pcode, login, secret) and ends exactly after
    // the secret — encodeConfig can no longer emit this shape (it always writes
    // the URL length word), so the bytes are hand-built. parseConfig's
    // forward-compat arm must accept it with remote_url empty, never Corrupt.
    const pre_remote = [_]u8{
        0x04, // hash_policy
        0x05, 0x00, 't', 'r', 'u', 'n', 'k', // branch_len=5
        0x03, 0x00, 'p', 'i', 'd', // pid_len=3
        0x05, 0x00, 'c', 'o', 'd', 'e', '9', // pcode_len=5
        0x03, 0x00, 'b', 'o', 'b', // login_len=3
        0x02, 0x00, 'a', 'b', // secret_len=2 (parseConfig caps, does not fix at 40)
    };
    const cfg = try parseConfig(&pre_remote);
    try std.testing.expectEqualStrings("bob", cfg.login_user);
    try std.testing.expectEqualStrings("ab", cfg.login_secret_hex);
    try std.testing.expectEqual(@as(usize, 0), cfg.remote_url.len);
}

test "config submanifests §6.3/§8.1: the two settings round-trip through encodeConfig/parseConfig" {
    var buf: [max_config_payload]u8 = undefined;
    const cfg: Config = .{
        .hash_policy = 4,
        .default_branch = "trunk",
        .project_id = "pid",
        .project_code = "code9",
        .seen_submanifest = true,
        .enable_submanifests = true,
        .submanifest_k = 3, // the A.5 conformance vector's value
    };
    const back = try parseConfig(encodeConfig(&buf, cfg));
    try std.testing.expect(back.seen_submanifest);
    try std.testing.expect(back.enable_submanifests);
    try std.testing.expectEqual(@as(u8, 3), back.submanifest_k);
    // Every neighbouring field survives the widened trailing block.
    try std.testing.expectEqualStrings("trunk", back.default_branch);
    try std.testing.expectEqualStrings("code9", back.project_code);
}

test "config submanifests §8.1/§6.4: the defaults are off, unpinned, and manual, and survive a round trip" {
    // §8.1 "a project-scope boolean, default false", §6.3's "no pinned
    // parameter" state (which §8.1 forbids enabling emission on), and §6.4's
    // `submanifest-auto-retune` setting, also default false. A Config built
    // with none of the three fields named must decode back to exactly that.
    var buf: [max_config_payload]u8 = undefined;
    const cfg: Config = .{ .hash_policy = 4, .default_branch = "trunk", .project_id = "pid" };
    try std.testing.expect(!cfg.enable_submanifests);
    try std.testing.expectEqual(@as(u8, 0), cfg.submanifest_k);
    try std.testing.expect(!cfg.submanifest_auto_retune);
    const back = try parseConfig(encodeConfig(&buf, cfg));
    try std.testing.expect(!back.enable_submanifests);
    try std.testing.expectEqual(@as(u8, 0), back.submanifest_k);
    try std.testing.expect(!back.submanifest_auto_retune);
}

test "config submanifests §6.4: auto-retune round-trips through encodeConfig/parseConfig" {
    var buf: [max_config_payload]u8 = undefined;
    const cfg: Config = .{
        .hash_policy = 4,
        .default_branch = "trunk",
        .project_id = "pid",
        .enable_submanifests = true,
        .submanifest_k = 3,
        .submanifest_auto_retune = true,
    };
    const back = try parseConfig(encodeConfig(&buf, cfg));
    try std.testing.expect(back.submanifest_auto_retune);
    // Every neighbouring field survives the widened trailing block.
    try std.testing.expect(back.enable_submanifests);
    try std.testing.expectEqual(@as(u8, 3), back.submanifest_k);
}

test "config submanifests §6.4: a pre-auto-retune (3-byte-tail) frame decodes with auto-retune off" {
    // A frame written by a writer before the auto-retune slice existed ends at
    // the 3-byte tail (seen + enable + k); the fourth field must default false
    // on decode, never Corrupt, and never a spurious "enabled".
    var buf: [max_config_payload]u8 = undefined;
    const cfg: Config = .{
        .hash_policy = 4,
        .default_branch = "trunk",
        .project_id = "pid",
        .enable_submanifests = true,
        .submanifest_k = 5,
    };
    const full = encodeConfig(&buf, cfg);
    const back = try parseConfig(full[0 .. full.len - 1]);
    try std.testing.expect(back.enable_submanifests);
    try std.testing.expectEqual(@as(u8, 5), back.submanifest_k);
    try std.testing.expect(!back.submanifest_auto_retune);
}

test "config submanifests §6.3: an out-of-range k is stored verbatim, never repaired by the storage layer" {
    // §6.3 puts the out-of-range fallback on the WRITER ("it MUST fall back to
    // emitting flat manifests rather than substitute a parameter of its own").
    // Storage clamping or zeroing the value here would take that decision away
    // and turn a misconfiguration into a silent, different misconfiguration.
    var buf: [max_config_payload]u8 = undefined;
    for ([_]u8{ 13, 200, 255 }) |k| {
        const cfg: Config = .{
            .hash_policy = 4,
            .default_branch = "trunk",
            .project_id = "pid",
            .enable_submanifests = true,
            .submanifest_k = k,
        };
        const back = try parseConfig(encodeConfig(&buf, cfg));
        try std.testing.expectEqual(k, back.submanifest_k);
    }
}

test "config submanifests §8.1: a seen-indicator-only frame decodes with the settings off" {
    // The frame shape the slice that introduced `seen_submanifest` wrote: a
    // ONE-byte trailing block. A repository carrying one must still open, with
    // the two settings this slice adds reading their defaults — never Corrupt,
    // and never a spurious "enabled".
    const one_byte_tail = [_]u8{
        0x04, // hash_policy
        0x05,
        0x00,
        't',
        'r',
        'u',
        'n',
        'k',
        0x03,
        0x00,
        'p',
        'i',
        'd',
        0x05,
        0x00,
        'c',
        'o',
        'd',
        'e',
        '9',
        0x03,
        0x00,
        'b',
        'o',
        'b',
        0x02,
        0x00,
        'a',
        'b',
        0x00, 0x00, // url_len = 0
        0x01, // seen_submanifest = true, and nothing after it
    };
    const cfg = try parseConfig(&one_byte_tail);
    try std.testing.expect(cfg.seen_submanifest);
    try std.testing.expect(!cfg.enable_submanifests);
    try std.testing.expectEqual(@as(u8, 0), cfg.submanifest_k);
}

test "config submanifests: legacy settings tails parse, the tagged digest round-trips, and any unreadable shape re-scans instead of failing" {
    // A reader accepts the three prior settings widths and the untagged legacy
    // digest by width, and everything newer by shape tag. Any shape it cannot
    // read — an unknown tag, or a known tag whose body width disagrees — is
    // dropped, which costs a re-scan and keeps the rest of the config readable.
    // Only a tail too short to be a legal settings block is corruption.
    const prefix = [_]u8{
        0x04,
        0x05,
        0x00,
        't',
        'r',
        'u',
        'n',
        'k',
        0x03,
        0x00,
        'p',
        'i',
        'd',
        0x05,
        0x00,
        'c',
        'o',
        'd',
        'e',
        '9',
        0x03,
        0x00,
        'b',
        'o',
        'b',
        0x02,
        0x00,
        'a',
        'b',
        0x00, 0x00, // url_len = 0
    };
    // A tail too short to be any legal settings block stays corruption: there is
    // no shape byte to consult, so nothing can be dropped in favour of a re-scan.
    for ([_]usize{2}) |width| {
        var payload: [80]u8 = undefined;
        @memcpy(payload[0..prefix.len], &prefix);
        @memset(payload[prefix.len..][0..width], 1);
        try std.testing.expectError(error.Corrupt, parseConfig(payload[0 .. prefix.len + width]));
    }
    // Every byte is 1 here, so these widths carry the KNOWN sha256 tag with a
    // body of the wrong width (0, 1, 30 and 33 bytes). That is a writer this
    // reader disagrees with about the format, not media damage — `readFrameAt`
    // already checked the CRC — so the shape is dropped exactly like an unknown
    // tag and the surrounding fields must survive. Failing the whole frame here
    // would lose the login, remote and project fields to save one scan.
    for ([_]usize{ 5, 6, 35, 38 }) |width| {
        var payload: [80]u8 = undefined;
        @memcpy(payload[0..prefix.len], &prefix);
        @memset(payload[prefix.len..][0..width], 1);
        const back = try parseConfig(payload[0 .. prefix.len + width]);
        try std.testing.expect(back.submanifest_settled_cost_shape == null);
        try std.testing.expect(back.submanifest_auto_retune);
        try std.testing.expectEqualStrings("trunk", back.default_branch);
        try std.testing.expectEqualStrings("bob", back.login_user);
        try std.testing.expectEqualStrings("pid", back.project_id);
    }
    // …and every legacy legal width still parses.
    var ok1: [64]u8 = undefined;
    @memcpy(ok1[0..prefix.len], &prefix);
    @memset(ok1[prefix.len..][0..1], 1);
    _ = try parseConfig(ok1[0 .. prefix.len + 1]);
    var ok3: [64]u8 = undefined;
    @memcpy(ok3[0..prefix.len], &prefix);
    @memset(ok3[prefix.len..][0..3], 1);
    _ = try parseConfig(ok3[0 .. prefix.len + 3]);
    var ok4: [64]u8 = undefined;
    @memcpy(ok4[0..prefix.len], &prefix);
    @memset(ok4[prefix.len..][0..4], 1);
    const back4 = try parseConfig(ok4[0 .. prefix.len + 4]);
    try std.testing.expect(back4.submanifest_auto_retune);
    // …including the untagged digest a pre-tag branch binary wrote.
    var settled: [80]u8 = undefined;
    @memcpy(settled[0..prefix.len], &prefix);
    @memset(settled[prefix.len..][0..36], 1);
    const parsed = try parseConfig(settled[0 .. prefix.len + 36]);
    const digest = parsed.submanifest_settled_cost_shape orelse return error.TestUnexpectedResult;
    try std.testing.expectEqual(@as(usize, 32), digest.len);

    // The written shape: four settings bytes, the sha256 tag, then the digest.
    var tagged: [80]u8 = undefined;
    @memcpy(tagged[0..prefix.len], &prefix);
    @memset(tagged[prefix.len..][0..37], 0xab);
    tagged[prefix.len + 4] = config_shape_cost_shape_sha256;
    const tagged_back = try parseConfig(tagged[0 .. prefix.len + 37]);
    const tagged_digest = tagged_back.submanifest_settled_cost_shape orelse return error.TestUnexpectedResult;
    try std.testing.expectEqualSlices(u8, &([_]u8{0xab} ** 32), &tagged_digest);

    // An unknown shape is dropped rather than rejected, at any body width: the
    // settings bytes still parse and the digest reads absent, so the next commit
    // re-scans. This is the property the tag exists to buy — under the old bare
    // width whitelist each of these was `error.Corrupt`, taking the login,
    // remote and project fields down with it.
    for ([_]usize{ 5, 6, 35, 37, 64 }) |width| {
        var unknown: [96]u8 = undefined;
        @memcpy(unknown[0..prefix.len], &prefix);
        @memset(unknown[prefix.len..][0..width], 1);
        unknown[prefix.len + 4] = 0x7f; // a shape no writer has defined
        const back = try parseConfig(unknown[0 .. prefix.len + width]);
        try std.testing.expect(back.submanifest_settled_cost_shape == null);
        try std.testing.expect(back.seen_submanifest);
        try std.testing.expect(back.enable_submanifests);
        try std.testing.expect(back.submanifest_auto_retune);
        try std.testing.expectEqualStrings("trunk", back.default_branch);
        try std.testing.expectEqualStrings("bob", back.login_user);
    }
}

test "config submanifests: a payload over max_config_payload is corruption, however well-formed its head" {
    // The shape block is open-ended by design — an unreadable shape is dropped,
    // not rejected on width — so the only thing bounding the trailing run is the
    // record limit itself. Without the reader restating it, a frame whose head
    // parses cleanly could carry a trailing run up to `readFrameAt`'s 1 GiB
    // payload cap and still be accepted, with the writer asserting a limit the
    // reader never checked.
    var payload: [max_config_payload + 64]u8 = undefined;
    @memset(&payload, 0);
    payload[0] = 0x04; // hash_policy; all six length fields then read as zero
    // One byte inside the limit still parses, so this pins the boundary rather
    // than merely proving that some large payload fails.
    _ = try parseConfig(payload[0..max_config_payload]);
    try std.testing.expectError(error.Corrupt, parseConfig(payload[0 .. max_config_payload + 1]));
}

test "config logstore plan E: remote_url over max_config_str is rejected" {
    // A url_len declaring more than the per-field cap is corruption: the same
    // guard every other length field carries (parity, not a special case).
    const pre_remote = [_]u8{
        0x04, // hash_policy
        0x05, 0x00, 't', 'r', 'u', 'n', 'k', // branch
        0x03, 0x00, 'p', 'i', 'd', // pid
        0x05, 0x00, 'c', 'o', 'd', 'e', '9', // pcode
        0x03, 0x00, 'b', 'o', 'b', // login
        0x02, 0x00, 'a', 'b', // secret
    };
    // url_len = max_config_str + 1 = 256 = 0x0100 (LE 0x00, 0x01).
    const oversized = pre_remote ++ [_]u8{ 0x00, 0x01 };
    try std.testing.expectError(error.Corrupt, parseConfig(&oversized));
}

test "logstore §B: a torn config-frame write reverts to the prior committed config" {
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    const io = std.testing.io;
    var fs = try openTmp(&tmp, io, "log.bin");
    defer fs.file.close(io);
    const file = fs.storageFile();

    var store = try LogStore.open(std.testing.allocator, file, 0);
    try store.putConfig(.{ .hash_policy = 9, .default_branch = "trunk", .project_id = "keep" });
    try store.commit();
    store.deinit(std.testing.allocator);

    // Append a torn config B frame (prefix only): scan truncates it away, so
    // the surviving latest config is still A.
    var pbuf: [max_config_payload]u8 = undefined;
    const cfgB = encodeConfig(&pbuf, .{ .hash_policy = 1, .default_branch = "x", .project_id = "y" });
    var fbuf: [64]u8 = undefined;
    const frameB = encodeFrame(&fbuf, config_sentinel, cfgB, @intFromEnum(FrameKind.config), 0);
    try file.append(frameB[0..10]); // torn: short of the fixed header+crc window

    var reopened = try LogStore.open(std.testing.allocator, file, 0);
    defer reopened.deinit(std.testing.allocator);
    var oc = (try reopened.readConfig(std.testing.allocator)).?;
    defer oc.deinit(std.testing.allocator);
    try std.testing.expectEqual(@as(u8, 9), oc.config.hash_policy);
    try std.testing.expectEqualStrings("trunk", oc.config.default_branch);
    try std.testing.expectEqualStrings("keep", oc.config.project_id);
}

// Seeded deterministic fuzz: parseConfig on adversarial wire bytes yields a
// value or error.Corrupt, never a crash.
test "logstore plan E fuzz: parseConfig never crashes on adversarial payloads" {
    var prng = std.Random.DefaultPrng.init(0xC04F16);
    const random = prng.random();
    var iter: usize = 0;
    while (iter < 400) : (iter += 1) {
        var garbage: [64]u8 = undefined;
        const gn = random.uintAtMost(usize, garbage.len);
        random.bytes(garbage[0..gn]);
        if (parseConfig(garbage[0..gn])) |_| {} else |err| switch (err) {
            error.Corrupt => {},
        }
    }
}

test "logstore §5: deltaProfitable is the exact <75% integer boundary" {
    // 75/100 is 75% exactly → NOT strictly under → reject; 74/100 accepts.
    try std.testing.expect(!deltaProfitable(75, 100));
    try std.testing.expect(deltaProfitable(74, 100));
    // Small-integer edges, where a rounded/float ratio would smear the line:
    // 3/4 == 75% rejects; 2/3 < 75% accepts; 1/1 (4 < 3) rejects.
    try std.testing.expect(!deltaProfitable(3, 4));
    try std.testing.expect(deltaProfitable(2, 3));
    try std.testing.expect(!deltaProfitable(1, 1));
    // Degenerate ends: a zero-length delta beats any non-empty target; an
    // empty target admits no profitable delta.
    try std.testing.expect(deltaProfitable(0, 1));
    try std.testing.expect(!deltaProfitable(0, 0));
}

// ---------------------------------------------------------------------------
// Shun frames. Tests reaching file-private decls (`encodeFrame`,
// `appendCrafted`, `encodeShun`) stay here per the kind-separation convention;
// the public-surface shun tests live in `logstore_test.zig`.
// ---------------------------------------------------------------------------

test "logstore §C: a delta base offset pointing at a shun frame is rejected as corrupt" {
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    const io = std.testing.io;
    var fs = try openTmp(&tmp, io, "log.bin");
    defer fs.file.close(io);
    const file = fs.storageFile();

    var seed = try LogStore.open(std.testing.allocator, file, 0);
    seed.deinit(std.testing.allocator); // header only

    // A shun frame at header_len, then a delta whose (valid) container base
    // offset points back at it: resolving the base hits the .shun arm. The
    // twin of the config case — a tombstone is not content and never a basis.
    var sbuf: [max_shun_payload]u8 = undefined;
    const shun_payload = encodeShun(&sbuf, 1700000000, "takedown");
    try appendCrafted(std.testing.allocator, file, "shunned-artifact-hash-0000000001", shun_payload, .shun, 0);
    const container = try codec.encode(std.testing.allocator, "delta-ish bytes; only the base offset matters here", .default);
    defer std.testing.allocator.free(container);
    try appendCrafted(std.testing.allocator, file, "delta-over-shun-base-hash-000001", container, .delta, header_len);

    var store = try LogStore.open(std.testing.allocator, file, 0);
    defer store.deinit(std.testing.allocator);
    const rid = store.lookup("delta-over-shun-base-hash-000001").?;
    try std.testing.expectError(error.Corrupt, store.content(std.testing.allocator, rid));
}

test "logstore §C(inv2): a torn shun frame truncates to exactly the last valid boundary" {
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    const io = std.testing.io;
    var fs = try openTmp(&tmp, io, "log.bin");
    defer fs.file.close(io);
    const file = fs.storageFile();

    const durable = "s" ** 32;
    const torn = "t" ** 32;
    var store = try LogStore.open(std.testing.allocator, file, 0);
    _ = try store.put("frame-one-hash", "payload-one");
    try store.putShun(durable, "durable tombstone", 42);
    try store.commit();
    const boundary = store.size();
    store.deinit(std.testing.allocator);

    var pbuf: [max_shun_payload]u8 = undefined;
    const torn_payload = encodeShun(&pbuf, 99, "never lands");
    var fbuf: [128]u8 = undefined;
    const frame = encodeFrame(&fbuf, torn, torn_payload, @intFromEnum(FrameKind.shun), 0);
    try file.append(frame[0..10]); // short of the fixed header+crc window

    var reopened = try LogStore.open(std.testing.allocator, file, 0);
    defer reopened.deinit(std.testing.allocator);
    try std.testing.expectEqual(boundary, reopened.size());
    try std.testing.expect(reopened.isShunned(durable)); // committed tombstone stands
    try std.testing.expect(!reopened.isShunned(torn)); // the torn one never existed
    try std.testing.expectEqual(@as(usize, 1), reopened.shunList().len);
}

test "logstore §C: a shun frame over the max_shun_reason bound is rejected on open" {
    // (a) a declared reason length past the cap, inside an otherwise plausible
    //     payload — `parseShun`'s per-field guard (parity with `parseConfig`);
    // (b) a payload longer than any conforming shun record — rejected on the
    //     frame's declared length, before a byte of it is parsed.
    var over_cap: [13]u8 = undefined;
    std.mem.writeInt(i64, over_cap[0..8], 5, .little);
    std.mem.writeInt(u16, over_cap[8..10], @intCast(max_shun_reason + 1), .little);
    @memcpy(over_cap[10..13], "abc");

    var over_frame: [max_shun_payload + 1]u8 = [_]u8{0} ** (max_shun_payload + 1);
    std.mem.writeInt(u16, over_frame[8..10], @intCast(max_shun_reason + 1), .little);

    const cases = [_][]const u8{ &over_cap, &over_frame };
    for (cases) |payload| {
        var tmp = std.testing.tmpDir(.{});
        defer tmp.cleanup();
        const io = std.testing.io;
        var fs = try openTmp(&tmp, io, "log.bin");
        defer fs.file.close(io);
        const file = fs.storageFile();

        var seed = try LogStore.open(std.testing.allocator, file, 0);
        seed.deinit(std.testing.allocator); // header only

        try appendCrafted(std.testing.allocator, file, "oversize-shun-reason-hash-000001", payload, .shun, 0);
        try std.testing.expectError(error.Corrupt, LogStore.open(std.testing.allocator, file, 0));
    }
}

test "logstore §C: a shun frame keyed by an incomplete digest is rejected on open" {
    // The complete-artifact-name key rule on the READ path. `putShun` rejects
    // such a key with `error.BadKey`, but a frame that is already in the file
    // never went through `putShun` — it was hand-built, written by a foreign
    // implementation, or written by a future kind — so `scan` states the rule
    // too. Without this the invariant would hold only for logs this binary
    // wrote, and `open` could rebuild a set holding an abbreviated key — the
    // same fail-open hole, reached through the file instead of the API.
    //
    // 31 bytes is inside `min_hash_len .. max_hash_len` — a width `put` accepts
    // happily — so this pins the shun-specific rule, not the generic bound.
    const incomplete = "shun-key-that-is-31-bytes-long!";
    try std.testing.expectEqual(@as(usize, 31), incomplete.len);
    try std.testing.expect(!isCompleteDigest(incomplete.len));

    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    const io = std.testing.io;
    var fs = try openTmp(&tmp, io, "log.bin");
    defer fs.file.close(io);
    const file = fs.storageFile();

    var seed = try LogStore.open(std.testing.allocator, file, 0);
    seed.deinit(std.testing.allocator); // header only

    var sbuf: [max_shun_payload]u8 = undefined;
    const payload = encodeShun(&sbuf, 1700000000, "prefix key");
    try appendCrafted(std.testing.allocator, file, incomplete, payload, .shun, 0);

    // Fail-CLOSED: the whole open fails rather than the row being skipped.
    try std.testing.expectError(error.Corrupt, LogStore.open(std.testing.allocator, file, 0));
}

test "logstore §C: two shun frames naming the same digest collapse to one set entry" {
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    const io = std.testing.io;
    var fs = try openTmp(&tmp, io, "log.bin");
    defer fs.file.close(io);
    const file = fs.storageFile();

    var seed = try LogStore.open(std.testing.allocator, file, 0);
    seed.deinit(std.testing.allocator); // header only

    // A log CAN carry a repeat — a concurrent writer, or a hand-built fixture
    // like this one. `putShun`'s idempotence check cannot prevent it, so the
    // scan must: the first record stands and the duplicate's owned copies are
    // freed (`std.testing.allocator` fails the test on the leak otherwise).
    const dup = "d" ** 32;
    var b1: [max_shun_payload]u8 = undefined;
    try appendCrafted(std.testing.allocator, file, dup, encodeShun(&b1, 11, "first"), .shun, 0);
    var b2: [max_shun_payload]u8 = undefined;
    try appendCrafted(std.testing.allocator, file, dup, encodeShun(&b2, 22, "second"), .shun, 0);

    var store = try LogStore.open(std.testing.allocator, file, 0);
    defer store.deinit(std.testing.allocator);
    try std.testing.expect(store.isShunned(dup));
    const list = store.shunList();
    try std.testing.expectEqual(@as(usize, 1), list.len);
    try std.testing.expectEqual(@as(i64, 11), list[0].epoch);
    try std.testing.expectEqualStrings("first", list[0].reason);
}

test "logstore plan E/shun: encodeShun/parseShun round-trip, and parseShun rejects malformed payloads" {
    var buf: [max_shun_payload]u8 = undefined;
    const wire = encodeShun(&buf, -1, "a" ** max_shun_reason);
    const rec = try parseShun(wire);
    try std.testing.expectEqual(@as(i64, -1), rec.epoch); // epoch is signed and round-trips
    try std.testing.expectEqualStrings("a" ** max_shun_reason, rec.reason);

    const empty = encodeShun(&buf, 0, "");
    const rec2 = try parseShun(empty);
    try std.testing.expectEqual(@as(i64, 0), rec2.epoch);
    try std.testing.expectEqual(@as(usize, 0), rec2.reason.len);

    var trailing: [11]u8 = [_]u8{0} ** 11; // epoch + reason_len=0 + one stray byte

    // (e) an over-cap reason that is FULLY BACKED: reason_len = max_shun_reason
    // + 1 with that many real bytes behind it, so every other guard in
    // `parseShun` is satisfied — the short-payload check (`payload.len - pos >=
    // rlen`) and the exact-consumption check both pass, and only the per-field
    // `rlen > max_shun_reason` cap can reject it. Deleting that one line leaves
    // every other shun test green (the frame-level `max_shun_payload` guard
    // catches the oversize-FRAME case, not this one), so without this case the
    // cap is untested and free to rot when `max_shun_payload` is widened.
    var backed_over_cap: [8 + 2 + max_shun_reason + 1]u8 = [_]u8{'a'} ** (8 + 2 + max_shun_reason + 1);
    std.mem.writeInt(i64, backed_over_cap[0..8], 5, .little);
    std.mem.writeInt(u16, backed_over_cap[8..10], @intCast(max_shun_reason + 1), .little);

    const cases = [_][]const u8{
        &[_]u8{}, // (a) empty: no epoch
        &[_]u8{ 1, 2, 3, 4, 5, 6, 7, 8 }, // (b) epoch only, no reason_len
        &[_]u8{ 0, 0, 0, 0, 0, 0, 0, 0, 5, 0, 'a' }, // (c) reason_len 5 exceeds remaining
        &trailing, // (d) trailing byte after an empty reason
        &backed_over_cap, // (e) over-cap reason with the bytes to back it
    };
    for (cases) |c| try std.testing.expectError(error.Corrupt, parseShun(c));
}

test "PlaintextCache: the entry cap is derived from the byte budget, not picked" {
    // The defect this pins: two independent budgets, one silently holding the
    // cache to a fraction of the other. On a real import the entry cap forced
    // 99,097 evictions while the byte cap forced none and the resident set sat
    // at about a third of it.
    //
    // Stated as a relationship rather than a number so changing one budget
    // cannot leave the other stranded: the entry cap must be reachable only by
    // entries SMALLER than the planning size, which is what "the byte budget
    // does the bounding for normal data" means.
    try std.testing.expectEqual(
        PlaintextCache.default_max_bytes / PlaintextCache.planning_entry_bytes,
        PlaintextCache.default_max_entries,
    );

    // At the planning size the two caps are reached together; anything larger
    // makes bytes bind first, which is the intended regime.
    const at_planning = PlaintextCache.default_max_entries * PlaintextCache.planning_entry_bytes;
    try std.testing.expectEqual(PlaintextCache.default_max_bytes, at_planning);
}

test "logstore prefix key: a captured prefix survives appends, exact length, and every invalidation" {
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    const io = std.testing.io;
    var fs = try openTmp(&tmp, io, "prefix.bin");
    defer fs.file.close(io);
    const file = fs.storageFile();

    const h1 = [_]u8{0x11} ** 32;
    const h2 = [_]u8{0x22} ** 32;
    const h3 = [_]u8{0x33} ** 32;
    var store = try LogStore.open(std.testing.allocator, file, 0);
    defer store.deinit(std.testing.allocator);
    const rid1 = try store.put(&h1, "first artifact");
    try store.commit();

    // Capture the one-frame prefix, then grow the log past it.
    const key1 = try store.logKeyAt(rid1);
    try std.testing.expect(LogStore.validatePrefix(file, key1)); // at exactly L
    _ = try store.put(&h2, "second artifact");
    const rid3 = try store.put(&h3, "third artifact");
    try store.commit();
    try std.testing.expect(LogStore.validatePrefix(file, key1)); // after appends — the whole point
    try std.testing.expect(!LogStore.validateTail(file, key1)); // and exactly why validateTail cannot serve

    // Captured at the live tail, the prefix key IS the tail key: one walk and
    // one scan agree on end, count, boundary, and header crc.
    const key3 = try store.logKeyAt(rid3);
    const tail = try store.logKey();
    try std.testing.expectEqual(tail.end, key3.end);
    try std.testing.expectEqual(tail.frame_count, key3.frame_count);
    try std.testing.expectEqual(tail.last_frame_offset, key3.last_frame_offset);
    try std.testing.expectEqual(tail.last_header_crc, key3.last_header_crc);

    // Invalidations, one per condition — each a MISS (false), never an error.
    var bad = key1;
    bad.last_frame_offset += 1; // not a real interior boundary
    try std.testing.expect(!LogStore.validatePrefix(file, bad));
    bad = key1;
    bad.last_header_crc ^= 1; // boundary frame header differs
    try std.testing.expect(!LogStore.validatePrefix(file, bad));
    bad = key1;
    bad.end -= 8; // the boundary frame no longer ends at key.end
    try std.testing.expect(!LogStore.validatePrefix(file, bad));

    // Truncation below key.end invalidates (use the FULL key so the truncated
    // file is genuinely shorter than the prefix it claims).
    try file.truncate(key1.end - 8);
    try std.testing.expect(!LogStore.validatePrefix(file, key1));
}

test "logstore prefix key: an empty-log key validates on length alone" {
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    const io = std.testing.io;
    var fs = try openTmp(&tmp, io, "empty.bin");
    defer fs.file.close(io);
    const file = fs.storageFile();

    var store = try LogStore.open(std.testing.allocator, file, 0);
    defer store.deinit(std.testing.allocator);
    const empty_key = try store.logKey();
    try std.testing.expectEqual(@as(u64, 0), empty_key.frame_count);
    try std.testing.expect(LogStore.validatePrefix(file, empty_key));

    // Still a valid PREFIX after the log grows (the empty prefix of any log).
    const h = [_]u8{0x44} ** 32;
    _ = try store.put(&h, "growth");
    try store.commit();
    try std.testing.expect(LogStore.validatePrefix(file, empty_key));
    // A claimed-empty key whose end is not the bare header is nonsense.
    var bad = empty_key;
    bad.end += 8;
    try std.testing.expect(!LogStore.validatePrefix(file, bad));
}

test "PlaintextCache: O(1) recency list evicts strictly least-recently-used, get refreshes, clear resets" {
    const gpa = std.testing.allocator;
    var cache = PlaintextCache.init();
    defer cache.deinit(gpa);
    cache.max_entries = 3;

    try cache.put(gpa, 1, "one");
    try cache.put(gpa, 2, "two");
    try cache.put(gpa, 3, "three");
    // Touch 1: the LRU victim becomes 2.
    try std.testing.expectEqualStrings("one", cache.get(1).?);
    try cache.put(gpa, 4, "four"); // evicts 2
    try std.testing.expectEqual(@as(?[]const u8, null), cache.get(2));
    try std.testing.expect(cache.get(1) != null);
    try std.testing.expect(cache.get(3) != null);
    try std.testing.expect(cache.get(4) != null);
    try std.testing.expectEqual(@as(u64, 1), cache.evicted);

    // After the gets above the recency order (MRU first) is 4, 3, 1 — the
    // next insert evicts 1, the strict LRU.
    try cache.put(gpa, 5, "five");
    try std.testing.expectEqual(@as(?[]const u8, null), cache.get(1));
    try std.testing.expectEqual(@as(u64, 2), cache.evicted);
    try std.testing.expectEqual(@as(usize, 3), cache.map.count());

    cache.clear(gpa);
    try std.testing.expectEqual(@as(usize, 0), cache.map.count());
    try std.testing.expectEqual(@as(usize, 0), cache.bytes_used);
    try std.testing.expect(cache.mru == null and cache.lru == null);
    // Usable after clear.
    try cache.put(gpa, 9, "nine");
    try std.testing.expectEqualStrings("nine", cache.get(9).?);
}