Skip to content
zit Artifact [2bb24aa3f0]
Login

Artifact [2bb24aa3f0]

Check-ins Using Download Hex

Artifact 2bb24aa3f0cb13a75591bbc7b906d7b013c61990707d6ffca5bfe18c2b460eed:

//! zit sync / network / store-transfer verbs (serve, clone, open, pack, pull,
//! push, sync, remote) and the helpers and tests private to them.
//!
//! Extracted verbatim from src/main.zig (release step S2). Every declaration
//! below is byte-identical to its main.zig source; only this import prologue is
//! new. Helpers still shared with other verbs stay in main.zig and are aliased
//! back here so the moved bodies call them unqualified. main.zig's verb-dispatch
//! switch calls into this file (`cmd_sync.runServe` ...); its test block imports
//! it so these inline tests keep running under `zig build test`.

const std = @import("std");
const zit = @import("zit");
const cli = @import("../cli.zig");
const extras = @import("../extras.zig");
const logstore = @import("zit").logstore;
const storage = @import("zit").storage;
const update = @import("zit").update;
const checkout = @import("zit").checkout;
const Repository = @import("zit").Repository;
const serve_mod = @import("../serve.zig");
const transport = @import("../transport.zig");
const sync_driver = @import("../sync_driver.zig");

const Io = std.Io;
const File = Io.File;
const repostate = zit.repostate;
const managed = zit.managed;
const artifact = zit.artifact;

// Shared CLI helpers and test fixtures that stay in main.zig (other verbs use
// them too); aliased back so the moved code and tests reference them by their
// original unqualified names.
const main = @import("../main.zig");
const runInit = main.runInit;
const repo_filename = main.repo_filename;
const renderStorage = main.renderStorage;
const reasonOf = main.reasonOf;
const stateOom = main.stateOom;
const not_a_repo_msg = main.not_a_repo_msg;
const repoReadFail = main.repoReadFail;
const repoOpenFail = main.repoOpenFail;
const openRepo = main.openRepo;
const updateApplyFail = main.updateApplyFail;
const digest_len = main.digest_len;
const hexToDigest = main.hexToDigest;
const digestToWireName = main.digestToWireName;
const decodeWireName = main.decodeWireName;
const open_help = main.open_help;
const pack_help = main.pack_help;
const serve_help = main.serve_help;
const clone_help = main.clone_help;
const pull_help = main.pull_help;
const push_help = main.push_help;
const sync_help = main.sync_help;
const remote_help = main.remote_help;
const packSeedBlob = main.packSeedBlob;
const packSeedCheckin = main.packSeedCheckin;
const packEvolvingBlob = main.packEvolvingBlob;
const packOpenWritable = main.packOpenWritable;
const packOpenReadOnly = main.packOpenReadOnly;

/// `zit serve [--port N]` — put an HTTP face on the stateless sync responder.
/// Opens `.zit` read-only and keeps it open for the server's lifetime; every
/// request reads a fresh content view of it and is answered independently
/// (sync §3.8). `std.http`/socket I/O lives entirely in `serve.zig`; this only
/// wires the repository and the listen address.
pub fn runServe(
    gpa: std.mem.Allocator,
    io: std.Io,
    arena: std.mem.Allocator,
    verb_args: []const []const u8,
    streams: cli.Streams,
) cli.Exit {
    const parsed = switch (cli.parseServeArgs(verb_args)) {
        .ok => |p| p,
        .unknown_option => |tok| return cli.usageError(streams, "serve", .unknown_option, tok),
        .extra_argument => |tok| return cli.usageError(streams, "serve", .extra_argument, tok),
        .bad_port => |tok| {
            streams.err.print("zit serve: invalid --port value '{s}'\n", .{tok}) catch {};
            return .usage;
        },
    };
    if (parsed.help) {
        streams.out.writeAll(serve_help) catch {};
        return .ok;
    }

    const root = (cli.discoverRoot(arena, io) catch return stateOom(streams, "serve")) orelse {
        streams.err.writeAll(not_a_repo_msg) catch {};
        return .failure;
    };
    var root_dir = root.dir;
    var file = root_dir.openFile(io, repo_filename, .{}) catch {
        root_dir.close(io);
        return repoReadFail(streams, "serve");
    };
    defer file.close(io);
    const log_st = file.stat(io) catch {
        root_dir.close(io);
        return repoReadFail(streams, "serve");
    };
    // Bring-up order is load-bearing for #13 (rationale in `serve_mod.listen`'s
    // doc): open the listener FIRST — before ANY O(store) work, which here means
    // both the config read below and the prime, each a full-log scan. A client
    // that connects during either queues in the accept backlog instead of being
    // refused, so the readiness window starts at the bind rather than after the
    // scans. Announce "listening" only once the bind has actually succeeded (so
    // it never lies about a listener that failed to come up); the accept loop
    // then arms each connection's deadline on accept. The backlog property this
    // rests on is pinned by serve.zig's "listen opens the accept backlog before
    // serve" test.
    const address = std.Io.net.IpAddress.parse("127.0.0.1", parsed.port) catch {
        streams.err.print("zit serve: invalid --port value '{d}'\n", .{parsed.port}) catch {};
        return .usage;
    };
    var listener = serve_mod.listen(io, address) catch {
        streams.err.print("zit serve: cannot bind 127.0.0.1:{d}\n", .{parsed.port}) catch {};
        return .failure;
    };
    defer listener.deinit(io);
    streams.err.print("zit serve: listening on http://127.0.0.1:{d}/xfer (Ctrl-C to stop)\n", .{parsed.port}) catch {};
    streams.err.flush() catch {};

    var fs = storage.FileStorage.init(io, file, log_st.size);
    var log = logstore.LogStore.openReadOnly(gpa, fs.storageFile()) catch |e| {
        root_dir.close(io);
        switch (e) {
            error.OutOfMemory => streams.err.writeAll("zit serve: out of memory\n") catch {},
            else => streams.err.print("zit serve: cannot read repository: {s}\n", .{reasonOf(e)}) catch {},
        }
        return .failure;
    };
    defer log.deinit(gpa);
    defer root_dir.close(io);

    // §9.1/§8.2 — emit the repository's stored project code in the clone
    // reply's `push` card, so a clone learns the shared project identity. A
    // repo predating the stored code, or an unreadable config, falls back to
    // the fixed placeholder (informational; servercode never gates a clone).
    var maybe_cfg: ?logstore.OwnedConfig = log.readConfig(gpa) catch null;
    defer if (maybe_cfg) |*oc| oc.deinit(gpa);
    const project_code: []const u8 = if (maybe_cfg) |oc|
        (if (oc.config.project_code.len > 0) oc.config.project_code else serve_mod.server_projectcode)
    else
        serve_mod.server_projectcode;

    const server = zit.sync.Server{
        .version = serve_mod.server_version,
        .credential = null,
        .self_servercode = serve_mod.server_servercode,
        .project_code = project_code,
    };
    var ctx = ServeCtx{ .io = io, .root_dir = root_dir, .file = file, .server = &server };
    defer ctx.dropSnapshot();
    const responder = serve_mod.Responder{ .ctx = &ctx, .answerFn = serveAnswer, .maintainFn = ServeCtx.maintain };

    if (ctx.baseSnapshot(gpa)) |_| {} else |_| return repoReadFail(streams, "serve");
    serve_mod.serve(io, gpa, &listener, responder, .{}) catch |e| switch (e) {
        error.Accept => {
            streams.err.writeAll("zit serve: connection accept failed\n") catch {};
            return .failure;
        },
    };
    return .ok;
}

/// The per-request context threaded into `serve.Responder`: the open log to read
/// a fresh store from, the server identity to answer with, and the injected IO +
/// repository directory used to make pushed content durable (a push request
/// ingests new artifacts into the per-request store, which must be persisted into
/// the served repository's on-disk log — sync §3.3).
const ServeCtx = struct {
    io: std.Io,
    root_dir: std.Io.Dir,
    /// The open read handle on the served `.zit`. `respond` re-reads the CURRENT
    /// on-disk log from it (§3.8): a fresh stat fixes the committed end-of-log,
    /// so a check-in appended by a separate `zit commit` while `serve` runs is
    /// advertised on the next request — no restart.
    ///
    /// The handle pins an INODE, not a path. `zit pack` replaces `.zit` by
    /// rename, so a pack performed under a running server leaves that server
    /// reading the replaced inode until it is restarted. That is a property of
    /// holding the handle open, unchanged by whether the store read from it is
    /// cached: before caching, every request re-read the same dead inode. It is
    /// recorded here because "a changed size is a changed log" below is a
    /// statement about THIS inode.
    file: File,
    server: *const zit.sync.Server,
    /// The most recent snapshot of the on-disk log, reused across requests. A
    /// log-backed INDEX (not a full-container copy): it reads the containers it
    /// serves on demand and holds only the index between rounds (#14). Built from
    /// the long-lived allocator `serve` threads into every answer, so it outlives
    /// the request that built it, and owned here — heap-allocated because it owns
    /// the `FileStorage` its log borrows; `dropSnapshot` frees it. Never handed to
    /// `respond` directly — see `serveAnswer`.
    snapshot: ?*LogIndexStore = null,
    /// The log size `snapshot` was built from. The log is append-only, so a
    /// changed size is a changed log and an unchanged size is an unchanged log:
    /// one `stat` per request decides whether the snapshot still describes it.
    snapshot_log_size: u64 = 0,
    /// How many snapshots this server has built. Not used by the server; it is
    /// the observable that makes "snapshot builds do not grow with the round
    /// count" a testable property rather than a wall-clock claim.
    loads: u64 = 0,

    /// Free the cached snapshot, leaving the context usable: the next
    /// `baseSnapshot` rebuilds. Both a caller finishing with the context and the
    /// rebuild path itself want exactly this.
    fn dropSnapshot(self: *ServeCtx) void {
        if (self.snapshot) |s| s.destroy();
        self.snapshot = null;
        self.snapshot_log_size = 0;
    }

    /// Drop plaintext the cached snapshot memoised while answering, keeping the
    /// retained containers it serves from. Called once per round, after every
    /// borrow of the snapshot has ended.
    ///
    /// Inflating is how `content()` answers a wire-backed entry, and the
    /// responder's own §7.2 size proofs call `content()` on every artifact they
    /// emit. Against a throwaway per-request store that plaintext died with the
    /// request; against a store that outlives the request it would accumulate
    /// until the server held the whole repository's uncompressed history. The
    /// containers stay, so the next round answers identically.
    ///
    /// The memo is given up rather than capped because in an optimized build
    /// there is almost nothing to give up: the size proofs are `runtime_safety`
    /// only, so a clone round inflates nothing and the memo would stay empty
    /// regardless. Measured over four clones of a 3001-artifact repository,
    /// ReleaseFast: 12 MB resident and 0.09 s of server CPU with this release,
    /// against 26 MB and 0.15 s without it. A safety-checked build does re-prove
    /// every emitted artifact each round and so pays to re-inflate; that cost
    /// belongs to the proofs, not to the cache.
    fn releaseSnapshotPlaintext(self: *ServeCtx) void {
        if (self.snapshot) |s| s.releaseTransient();
    }

    /// §3.8 — the store describing the CURRENT on-disk log, rebuilt only when the
    /// log has actually grown. The §3.8 guarantee is that a check-in appended by
    /// a separate `zit commit` is advertised on the NEXT request with no restart;
    /// that needs a fresh `stat` per request, not a fresh rebuild per request.
    /// Rebuilding regardless made every round cost the whole repository, which is
    /// what turned a clone of a large store into O(rounds x store) server work.
    ///
    /// A torn tail from an append caught mid-write is rejected by the read-only
    /// scanner exactly as before; the previous snapshot is dropped only once a
    /// replacement is fully built, so a failed rebuild leaves the cache coherent.
    ///
    /// The rebuild itself is O(store) and must therefore not fall inside a
    /// request whose connection carries a deadline. `runServe` primes the cache
    /// before it accepts anything. A round that appends to the log does NOT
    /// rebuild before it returns — it deliberately leaves the snapshot stale
    /// (describing the pre-append log) rather than pay the O(store) rebuild inside
    /// its own reply deadline (#13). The append-only stat gate above makes that
    /// stale snapshot self-correcting: the size no longer matches, so the next
    /// `baseSnapshot` rebuilds. That next call is the accept loop's `maintain`
    /// hook (see `ServeCtx.maintain`), run between connections with no connection
    /// open — so the deferred rebuild still lands outside every connection
    /// deadline, never on an innocent later request. Nothing here can bound the
    /// startup build itself; that is bounded by the store, and always was.
    fn baseSnapshot(self: *ServeCtx, gpa: std.mem.Allocator) serve_mod.AnswerError!repostate.ArtifactStore {
        const st = self.file.stat(self.io) catch return error.ServerStorage;
        if (self.snapshot) |s| {
            if (st.size == self.snapshot_log_size) return s.store();
        }

        // Build the fresh index BEFORE dropping the old one: a failed rebuild
        // leaves the previous snapshot (and its size) intact, so the stat gate
        // keeps mismatching until a rebuild succeeds and stale content is never
        // served in the meantime. `openFromFile` fully frees itself on any error.
        const fresh = LogIndexStore.openFromFile(gpa, self.io, self.file, st.size) catch |e| switch (e) {
            error.OutOfMemory => return error.OutOfMemory,
            error.Storage, error.Corrupt => return error.ServerStorage,
        };
        self.dropSnapshot();
        self.snapshot = fresh;
        self.snapshot_log_size = st.size;
        self.loads += 1;
        return fresh.store();
    }

    /// The accept loop's between-connections maintenance hook (see
    /// `serve.Responder.maintainFn`). Runs after a connection has closed and
    /// before the next accept, so no connection is open to charge its O(store)
    /// work to. It re-primes the cache via the same stat-gated `baseSnapshot`:
    /// after a pure read round the stat is unchanged and this is one `stat`; after
    /// a round that appended (a push) the log grew, so this rebuilds the snapshot
    /// the pushing round deliberately left stale rather than pay inside its
    /// deadline (#13). A failed rebuild is swallowed for the same reason
    /// `baseSnapshot` leaves the cache coherent on failure: `baseSnapshot` drops
    /// the old snapshot only after a fresh one is fully built and updates
    /// `snapshot_log_size` only on success, so a failed rebuild RETAINS the
    /// previous (stale) snapshot with its old size — the stat gate keeps
    /// mismatching until a rebuild succeeds, and stale content is never served in
    /// the meantime. The next round (or the next `maintain`) retries, still
    /// outside any earlier round's budget. Type-erased to match the `maintainFn`
    /// seam.
    fn maintain(ctx_ptr: *anyopaque, gpa: std.mem.Allocator) void {
        const self: *ServeCtx = @ptrCast(@alignCast(ctx_ptr));
        if (self.baseSnapshot(gpa)) |_| {} else |_| {}
    }
};

/// The `serve.Responder` seam: snapshot the repository into an in-memory store,
/// answer the framed request over it, and — when the request pushed new content —
/// make that content durable in the served repository's on-disk log (sync §3.3).
///
/// `respond` stays sans-IO: it only `put`s pushed artifacts into the snapshot
/// store it is given. The durable write is this serve-layer's job and reuses the
/// idempotent `persistStore` (a re-push of an already-stored hash appends nothing
/// — content is addressed by hash), the very path clone/pull persist through. A
/// pure gimme/pull request makes nothing new servable, so the overlay stays
/// empty and the log is never reopened; only a genuine push touches disk.
///
/// `respond` is answered over a per-request WRITE OVERLAY of the cached snapshot
/// (`repostate.OverlayStore`), never over the snapshot itself: the responder
/// interns phantoms and upgrades entries as it works, and letting those survive
/// into the next round changes what that round serves. Reads fall through to the
/// snapshot, so reuse costs nothing; writes are confined to the overlay, and the
/// persist pass — which is idempotent over everything already in the log — can
/// only append what the overlay holds.
///
/// The pushed bytes were size-validated by `respond`'s request scan before they
/// entered the store, so `persistStore` only ever writes well-formed content, and
/// a mid-write fault surfaces before `commit`, never as a partial log append.
fn serveAnswer(ctx_ptr: *anyopaque, gpa: std.mem.Allocator, request_container: []const u8) serve_mod.AnswerError!serve_mod.XferReply {
    const ctx: *ServeCtx = @ptrCast(@alignCast(ctx_ptr));
    const reply = try serveRound(ctx, gpa, request_container);

    // Past this point nothing borrows the snapshot: the overlay is gone with
    // `serveRound`'s frame and the reply owns its own bytes. Dropping the
    // plaintext memo is bounded (it frees only what this round inflated), so it
    // is safe inside the deadline.
    //
    // A round that appended (a push) has left the snapshot stale — it still
    // describes the pre-push log. It is NOT rebuilt here: an O(store) rebuild
    // inside the reply's connection deadline is exactly #13's failure. The
    // rebuild is the accept loop's job, via `ServeCtx.maintain`, run after this
    // connection closes and before the next accept — outside every deadline. The
    // append-only stat gate in `baseSnapshot` makes the stale snapshot self-
    // correcting: the next `baseSnapshot` (in `maintain` or the next round) sees
    // the grown size and rebuilds, so no `appended` signal needs to be threaded
    // out of the round.
    //
    // A post-scan read fault this round mapped a held artifact to null. The §8.2
    // clone pager skips such a rid and still reports the sequence complete
    // (NEXT 0), silently finishing a clone MISSING an artifact — a failure the
    // copy-based prime could not produce (a fault surfaced at prime, and a
    // retained-but-uninflatable container was still served verbatim). Read the
    // flag BEFORE releasing the round's memos (which clears it), then surface it
    // as a storage error so the client retries; the next round re-reads.
    const faulted = if (ctx.snapshot) |s| s.read_fault else false;
    ctx.releaseSnapshotPlaintext();
    if (faulted) {
        gpa.free(reply.body);
        return error.ServerStorage;
    }
    return reply;
}

/// One round over a per-request write overlay. Split out of `serveAnswer` so the
/// overlay — which borrows the cached snapshot — is provably destroyed before
/// `serveAnswer` touches that snapshot's storage. A round that pushes new content
/// persists it durably here; the snapshot it read from is left describing the
/// pre-push log, and `baseSnapshot`'s append-only stat gate rebuilds it on the
/// next call (in the accept loop's `maintain`, never inside this round's
/// deadline — #13), so this round signals nothing back to its caller.
fn serveRound(
    ctx: *ServeCtx,
    gpa: std.mem.Allocator,
    request_container: []const u8,
) serve_mod.AnswerError!serve_mod.XferReply {
    var overlay = repostate.OverlayStore.init(gpa, try ctx.baseSnapshot(gpa));
    defer overlay.deinit();

    const reply = try serve_mod.handleXfer(gpa, ctx.server, overlay.store(), request_container);
    errdefer gpa.free(reply.body);

    if (overlay.writtenCount() > 0) {
        _ = persistStore(gpa, ctx.io, ctx.root_dir, overlay.store()) catch |e| switch (e) {
            error.OutOfMemory => return error.OutOfMemory,
            error.Storage, error.Corrupt => return error.ServerStorage,
        };
    }
    return reply;
}

/// A serve snapshot that reads container bytes from the log ON DEMAND instead of
/// copying the whole repository into RAM at prime (#14). It owns a read-only
/// `LogStore` — whose open scan already whole-frame-CRC'd every byte, so the
/// single read stays; the second read `loadContentStore` performed does not — and
/// a resident INDEX: the adapter-rid→log-rid map plus the lowercase-hex wire names
/// the responder walks. `held()` answers from that index with zero I/O; `wire()`
/// and `content()` read the one frame they are asked for and memoise it for the
/// round; `releaseTransient` drops those memos between rounds, so a round's
/// resident bytes are bounded by what it emitted, not by the store.
///
/// Heap-allocated because it owns a `storage.FileStorage` whose `storageFile()`
/// vtable points back into the struct — the box must not move once the log borrows
/// it. `openFromFile` takes that ownership (the serve path); `open` borrows a
/// caller-owned `StorageFile` (a test instrumenting read traffic).
///
/// Adapter rids `1..=rids.len` name the log's non-config artifact frames in
/// `ridsInPersistedOrder` — the SAME enumeration `loadContentStore` assigns, so
/// adapter rid r is byte-for-byte the rid the old MemStore snapshot gave, which is
/// what keeps the clone pager's persisted-order cursor (sync §8.2) stable. Rids
/// above `rids.len` are an appendix `MemStore` holding any hash a caller `intern`s
/// or `put`s that the log does not carry. The serve path never writes the base
/// (every write lands in the per-request `OverlayStore`), so the appendix stays
/// empty in production; it exists so this store honours the whole `ArtifactStore`
/// interface — composed over the existing `MemStore` rather than hand-rolled —
/// instead of trapping on a write it will never see.
const LogIndexStore = struct {
    gpa: std.mem.Allocator,
    /// Owned storage binding on the `openFromFile` path (the log borrows
    /// `&self.fs`); `undefined` on the `open` path, where the caller owns the
    /// storage and keeps it alive for this store's life.
    fs: storage.FileStorage,
    log: logstore.LogStore,
    /// Adapter rid a (1-based) → `rids[a-1]` = the log rid. Persisted order, config
    /// frames removed.
    rids: []logstore.Rid,
    /// Adapter rid a → `names[a-1]` = its lowercase-hex wire name (sync §3.3),
    /// backed by `name_arena`. The single site the binary-log digest becomes the
    /// wire name — the boundary `loadContentStore` held, kept to one place.
    names: [][]const u8,
    name_arena: std.heap.ArenaAllocator,
    by_name: std.StringHashMapUnmanaged(repostate.Rid) = .empty,
    /// Per-round transient memos over the persisted rids: the owned plaintext /
    /// wire container that `content()`/`wire()` read on demand, released each round
    /// by `releaseTransient`.
    content_memo: []?[]u8,
    wire_memo: []?WireHold,
    /// Hashes not in the log, interned/put by a caller. Empty on the serve path.
    appendix: repostate.MemStore,
    /// Total plaintext this store has read on demand from the log, monotonic
    /// across `releaseTransient` — the work counter that makes "answering a round
    /// materialises only what it emits" testable (the log-backed twin of
    /// `MemStore.inflated_bytes`). Container reads are excluded, as they were
    /// there: this measures inflation, not verbatim retransmission.
    materialized_bytes: usize = 0,
    /// Set true when an on-demand `content()`/`wire()` read this round mapped a
    /// real error to null (a post-scan I/O fault or an OOM building the wire
    /// form) — NOT when a frame legitimately has no plaintext or no wire form.
    /// The vtable cannot signal an error, so the §8.2 clone pager would skip the
    /// null rid and still report the sequence complete, shipping a clone missing
    /// an artifact; `serveAnswer` reads this after the round and converts a set
    /// flag to `error.ServerStorage` so the client retries instead. Cleared with
    /// the round's memos by `releaseTransient`, so every round starts clean.
    read_fault: bool = false,

    /// A memoised wire form for one persisted rid: the owned §4.1 container and,
    /// for the delta form, the basis' owned hex name (DELTASRC) and the
    /// reconstructed target size (delta §3.2). Freed by `releaseTransient`.
    const WireHold = struct {
        container: []u8,
        delta: ?Delta = null,
        const Delta = struct { src_hex: []u8, target_size: u64 };
    };

    /// Build over a caller-owned `StorageFile` (the log borrows it; `fs` unused).
    fn open(gpa: std.mem.Allocator, storage_file: storage.StorageFile) logstore.OpenError!*LogIndexStore {
        const self = try gpa.create(LogIndexStore);
        errdefer gpa.destroy(self);
        self.gpa = gpa;
        self.fs = undefined;
        try self.build(storage_file);
        return self;
    }

    /// Build over an owned `FileStorage` bound to `file` (the serve path). `size`
    /// must equal `file`'s current length (`FileStorage.init`'s precondition).
    fn openFromFile(gpa: std.mem.Allocator, io: std.Io, file: File, size: u64) logstore.OpenError!*LogIndexStore {
        const self = try gpa.create(LogIndexStore);
        errdefer gpa.destroy(self);
        self.gpa = gpa;
        self.fs = storage.FileStorage.init(io, file, size);
        try self.build(self.fs.storageFile());
        return self;
    }

    fn build(self: *LogIndexStore, storage_file: storage.StorageFile) logstore.OpenError!void {
        self.materialized_bytes = 0;
        // `self` came from `gpa.create` (undefined memory) and is filled field by
        // field, so the struct-declaration default `read_fault = false` never runs.
        // `serveAnswer` reads this flag on the FIRST round after a prime/rebuild,
        // before any `releaseTransient` clears it — leave it uninitialised and a
        // garbage-true value spuriously fails a healthy round. Zero it here.
        self.read_fault = false;
        self.name_arena = std.heap.ArenaAllocator.init(self.gpa);
        errdefer self.name_arena.deinit();
        self.by_name = .empty;
        errdefer self.by_name.deinit(self.gpa);
        self.appendix = repostate.MemStore.init(self.gpa);
        errdefer self.appendix.deinit();

        self.log = try logstore.LogStore.openReadOnly(self.gpa, storage_file);
        errdefer self.log.deinit(self.gpa);

        const persisted = try self.log.ridsInPersistedOrder(self.gpa);
        defer self.gpa.free(persisted);

        var n: usize = 0;
        for (persisted) |lr| {
            if (self.log.kindOf(lr) != .config) n += 1;
        }

        self.rids = try self.gpa.alloc(logstore.Rid, n);
        errdefer self.gpa.free(self.rids);
        self.names = try self.gpa.alloc([]const u8, n);
        errdefer self.gpa.free(self.names);
        self.content_memo = try self.gpa.alloc(?[]u8, n);
        errdefer self.gpa.free(self.content_memo);
        self.wire_memo = try self.gpa.alloc(?WireHold, n);
        errdefer self.gpa.free(self.wire_memo);
        try self.by_name.ensureTotalCapacity(self.gpa, @intCast(n));

        var a: usize = 0;
        for (persisted) |lr| {
            if (self.log.kindOf(lr) == .config) continue;
            // The one site the binary-log digest becomes the wire name (sync §3.3),
            // exactly as `loadContentStore` did — held to one place.
            var name_buf: [2 * digest_len]u8 = undefined;
            const wire_name = digestToWireName(self.log.hashOf(lr), &name_buf);
            const owned = try self.name_arena.allocator().dupe(u8, wire_name);
            self.rids[a] = lr;
            self.names[a] = owned;
            self.content_memo[a] = null;
            self.wire_memo[a] = null;
            // Hashes are unique per log entry (content-addressed intern), so no key
            // collides; adapter rid is a+1.
            self.by_name.putAssumeCapacity(owned, @intCast(a + 1));
            a += 1;
        }
        std.debug.assert(a == n);
    }

    /// Free every per-round transient memo, restoring the store to "index only,
    /// no materialised container bytes" — the shape between rounds. Every borrowed
    /// `content()`/`wire()` slice handed out this round is invalidated here, so it
    /// runs only once every borrow of the snapshot has ended (see
    /// `ServeCtx.releaseSnapshotPlaintext`).
    fn releaseTransient(self: *LogIndexStore) void {
        for (self.content_memo) |*slot| {
            if (slot.*) |b| self.gpa.free(b);
            slot.* = null;
        }
        for (self.wire_memo) |*slot| {
            if (slot.*) |wh| {
                self.gpa.free(wh.container);
                if (wh.delta) |d| self.gpa.free(d.src_hex);
            }
            slot.* = null;
        }
        // A round's fault is a property of that round's reads; clearing it here
        // (once every borrow of the round's memos has ended) lets a retry start
        // clean over the same snapshot.
        self.read_fault = false;
    }

    /// Total container/plaintext bytes currently memoised — a test/accounting
    /// probe. ~0 right after a prime (the whole point: no whole-store copy) and
    /// after `releaseTransient`; rises only as a round materialises what it emits.
    fn residentBytes(self: *const LogIndexStore) usize {
        var total: usize = 0;
        for (self.content_memo) |slot| {
            if (slot) |b| total += b.len;
        }
        for (self.wire_memo) |slot| {
            if (slot) |wh| total += wh.container.len;
        }
        return total;
    }

    fn deinit(self: *LogIndexStore) void {
        self.releaseTransient();
        self.gpa.free(self.content_memo);
        self.gpa.free(self.wire_memo);
        self.gpa.free(self.rids);
        self.gpa.free(self.names);
        self.by_name.deinit(self.gpa);
        self.name_arena.deinit();
        self.appendix.deinit();
        self.log.deinit(self.gpa);
    }

    /// Free the store's internals and the heap box. `self` is invalid after.
    fn destroy(self: *LogIndexStore) void {
        const gpa = self.gpa;
        self.deinit();
        gpa.destroy(self);
    }

    fn store(self: *LogIndexStore) repostate.ArtifactStore {
        return .{ .context = self, .vtable = &vtable };
    }

    const vtable = repostate.ArtifactStore.VTable{
        .intern = internImpl,
        .put = putImpl,
        .content = contentImpl,
        .hashOf = hashOfImpl,
        .lookup = lookupImpl,
        .count = countImpl,
        .wire = wireImpl,
        .retain = retainImpl,
        .held = heldImpl,
    };

    fn wireFrameOf(wh: *const WireHold) repostate.WireFrame {
        if (wh.delta) |d| return .{ .container = wh.container, .delta = .{ .src_hash = d.src_hex, .target_size = d.target_size } };
        return .{ .container = wh.container };
    }

    fn countImpl(context: *anyopaque) usize {
        const self: *LogIndexStore = @ptrCast(@alignCast(context));
        return self.rids.len + self.appendix.store().count();
    }

    fn internImpl(context: *anyopaque, hash: []const u8) error{OutOfMemory}!repostate.Rid {
        const self: *LogIndexStore = @ptrCast(@alignCast(context));
        if (self.by_name.get(hash)) |rid| return rid;
        return self.rids.len + try self.appendix.store().intern(hash);
    }

    fn lookupImpl(context: *anyopaque, hash: []const u8) ?repostate.Rid {
        const self: *LogIndexStore = @ptrCast(@alignCast(context));
        if (self.by_name.get(hash)) |rid| return rid;
        if (self.appendix.store().lookup(hash)) |l| return self.rids.len + l;
        return null;
    }

    fn putImpl(context: *anyopaque, hash: []const u8, content_bytes: []const u8) error{OutOfMemory}!repostate.Rid {
        const self: *LogIndexStore = @ptrCast(@alignCast(context));
        // A persisted hash is content-addressed and already carries these bytes,
        // so `put` is idempotent and `content(rid)` still reads them from the log
        // — the serve path never reaches this arm (writes go to the overlay).
        if (self.by_name.get(hash)) |rid| {
            // Parity with `MemStore.putImpl`: a repeat put must carry identical
            // bytes (a differing answer is a hash collision). Verified only under
            // runtime safety, and only when the log frame actually inflates — a
            // read fault leaves `content` null and the assert is skipped rather
            // than firing spuriously. Inflate through a THROWAWAY read (not
            // `contentImpl`), so this assertion path never memoises a container or
            // bumps `materialized_bytes`/`residentBytes` — the probes the round
            // tests assert on stay untouched by an assert that never runs on serve.
            if (std.debug.runtime_safety) {
                if (self.log.content(self.gpa, self.rids[@intCast(rid - 1)]) catch null) |existing| {
                    defer self.gpa.free(existing);
                    std.debug.assert(std.mem.eql(u8, existing, content_bytes));
                }
            }
            return rid;
        }
        return self.rids.len + try self.appendix.store().put(hash, content_bytes);
    }

    fn hashOfImpl(context: *anyopaque, rid: repostate.Rid) []const u8 {
        const self: *LogIndexStore = @ptrCast(@alignCast(context));
        if (rid <= self.rids.len) return self.names[@intCast(rid - 1)];
        return self.appendix.store().hashOf(rid - self.rids.len);
    }

    fn heldImpl(context: *anyopaque, rid: repostate.Rid) bool {
        const self: *LogIndexStore = @ptrCast(@alignCast(context));
        // Every persisted (non-phantom, non-config) frame carries a form to serve
        // verbatim, so it is held — answered from the index with no frame read.
        if (rid <= self.rids.len) return true;
        return self.appendix.store().held(rid - self.rids.len);
    }

    fn retainImpl(context: *anyopaque, rid: repostate.Rid, ingest: repostate.WireIngest) error{OutOfMemory}!void {
        const self: *LogIndexStore = @ptrCast(@alignCast(context));
        // Advisory (sync §7.2): a persisted rid already carries a stored form, so
        // first-form-wins makes this a no-op — the log is the retained form.
        if (rid <= self.rids.len) return;
        return self.appendix.store().retain(rid - self.rids.len, ingest);
    }

    fn contentImpl(context: *anyopaque, rid: repostate.Rid) ?[]const u8 {
        const self: *LogIndexStore = @ptrCast(@alignCast(context));
        if (rid > self.rids.len) return self.appendix.store().content(rid - self.rids.len);
        const i: usize = @intCast(rid - 1);
        if (self.content_memo[i]) |b| return b;
        // Read the frame on demand and memoise for the round. The vtable cannot
        // signal an error, so a post-scan I/O fault (frames were CRC-verified at
        // open, so this is transient EIO or bitrot after the scan) is recorded on
        // `read_fault` and returned as null; `serveAnswer` turns a set flag into
        // `error.ServerStorage`. A frame that legitimately has no plaintext is a
        // plain null and does NOT set the flag.
        const maybe = self.log.content(self.gpa, self.rids[i]) catch {
            self.read_fault = true;
            return null;
        };
        const bytes = maybe orelse return null;
        self.content_memo[i] = bytes;
        self.materialized_bytes += bytes.len;
        return bytes;
    }

    fn wireImpl(context: *anyopaque, rid: repostate.Rid) ?repostate.WireFrame {
        const self: *LogIndexStore = @ptrCast(@alignCast(context));
        if (rid > self.rids.len) return self.appendix.store().wire(rid - self.rids.len);
        const i: usize = @intCast(rid - 1);
        if (self.wire_memo[i]) |*wh| return wireFrameOf(wh);
        // A frame read fault is a real error (post-scan EIO/bitrot) recorded on
        // `read_fault` so `serveAnswer` can fail the round; a `.raw`/phantom frame
        // legitimately has no wire form (the responder compresses `content()`
        // instead) and yields a plain null that does NOT set the flag.
        var w = (self.log.wire(self.gpa, self.rids[i]) catch {
            self.read_fault = true;
            return null;
        }) orelse return null;
        switch (w) {
            .full => |c| {
                self.wire_memo[i] = .{ .container = c };
            },
            .delta => |d| {
                // The basis precedes this frame in the log, so its wire name is the
                // hex of the base digest the wire read carried (delta §3.2 target
                // size alongside).
                var src_buf: [2 * digest_len]u8 = undefined;
                const src_hex = digestToWireName(d.baseHash(), &src_buf);
                const owned_src = self.gpa.dupe(u8, src_hex) catch {
                    // Failing to produce the wire form here would silently drop a
                    // held delta from the clone; record it so the round errors.
                    w.deinit(self.gpa);
                    self.read_fault = true;
                    return null;
                };
                self.wire_memo[i] = .{ .container = d.container, .delta = .{ .src_hex = owned_src, .target_size = d.target_size } };
            },
        }
        return wireFrameOf(&self.wire_memo[i].?);
    }
};

/// Snapshot every stored artifact from the persisted log into `ms`, skipping
/// config frames (not artifacts). A content-bearing frame becomes a held entry;
/// a content-absent frame becomes a phantom — exactly the held/phantom split
/// `respond` walks (sync §3.2/§3.3).
///
/// The log keys frames by the RAW binary digest, but the sync wire names an
/// artifact by that digest's lowercase hex (sync §3.3 — a `file`/`igot`/`gimme`
/// HASH is 40/64 hex characters). The store the responder walks must therefore
/// hold the hex name, or `respond` would emit non-hex bytes that break the
/// peer's card tokenizer. Every hash crossing this boundary is hex-encoded once,
/// here — the single place binary-log identity becomes wire identity.
fn loadContentStore(
    gpa: std.mem.Allocator,
    log: *logstore.LogStore,
    ms: *repostate.MemStore,
) serve_mod.AnswerError!void {
    const rids = log.ridsInPersistedOrder(gpa) catch return error.OutOfMemory;
    defer gpa.free(rids);
    for (rids) |rid| {
        if (log.kindOf(rid) == .config) continue;
        var name_buf: [2 * digest_len]u8 = undefined;
        const name = digestToWireName(log.hashOf(rid), &name_buf);
        // sync §4.1/§7.2 — copy the stored container into the snapshot verbatim
        // so the clone serves it without inflating here and re-compressing at
        // serve. `wire()` is null only for a `.raw` frame or a phantom; those
        // fall back to the plaintext path below. A `.zlib` frame carries the
        // full form; a `.delta` frame carries the stored delta container plus
        // its basis name and reconstructed size (the §7.2 USIZE, delta §3.2).
        var maybe_wire = log.wire(gpa, rid) catch |e| switch (e) {
            error.OutOfMemory => return error.OutOfMemory,
            error.Storage, error.Corrupt => return error.ServerStorage,
        };
        if (maybe_wire) |*w| {
            defer w.deinit(gpa);
            switch (w.*) {
                .full => |container| {
                    _ = ms.putWire(name, container) catch return error.OutOfMemory;
                },
                .delta => |d| {
                    // The basis frame precedes this one in the log (delta_of <
                    // offset), so it was already inserted at an earlier rid; its
                    // wire name is the hex of the base's binary digest.
                    var src_name_buf: [2 * digest_len]u8 = undefined;
                    const src_name = digestToWireName(d.baseHash(), &src_name_buf);
                    _ = ms.putWireDelta(name, d.container, src_name, d.target_size) catch return error.OutOfMemory;
                },
            }
            continue;
        }
        const maybe = log.content(gpa, rid) catch |e| switch (e) {
            error.OutOfMemory => return error.OutOfMemory,
            error.Storage, error.Corrupt => return error.ServerStorage,
        };
        if (maybe) |bytes| {
            defer gpa.free(bytes);
            _ = ms.store().put(name, bytes) catch return error.OutOfMemory;
        } else {
            _ = ms.store().intern(name) catch return error.OutOfMemory;
        }
    }
}

// ---------------------------------------------------------------------------
// `zit clone URL DIR` — the interop milestone: a real client converging against
// a real `zit serve` over HTTP. Create a fresh repo at DIR, open an HTTP
// transport to URL, pump the sans-IO clone client (`sync.Client`) to §3.7
// convergence through `sync_driver.runSync`, then persist the received
// artifacts into DIR's on-disk log so the clone is durable. Anonymous only.
// ---------------------------------------------------------------------------

/// §9.1 — a fixed placeholder repository identity emitted in the client's
/// pull/push operation cards. Neither a fresh clone nor a pulled repository
/// carries a persistent SERVERCODE/PROJECTCODE of its own yet (real per-repo
/// sync identity is future work); the stateless server echoes the peer's codes
/// and never gates on a configured project code (sync §9.1), so any well-formed
/// 16-hex pair carries a content-only exchange. Shared by clone and pull.
const sync_servercode = "0000000000000000";

const sync_projectcode = "0000000000000000";

/// §10.1 — the release integer the sync client announces as `pragma
/// client-version`. Undated (no server-version is solicited); a fixed build
/// identity, matching `serve`'s own version constant style.
const sync_client_version: u32 = 22700;

/// An explicit upper bound on sync rounds (TigerStyle: every loop is bounded).
/// A single-hop reconcile converges in a handful of rounds; this is deep slack,
/// so a misbehaving peer yields `error.RoundLimitExceeded` instead of hanging.
const max_sync_rounds: usize = 1024;

/// `zit clone URL DIR` — thin wiring over three deep modules: `transport`
/// (HTTP/IO), `sync.Client` + `sync_driver.runSync` (sans-IO §3.7 loop), and
/// `LogStore` (persistence). The URL is used verbatim (sync §4 r1 — the client
/// never appends /xfer). All-or-nothing failures surface an honest exit-1
/// diagnostic; a bad URL scheme is rejected before any repo is created.
pub fn runClone(
    gpa: std.mem.Allocator,
    io: std.Io,
    arena: std.mem.Allocator,
    verb_args: []const []const u8,
    streams: cli.Streams,
) cli.Exit {
    const parsed = switch (cli.parseCloneArgs(verb_args)) {
        .ok => |p| p,
        .unknown_option => |tok| return cli.usageError(streams, "clone", .unknown_option, tok),
        .extra_argument => |tok| return cli.usageError(streams, "clone", .extra_argument, tok),
        .no_url => return cli.usageError(streams, "clone", .no_url, ""),
        .no_dir => return cli.usageError(streams, "clone", .no_dir, ""),
    };
    if (parsed.help) {
        streams.out.writeAll(clone_help) catch {};
        return .ok;
    }

    // Open the transport first: a malformed URL fails before any repo lands on
    // disk (nothing half-made to roll back). `open` performs no network I/O.
    var tr = transport.Transport.open(io, gpa, parsed.url) catch |e|
        return syncOpenFail(streams, "clone", parsed.url, e);
    defer tr.close();

    // Create a fresh `.zit` at DIR by reusing the init sequence, quietly (its
    // own success line is suppressed; clone prints its own). A pre-existing repo
    // there is init's "already exists" error, propagated verbatim.
    const init_rc = runInit(gpa, io, arena, &.{ "-q", "--", parsed.dir }, streams);
    if (init_rc != .ok) return init_rc;

    // Converge an in-memory store against the peer, then persist it into DIR.
    var ms = repostate.MemStore.init(gpa);
    defer ms.deinit();
    const outcome = cloneRun(gpa, &tr, &ms) catch |e|
        return syncRunFail(streams, "clone", e);

    var target = std.Io.Dir.cwd().openDir(io, parsed.dir, .{}) catch
        return renderStorage(streams, parsed.dir, "cannot reopen the clone target directory");
    defer target.close(io);
    const written = persistStore(gpa, io, target, ms.store()) catch |e| switch (e) {
        error.OutOfMemory => return stateOom(streams, "clone"),
        error.Storage => return renderStorage(streams, parsed.dir, "cannot write the cloned repository"),
        error.Corrupt => return renderStorage(streams, parsed.dir, "the peer sent a malformed artifact name"),
    };

    // §9.1 — adopt the origin's project code as this repository's shared
    // identity (a later pull refuses a PROJECTCODE mismatch), overwriting the
    // local code `init` seeded (last config frame wins). A peer that sent no
    // push card leaves the local code in place.
    storeClonedProjectCode(gpa, io, target, outcome.projectCode()) catch |e| switch (e) {
        error.OutOfMemory => return stateOom(streams, "clone"),
        error.Storage => return renderStorage(streams, parsed.dir, "cannot write the cloned repository"),
        error.Corrupt => return renderStorage(streams, parsed.dir, "the cloned repository config is unreadable"),
    };

    // Eagerly build the derivation cache so the FIRST read verb after this clone
    // (`leaves`/`log`/…) takes the zero-parse image path instead of paying the
    // full cold derive. Run after the project-code frame above — the clone's last
    // log append — so the cache keys to the settled log. Best-effort (see
    // `Repository.warmCache`): a failure here leaves the clone correct, with the
    // first read deriving lazily as it did before; it never fails the clone.
    Repository.warmCache(gpa, io, target);

    streams.out.print("cloned {d} artifacts into {s} in {d} rounds\n", .{ written, parsed.dir, outcome.rounds }) catch {};
    return .ok;
}

/// Render an `open` diagnostic to stderr and fail. Best-effort on the write: a
/// diagnostic that cannot be printed must not change the exit code.
fn openDiag(s: cli.Streams, msg: []const u8) cli.Exit {
    s.err.print("zit open: {s}\n", .{msg}) catch {};
    return .failure;
}

/// Map a materialization fault to its `open` diagnostic, one arm per
/// `update.ApplyError` variant and NO `else`: a new variant must fail this
/// compile and force a decision here, exactly as it does in `updateApplyFail`.
///
/// Only `InvalidCheckinId` and `CheckinNotFound` are genuinely unreachable —
/// `open` hands the engine a hash it just resolved out of this repository's own
/// derived state, never a caller-supplied name — and even those get honest
/// prose rather than a corruption claim. Everything else is reachable on an
/// intact repository: the content/manifest arms describe a repository that
/// really is damaged, while `SymlinkUnsupported` / `InvalidPath` /
/// `ReservedPath` are raised while planning the TARGET manifest, so a clone of
/// any repository containing a symlink or a path this platform cannot represent
/// reaches them with nothing corrupt at all. Reporting those as corruption
/// would be a false accusation against a healthy repository; the prose mirrors
/// `updateApplyFail`'s.
fn openApplyFail(s: cli.Streams, err: update.ApplyError) cli.Exit {
    return switch (err) {
        error.OutOfMemory => openDiag(s, "out of memory"),
        error.InvalidCheckinId => openDiag(s, "cannot read repository: the resolved check-in ID is not a valid artifact name"),
        error.CheckinNotFound => openDiag(s, "cannot read repository: the resolved check-in is not present"),
        error.CheckinContentMissing => openDiag(s, "cannot read repository: check-in content is missing"),
        error.CheckinNameMismatch => openDiag(s, "cannot read repository: check-in content does not match its name"),
        error.TargetNotManifest => openDiag(s, "cannot read repository: the check-in is not a valid manifest"),
        error.BaselineNotFound => openDiag(s, "cannot read repository: the check-in baseline is not found"),
        error.BaselineContentMissing => openDiag(s, "cannot read repository: the check-in baseline content is missing"),
        error.BaselineNameMismatch => openDiag(s, "cannot read repository: the check-in baseline content does not match its name"),
        error.BaselineNotManifest => openDiag(s, "cannot read repository: the check-in baseline is not a valid manifest"),
        error.BaselineIsDelta => openDiag(s, "cannot read repository: the check-in baseline is not fully resolvable"),
        error.StateRecordInvalid => openDiag(s, "cannot read repository: " ++ update.state_record_invalid_reason),
        error.Corrupt => openDiag(s, "cannot read repository: repository is corrupt"),
        error.SymlinkUnsupported => openDiag(s, "the check-in contains an unsupported symlink entry"),
        error.InvalidPath => openDiag(s, "the check-in names an invalid file path"),
        error.ReservedPath => openDiag(s, "the check-in names a reserved file path"),
        error.WorktreeNotClean => openDiag(s, "working tree is not clean"),
        error.UnmanagedCollision => openDiag(s, "an untracked file or directory is in the way"),
        error.CurrentStateInvalid => openDiag(s, "current working-tree state is invalid"),
        error.Storage => openDiag(s, "cannot update working tree: input/output error"),
        error.TransactionUnavailable => openDiag(s, "cannot create a transaction directory"),
        error.RollbackFailed => openDiag(s, "open failed and could not be rolled back; the previous working-tree state is retained for recovery"),
        error.CleanupFailed => openDiag(s, "open committed, but the transaction directory could not be removed"),
    };
}

/// `zit open` — the second half of `zit clone URL DIR`: turn a repository whose
/// checkout is still UNBORN into a materialized working tree. Fossil's own UX is
/// the same two steps (`fossil clone` then `fossil open`), and splitting them
/// keeps `runClone` strictly repo-only.
///
/// Thin wiring over the shipped machinery, deliberately: probe the checkout
/// record, resolve the tip of the main branch out of the derived repo state, and
/// hand materialization to the SAME `update.applyCheckin` engine every other
/// working-tree move rides (journal-armed, transactional, rollback-safe). Only
/// the tip resolution is open's own work, and it is load-bearing: `update`
/// accepts nothing but a literal 40/64-hex check-in ID, so a freshly cloned tree
/// has no way to name its own tip.
///
/// It writes NO checkout record and NO managed set of its own. `init` — which
/// `clone` delegates repo creation to — already wrote the unborn record; an
/// absent managed set already reads as the empty set; and the update engine is
/// the single writer of the born record and the managed set. Adding a second
/// writer of that bootstrap state would only create two things to keep in sync.
pub fn runOpen(
    gpa: std.mem.Allocator,
    io: std.Io,
    arena: std.mem.Allocator,
    verb_args: []const []const u8,
    streams: cli.Streams,
) cli.Exit {
    const parsed = switch (cli.parseOpenArgs(verb_args)) {
        .ok => |p| p,
        .unknown_option => |tok| return cli.usageError(streams, "open", .unknown_option, tok),
        .extra_argument => |tok| return cli.usageError(streams, "open", .extra_argument, tok),
        .repeated_option => |tok| return cli.usageError(streams, "open", .repeated_option, tok),
        .empty_value => |tok| return cli.usageError(streams, "open", .empty_value, tok),
        .bad_manifest => |tok| return cli.usageError(streams, "open", .bad_manifest, tok),
    };
    if (parsed.help) {
        streams.out.writeAll(open_help) catch {};
        return .ok;
    }

    // The full derive: resolving the main branch reads the tag tables, which the
    // `.fast_read` image path does not carry.
    var repo: Repository = undefined;
    var root_dir: std.Io.Dir = undefined;
    const rc = openRepo(gpa, io, arena, streams, "open", null, false, false, &repo, &root_dir);
    if (rc != .ok) return rc;
    defer root_dir.close(io);
    defer repo.deinit(gpa);

    // BORN-ness, not presence, is the "already open" signal. Every fresh
    // repository carries a checkout record — `init` writes an unborn one
    // (`version == ""`, branch `trunk`) as its step 8, and `clone` delegates to
    // `init` — so a presence-based gate would reject `zit clone && zit open` on
    // every clone, the primary path this verb exists to serve. Only a non-empty
    // version means a tree has actually been materialized against a check-in.
    var ck_buf: [checkout.max_ck_payload + 1]u8 = undefined;
    const current = checkout.read(io, root_dir, &ck_buf) catch |e| switch (e) {
        error.Storage => return openDiag(streams, "cannot read the checkout state: input/output error"),
        error.Corrupt => return openDiag(streams, "the checkout state is corrupt"),
    };
    if (current) |state| if (state.version.len != 0) {
        streams.err.print(
            "zit open: this working tree is already open at check-in '{s}'; use 'zit update' to change check-ins\n",
            .{state.version},
        ) catch {};
        return .failure;
    };

    // The tip of the main branch (repo-state §9 current-trunk).
    const main_branch = repo.mainBranch();
    // Switched, not an unbound `catch`: the success message below is an exit-0
    // claim that the tree is already correct, and it is only honest for
    // `NotFound`. Naming the arm means a later Storage/OOM arm on `currentTrunk`
    // fails this compile instead of silently reporting a genuine failure as
    // success on a materialization verb.
    const tip = repo.state().currentTrunk(main_branch) catch |e| switch (e) {
        // A repository with no check-in on its main branch is ALREADY in its
        // correct end state — the unborn record and an empty tree. There is
        // nothing to materialize and nothing wrong, so this is reported, not
        // failed, and the record is left exactly as it was found.
        error.NotFound => {
            streams.out.print(
                "open: no check-in on '{s}' yet; the working tree is already in its correct (empty) state\n",
                .{main_branch},
            ) catch {};
            return .ok;
        },
        // Settling an equal-mtime tie by ancestry allocates (repo-state
        // `MtimeMaxPick`). A failure here is a genuine failure — reporting it as
        // the exit-0 "already correct" message would be the lie the switch
        // above exists to prevent.
        error.OutOfMemory => return openDiag(streams, "out of memory"),
    };
    // Both borrow the derived view, which outlives the update below.
    const target = repo.hashOf(tip);
    const branch = repo.branchOf(tip);
    // An assertion, not a check: unlike `update`, whose target is user-supplied
    // and therefore validated, open's target comes out of this repository's own
    // derived state. A malformed name here is a defect in the derivation, not
    // bad input, so it belongs in the negative space rather than in a
    // diagnostic path no correct run can reach.
    std.debug.assert(artifact.validateName(target));

    const log = repo.logStore(gpa) catch |e| return repoOpenFail(streams, "open", e);
    // `open` is the ONE verb that SETS the manifest opt-in; absent flag records
    // "none", which is what an untouched checkout already carried.
    const manifest: update.ManifestOpt = .{ .set = parsed.manifest orelse "" };
    update.applyCheckin(gpa, io, root_dir, log, target, manifest) catch |e| return openApplyFail(streams, e);

    streams.out.print("opened {s} on {s}\n", .{ target, branch }) catch {};
    return .ok;
}

/// The fields a `updateConfig` overlay may change; a null field is left exactly
/// as the stored config has it. Naming a single change site keeps the "copy the
/// whole config, overlay one field" discipline in one place — a caller that
/// means to change only the project code can never silently drop the credential
/// or the saved remote (the class of bug the pre-`updateConfig` re-puts had).
const ConfigPatch = struct {
    project_code: ?[]const u8 = null,
    login: ?zit.sync.Credential = null,
    remote_url: ?[]const u8 = null,
};

/// Read the live repo config at `dir/.zit`, overlay `patch`, and append the
/// result as the new live config (§E last-frame-wins), preserving every field
/// the patch does not name. This is the SOLE read-modify-write of the repo
/// config; storing a clone's project code, a push credential, or a saved remote
/// all route through it. The repo must already carry a config frame (`init`
/// always writes one); its absence is `Corrupt`. Shares `PersistError` with the
/// artifact-write path so the caller maps one storage-diagnostic surface.
fn updateConfig(
    gpa: std.mem.Allocator,
    io: std.Io,
    dir: std.Io.Dir,
    patch: ConfigPatch,
) PersistError!void {
    var file = dir.openFile(io, repo_filename, .{ .mode = .read_write }) catch return error.Storage;
    defer file.close(io);
    const st = file.stat(io) catch return error.Storage;
    var fs = storage.FileStorage.init(io, file, st.size);
    var log = logstore.LogStore.open(gpa, fs.storageFile(), 0) catch |e| switch (e) {
        error.OutOfMemory => return error.OutOfMemory,
        error.Storage => return error.Storage,
        error.Corrupt => return error.Corrupt,
    };
    defer log.deinit(gpa);
    var oc = (log.readConfig(gpa) catch |e| switch (e) {
        error.OutOfMemory => return error.OutOfMemory,
        error.Storage => return error.Storage,
        error.Corrupt => return error.Corrupt,
    }) orelse return error.Corrupt; // init always writes a config frame
    defer oc.deinit(gpa);
    // Copy every field, then overlay only the named ones. `next`'s string slices
    // borrow `oc.backing` / `patch`; both outlive the `putConfig` memcpy below.
    var next = oc.config;
    if (patch.project_code) |c| next.project_code = c;
    if (patch.login) |cred| {
        next.login_user = cred.user;
        // Point at the function-scoped `patch` storage, not the block-local
        // capture `cred`: `next` is read by `putConfig` after this block closes.
        next.login_secret_hex = &patch.login.?.secret_hex;
    }
    if (patch.remote_url) |u| next.remote_url = u;
    log.putConfig(next) catch |e| switch (e) {
        error.OutOfMemory => return error.OutOfMemory,
        error.Storage => return error.Storage,
    };
    log.commit() catch return error.Storage;
}

/// §9.1 — persist the origin's project code into the freshly cloned repo at
/// `dir/.zit`, preserving every other field `init` wrote. A no-op when the peer
/// sent no code. Thin wrapper over `updateConfig` (the field-preserving RMW).
fn storeClonedProjectCode(
    gpa: std.mem.Allocator,
    io: std.Io,
    dir: std.Io.Dir,
    project_code: []const u8,
) PersistError!void {
    if (project_code.len == 0) return; // peer sent none; keep the local init code
    return updateConfig(gpa, io, dir, .{ .project_code = project_code });
}

/// Drives a modern-clone client (§3.8 CLONE_ROUND0 → EXCHANGE) to convergence
/// over `tr`, filling `ms` with the peer's artifacts; returns the round count.
/// Sans-IO on this side of `tr.exchange`: the state machine only moves
/// `[]const u8` bytes. Its closed error set is the driver's own causes unioned
/// with the transport's, so `runClone` switches the whole surface exhaustively.
/// The result of a clone run: the round count plus the origin's project code
/// (§9.1), copied out of the sans-IO client so the caller can persist the
/// shared identity after the client is torn down.
const CloneOutcome = struct {
    rounds: usize,
    project_code_buf: [zit.sync.max_projectcode_len]u8,
    project_code_len: usize,

    fn projectCode(self: *const CloneOutcome) []const u8 {
        return self.project_code_buf[0..self.project_code_len];
    }
};

fn cloneRun(
    gpa: std.mem.Allocator,
    tr: *transport.Transport,
    ms: *repostate.MemStore,
) sync_driver.RunError(*transport.Transport)!CloneOutcome {
    var client = zit.sync.Client{
        .state = zit.sync.initialState(.clone),
        .operation = .clone,
        .self_servercode = sync_servercode,
        .project_code = sync_projectcode,
        .version = .{ .version = sync_client_version },
    };
    defer client.deinit(gpa);
    const stats = try sync_driver.runSync(gpa, &client, ms.store(), tr, max_sync_rounds);
    // §8.3/§9.1 — the peer's project code, learned from the clone reply's
    // `push` card, copied out before the client is deinitialized.
    var outcome = CloneOutcome{ .rounds = stats.rounds, .project_code_buf = undefined, .project_code_len = 0 };
    const pc = client.peerProjectCode();
    @memcpy(outcome.project_code_buf[0..pc.len], pc);
    outcome.project_code_len = pc.len;
    return outcome;
}

/// The closed error set of `persistStore`: memory, a storage write fault, or a
/// corrupt/oversized peer artifact name that will not decode to a log key.
const PersistError = error{ OutOfMemory, Storage, Corrupt };

/// The disposition of one artifact in `persistStore`'s ordered pass.
const PersistOutcome = union(enum) {
    /// A non-phantom frame for this artifact now exists in the log — written
    /// here, or already present from a prior cascade / a pull re-persist.
    persisted,
    /// A reverse-delta whose basis is a content-bearing store entry not YET
    /// written to the log. It is deferred on the basis wire-hash and re-driven
    /// once that basis lands, so its `delta_of` offset points strictly backward.
    deferred: []const u8,
};

/// Persist one artifact `rid` into `log`, choosing its stored form, and verify
/// §7.7 integrity (the reconstructed content MUST hash to its wire name) BEFORE
/// any durable write. A retained delta whose basis is already a held log frame
/// under the read depth bound is stored VERBATIM as a `.delta`; a retained delta
/// whose basis is a not-yet-written content-bearing store entry is DEFERRED
/// (when `allow_defer`) so the basis frame is written first (reverse-delta
/// ordering); every other case — no wire form, a full container, or an
/// absent / phantom / over-deep basis — lands as the full form, never a
/// phantom-based or unreadable delta. Idempotent: an artifact already holding a
/// non-phantom frame returns `.persisted` without a re-write or re-verification
/// (§7.7 was proven when it was first persisted).
fn persistArtifact(
    gpa: std.mem.Allocator,
    store: repostate.ArtifactStore,
    log: *logstore.LogStore,
    rid: repostate.Rid,
    allow_defer: bool,
    /// A bounded plaintext memo shared across `persistStore`'s whole sweep. Each
    /// artifact reconstructs its own plaintext exactly once here — for the §7.7
    /// verify — and primes it under its just-written frame's offset, so a later
    /// reverse-delta naming this frame as basis reuses that plaintext instead of
    /// re-walking (re-inflating, re-applying) the chain. Bounded (LRU, fixed byte
    /// + entry caps): a cache miss just costs a fresh reconstruction, never a
    /// wrong answer, so the stored bytes are identical with or without it.
    cache: *logstore.PlaintextCache,
) PersistError!PersistOutcome {
    const name = store.hashOf(rid);
    var digest_buf: [digest_len]u8 = undefined;
    const digest = decodeWireName(name, &digest_buf) orelse return error.Corrupt;
    if (log.lookup(digest)) |ex| {
        if (!log.isPhantom(ex)) return .persisted;
    }
    // A genuine phantom carries nothing durable — skip it (never a drop of held
    // content). `held` classifies without inflating a wire-backed entry.
    if (!store.held(rid)) return .persisted;
    // §7.7 L1332/L1336-1338 — the repository transaction-commit hook. Recompute
    // the artifact hash over the reconstructed bytes and abort BEFORE any frame is
    // written on a mismatch (a lying container inflates but does not name its
    // content). The verify plaintext is reconstructed against the LOG basis (a
    // read bounded by `max_delta_chain`) — NOT the in-memory wire chain, which,
    // now a clone stores every artifact wire-backed, is the full reverse-delta
    // history and would exceed the read depth bound. `persistStore` writes basis
    // frames before the deltas that name them (reverse-delta order), so a delta's
    // basis is already a log frame here.
    if (store.wire(rid)) |wf| {
        if (wf.delta) |d| {
            var base_buf: [digest_len]u8 = undefined;
            const base_digest = decodeWireName(d.src_hash, &base_buf) orelse return error.Corrupt;
            const maybe_base = log.lookup(base_digest);
            if (maybe_base == null or log.isPhantom(maybe_base.?)) {
                // The basis is not yet a log frame. Defer on it (reverse-delta
                // order: the newer basis frame must be written first) when it is a
                // held store entry that WILL be persisted.
                if (allow_defer) {
                    if (store.lookup(d.src_hash)) |bs_rid| {
                        if (store.held(bs_rid)) return .{ .deferred = d.src_hash };
                    }
                }
                // Basis absent and not deferrable (§3.3): fall back to FULL form
                // from the entry's OWN plaintext — a put+retain entry carries it
                // directly (never a phantom-based delta). In a converged clone this
                // is unreachable (every delta's basis is a written frame, handled
                // below); a wire-delta entry whose basis truly never arrived cannot
                // reconstruct — a hard error, never a fabricated artifact.
                const content = store.content(rid) orelse return error.Corrupt;
                if (!zit.artifact.verify(name, content)) return error.Corrupt;
                const new_rid = log.putZlib(digest, content) catch |e| switch (e) {
                    error.OutOfMemory => return error.OutOfMemory,
                    error.Storage => return error.Storage,
                };
                try cache.put(gpa, log.offsetOf(new_rid), content);
                return .persisted;
            }
            const base_rid = maybe_base.?;
            const base_off = log.offsetOf(base_rid);
            // Reuse the basis plaintext memoised when the basis frame was
            // persisted — reverse-delta order writes every basis before the
            // delta that names it, so the memo is normally warm. Only a miss
            // (first use, or an LRU eviction) reconstructs it by walking the
            // chain; either way `base_plain` is owned and freed below.
            const base_plain = if (cache.get(base_off)) |hit| blk: {
                // §7.7 positive space: a memo entry is the exact plaintext that
                // named its wire hash at the basis's own persist, so the reused
                // bytes still hash to the basis name (debug-only assertion; the
                // hash is never recomputed on the hot path).
                std.debug.assert(zit.artifact.verify(d.src_hash, hit));
                break :blk gpa.dupe(u8, hit) catch return error.OutOfMemory;
            } else blk: {
                const p = (log.contentCached(gpa, base_rid, cache) catch |e| switch (e) {
                    error.OutOfMemory => return error.OutOfMemory,
                    error.Storage => return error.Storage,
                    error.Corrupt => return error.Corrupt,
                }) orelse return error.Corrupt;
                // `contentCached` does not memoise a top-level full-form basis;
                // prime it here so sibling reverse-deltas on the same basis reuse
                // this reconstruction instead of repeating it.
                try cache.put(gpa, base_off, p);
                break :blk p;
            };
            defer gpa.free(base_plain);
            const raw = zit.container.decode(gpa, wf.container, @intCast(logstore.max_payload)) catch |e| switch (e) {
                error.OutOfMemory => return error.OutOfMemory,
                else => return error.Corrupt,
            };
            defer gpa.free(raw);
            const content = zit.delta.apply(gpa, base_plain, raw) catch |e| switch (e) {
                error.OutOfMemory => return error.OutOfMemory,
                else => return error.Corrupt,
            };
            defer gpa.free(content);
            if (!zit.artifact.verify(name, content)) return error.Corrupt;
            // Store the delta VERBATIM iff its basis is a held frame under the READ
            // depth bound; an over-deep basis falls back to full form (never an
            // unreadable chain). `delta_of` is the basis's backward offset —
            // strictly lower than this new frame, the invariant `putDeltaContainer`
            // asserts.
            if (log.depthOf(base_rid) < logstore.max_delta_chain) {
                const new_rid = log.putDeltaContainer(digest, base_rid, wf.container) catch |e| switch (e) {
                    error.OutOfMemory => return error.OutOfMemory,
                    error.Storage => return error.Storage,
                };
                // Prime the just-verified plaintext so an older reverse-delta
                // naming this frame as basis reuses it (no chain re-walk).
                try cache.put(gpa, log.offsetOf(new_rid), content);
            } else {
                const new_rid = log.putZlib(digest, content) catch |e| switch (e) {
                    error.OutOfMemory => return error.OutOfMemory,
                    error.Storage => return error.Storage,
                };
                try cache.put(gpa, log.offsetOf(new_rid), content);
            }
            return .persisted;
        }
        // §7.2 full form — decode to verify, then store the container verbatim.
        const content = zit.container.decode(gpa, wf.container, @intCast(logstore.max_payload)) catch |e| switch (e) {
            error.OutOfMemory => return error.OutOfMemory,
            else => return error.Corrupt,
        };
        defer gpa.free(content);
        if (!zit.artifact.verify(name, content)) return error.Corrupt;
        const new_rid = log.putContainer(digest, wf.container) catch |e| switch (e) {
            error.OutOfMemory => return error.OutOfMemory,
            error.Storage => return error.Storage,
        };
        // Prime the just-verified full-form plaintext so a reverse-delta naming
        // this frame as basis reuses it instead of re-inflating the container.
        try cache.put(gpa, log.offsetOf(new_rid), content);
        return .persisted;
    }
    // §4.1 — no retained wire (a raw pull `put`): the plaintext is memoised in the
    // store, no chain to walk. Verify and compress it; the hash names the
    // UNCOMPRESSED bytes, so compression is address-transparent.
    const content = store.content(rid) orelse return error.Corrupt;
    if (!zit.artifact.verify(name, content)) return error.Corrupt;
    const new_rid = log.putZlib(digest, content) catch |e| switch (e) {
        error.OutOfMemory => return error.OutOfMemory,
        error.Storage => return error.Storage,
    };
    // Prime the plaintext so a reverse-delta naming this frame as basis reuses it.
    try cache.put(gpa, log.offsetOf(new_rid), content);
    return .persisted;
}

/// Persists every content-bearing artifact in the converged store `ms` into the
/// on-disk log at `dir/.zit`, then fsyncs (durability before success). The
/// store names artifacts by their wire hex; the log keys frames by the raw
/// binary digest, so each name is decoded once here — the inverse of the
/// hex-encode in `loadContentStore`. Idempotent per `LogStore.put` (an artifact
/// already held returns its rid without a re-append), so a pull re-persisting a
/// store that mixes prior holdings with freshly received content writes only the
/// new frames. Returns the number of content-bearing artifacts in the store.
/// Shared by clone (fresh repo) and pull (existing repo).
fn persistStore(
    gpa: std.mem.Allocator,
    io: std.Io,
    dir: std.Io.Dir,
    store: repostate.ArtifactStore,
) PersistError!usize {
    var file = dir.openFile(io, repo_filename, .{ .mode = .read_write }) catch return error.Storage;
    defer file.close(io);
    const st = file.stat(io) catch return error.Storage;
    var fs = storage.FileStorage.init(io, file, st.size);
    // The header of the just-created repo carries the hash kind; `open` reads it
    // from the non-empty file, so the passed value is a don't-care (0).
    var log = logstore.LogStore.open(gpa, fs.storageFile(), 0) catch |e| switch (e) {
        error.OutOfMemory => return error.OutOfMemory,
        error.Storage => return error.Storage,
        error.Corrupt => return error.Corrupt,
    };
    defer log.deinit(gpa);

    var written: usize = 0;
    const n = store.count();

    // Reverse-delta persist ordering (§7.2/§8.2). Fossil stores an OLDER
    // artifact as a delta of a NEWER one, so a delta cfile can land in the store
    // at a LOWER rid than the basis it points at, while the log's `delta_of` is a
    // BACKWARD offset that REQUIRES the basis frame to precede the delta. Persist
    // therefore drains a worklist: a delta whose basis is not yet a frame is
    // deferred (keyed by the basis wire-hash) and re-driven the instant that
    // basis lands. Each content-bearing artifact seeds the drain once in rid
    // order; each deferred delta is re-driven exactly once, so the pass is O(n)
    // and terminates — a chain bottoms out at fossil's full newest artifact, and
    // content addressing forbids a cycle.
    var waiting: std.StringHashMapUnmanaged(std.ArrayListUnmanaged(repostate.Rid)) = .empty;
    defer {
        var it = waiting.valueIterator();
        while (it.next()) |list| list.deinit(gpa);
        waiting.deinit(gpa);
    }
    var ready: std.ArrayListUnmanaged(repostate.Rid) = .empty;
    defer ready.deinit(gpa);

    // One bounded plaintext memo for the whole sweep: a basis reconstructed for
    // its own §7.7 verify is reused by every reverse-delta that names it, so a
    // delta family reconstructs each basis ~once rather than once per referencing
    // delta. Bounded (LRU byte + entry caps) — it never retains the whole repo.
    var cache = logstore.PlaintextCache.init();
    defer cache.deinit(gpa);

    var seed: repostate.Rid = 1;
    while (seed <= n) : (seed += 1) {
        if (!store.held(seed)) continue; // residual phantom: not durable
        written += 1; // count each content-bearing artifact exactly once, at its rid
        try ready.append(gpa, seed);
        while (ready.pop()) |rid| {
            switch (try persistArtifact(gpa, store, &log, rid, true, &cache)) {
                .persisted => {
                    // `rid`'s frame now exists: re-drive any reverse-deltas that
                    // deferred on it as their (now-written) basis.
                    if (waiting.fetchRemove(store.hashOf(rid))) |kv| {
                        var list = kv.value;
                        defer list.deinit(gpa);
                        try ready.appendSlice(gpa, list.items);
                    }
                },
                .deferred => |base_hash| {
                    const gop = try waiting.getOrPut(gpa, base_hash);
                    if (!gop.found_existing) gop.value_ptr.* = .empty;
                    try gop.value_ptr.append(gpa, rid);
                },
            }
        }
    }
    // Defensive drain: a deferred basis is itself content-bearing and therefore
    // persisted above (draining its waiters), so `waiting` is normally empty
    // here. Should a basis nonetheless be missing, its waiters land as full form
    // rather than be dropped — every content-bearing artifact stays durable.
    var leftover = waiting.iterator();
    while (leftover.next()) |e| {
        for (e.value_ptr.items) |rid| {
            switch (try persistArtifact(gpa, store, &log, rid, false, &cache)) {
                .persisted => {},
                .deferred => unreachable, // allow_defer=false never defers
            }
        }
    }

    log.commit() catch return error.Storage;
    return written;
}

// ---------------------------------------------------------------------------
// `zit pack` — repack the store into a compact new generation.
//
// On-disk storage is UNSPECIFIED by the RFCs (they fix the wire format and
// content-addressed naming, §4.1 L1206-1208), so pack is free to reorganize
// the log as long as every artifact's exact content and name survive and the
// §7.2 verbatim wire bytes are still servable. It re-ingests every artifact,
// in the source's offset order, into a FRESH log through `putArtifact` with a
// lineage base hint — a check-in against its parent manifest, a file blob
// against the same path's blob in the parent check-in (the commit path's own
// §4.1/§7.2 compaction heuristic, applied offline). Offset order plus the
// strictly-backward `delta_of` invariant means every base is already written
// before the delta that names it, so no worklist is needed (unlike
// `persistStore`, whose input is rid-ordered).
//
// The new log is written to a temporary file beside `.zit`, fsynced, and then
// atomically renamed over `.zit`: a crash anywhere before the rename leaves the
// original byte-identical (the temp is discarded next run). The `.zit-cache`
// invalidates itself on the tail change and rebuilds on the next read verb.
// ---------------------------------------------------------------------------

/// pack's core store-to-store transform lives in the `zit` module (pack.zig)
/// so the DST/fuzz harnesses drive the REAL pass in-process; the CLI wiring
/// below (temp file, atomic rename, stats rendering) stays here. These aliases
/// keep every call site and test reading `packLog`/`PackError`/`PackStats`.
const PackError = zit.pack.PackError;

const PackStats = zit.pack.PackStats;

const packLog = zit.pack.packLog;

/// The name of pack's transient new-generation file, beside `.zit`. A fixed
/// name (never a random one) so a temp left by a crashed pack is simply
/// overwritten by the next run rather than accumulating. Sourced from
/// `extras.zig`, which owns the literal for the single never-tracked predicate.
const pack_tmp_filename = extras.pack_tmp_filename;

/// `zit pack` — repack the current repository's store into a compact new
/// generation and atomically swap it in. Thin CLI wiring over `packLog`: parse
/// (no positional), discover the checkout root, open `.zit` read-only (which
/// whole-frame-CRC-verifies every frame — the integrity pre-flight) and capture
/// its tail key, stream the packed log into a temp file, fsync, re-check the
/// source tail (the no-repo-lock concurrency guard), then rename the temp over
/// `.zit`. On any failure the temp is deleted and the original store is intact.
pub fn runPack(
    gpa: std.mem.Allocator,
    io: std.Io,
    arena: std.mem.Allocator,
    verb_args: []const []const u8,
    streams: cli.Streams,
) cli.Exit {
    const help = switch (cli.parseNoPositionalArgs(verb_args)) {
        .ok => |a| a.help,
        .unknown_option => |tok| return cli.usageError(streams, "pack", .unknown_option, tok),
        .extra_argument => |tok| return cli.usageError(streams, "pack", .extra_argument, tok),
    };
    if (help) {
        streams.out.writeAll(pack_help) catch {};
        return .ok;
    }

    const root = (cli.discoverRoot(arena, io) catch return stateOom(streams, "pack")) orelse {
        streams.err.writeAll(not_a_repo_msg) catch {};
        return .failure;
    };
    var root_dir = root.dir;
    defer root_dir.close(io);

    // Open the source read-only: `openReadOnly` scans and whole-frame-CRC-checks
    // every frame, so a corrupt source aborts here (no temp written). The tail
    // key pins the log we packed for the pre-rename concurrency re-check.
    var src_file = root_dir.openFile(io, repo_filename, .{}) catch return repoReadFail(streams, "pack");
    defer src_file.close(io);
    const src_st = src_file.stat(io) catch return repoReadFail(streams, "pack");
    var src_fs = storage.FileStorage.init(io, src_file, src_st.size);
    var src = logstore.LogStore.openReadOnly(gpa, src_fs.storageFile()) catch |e| {
        switch (e) {
            error.OutOfMemory => streams.err.writeAll("zit pack: out of memory\n") catch {},
            else => streams.err.print("zit pack: cannot read repository: {s}\n", .{reasonOf(e)}) catch {},
        }
        return .failure;
    };
    defer src.deinit(gpa);
    const src_key = src.logKey() catch return packFail(streams, error.Corrupt);

    // New-generation file: a readable temp beside `.zit` (createFileAtomic opens
    // write-only, but pack reads its own delta bases back, so we roll our own
    // readable temp and rename it). `truncate` overwrites a temp left by a prior
    // crashed pack. Cleaned up on every path except the successful rename.
    var tmp_file = root_dir.createFile(io, pack_tmp_filename, .{ .read = true, .truncate = true }) catch
        return packFail(streams, error.Storage);
    var renamed = false;
    defer {
        tmp_file.close(io);
        if (!renamed) root_dir.deleteFile(io, pack_tmp_filename) catch {};
    }

    var dst_fs = storage.FileStorage.init(io, tmp_file, 0);
    var dst = logstore.LogStore.open(gpa, dst_fs.storageFile(), src.hash_kind) catch |e| switch (e) {
        error.OutOfMemory => return packFail(streams, error.OutOfMemory),
        error.Storage => return packFail(streams, error.Storage),
        error.Corrupt => return packFail(streams, error.Corrupt),
    };
    const stats = packLog(gpa, &src, &dst) catch |e| {
        dst.deinit(gpa);
        return packFail(streams, e);
    };
    dst.deinit(gpa); // `packLog` already fsynced; deinit only frees memory

    // No repo lock exists: if the source log changed under us since we scanned
    // it, abort rather than clobber a concurrent write with our stale snapshot.
    // O(1) (one stat + one header read); the original stays intact.
    if (!logstore.LogStore.validateTail(src_fs.storageFile(), src_key)) {
        streams.err.writeAll(
            "zit pack: repository changed during pack; aborted (original is intact)\n",
        ) catch {};
        return .failure;
    }

    // The single durable transition: atomically replace `.zit` with the packed
    // log. A crash before this leaves the original untouched; after it, the
    // complete new log. There is no window where `.zit` is partial.
    root_dir.rename(pack_tmp_filename, root_dir, repo_filename, io) catch
        return packFail(streams, error.Storage);
    renamed = true;

    renderPackStats(streams.out, stats);
    return .ok;
}

/// Render pack's failure `err` as `zit pack: <reason>` on stderr, exit 1. The
/// original store is always intact when this runs (every write went to the
/// temp, which the caller's `defer` deletes).
fn packFail(s: cli.Streams, err: PackError) cli.Exit {
    s.err.print("zit pack: {s}\n", .{reasonOf(err)}) catch {};
    return .failure;
}

/// One human line summarizing a repack. "already packed" when the new log did
/// not shrink (the store was already compact — pack still verified every frame,
/// its integrity-scrub value). Split from `runPack` so the exact wording is unit
/// tested without a filesystem.
fn renderPackStats(out: *std.Io.Writer, stats: PackStats) void {
    const verb = if (stats.bytes_after < stats.bytes_before) "packed" else "already packed";
    out.print("{s}: {d} artifacts, {d} re-deltaed, {d} -> {d} bytes\n", .{
        verb, stats.artifacts, stats.deltas, stats.bytes_before, stats.bytes_after,
    }) catch {};
}

/// Maps a transport-open failure (bad URL) to the `verb` diagnostic + exit 1.
/// Shared by clone and pull — the only difference is the verb label.
fn syncOpenFail(s: cli.Streams, verb: []const u8, url: []const u8, err: transport.TransportError) cli.Exit {
    switch (err) {
        error.InvalidUrl => s.err.print("zit {s}: '{s}' is not a valid URL\n", .{ verb, url }) catch {},
        error.UnsupportedScheme => s.err.print("zit {s}: unsupported URL scheme in '{s}' (use http or https)\n", .{ verb, url }) catch {},
        error.OutOfMemory => s.err.print("zit {s}: out of memory\n", .{verb}) catch {},
        else => s.err.print("zit {s}: cannot open '{s}'\n", .{ verb, url }) catch {},
    }
    return .failure;
}

/// Maps a sync run (transport + driver) failure to the `verb` diagnostic + exit
/// 1. Exhaustive over the driver's own causes and the transport's, so the whole
/// closed error set is accounted for (no `else`). Shared by clone and pull.
fn syncRunFail(s: cli.Streams, verb: []const u8, err: sync_driver.RunError(*transport.Transport)) cli.Exit {
    const reason: []const u8 = switch (err) {
        error.OutOfMemory => "out of memory",
        error.RoundLimitExceeded => "the peer did not converge within the round limit",
        error.FatalProtocolError => "the peer sent a fatal protocol error",
        error.NotImplemented => "the peer used an unsupported protocol feature",
        error.RequestTooLarge => "the sync request exceeded the transport limit",
        error.ReplyTruncated, error.ReplyMalformed => "the peer sent a malformed reply",
        error.ReplyTooLarge => "the peer's reply exceeded the size limit",
        error.RedirectLimit => "too many HTTP redirects",
        error.InsecureRedirect => "the peer redirected HTTPS to plaintext HTTP",
        error.RedirectNotAllowed => "the peer redirected on a transport that forbids it",
        error.MissingLocation => "the peer sent a redirect with no location",
        error.InvalidUrl => "a redirect target was not a valid URL",
        error.UnsupportedScheme => "a redirect named an unsupported URL scheme",
        error.ProtocolError => "the peer answered with an unexpected HTTP status",
        error.ConnectionFailed => "cannot connect to the peer",
    };
    s.err.print("zit {s}: {s}\n", .{ verb, reason }) catch {};
    return .failure;
}

/// The causes `resolveRemoteUrl` can surface: a store read failure, OOM, or the
/// repository having no URL to use (`NoRemote` — neither an explicit arg nor a
/// saved remote). Closed set so the caller switches exhaustively.
const RemoteResolveError = error{ OutOfMemory, Storage, NoRemote };

/// sync §4/§6 — resolve the peer URL for a sync-family verb: an explicit `arg`
/// (non-empty) wins; otherwise the saved remote (`config.remote_url`) is used;
/// a repository with neither is a usage error (`NoRemote`). The saved URL is
/// copied into `buf` so it outlives the read-only config frame. `buf` need only
/// live until `transport.Transport.open` (which dupes the URL). An out-of-spec
/// stored URL longer than the config-field cap is treated as none.
fn resolveRemoteUrl(
    gpa: std.mem.Allocator,
    io: std.Io,
    dir: std.Io.Dir,
    arg: []const u8,
    buf: *[logstore.max_config_str]u8,
) RemoteResolveError![]const u8 {
    if (arg.len > 0) return arg;
    var file = dir.openFile(io, repo_filename, .{}) catch return error.Storage;
    defer file.close(io);
    const st = file.stat(io) catch return error.Storage;
    var fs = storage.FileStorage.init(io, file, st.size);
    var log = logstore.LogStore.openReadOnly(gpa, fs.storageFile()) catch |e| switch (e) {
        error.OutOfMemory => return error.OutOfMemory,
        else => return error.Storage,
    };
    defer log.deinit(gpa);
    var oc = (log.readConfig(gpa) catch |e| switch (e) {
        error.OutOfMemory => return error.OutOfMemory,
        else => return error.Storage,
    }) orelse return error.NoRemote;
    defer oc.deinit(gpa);
    const saved = oc.config.remote_url;
    if (saved.len == 0 or saved.len > buf.len) return error.NoRemote;
    @memcpy(buf[0..saved.len], saved);
    return buf[0..saved.len];
}

/// sync §4/§6 — the usage error (exit 2) when a sync-family `verb` is given no
/// URL and the repository has no saved remote to fall back to. Names the fix.
/// Shared by pull, push, and sync so the wording lives in one place.
fn noRemoteError(s: cli.Streams, verb: []const u8) cli.Exit {
    s.err.print("zit {s}: no repository URL given and no saved remote\n", .{verb}) catch {};
    s.err.writeAll("set one with 'zit remote URL', or pass the URL explicitly\n") catch {};
    return .usage;
}

// ---------------------------------------------------------------------------
// `zit pull URL` — incremental sync into an EXISTING repository. The sibling of
// clone: it reuses the same transport + `runSync` + persist machinery, but opens
// the cwd's `.zit` (never creating one), announces what it already holds so the
// peer sends only the missing artifacts (sync §3.2/§3.3 igot/gimme), and drives
// to §3.7 termination. Receive-only (§3.3 r24: pull obtains but never uploads).
// ---------------------------------------------------------------------------

/// `zit pull URL` — open the current repository, converge it toward the peer at
/// URL, and persist the newly received artifacts. Thin wiring over the same deep
/// modules clone uses: `transport` (HTTP/IO), `sync.Client` + `runSync` (sans-IO
/// §3.7 loop), and `LogStore` (persistence). With URL omitted the saved remote
/// (`zit remote`) is used; a repo with neither is a usage error. Anonymous only.
pub fn runPull(
    gpa: std.mem.Allocator,
    io: std.Io,
    arena: std.mem.Allocator,
    verb_args: []const []const u8,
    streams: cli.Streams,
) cli.Exit {
    const parsed = switch (cli.parsePullArgs(verb_args)) {
        .ok => |p| p,
        .unknown_option => |tok| return cli.usageError(streams, "pull", .unknown_option, tok),
        .extra_argument => |tok| return cli.usageError(streams, "pull", .extra_argument, tok),
        // An omitted URL is not an error: it falls back to the saved remote below.
        .no_url => cli.PullArgs{ .url = "" },
    };
    if (parsed.help) {
        streams.out.writeAll(pull_help) catch {};
        return .ok;
    }

    // Locate the current repository before any network I/O: a pull outside a repo
    // is a usage failure, not a wasted connection.
    const root = (cli.discoverRoot(arena, io) catch return stateOom(streams, "pull")) orelse {
        streams.err.writeAll(not_a_repo_msg) catch {};
        return .failure;
    };
    var root_dir = root.dir;
    defer root_dir.close(io);

    // sync §4/§6 — the explicit URL wins; an omitted one falls back to the saved
    // remote; a repo with neither is a usage error. `url_buf` outlives `open`.
    var url_buf: [logstore.max_config_str]u8 = undefined;
    const url = resolveRemoteUrl(gpa, io, root_dir, parsed.url, &url_buf) catch |e| switch (e) {
        error.OutOfMemory => return stateOom(streams, "pull"),
        error.Storage => return repoReadFail(streams, "pull"),
        error.NoRemote => return noRemoteError(streams, "pull"),
    };

    var tr = transport.Transport.open(io, gpa, url) catch |e|
        return syncOpenFail(streams, "pull", url, e);
    defer tr.close();

    // Snapshot what this repository already holds into the in-memory store: those
    // entries become the client's `igot` announcement, so the peer sends only the
    // artifacts this side lacks (sync §3.3 — the incremental of the exchange).
    var ms = repostate.MemStore.init(gpa);
    defer ms.deinit();
    loadRepoHoldings(gpa, io, root_dir, &ms) catch |e| switch (e) {
        error.OutOfMemory => return stateOom(streams, "pull"),
        error.Storage => return repoReadFail(streams, "pull"),
    };
    const held_before = contentCount(ms.store());

    // §9.1 — announce this repository's adopted project code in the pull card;
    // a real peer refuses a PROJECTCODE mismatch, so the placeholder cannot be
    // sent against fossil. The buffer outlives the run (the client borrows it).
    var pc_buf: [zit.sync.max_projectcode_len]u8 = undefined;
    const project_code = repoProjectCode(gpa, io, root_dir, &pc_buf) catch |e| switch (e) {
        error.OutOfMemory => return stateOom(streams, "pull"),
        error.Storage => return repoReadFail(streams, "pull"),
    };

    const outcome = pullRun(gpa, &tr, &ms, project_code) catch |e|
        return syncRunFail(streams, "pull", e);

    const written = persistStore(gpa, io, root_dir, ms.store()) catch |e| switch (e) {
        error.OutOfMemory => return stateOom(streams, "pull"),
        error.Storage => return repoReadFail(streams, "pull"),
        error.Corrupt => return renderStorage(streams, root.abs, "the peer sent a malformed artifact name"),
    };
    // A pull only adds content to what was already held (§3.3 r24: it never drops
    // an artifact), so the persisted count can never fall below the starting set.
    std.debug.assert(written >= held_before);
    const received = written - held_before;

    // The peer named artifacts this run never obtained. Everything is persisted
    // — a partial pull is still progress — but the exit status says so, because
    // "exit 0 with nothing said" is indistinguishable from a converged sync and
    // leaves a repository quietly out of date. This holds whether `received` is
    // zero or not: the invariant is convergence, not motion.
    if (outcome.unresolved > 0) {
        streams.err.print(
            "zit pull: the peer holds {d} artifacts this pull did not obtain\n",
            .{outcome.unresolved},
        ) catch {};
        return .failure;
    }

    if (received == 0) {
        streams.out.writeAll("already up to date\n") catch {};
    } else {
        streams.out.print("received {d} artifacts in {d} rounds\n", .{ received, outcome.rounds }) catch {};
    }
    return .ok;
}

/// Drives a pull client (§3.2/§3.3) to convergence over `tr`, filling `ms` with
/// the artifacts the peer holds and this side lacked; returns the round count.
/// The client announces `ms`'s holdings via igot and requests the rest via
/// gimme (receive-only — §3.3 r24: pull obtains content but never uploads).
/// Sans-IO on this side of `tr.exchange`. Its closed error set is the driver's
/// own causes unioned with the transport's, so `runPull` switches it exhaustively.
fn pullRun(
    gpa: std.mem.Allocator,
    tr: *transport.Transport,
    ms: *repostate.MemStore,
    project_code: []const u8,
) sync_driver.RunError(*transport.Transport)!PullOutcome {
    var client = zit.sync.Client{
        .state = zit.sync.initialState(.pull),
        .operation = .pull,
        .self_servercode = sync_servercode,
        .project_code = project_code,
        .version = .{ .version = sync_client_version },
    };
    defer client.deinit(gpa);
    const stats = try sync_driver.runSync(gpa, &client, ms.store(), tr, max_sync_rounds);
    return .{ .rounds = stats.rounds, .unresolved = client.unresolvedPhantoms(ms.store()) };
}

/// What one pull run reports: the round count, and how many artifacts the peer
/// announced (§3.3 r24 — by igot, or as a member of a cluster it sent) that this
/// side still lacks when §3.7 reached its clean terminal. A nonzero `unresolved`
/// is a pull that stopped short of what it was told existed; the caller must not
/// report success for it, however well-formed every card on the wire was.
const PullOutcome = struct { rounds: usize, unresolved: usize };

/// Opens `dir/.zit` read-only and snapshots its stored artifacts into `ms` — the
/// held/phantom split the pull client announces via `igot` (sync §3.2/§3.3).
/// Reuses `loadContentStore` (the very snapshot `serve` builds per request), so
/// the binary-log→wire-hex name encoding lives in exactly one place. The
/// read-only handle is closed before `persistStore` reopens the log read-write.
fn loadRepoHoldings(
    gpa: std.mem.Allocator,
    io: std.Io,
    dir: std.Io.Dir,
    ms: *repostate.MemStore,
) error{ OutOfMemory, Storage }!void {
    var file = dir.openFile(io, repo_filename, .{}) catch return error.Storage;
    defer file.close(io);
    const st = file.stat(io) catch return error.Storage;
    var fs = storage.FileStorage.init(io, file, st.size);
    var log = logstore.LogStore.openReadOnly(gpa, fs.storageFile()) catch |e| switch (e) {
        error.OutOfMemory => return error.OutOfMemory,
        else => return error.Storage,
    };
    defer log.deinit(gpa);
    loadContentStore(gpa, &log, ms) catch |e| switch (e) {
        error.OutOfMemory => return error.OutOfMemory,
        else => return error.Storage,
    };
}

/// sync §9.1 — the repository's stored project code, copied into `buf`, or the
/// zero placeholder when the repo carries none (a repo predating the stored
/// code) or an out-of-spec-length one. `init` derives a project code and a
/// clone adopts the origin's (`storeClonedProjectCode`), so a real repo
/// announces its adopted code here; the pull/push client emits it verbatim as
/// the operation card's PROJECTCODE, and a real peer (fossil §9.1, ABNF
/// `pull-card = "pull" SP servercode SP projectcode`) refuses a mismatch — so a
/// zero placeholder would be rejected. Bounded by `max_projectcode_len`: a
/// longer stored field is out of spec and never truncated into the wire.
fn repoProjectCode(
    gpa: std.mem.Allocator,
    io: std.Io,
    dir: std.Io.Dir,
    buf: *[zit.sync.max_projectcode_len]u8,
) error{ OutOfMemory, Storage }![]const u8 {
    var file = dir.openFile(io, repo_filename, .{}) catch return error.Storage;
    defer file.close(io);
    const st = file.stat(io) catch return error.Storage;
    var fs = storage.FileStorage.init(io, file, st.size);
    var log = logstore.LogStore.openReadOnly(gpa, fs.storageFile()) catch |e| switch (e) {
        error.OutOfMemory => return error.OutOfMemory,
        else => return error.Storage,
    };
    defer log.deinit(gpa);
    var oc = (log.readConfig(gpa) catch |e| switch (e) {
        error.OutOfMemory => return error.OutOfMemory,
        else => return error.Storage,
    }) orelse return sync_projectcode;
    defer oc.deinit(gpa);
    const code = oc.config.project_code;
    if (code.len == 0 or code.len > buf.len) return sync_projectcode;
    @memcpy(buf[0..code.len], code);
    return buf[0..code.len];
}

/// The number of content-bearing (non-phantom) artifacts in `store` — the held
/// set (§3.3 r24). A pull measures its yield as the held count after convergence
/// minus the count before, so "already up to date" is an exact zero, never a guess.
fn contentCount(store: repostate.ArtifactStore) usize {
    var n: usize = 0;
    const total = store.count();
    var rid: repostate.Rid = 1;
    while (rid <= total) : (rid += 1) {
        if (store.content(rid) != null) n += 1;
    }
    return n;
}

// ---------------------------------------------------------------------------
// `zit push URL` — the mirror of pull: UPLOAD this repository's artifacts into an
// EXISTING served repo. It opens the cwd's `.zit`, announces what it holds so the
// peer requests only what it lacks (sync §3.2/§3.3 igot/gimme), and uploads the
// missing artifacts as `file` cards, driving to §3.7 termination. The peer's
// `zit serve` persists what it receives; nothing changes on this side (§3.3 r24).
// Anonymous only.
// ---------------------------------------------------------------------------

/// What one push run reports: the number of artifacts uploaded and the round
/// count it took. `sent` reads the client's own upload counter, so it is the
/// exact count of `file` cards emitted — never an estimate.
const PushOutcome = struct { rounds: usize, sent: usize };

/// `zit push URL` — open the current repository, announce its holdings to the
/// peer at URL, and upload every artifact the peer lacks until §3.7 termination.
/// Thin wiring over the same deep modules pull uses: `transport` (HTTP/IO),
/// `sync.Client` + `runSync` (sans-IO §3.7 loop). With URL omitted the saved
/// remote (`zit remote`) is used; a repo with neither is a usage error.
pub fn runPush(
    gpa: std.mem.Allocator,
    io: std.Io,
    arena: std.mem.Allocator,
    verb_args: []const []const u8,
    streams: cli.Streams,
    amb: cli.Ambient,
) cli.Exit {
    const parsed = switch (cli.parsePushArgs(verb_args)) {
        .ok => |p| p,
        .unknown_option => |tok| return cli.usageError(streams, "push", .unknown_option, tok),
        .extra_argument => |tok| return cli.usageError(streams, "push", .extra_argument, tok),
        // An omitted URL is not an error: it falls back to the saved remote below.
        .no_url => cli.PushArgs{ .url = "" },
    };
    if (parsed.help) {
        streams.out.writeAll(push_help) catch {};
        return .ok;
    }

    // sync §6.3 — a login name is a whitespace-free token (the shared-secret
    // derivation embeds it between "/" separators). Reject a malformed --user
    // as a usage failure rather than reaching the derivation's assertion.
    if (parsed.user) |u| {
        if (u.len > 0 and zit.sync.containsFossilWhitespace(u)) {
            streams.err.writeAll("zit push: --user must not contain whitespace\n") catch {};
            return .failure;
        }
    }

    // Locate the current repository before any network I/O: a push outside a repo
    // is a usage failure, not a wasted connection.
    const root = (cli.discoverRoot(arena, io) catch return stateOom(streams, "push")) orelse {
        streams.err.writeAll(not_a_repo_msg) catch {};
        return .failure;
    };
    var root_dir = root.dir;
    defer root_dir.close(io);

    // sync §4/§6 — the explicit URL wins; an omitted one falls back to the saved
    // remote; a repo with neither is a usage error. `url_buf` outlives `open`.
    var url_buf: [logstore.max_config_str]u8 = undefined;
    const url = resolveRemoteUrl(gpa, io, root_dir, parsed.url, &url_buf) catch |e| switch (e) {
        error.OutOfMemory => return stateOom(streams, "push"),
        error.Storage => return repoReadFail(streams, "push"),
        error.NoRemote => return noRemoteError(streams, "push"),
    };

    var tr = transport.Transport.open(io, gpa, url) catch |e|
        return syncOpenFail(streams, "push", url, e);
    defer tr.close();

    // Snapshot what this repository holds: those entries become the client's igot
    // announcement, so the peer requests (and this side uploads) only the artifacts
    // the peer lacks (sync §3.2/§3.3 — the incremental of the exchange).
    var ms = repostate.MemStore.init(gpa);
    defer ms.deinit();
    loadRepoHoldings(gpa, io, root_dir, &ms) catch |e| switch (e) {
        error.OutOfMemory => return stateOom(streams, "push"),
        error.Storage => return repoReadFail(streams, "push"),
    };

    // §9.1 — announce this repository's adopted project code in the push card
    // (same identity gate as pull; a real peer refuses a mismatch). The buffer
    // outlives the run — the client borrows it.
    var pc_buf: [zit.sync.max_projectcode_len]u8 = undefined;
    const project_code = repoProjectCode(gpa, io, root_dir, &pc_buf) catch |e| switch (e) {
        error.OutOfMemory => return stateOom(streams, "push"),
        error.Storage => return repoReadFail(streams, "push"),
    };

    // sync §6 — resolve the login credential (null for an anonymous push). The
    // password plaintext is confined to `pw_buf`/argv/env; only the derived
    // 40-hex secret (§6.3) is signed with and — on success — persisted.
    var pw_buf: [max_password_len]u8 = undefined;
    // Defense-in-depth: once the 40-hex secret is derived the plaintext in
    // pw_buf is spent, so scrub it on every exit from this scope — success and
    // error alike — rather than leave it resident until the stack is reused.
    defer std.crypto.secureZero(u8, &pw_buf);
    const resolved = resolvePushCredential(gpa, io, root_dir, streams, project_code, parsed, amb, &pw_buf) catch |e| switch (e) {
        error.OutOfMemory => return stateOom(streams, "push"),
        error.Storage => return repoReadFail(streams, "push"),
        error.NoPassword => {
            streams.err.writeAll("zit push: no password supplied for --user\n") catch {};
            return .failure;
        },
    };

    const outcome = pushRun(gpa, &tr, &ms, project_code, if (resolved) |r| r.credential else null) catch |e|
        return syncRunFail(streams, "push", e);

    // DECISION-3 — persist the derived secret (never the plaintext) at 0600 only
    // after the peer accepted the authenticated push, and only when it was
    // freshly derived: a reused stored secret is already durable.
    if (resolved) |r| {
        if (r.persist) {
            storePushCredential(gpa, io, root_dir, r.credential) catch |e| switch (e) {
                error.OutOfMemory => return stateOom(streams, "push"),
                else => return repoReadFail(streams, "push"),
            };
        }
    }

    if (outcome.sent == 0) {
        streams.out.writeAll("nothing to push\n") catch {};
    } else {
        streams.out.print("sent {d} artifacts in {d} rounds\n", .{ outcome.sent, outcome.rounds }) catch {};
    }
    return .ok;
}

/// The upper bound on an interactively-typed or `$ZIT_PASSWORD` password. A
/// login password far shorter than this in practice; the fixed bound keeps the
/// no-echo read on the stack with no unbounded allocation (TigerStyle).
const max_password_len: usize = 1024;

/// sync §6.3 — a resolved push credential and whether its secret was freshly
/// derived (so it should be persisted after a successful push).
const ResolvedCredential = struct {
    credential: zit.sync.Credential,
    persist: bool,
};

/// The causes `resolvePushCredential` can surface: store read failure, OOM, or
/// a `--user` with no obtainable password (`NoPassword`). Closed set so the
/// caller switches exhaustively.
const ResolveError = error{ OutOfMemory, Storage, NoPassword };

/// sync §6.3/DECISION-3 — turn `push`'s `--user`/`--password` into a signing
/// credential. A null (or empty) `--user` is an anonymous push (null result).
/// Otherwise the 40-hex secret is resolved, in order, from: a stored credential
/// for this exact login (reused verbatim, already durable), `--password`,
/// `$ZIT_PASSWORD`, or a no-echo terminal prompt. A freshly derived secret is
/// flagged `persist`; the plaintext lives only in `pw_buf`/argv/env and is
/// never returned or stored.
fn resolvePushCredential(
    gpa: std.mem.Allocator,
    io: std.Io,
    dir: std.Io.Dir,
    streams: cli.Streams,
    project_code: []const u8,
    parsed: cli.PushArgs,
    amb: cli.Ambient,
    pw_buf: *[max_password_len]u8,
) ResolveError!?ResolvedCredential {
    const user = parsed.user orelse return null;
    if (user.len == 0) return null; // an empty --user is an anonymous push
    std.debug.assert(!zit.sync.containsFossilWhitespace(user)); // pre-validated in runPush

    // A stored secret for this exact login is reused as-is: no re-derive, no
    // prompt, no re-persist. Only consulted when no explicit password is given.
    if (parsed.password == null and amb.push_password == null) {
        if (try storedSecretFor(gpa, io, dir, user)) |sec| {
            return .{ .credential = .{ .user = user, .secret_hex = sec }, .persist = false };
        }
    }

    // Otherwise derive from a plaintext password. Precedence: --password, then
    // $ZIT_PASSWORD, then an interactive no-echo prompt.
    const password: []const u8 = parsed.password orelse amb.push_password orelse
        (readPasswordNoEcho(streams, user, pw_buf) orelse return error.NoPassword);
    const secret = zit.sync.deriveSharedSecret(project_code, user, password);
    return .{ .credential = .{ .user = user, .secret_hex = secret }, .persist = true };
}

/// DECISION-3 — the 40-hex secret stored for `user` in this repo's config, or
/// null when none is stored or it belongs to a different login. Read-only; the
/// returned value is an owned copy (the config frame is freed here).
fn storedSecretFor(
    gpa: std.mem.Allocator,
    io: std.Io,
    dir: std.Io.Dir,
    user: []const u8,
) error{ OutOfMemory, Storage }!?[40]u8 {
    var file = dir.openFile(io, repo_filename, .{}) catch return error.Storage;
    defer file.close(io);
    const st = file.stat(io) catch return error.Storage;
    var fs = storage.FileStorage.init(io, file, st.size);
    var log = logstore.LogStore.openReadOnly(gpa, fs.storageFile()) catch |e| switch (e) {
        error.OutOfMemory => return error.OutOfMemory,
        else => return error.Storage,
    };
    defer log.deinit(gpa);
    var oc = (log.readConfig(gpa) catch |e| switch (e) {
        error.OutOfMemory => return error.OutOfMemory,
        else => return error.Storage,
    }) orelse return null;
    defer oc.deinit(gpa);
    if (oc.config.login_secret_hex.len != 40) return null;
    if (!std.mem.eql(u8, oc.config.login_user, user)) return null;
    var out: [40]u8 = undefined;
    @memcpy(&out, oc.config.login_secret_hex[0..40]);
    return out;
}

/// DECISION-3 — persist the derived login credential into the repo config,
/// preserving every other field, then tighten the repository file to 0600 so
/// the stored secret is owner-only at rest. The permission tightening is
/// best-effort: the secret is already durable, so a filesystem that refuses
/// chmod does not fail the push (perms + FDE are the boundary, not the sole
/// guarantee — per DECISION-3). NEVER writes the plaintext password.
fn storePushCredential(
    gpa: std.mem.Allocator,
    io: std.Io,
    dir: std.Io.Dir,
    credential: zit.sync.Credential,
) PersistError!void {
    std.debug.assert(credential.secret_hex.len == 40);
    // The field-preserving RMW writes the credential without disturbing the
    // project code or a saved remote (a bare `putConfig` here would drop them).
    try updateConfig(gpa, io, dir, .{ .login = credential });
    // Owner-only now that the file carries a credential (best-effort; see above).
    var file = dir.openFile(io, repo_filename, .{ .mode = .read_write }) catch return;
    defer file.close(io);
    file.setPermissions(io, File.Permissions.fromMode(0o600)) catch {};
}

/// CLI/IO seam — prompt on stderr and read one password line from the terminal
/// with echo disabled, returning the trimmed line in `buf` (never its newline),
/// or null on a read failure or an empty line. Echo is restored on every exit
/// path; when stdin is not a terminal (`tcgetattr` fails) the line is read with
/// no termios toggle. Not exercised by the automated gates (they pass
/// `--password`/`$ZIT_PASSWORD`) — this is the interactive path.
fn readPasswordNoEcho(
    streams: cli.Streams,
    user: []const u8,
    buf: *[max_password_len]u8,
) ?[]const u8 {
    streams.err.print("Password for {s}: ", .{user}) catch {};
    streams.err.flush() catch {};
    const fd = std.posix.STDIN_FILENO;
    const restore: ?std.posix.termios = blk: {
        var t = std.posix.tcgetattr(fd) catch break :blk null;
        const saved = t;
        t.lflag.ECHO = false;
        std.posix.tcsetattr(fd, .FLUSH, t) catch break :blk null;
        break :blk saved;
    };
    defer if (restore) |saved| {
        std.posix.tcsetattr(fd, .FLUSH, saved) catch {};
        streams.err.writeAll("\n") catch {};
        streams.err.flush() catch {};
    };
    const n = std.posix.read(fd, buf) catch return null;
    if (n == 0) return null;
    var line = buf[0..n];
    // Trim a single trailing CRLF/LF; interior bytes (including spaces) are kept.
    if (line.len > 0 and line[line.len - 1] == '\n') line = line[0 .. line.len - 1];
    if (line.len > 0 and line[line.len - 1] == '\r') line = line[0 .. line.len - 1];
    if (line.len == 0) return null;
    return line;
}

/// Drives a push client (§3.2/§3.3) to convergence over `tr`, uploading the
/// artifacts the peer lacks; returns the upload count and round count. The client
/// announces `ms`'s holdings via igot and uploads what the peer gimme's (§3.3
/// r24: push sends content but never obtains it). Sans-IO on this side of
/// `tr.exchange`. Its closed error set is the driver's own causes unioned with
/// the transport's, so `runPush` switches it exhaustively.
fn pushRun(
    gpa: std.mem.Allocator,
    tr: *transport.Transport,
    ms: *repostate.MemStore,
    project_code: []const u8,
    credential: ?zit.sync.Credential,
) sync_driver.RunError(*transport.Transport)!PushOutcome {
    var client = zit.sync.Client{
        .state = zit.sync.initialState(.push),
        .operation = .push,
        .self_servercode = sync_servercode,
        .project_code = project_code,
        .version = .{ .version = sync_client_version },
        // sync §6 — a named credential makes `buildRequest` emit the §6.1 login
        // card; null leaves the push anonymous (unchanged behavior).
        .credential = credential,
    };
    defer client.deinit(gpa);
    const stats = try sync_driver.runSync(gpa, &client, ms.store(), tr, max_sync_rounds);
    // Read the upload counter before `deinit` runs (the return expression is
    // evaluated first): the exact number of `file` cards this session emitted.
    return .{ .rounds = stats.rounds, .sent = client.content_sent };
}

// ---------------------------------------------------------------------------
// `zit sync [URL]` — BIDIRECTIONAL convergence with a peer in ONE session.
// Unlike pull-then-push, a single `operation=.sync` client announces igot,
// requests gimme, AND uploads on one §3.7 exchange (§3.2 round structure: a sync
// request carries both push and pull each round) — not a third copy of the driver
// loop over two transport sessions.
// Opens the cwd's `.zit`, resolves the URL (arg → saved remote), may
// authenticate (sync uploads), persists what it receives like pull, and reports
// both directions. A saved `zit remote` lets the URL be omitted.
// ---------------------------------------------------------------------------

/// What one sync run reports: uploads (`sent`) and the round count. Received is
/// measured by the caller as the persisted-count growth (as pull does), so it is
/// not carried here.
const SyncOutcome = struct { rounds: usize, sent: usize };

pub fn runSync(
    gpa: std.mem.Allocator,
    io: std.Io,
    arena: std.mem.Allocator,
    verb_args: []const []const u8,
    streams: cli.Streams,
    amb: cli.Ambient,
) cli.Exit {
    const parsed = switch (cli.parseSyncArgs(verb_args)) {
        .ok => |p| p,
        .unknown_option => |tok| return cli.usageError(streams, "sync", .unknown_option, tok),
        .extra_argument => |tok| return cli.usageError(streams, "sync", .extra_argument, tok),
    };
    if (parsed.help) {
        streams.out.writeAll(sync_help) catch {};
        return .ok;
    }

    // sync §6.3 — a login name is whitespace-free (mirrors push). Reject a
    // malformed --user as usage before the secret derivation's assertion.
    if (parsed.user) |u| {
        if (u.len > 0 and zit.sync.containsFossilWhitespace(u)) {
            streams.err.writeAll("zit sync: --user must not contain whitespace\n") catch {};
            return .failure;
        }
    }

    // Locate the current repository before any network I/O.
    const root = (cli.discoverRoot(arena, io) catch return stateOom(streams, "sync")) orelse {
        streams.err.writeAll(not_a_repo_msg) catch {};
        return .failure;
    };
    var root_dir = root.dir;
    defer root_dir.close(io);

    // sync §4/§6 — the explicit URL wins; an omitted one falls back to the saved
    // remote; a repo with neither is a usage error. `url_buf` outlives `open`.
    var url_buf: [logstore.max_config_str]u8 = undefined;
    const url = resolveRemoteUrl(gpa, io, root_dir, parsed.url, &url_buf) catch |e| switch (e) {
        error.OutOfMemory => return stateOom(streams, "sync"),
        error.Storage => return repoReadFail(streams, "sync"),
        error.NoRemote => return noRemoteError(streams, "sync"),
    };

    var tr = transport.Transport.open(io, gpa, url) catch |e|
        return syncOpenFail(streams, "sync", url, e);
    defer tr.close();

    // Snapshot holdings: the igot announcement and the baseline for the received
    // count. A sync both uploads what the peer lacks and receives what this side
    // lacks — the held set can only grow (§3.3 storage only adds content).
    var ms = repostate.MemStore.init(gpa);
    defer ms.deinit();
    loadRepoHoldings(gpa, io, root_dir, &ms) catch |e| switch (e) {
        error.OutOfMemory => return stateOom(streams, "sync"),
        error.Storage => return repoReadFail(streams, "sync"),
    };
    const held_before = contentCount(ms.store());

    // §9.1 — announce this repository's adopted project code (a real peer refuses
    // a mismatch). The buffer outlives the run — the client borrows it.
    var pc_buf: [zit.sync.max_projectcode_len]u8 = undefined;
    const project_code = repoProjectCode(gpa, io, root_dir, &pc_buf) catch |e| switch (e) {
        error.OutOfMemory => return stateOom(streams, "sync"),
        error.Storage => return repoReadFail(streams, "sync"),
    };

    // sync §6 — resolve the login credential (null for anonymous). A sync
    // uploads, so it may authenticate exactly as push does; the plaintext lives
    // only in `pw_buf`/argv/env and is scrubbed on every exit from this scope.
    var pw_buf: [max_password_len]u8 = undefined;
    defer std.crypto.secureZero(u8, &pw_buf);
    const resolved = resolvePushCredential(gpa, io, root_dir, streams, project_code, .{ .url = url, .user = parsed.user, .password = parsed.password }, amb, &pw_buf) catch |e| switch (e) {
        error.OutOfMemory => return stateOom(streams, "sync"),
        error.Storage => return repoReadFail(streams, "sync"),
        error.NoPassword => {
            streams.err.writeAll("zit sync: no password supplied for --user\n") catch {};
            return .failure;
        },
    };

    const outcome = syncRun(gpa, &tr, &ms, project_code, if (resolved) |r| r.credential else null) catch |e|
        return syncRunFail(streams, "sync", e);

    const written = persistStore(gpa, io, root_dir, ms.store()) catch |e| switch (e) {
        error.OutOfMemory => return stateOom(streams, "sync"),
        error.Storage => return repoReadFail(streams, "sync"),
        error.Corrupt => return renderStorage(streams, root.abs, "the peer sent a malformed artifact name"),
    };
    std.debug.assert(written >= held_before);
    const received = written - held_before;

    // DECISION-3 — persist a freshly derived secret only after the peer accepted
    // the authenticated session, and only when it was freshly derived (mirrors push).
    if (resolved) |r| {
        if (r.persist) {
            storePushCredential(gpa, io, root_dir, r.credential) catch |e| switch (e) {
                error.OutOfMemory => return stateOom(streams, "sync"),
                else => return repoReadFail(streams, "sync"),
            };
        }
    }

    if (outcome.sent == 0 and received == 0) {
        streams.out.writeAll("already in sync\n") catch {};
    } else {
        streams.out.print("sent {d}, received {d} in {d} rounds\n", .{ outcome.sent, received, outcome.rounds }) catch {};
    }
    return .ok;
}

/// Drives a bidirectional sync client (§3.2 round structure — both directions on
/// ONE §3.7 session) to convergence over `tr`; returns the upload count and round
/// count.
/// The received count is the caller's persisted-count growth. Sans-IO on this
/// side of `tr.exchange`; its closed error set is switched exhaustively by
/// `runSync`. The sibling of `pullRun`/`pushRun`, differing only in the
/// operation tag (`.sync`, which drives igot + gimme + upload on one session).
fn syncRun(
    gpa: std.mem.Allocator,
    tr: *transport.Transport,
    ms: *repostate.MemStore,
    project_code: []const u8,
    credential: ?zit.sync.Credential,
) sync_driver.RunError(*transport.Transport)!SyncOutcome {
    var client = zit.sync.Client{
        .state = zit.sync.initialState(.sync),
        .operation = .sync,
        .self_servercode = sync_servercode,
        .project_code = project_code,
        .version = .{ .version = sync_client_version },
        // sync §6 — a named credential emits the §6.1 login card; null is anonymous.
        .credential = credential,
    };
    defer client.deinit(gpa);
    const stats = try sync_driver.runSync(gpa, &client, ms.store(), tr, max_sync_rounds);
    return .{ .rounds = stats.rounds, .sent = client.content_sent };
}

// ---------------------------------------------------------------------------
// `zit remote [URL | --unset]` — read, set, or clear the single default peer URL
// (sync §4/§6). Bare prints the saved remote; a URL saves it; --unset clears it.
// A local, non-synced config value (`config.remote_url`) — anonymous only, no
// credential in the URL — that lets pull/push/sync omit their URL argument. Not
// a git-style named-remote registry: one value, matching the single-value config.
// ---------------------------------------------------------------------------

pub fn runRemote(
    gpa: std.mem.Allocator,
    io: std.Io,
    arena: std.mem.Allocator,
    verb_args: []const []const u8,
    streams: cli.Streams,
) cli.Exit {
    const parsed = switch (cli.parseRemoteArgs(verb_args)) {
        .ok => |p| p,
        .unknown_option => |tok| return cli.usageError(streams, "remote", .unknown_option, tok),
        .extra_argument => |tok| return cli.usageError(streams, "remote", .extra_argument, tok),
    };
    if (parsed.help) {
        streams.out.writeAll(remote_help) catch {};
        return .ok;
    }

    const root = (cli.discoverRoot(arena, io) catch return stateOom(streams, "remote")) orelse {
        streams.err.writeAll(not_a_repo_msg) catch {};
        return .failure;
    };
    var root_dir = root.dir;
    defer root_dir.close(io);

    if (parsed.unset) {
        // Overlay remote_url with empty — clears the saved value, every other
        // config field preserved by the field-preserving RMW.
        updateConfig(gpa, io, root_dir, .{ .remote_url = "" }) catch |e| switch (e) {
            error.OutOfMemory => return stateOom(streams, "remote"),
            else => return repoReadFail(streams, "remote"),
        };
        streams.out.writeAll("remote cleared\n") catch {};
        return .ok;
    }

    if (parsed.url.len > 0) {
        // A URL over the config-field cap cannot be stored (the putConfig assert);
        // reject it as usage rather than trip the assertion.
        if (parsed.url.len > logstore.max_config_str) {
            streams.err.print("zit remote: URL too long (max {d} bytes)\n", .{logstore.max_config_str}) catch {};
            return .failure;
        }
        updateConfig(gpa, io, root_dir, .{ .remote_url = parsed.url }) catch |e| switch (e) {
            error.OutOfMemory => return stateOom(streams, "remote"),
            else => return repoReadFail(streams, "remote"),
        };
        streams.out.print("remote set to {s}\n", .{parsed.url}) catch {};
        return .ok;
    }

    // Bare `zit remote` — print the saved remote (reusing resolveRemoteUrl's read
    // with an empty arg; NoRemote means none is saved).
    var url_buf: [logstore.max_config_str]u8 = undefined;
    const url = resolveRemoteUrl(gpa, io, root_dir, "", &url_buf) catch |e| switch (e) {
        error.OutOfMemory => return stateOom(streams, "remote"),
        error.Storage => return repoReadFail(streams, "remote"),
        error.NoRemote => {
            streams.out.writeAll("no remote set\n") catch {};
            return .ok;
        },
    };
    streams.out.print("{s}\n", .{url}) catch {};
    return .ok;
}

test "serve §3.3: a pushed file card is persisted durably into the served repository's log" {
    const gpa = std.testing.allocator;
    const io = std.testing.io;

    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();

    // A fresh, empty on-disk repository (header only) — the "server" a peer pushes
    // new content into. Built through the real LogStore, not seeded from fossil.
    const hash_kind: u8 = @intFromEnum(zit.artifact.defaultPolicy(false, false));
    {
        var f = try tmp.dir.createFile(io, repo_filename, .{ .read = true, .exclusive = true });
        defer f.close(io);
        var fs = storage.FileStorage.init(io, f, 0);
        var seed = try logstore.LogStore.open(gpa, fs.storageFile(), hash_kind);
        defer seed.deinit(gpa);
        try seed.commit();
    }

    // A pushed full-content file card for a 64-hex (SHA3-256 width) artifact the
    // server lacks (sync §7.1: `file HASH SIZE\n` + payload; §4.1 zlib container).
    // The name is the artifact's real SHA3-256 (computed by our own hasher, not
    // fossil) so the server's §7.7 commit-hook recompute accepts it.
    const push_body = "pushed artifact bytes\n"; // 22 bytes
    var name_buf: [64]u8 = undefined;
    const push_hash = zit.artifact.artifactName(.sha3_256, push_body, &name_buf);
    const request_plain = try std.fmt.allocPrint(gpa, "pragma client-version 20000\nfile {s} {d}\n{s}", .{ push_hash, push_body.len, push_body });
    defer gpa.free(request_plain);
    const request = try zit.container.encode(gpa, request_plain, .default);
    defer gpa.free(request);

    var digest_buf: [digest_len]u8 = undefined;
    const push_digest = decodeWireName(push_hash, &digest_buf).?;

    const server = zit.sync.Server{ .version = serve_mod.server_version };

    // Answer the push twice through the real `serve.Responder` seam over a SINGLE
    // open handle (§3.8): the second answer re-reads the current on-disk log from
    // the same handle and sees the first push's frame, so it is an idempotent
    // re-push (content is addressed by hash — no dup frame). A stale startup
    // snapshot would instead re-append the artifact; this handle is never reopened.
    {
        var file = try tmp.dir.openFile(io, repo_filename, .{});
        defer file.close(io);
        var ctx = ServeCtx{ .io = io, .root_dir = tmp.dir, .file = file, .server = &server };
        defer ctx.dropSnapshot();
        for (0..2) |_| {
            const reply = try serveAnswer(&ctx, gpa, request);
            gpa.free(reply.body);
        }
    }

    // THE PROOF: re-open the log from disk in a fresh handle; the pushed artifact
    // is present with its exact content — durably stored, not discarded — and
    // appears exactly once (the idempotent re-push added no second frame).
    var file = try tmp.dir.openFile(io, repo_filename, .{});
    defer file.close(io);
    const st = try file.stat(io);
    var fs = storage.FileStorage.init(io, file, st.size);
    var log = try logstore.LogStore.openReadOnly(gpa, fs.storageFile());
    defer log.deinit(gpa);

    const rid = log.lookup(push_digest) orelse return error.PushedArtifactMissing;
    const stored = (try log.content(gpa, rid)).?;
    defer gpa.free(stored);
    try std.testing.expectEqualStrings(push_body, stored);

    const rids = try log.ridsInPersistedOrder(gpa);
    defer gpa.free(rids);
    var artifacts: usize = 0;
    for (rids) |r| {
        if (log.kindOf(r) != .config) artifacts += 1;
    }
    try std.testing.expectEqual(@as(usize, 1), artifacts);
}

// --- D4: client verbatim retention + the §7.7 persist hook ------------------

/// Create a fresh, empty (header-only, committed) `.zit` in `dir` so a
/// `persistStore` test can persist into it, exactly like a just-cloned repo.
fn d4SeedEmptyRepo(gpa: std.mem.Allocator, io: std.Io, dir: std.Io.Dir) !void {
    const hash_kind: u8 = @intFromEnum(zit.artifact.defaultPolicy(false, false));
    var f = try dir.createFile(io, repo_filename, .{ .read = true, .exclusive = true });
    defer f.close(io);
    var fs = storage.FileStorage.init(io, f, 0);
    var seed = try logstore.LogStore.open(gpa, fs.storageFile(), hash_kind);
    defer seed.deinit(gpa);
    try seed.commit();
}

/// Reopen the persisted `.zit` in `dir` for reading. `fs_out` is a caller-owned
/// `FileStorage` that MUST outlive the returned `LogStore` (its `storageFile()`
/// vtable borrows `fs_out`); the caller closes `fs_out.file` and deinits `log`.
fn d4OpenLog(gpa: std.mem.Allocator, io: std.Io, dir: std.Io.Dir, fs_out: *storage.FileStorage) !logstore.LogStore {
    const file = try dir.openFile(io, repo_filename, .{});
    const st = try file.stat(io);
    fs_out.* = storage.FileStorage.init(io, file, st.size);
    return try logstore.LogStore.open(gpa, fs_out.storageFile(), 0);
}

test "sync §7.7: persistStore recomputes the artifact hash and aborts on a lying container, writing no frame" {
    const gpa = std.testing.allocator;
    const io = std.testing.io;
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    try d4SeedEmptyRepo(gpa, io, tmp.dir);

    // A received full cfile whose wire NAME does not name its content (§7.7 —
    // a clone skips the transfer-layer pre-storage check, so a lying container
    // reaches the persist commit hook).
    var ms = repostate.MemStore.init(gpa);
    defer ms.deinit();
    const s = ms.store();
    const content = "content that the wire name does not actually name\n";
    const lying_name = "0" ** 64; // not SHA3-256(content)
    const rid = try s.put(lying_name, content);
    const container = try zit.container.encode(gpa, content, .default);
    defer gpa.free(container);
    try s.retain(rid, .{ .full = container });

    // §7.7 L1336-1338 — recompute and abort before ANY frame is written.
    try std.testing.expectError(error.Corrupt, persistStore(gpa, io, tmp.dir, ms.store()));

    var fs: storage.FileStorage = undefined;
    var log = try d4OpenLog(gpa, io, tmp.dir, &fs);
    defer fs.file.close(io);
    defer log.deinit(gpa);
    var digest_buf: [digest_len]u8 = undefined;
    const digest = decodeWireName(lying_name, &digest_buf).?;
    try std.testing.expect(log.lookup(digest) == null); // no frame landed
}

test "sync §7.2/§4.1: persistStore stores received full and delta wire forms verbatim; a re-serve preserves the delta container byte-for-byte" {
    const gpa = std.testing.allocator;
    const io = std.testing.io;
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    try d4SeedEmptyRepo(gpa, io, tmp.dir);

    // Model a clone reply in §8.2 RID order: a full base cfile then a delta
    // cfile against it. Names are the real SHA3-256 so the §7.7 hook accepts.
    var ms = repostate.MemStore.init(gpa);
    defer ms.deinit();
    const s = ms.store();
    const base = "the base artifact with a long shared run so the delta stays a compact copy payload distinct in size from the reconstructed target";
    const target = base ++ " plus a small trailing mutation";
    var bn: [64]u8 = undefined;
    var tn: [64]u8 = undefined;
    const base_name = zit.artifact.artifactName(.sha3_256, base, &bn);
    const target_name = zit.artifact.artifactName(.sha3_256, target, &tn);

    const base_container = try zit.container.encode(gpa, base, .default);
    defer gpa.free(base_container);
    const base_rid = try s.put(base_name, base);
    try s.retain(base_rid, .{ .full = base_container });

    const raw = try zit.delta.create(gpa, base, target);
    defer gpa.free(raw);
    const delta_container = try zit.container.encode(gpa, raw, .default);
    defer gpa.free(delta_container);
    const ts = try zit.delta.targetSize(raw);
    const delta_rid = try s.put(target_name, target);
    try s.retain(delta_rid, .{ .delta = .{ .src = base_name, .container = delta_container, .target_size = ts } });

    const written = try persistStore(gpa, io, tmp.dir, ms.store());
    try std.testing.expectEqual(@as(usize, 2), written);

    var fs: storage.FileStorage = undefined;
    var log = try d4OpenLog(gpa, io, tmp.dir, &fs);
    defer fs.file.close(io);
    defer log.deinit(gpa);

    // The target persisted as a `.delta` frame that reconstructs to the original.
    var tdig: [digest_len]u8 = undefined;
    const target_digest = decodeWireName(target_name, &tdig).?;
    const trid = log.lookup(target_digest).?;
    try std.testing.expectEqual(logstore.FrameKind.delta, log.kindOf(trid));
    const got = (try log.content(gpa, trid)).?;
    defer gpa.free(got);
    try std.testing.expectEqualStrings(target, got);

    // Clone-of-clone: a re-serve snapshot emits the SAME delta container bytes
    // (§7.2 retransmitted verbatim) — the compactness survives the second hop.
    var serve_ms = repostate.MemStore.init(gpa);
    defer serve_ms.deinit();
    try loadContentStore(gpa, &log, &serve_ms);
    const serve_rid = serve_ms.store().lookup(target_name).?;
    const wf = serve_ms.store().wire(serve_rid).?;
    try std.testing.expect(wf.delta != null);
    try std.testing.expectEqualSlices(u8, delta_container, wf.container);
    try std.testing.expectEqual(@as(u64, target.len), wf.delta.?.target_size);
}

test "sync §3.3: persistStore falls back to full form when a retained delta's basis is absent, never a phantom-based delta" {
    const gpa = std.testing.allocator;
    const io = std.testing.io;
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    try d4SeedEmptyRepo(gpa, io, tmp.dir);

    // Only the delta is present; its basis was never received/stored (a legal
    // §3.3 outcome that must not persist a delta over a phantom base).
    var ms = repostate.MemStore.init(gpa);
    defer ms.deinit();
    const s = ms.store();
    const base = "an absent basis whose long shared run would make a compact delta if it were here at all";
    const target = base ++ " mutated once";
    var bn: [64]u8 = undefined;
    var tn: [64]u8 = undefined;
    const base_name = zit.artifact.artifactName(.sha3_256, base, &bn);
    const target_name = zit.artifact.artifactName(.sha3_256, target, &tn);
    const raw = try zit.delta.create(gpa, base, target);
    defer gpa.free(raw);
    const dc = try zit.container.encode(gpa, raw, .default);
    defer gpa.free(dc);
    const ts = try zit.delta.targetSize(raw);
    const rid = try s.put(target_name, target);
    try s.retain(rid, .{ .delta = .{ .src = base_name, .container = dc, .target_size = ts } });

    const written = try persistStore(gpa, io, tmp.dir, ms.store());
    try std.testing.expectEqual(@as(usize, 1), written);

    var fs: storage.FileStorage = undefined;
    var log = try d4OpenLog(gpa, io, tmp.dir, &fs);
    defer fs.file.close(io);
    defer log.deinit(gpa);

    // The target persisted as a FULL (.zlib) frame — the fallback, not a delta.
    var tdig: [digest_len]u8 = undefined;
    const target_digest = decodeWireName(target_name, &tdig).?;
    const trid = log.lookup(target_digest).?;
    try std.testing.expectEqual(logstore.FrameKind.zlib, log.kindOf(trid));
    const got = (try log.content(gpa, trid)).?;
    defer gpa.free(got);
    try std.testing.expectEqualStrings(target, got);

    // No phantom basis frame was minted for the absent DELTASRC.
    var bdig: [digest_len]u8 = undefined;
    const base_digest = decodeWireName(base_name, &bdig).?;
    try std.testing.expect(log.lookup(base_digest) == null);
}

test "sync §7.7/§3.3: persistStore stores a retained delta chain up to the read bound and falls back to full form beyond it" {
    const gpa = std.testing.allocator;
    const io = std.testing.io;
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    try d4SeedEmptyRepo(gpa, io, tmp.dir);

    var ms = repostate.MemStore.init(gpa);
    defer ms.deinit();
    const s = ms.store();
    const cap: usize = logstore.max_delta_chain; // the READ bound (32)
    const n: usize = cap + 2; // links 0..cap+1: link cap+1's basis sits at depth cap

    // A chain grown one byte per link so each delta is tiny. Link i's basis is
    // link i-1 (§8.2 RID order); the persist path stores a `.delta` iff the base
    // depth is under `max_delta_chain`, so link cap+1 falls back to full form.
    var content: std.ArrayList(u8) = .empty;
    defer content.deinit(gpa);
    try content.appendSlice(gpa, "chain-root-artifact-with-a-long-shared-run-for-compact-forward-deltas");

    var prev_name: [64]u8 = undefined;
    var prev_name_len: usize = 0;
    var deep_name: [64]u8 = undefined; // link cap (deepest stored delta)
    var deep_len: usize = 0;
    var last_name: [64]u8 = undefined; // link cap+1 (the fallback)
    var last_len: usize = 0;

    var i: usize = 0;
    while (i < n) : (i += 1) {
        try content.append(gpa, 'a' + @as(u8, @intCast(i % 26)));
        const cur = content.items;
        const prev = cur[0 .. cur.len - 1];
        var nbuf: [64]u8 = undefined;
        const name = zit.artifact.artifactName(.sha3_256, cur, &nbuf);
        const rid = try s.put(name, cur);
        if (i == 0) {
            const c = try zit.container.encode(gpa, cur, .default);
            defer gpa.free(c);
            try s.retain(rid, .{ .full = c });
        } else {
            const raw = try zit.delta.create(gpa, prev, cur);
            defer gpa.free(raw);
            const c = try zit.container.encode(gpa, raw, .default);
            defer gpa.free(c);
            const tsz = try zit.delta.targetSize(raw);
            try s.retain(rid, .{ .delta = .{ .src = prev_name[0..prev_name_len], .container = c, .target_size = tsz } });
        }
        @memcpy(prev_name[0..name.len], name);
        prev_name_len = name.len;
        if (i == cap) {
            @memcpy(deep_name[0..name.len], name);
            deep_len = name.len;
        }
        if (i == n - 1) {
            @memcpy(last_name[0..name.len], name);
            last_len = name.len;
        }
    }

    _ = try persistStore(gpa, io, tmp.dir, ms.store());

    var fs: storage.FileStorage = undefined;
    var log = try d4OpenLog(gpa, io, tmp.dir, &fs);
    defer fs.file.close(io);
    defer log.deinit(gpa);

    // Link cap is the deepest `.delta` the write bound allows; link cap+1 fell
    // back to a full `.zlib` frame yet still reconstructs (never a Corrupt read).
    var ddig: [digest_len]u8 = undefined;
    const deep_rid = log.lookup(decodeWireName(deep_name[0..deep_len], &ddig).?).?;
    try std.testing.expectEqual(logstore.FrameKind.delta, log.kindOf(deep_rid));

    var ldig: [digest_len]u8 = undefined;
    const last_rid = log.lookup(decodeWireName(last_name[0..last_len], &ldig).?).?;
    try std.testing.expectEqual(logstore.FrameKind.zlib, log.kindOf(last_rid));
    const got = (try log.content(gpa, last_rid)).?;
    defer gpa.free(got);
    try std.testing.expectEqualStrings(content.items, got);
}

test "sync §7.2/§7.4/§3.3: a dangling reverse-delta cfile persists as a verbatim .delta frame that re-serves byte-identically" {
    const gpa = std.testing.allocator;
    const io = std.testing.io;
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    try d4SeedEmptyRepo(gpa, io, tmp.dir);

    // Fossil's clone stream: a reverse delta (the OLDER revision stored as a
    // delta of the NEWER one) arrives BEFORE its basis, so it is dangling. The
    // client retains the §4.1 container through the pending queue and, when the
    // basis lands, persists it as a compact `.delta` — not a full re-compression.
    const newer = "artifact NEWER: a long shared body kept verbatim so the reverse delta reconstructing the older revision stays a compact copy op";
    const older = "artifact NEWER: a long shared body kept verbatim so the reverse delta reconstructing the older revision stays a compact copy op minus tail";
    var nn: [64]u8 = undefined;
    var on: [64]u8 = undefined;
    const newer_name = zit.artifact.artifactName(.sha3_256, newer, &nn);
    const older_name = zit.artifact.artifactName(.sha3_256, older, &on);
    const newer_container = try zit.container.encode(gpa, newer, .default);
    defer gpa.free(newer_container);
    const raw = try zit.delta.create(gpa, newer, older); // OLDER = delta ∘ NEWER
    defer gpa.free(raw);
    const delta_container = try zit.container.encode(gpa, raw, .default);
    defer gpa.free(delta_container);

    var ms = repostate.MemStore.init(gpa);
    defer ms.deinit();
    var client = zit.sync.Client{ .operation = .clone, .state = .clone_round0 };
    defer client.deinit(gpa);

    var body: std.ArrayList(u8) = .empty;
    defer body.deinit(gpa);
    try body.appendSlice(gpa, "clone_seqno 0\npush 1111111111111111 aa87ea41b0bb05ba528f7a3bdba700d7e23d6318\n");
    // Reverse (dangling) order: the delta cfile precedes its basis cfile.
    const dh = try std.fmt.allocPrint(gpa, "cfile {s} {s} {d} {d}\n", .{ older_name, newer_name, older.len, delta_container.len });
    defer gpa.free(dh);
    try body.appendSlice(gpa, dh);
    try body.appendSlice(gpa, delta_container);
    try body.append(gpa, '\n');
    const bh = try std.fmt.allocPrint(gpa, "cfile {s} {d} {d}\n", .{ newer_name, newer.len, newer_container.len });
    defer gpa.free(bh);
    try body.appendSlice(gpa, bh);
    try body.appendSlice(gpa, newer_container);
    try body.append(gpa, '\n');

    _ = try client.consumeReply(gpa, ms.store(), body.items);

    const written = try persistStore(gpa, io, tmp.dir, ms.store());
    try std.testing.expectEqual(@as(usize, 2), written);

    var fs: storage.FileStorage = undefined;
    var log = try d4OpenLog(gpa, io, tmp.dir, &fs);
    defer fs.file.close(io);
    defer log.deinit(gpa);

    // §3.3 — the older revision persisted as a `.delta`, not a full re-compression,
    // and reconstructs the original bytes (delta ∘ basis).
    var od: [digest_len]u8 = undefined;
    const older_rid = log.lookup(decodeWireName(older_name, &od).?).?;
    try std.testing.expectEqual(logstore.FrameKind.delta, log.kindOf(older_rid));
    const got = (try log.content(gpa, older_rid)).?;
    defer gpa.free(got);
    try std.testing.expectEqualStrings(older, got);

    // §7.2 — a clone-of-clone re-serve re-emits the SAME delta container byte-for-byte.
    var serve_ms = repostate.MemStore.init(gpa);
    defer serve_ms.deinit();
    try loadContentStore(gpa, &log, &serve_ms);
    const srid = serve_ms.store().lookup(older_name).?;
    const wf = serve_ms.store().wire(srid).?;
    try std.testing.expect(wf.delta != null);
    try std.testing.expectEqualSlices(u8, delta_container, wf.container);
    try std.testing.expectEqualStrings(newer_name, wf.delta.?.src_hash);
}

test "sync §7.2/§3.3: persistStore stores a whole reverse-delta chain, ordering every basis frame before the delta that points at it" {
    const gpa = std.testing.allocator;
    const io = std.testing.io;
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    try d4SeedEmptyRepo(gpa, io, tmp.dir);

    // A reverse-delta chain in fossil's storage shape: revision i is stored as a
    // delta of the NEWER revision i+1, the newest stored full. The store rids run
    // oldest..newest (put order), so EVERY delta's basis sits at a HIGHER rid than
    // the delta — the worst case for physical ordering. persist's worklist must
    // still write each basis frame before its delta so `delta_of` points backward;
    // naive rid-order would degrade all but the oldest link to full form.
    const links = 6;
    var ms = repostate.MemStore.init(gpa);
    defer ms.deinit();
    const s = ms.store();

    var bodies: [links][]u8 = undefined;
    var made: usize = 0;
    defer for (bodies[0..made]) |b| gpa.free(b);
    var names: [links][64]u8 = undefined;
    var nlen: [links]usize = undefined;
    for (0..links) |i| {
        bodies[i] = try std.fmt.allocPrint(gpa, "chain body: one long shared prefix kept across every revision for compact reverse deltas -- revision {d}", .{i});
        made += 1;
        nlen[i] = zit.artifact.artifactName(.sha3_256, bodies[i], &names[i]).len;
    }
    // put oldest..newest (store rids 1..links); older i is a reverse delta of newer i+1.
    for (0..links) |i| {
        const rid = try s.put(names[i][0..nlen[i]], bodies[i]);
        if (i == links - 1) {
            const c = try zit.container.encode(gpa, bodies[i], .default);
            defer gpa.free(c);
            try s.retain(rid, .{ .full = c });
        } else {
            const dr = try zit.delta.create(gpa, bodies[i + 1], bodies[i]); // older from newer
            defer gpa.free(dr);
            const c = try zit.container.encode(gpa, dr, .default);
            defer gpa.free(c);
            try s.retain(rid, .{ .delta = .{ .src = names[i + 1][0..nlen[i + 1]], .container = c, .target_size = bodies[i].len } });
        }
    }

    _ = try persistStore(gpa, io, tmp.dir, ms.store());

    var fs: storage.FileStorage = undefined;
    var log = try d4OpenLog(gpa, io, tmp.dir, &fs);
    defer fs.file.close(io);
    defer log.deinit(gpa);

    // The newest link is full; every older link is a `.delta` whose basis frame is
    // physically earlier (a LOWER log rid = a lower offset), and all reconstruct.
    for (0..links) |i| {
        var d: [digest_len]u8 = undefined;
        const rid = log.lookup(decodeWireName(names[i][0..nlen[i]], &d).?).?;
        const rgot = (try log.content(gpa, rid)).?;
        defer gpa.free(rgot);
        try std.testing.expectEqualStrings(bodies[i], rgot);
        if (i == links - 1) {
            try std.testing.expectEqual(logstore.FrameKind.zlib, log.kindOf(rid));
        } else {
            try std.testing.expectEqual(logstore.FrameKind.delta, log.kindOf(rid));
            var bd: [digest_len]u8 = undefined;
            const brid = log.lookup(decodeWireName(names[i + 1][0..nlen[i + 1]], &bd).?).?;
            try std.testing.expect(brid < rid); // basis frame precedes the reverse-delta
        }
    }
}

test "sync §7.7: persistStore verifies a retained delta's reconstruction and refuses a lying delta container, writing no frame" {
    const gpa = std.testing.allocator;
    const io = std.testing.io;
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    try d4SeedEmptyRepo(gpa, io, tmp.dir);

    // A retained delta whose stored plaintext does NOT hash to its wire name (a
    // clone skips the transfer-layer pre-storage check, §7.7 r6, so a lying
    // delta reaches the persist commit hook). The §7.7 recompute runs BEFORE the
    // wire-form branch, so a delta-form entry is rejected exactly like a full one.
    var ms = repostate.MemStore.init(gpa);
    defer ms.deinit();
    const s = ms.store();
    const content = "content that the wire name does not actually name, delivered as a delta\n";
    const lying_name = "0" ** 64; // not SHA3-256(content)
    const some_basis = "1" ** 64; // an absent basis; verify fails before it is even consulted
    const raw = try zit.delta.create(gpa, "a plausible basis blob", content);
    defer gpa.free(raw);
    const dc = try zit.container.encode(gpa, raw, .default);
    defer gpa.free(dc);
    const rid = try s.put(lying_name, content);
    try s.retain(rid, .{ .delta = .{ .src = some_basis, .container = dc, .target_size = content.len } });

    try std.testing.expectError(error.Corrupt, persistStore(gpa, io, tmp.dir, ms.store()));

    var fs: storage.FileStorage = undefined;
    var log = try d4OpenLog(gpa, io, tmp.dir, &fs);
    defer fs.file.close(io);
    defer log.deinit(gpa);
    var digest_buf: [digest_len]u8 = undefined;
    try std.testing.expect(log.lookup(decodeWireName(lying_name, &digest_buf).?) == null); // no frame landed
}

fn persistReverseChainNoLeak(allocator: std.mem.Allocator) !void {
    const io = std.testing.io;
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    try d4SeedEmptyRepo(allocator, io, tmp.dir);

    // A small reverse-delta chain built on the (failing) allocator so injected
    // OOM is exercised in both the store build and the persist worklist.
    var ms = repostate.MemStore.init(allocator);
    defer ms.deinit();
    const s = ms.store();
    const links = 4;
    var bodies: [links][]u8 = undefined;
    var made: usize = 0;
    defer for (bodies[0..made]) |b| allocator.free(b);
    var names: [links][64]u8 = undefined;
    var nlen: [links]usize = undefined;
    for (0..links) |i| {
        bodies[i] = try std.fmt.allocPrint(allocator, "reverse chain oom: one shared body across revisions -- revision {d}", .{i});
        made += 1;
        nlen[i] = zit.artifact.artifactName(.sha3_256, bodies[i], &names[i]).len;
    }
    for (0..links) |i| {
        const rid = try s.put(names[i][0..nlen[i]], bodies[i]);
        if (i == links - 1) {
            const c = try zit.container.encode(allocator, bodies[i], .default);
            defer allocator.free(c);
            try s.retain(rid, .{ .full = c });
        } else {
            const dr = try zit.delta.create(allocator, bodies[i + 1], bodies[i]);
            defer allocator.free(dr);
            const c = try zit.container.encode(allocator, dr, .default);
            defer allocator.free(c);
            try s.retain(rid, .{ .delta = .{ .src = names[i + 1][0..nlen[i + 1]], .container = c, .target_size = bodies[i].len } });
        }
    }
    _ = try persistStore(allocator, io, tmp.dir, ms.store());
}

test "sync §7.2/§3.3: persistStore drains the reverse-delta worklist without leaking under allocation failure" {
    try std.testing.checkAllAllocationFailures(std.testing.allocator, persistReverseChainNoLeak, .{});
}

/// Drives `persistArtifact` over a FORWARD delta chain (each link a delta of the
/// immediately older, already-persisted lower-rid link) in rid order, so a test
/// owns the shared basis memo and can read its hit counter. Every basis precedes
/// its delta, so `allow_defer=false` never defers and no worklist is needed —
/// mirrors `persistStore`'s open/commit around a plain rid-order drive.
fn persistForwardChain(
    gpa: std.mem.Allocator,
    io: std.Io,
    dir: std.Io.Dir,
    ms: *repostate.MemStore,
    cache: *logstore.PlaintextCache,
) !void {
    var file = try dir.openFile(io, repo_filename, .{ .mode = .read_write });
    defer file.close(io);
    const st = try file.stat(io);
    var fs = storage.FileStorage.init(io, file, st.size);
    var log = try logstore.LogStore.open(gpa, fs.storageFile(), 0);
    defer log.deinit(gpa);
    const store = ms.store();
    const n = store.count();
    var rid: repostate.Rid = 1;
    while (rid <= n) : (rid += 1) {
        switch (try persistArtifact(gpa, ms.store(), &log, rid, false, cache)) {
            .persisted => {},
            .deferred => unreachable, // forward chain: no basis is ever absent
        }
    }
    try log.commit();
}

test "sync §7.2/§7.4: the basis memo reuses each basis (hits) and reconstructs a byte-identical store to the uncached path" {
    const gpa = std.testing.allocator;
    const io = std.testing.io;
    const links = 8;

    // A forward delta chain: link 0 is full, every later link is a delta that
    // reconstructs its body FROM the older body (its basis a lower, already-
    // persisted rid) — the case where the memo turns each basis reconstruction
    // into a lookup for the delta built on it. Store owns its copies (retain
    // dupes), but the bodies/names back the reconstructed-content assertions.
    var ms = repostate.MemStore.init(gpa);
    defer ms.deinit();
    const s = ms.store();
    var bodies: [links][]u8 = undefined;
    var made: usize = 0;
    defer for (bodies[0..made]) |b| gpa.free(b);
    var names: [links][64]u8 = undefined;
    var nlen: [links]usize = undefined;
    for (0..links) |i| {
        bodies[i] = try std.fmt.allocPrint(gpa, "forward chain shared prefix across every revision -- revision {d}", .{i});
        made += 1;
        nlen[i] = zit.artifact.artifactName(.sha3_256, bodies[i], &names[i]).len;
    }
    for (0..links) |i| {
        const rid = try s.put(names[i][0..nlen[i]], bodies[i]);
        if (i == 0) {
            const c = try zit.container.encode(gpa, bodies[i], .default);
            defer gpa.free(c);
            try s.retain(rid, .{ .full = c });
        } else {
            const dr = try zit.delta.create(gpa, bodies[i - 1], bodies[i]); // newer body FROM its older basis
            defer gpa.free(dr);
            const c = try zit.container.encode(gpa, dr, .default);
            defer gpa.free(c);
            try s.retain(rid, .{ .delta = .{ .src = names[i - 1][0..nlen[i - 1]], .container = c, .target_size = bodies[i].len } });
        }
    }

    // (A) a normal memo; (B) a zero-budget memo that can hold nothing — the exact
    // uncached reconstruction. Same store, two freshly-seeded repos.
    var tmp_a = std.testing.tmpDir(.{});
    defer tmp_a.cleanup();
    try d4SeedEmptyRepo(gpa, io, tmp_a.dir);
    var cache_a = logstore.PlaintextCache.init();
    defer cache_a.deinit(gpa);
    try persistForwardChain(gpa, io, tmp_a.dir, &ms, &cache_a);

    var tmp_b = std.testing.tmpDir(.{});
    defer tmp_b.cleanup();
    try d4SeedEmptyRepo(gpa, io, tmp_b.dir);
    var cache_b = logstore.PlaintextCache.init();
    cache_b.max_bytes = 0; // never memoises: forces a fresh reconstruction each time
    defer cache_b.deinit(gpa);
    try persistForwardChain(gpa, io, tmp_b.dir, &ms, &cache_b);

    // The memo reused every basis exactly once (one hit per delta link); the
    // zero-budget run reused nothing — the two ways the same bytes get built.
    try std.testing.expectEqual(@as(u64, links - 1), cache_a.hits);
    try std.testing.expectEqual(@as(u64, 0), cache_b.hits);

    // Byte-identical store: the memo changes only the work, never the bytes.
    const bytes_a = try tmp_a.dir.readFileAlloc(io, repo_filename, gpa, .limited(1 << 24));
    defer gpa.free(bytes_a);
    const bytes_b = try tmp_b.dir.readFileAlloc(io, repo_filename, gpa, .limited(1 << 24));
    defer gpa.free(bytes_b);
    try std.testing.expectEqualSlices(u8, bytes_b, bytes_a);

    // Every link reconstructs to its exact body (§7.7 chain correctness).
    var fs: storage.FileStorage = undefined;
    var log = try d4OpenLog(gpa, io, tmp_a.dir, &fs);
    defer fs.file.close(io);
    defer log.deinit(gpa);
    for (0..links) |i| {
        var d: [digest_len]u8 = undefined;
        const rid = log.lookup(decodeWireName(names[i][0..nlen[i]], &d).?).?;
        const got = (try log.content(gpa, rid)).?;
        defer gpa.free(got);
        try std.testing.expectEqualStrings(bodies[i], got);
    }
}

test "sync §9.1: repoProjectCode returns the stored code, else the zero placeholder" {
    const gpa = std.testing.allocator;
    const io = std.testing.io;
    const hash_kind: u8 = @intFromEnum(zit.artifact.defaultPolicy(false, false));

    const seed = struct {
        fn withCode(g: std.mem.Allocator, i: std.Io, dir: std.Io.Dir, hk: u8, code: []const u8) !void {
            var f = try dir.createFile(i, repo_filename, .{ .read = true, .exclusive = true });
            defer f.close(i);
            var fs = storage.FileStorage.init(i, f, 0);
            var log = try logstore.LogStore.open(g, fs.storageFile(), hk);
            defer log.deinit(g);
            try log.putConfig(.{ .hash_policy = hk, .default_branch = "trunk", .project_id = "", .project_code = code });
            try log.commit();
        }
    };

    // (a) a repo whose config stores a real 40-hex project code: it is returned
    // verbatim — the code the pull/push card announces (§9.1), never the zeros
    // fossil would refuse.
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    const code = "a1b2c3d4" ** 5; // 40 hex chars
    try seed.withCode(gpa, io, tmp.dir, hash_kind, code);
    var buf: [zit.sync.max_projectcode_len]u8 = undefined;
    const got = try repoProjectCode(gpa, io, tmp.dir, &buf);
    try std.testing.expectEqualStrings(code, got);
    try std.testing.expect(!std.mem.eql(u8, got, sync_projectcode));

    // (b) a repo predating the stored code (empty project_code): the placeholder
    // is the honest fallback (no adopted identity to announce yet).
    var tmp2 = std.testing.tmpDir(.{});
    defer tmp2.cleanup();
    try seed.withCode(gpa, io, tmp2.dir, hash_kind, "");
    var buf2: [zit.sync.max_projectcode_len]u8 = undefined;
    const got2 = try repoProjectCode(gpa, io, tmp2.dir, &buf2);
    try std.testing.expectEqualStrings(sync_projectcode, got2);
}

test "sync §9.1: the pull request card carries the adopted project code, not the placeholder" {
    const gpa = std.testing.allocator;

    // The pull client, constructed exactly as `pullRun` does, with a stored code.
    const code = "a1b2c3d4" ** 5; // a real (non-zero) 40-hex project code
    var client = zit.sync.Client{
        .state = zit.sync.initialState(.pull),
        .operation = .pull,
        .self_servercode = sync_servercode,
        .project_code = code,
        .version = .{ .version = sync_client_version },
    };
    defer client.deinit(gpa);

    var ms = repostate.MemStore.init(gpa);
    defer ms.deinit();
    var buf: std.ArrayListUnmanaged(u8) = .empty;
    defer buf.deinit(gpa);
    const request = try client.buildRequest(gpa, ms.store(), &buf);

    // §9.1 ABNF `pull-card = "pull" SP servercode SP projectcode`: the adopted
    // code MUST be the card's PROJECTCODE (a real peer refuses a mismatch).
    const expected = "pull " ++ sync_servercode ++ " " ++ ("a1b2c3d4" ** 5) ++ "\n";
    try std.testing.expect(std.mem.indexOf(u8, request, expected) != null);
    // And the all-zero placeholder is NOT the project code the card announces.
    const rejected = "pull " ++ sync_servercode ++ " " ++ sync_projectcode ++ "\n";
    try std.testing.expect(std.mem.indexOf(u8, request, rejected) == null);
}

test "serve §3.8: a per-request store-load on one serve handle sees a later append (no restart)" {
    const gpa = std.testing.allocator;
    const io = std.testing.io;
    const hash_kind: u8 = @intFromEnum(zit.artifact.defaultPolicy(false, false));

    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();

    const append = struct {
        fn artifactFrame(g: std.mem.Allocator, i: std.Io, dir: std.Io.Dir, hk: u8, digest: []const u8, bytes: []const u8) !void {
            var f = dir.openFile(i, repo_filename, .{ .mode = .read_write }) catch
                try dir.createFile(i, repo_filename, .{ .read = true });
            defer f.close(i);
            const st = try f.stat(i);
            var fs = storage.FileStorage.init(i, f, st.size);
            var log = try logstore.LogStore.open(g, fs.storageFile(), hk);
            defer log.deinit(g);
            _ = try log.put(digest, bytes);
            try log.commit();
        }
    };

    // Seed the repo holding exactly one content artifact (a 32-byte raw digest).
    const digest_a = "a" ** digest_len;
    try append.artifactFrame(gpa, io, tmp.dir, hash_kind, digest_a, "content a\n");

    // Open ONE read handle — the handle `serve` holds for its whole lifetime.
    var handle = try tmp.dir.openFile(io, repo_filename, .{});
    defer handle.close(io);

    // First per-request read: one content artifact.
    var index1 = try LogIndexStore.openFromFile(gpa, io, handle, (try handle.stat(io)).size);
    defer index1.destroy();
    const n1 = contentCount(index1.store());
    try std.testing.expectEqual(@as(usize, 1), n1);

    // A SEPARATE writer appends a second artifact while the handle stays open —
    // exactly a `zit commit` running alongside a live `zit serve`.
    const digest_b = "b" ** digest_len;
    try append.artifactFrame(gpa, io, tmp.dir, hash_kind, digest_b, "content b\n");

    // Second per-request read on the SAME handle sees the new artifact: proof the
    // server re-reads the current on-disk log per request (§3.8). A startup
    // snapshot would still report n1 — no restart is needed here.
    var index2 = try LogIndexStore.openFromFile(gpa, io, handle, (try handle.stat(io)).size);
    defer index2.destroy();
    try std.testing.expectEqual(n1 + 1, contentCount(index2.store()));
}

/// A test-only `storage.StorageFile` that wraps another and tallies the bytes
/// every `pread` actually returned — the instrument for "how many times does a
/// prime read the log?". Every other vtable method delegates verbatim, so a log
/// opened over it behaves identically while its read traffic is measured.
const CountingStorage = struct {
    inner: storage.StorageFile,
    read_bytes: usize = 0,

    fn storageFile(self: *CountingStorage) 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: *CountingStorage = @ptrCast(@alignCast(ctx));
        return self.inner.append(bytes);
    }
    fn preadImpl(ctx: *anyopaque, offset: u64, buf: []u8) storage.StorageError!usize {
        const self: *CountingStorage = @ptrCast(@alignCast(ctx));
        const n = try self.inner.pread(offset, buf);
        self.read_bytes += n;
        return n;
    }
    fn syncImpl(ctx: *anyopaque) storage.StorageError!void {
        const self: *CountingStorage = @ptrCast(@alignCast(ctx));
        return self.inner.sync();
    }
    fn sizeImpl(ctx: *anyopaque) storage.StorageError!u64 {
        const self: *CountingStorage = @ptrCast(@alignCast(ctx));
        return self.inner.size();
    }
    fn truncateImpl(ctx: *anyopaque, len: u64) storage.StorageError!void {
        const self: *CountingStorage = @ptrCast(@alignCast(ctx));
        return self.inner.truncate(len);
    }
};

/// Seed a `.zit` under `dir` holding a full+delta+manifest+config mix (via
/// `packSeedDeltaStore`) and return its committed byte size — the fixture the
/// serve-prime read-traffic and index-equivalence tests both build over.
fn seedServeRepo(gpa: std.mem.Allocator, io: std.Io, dir: std.Io.Dir) !u64 {
    const hash_kind: u8 = @intFromEnum(zit.artifact.defaultPolicy(false, false));
    var f = try dir.createFile(io, repo_filename, .{ .read = true });
    defer f.close(io);
    var fs = storage.FileStorage.init(io, f, 0);
    var log = try logstore.LogStore.open(gpa, fs.storageFile(), hash_kind);
    defer log.deinit(gpa);
    var tip: [64]u8 = undefined;
    try packSeedDeltaStore(gpa, &log, &tip);
    return log.size();
}

test "#14 serve prime reads the served log at most once (no second full-container copy)" {
    const gpa = std.testing.allocator;
    const io = std.testing.io;
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();

    const file_size = try seedServeRepo(gpa, io, tmp.dir);

    var handle = try tmp.dir.openFile(io, repo_filename, .{});
    defer handle.close(io);
    var fs = storage.FileStorage.init(io, handle, file_size);
    var counting = CountingStorage{ .inner = fs.storageFile() };

    // Build the serve base snapshot over the counting storage and measure how
    // many bytes priming read. `openReadOnly`'s scan already whole-frame-CRC'd
    // every byte once (#15's concern); an index prime adds NOTHING on top, so
    // total prime reads stay at roughly one pass over the log. The copy prime
    // (`loadContentStore`) read every container a SECOND time — this bound fails
    // for it, which is exactly the cost #14 removes.
    var index = try LogIndexStore.open(gpa, counting.storageFile());
    defer index.destroy();

    try std.testing.expect(counting.read_bytes <= file_size + file_size / 2);
}

test "#14 log-backed index answers count/hashOf/held/content/wire identically to the MemStore snapshot" {
    const gpa = std.testing.allocator;
    const io = std.testing.io;
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();

    const file_size = try seedServeRepo(gpa, io, tmp.dir);

    // Reference: the whole-container MemStore snapshot `loadContentStore` builds.
    var ref_handle = try tmp.dir.openFile(io, repo_filename, .{});
    defer ref_handle.close(io);
    var ref_fs = storage.FileStorage.init(io, ref_handle, file_size);
    var ref_log = try logstore.LogStore.openReadOnly(gpa, ref_fs.storageFile());
    defer ref_log.deinit(gpa);
    var ref = repostate.MemStore.init(gpa);
    defer ref.deinit();
    try loadContentStore(gpa, &ref_log, &ref);
    const rs = ref.store();

    // Subject: the log-backed index over the same file.
    var idx_handle = try tmp.dir.openFile(io, repo_filename, .{});
    defer idx_handle.close(io);
    var index = try LogIndexStore.openFromFile(gpa, io, idx_handle, file_size);
    defer index.destroy();
    const is = index.store();

    try std.testing.expectEqual(rs.count(), is.count());
    try std.testing.expect(rs.count() > 0);

    var rid: repostate.Rid = 1;
    while (rid <= rs.count()) : (rid += 1) {
        // Identity, held-ness, and lookup round-trip.
        try std.testing.expectEqualStrings(rs.hashOf(rid), is.hashOf(rid));
        try std.testing.expectEqual(rs.held(rid), is.held(rid));
        try std.testing.expectEqual(@as(?repostate.Rid, rid), is.lookup(rs.hashOf(rid)));

        // Plaintext byte-for-byte.
        const rc = rs.content(rid);
        const ic = is.content(rid);
        try std.testing.expectEqual(rc == null, ic == null);
        if (rc) |rb| try std.testing.expectEqualSlices(u8, rb, ic.?);

        // Wire form: container bytes, and the delta arm's DELTASRC + target size.
        const rw = rs.wire(rid);
        const iw = is.wire(rid);
        try std.testing.expectEqual(rw == null, iw == null);
        if (rw) |rwf| {
            try std.testing.expectEqualSlices(u8, rwf.container, iw.?.container);
            try std.testing.expectEqual(rwf.delta == null, iw.?.delta == null);
            if (rwf.delta) |rd| {
                try std.testing.expectEqualStrings(rd.src_hash, iw.?.delta.?.src_hash);
                try std.testing.expectEqual(rd.target_size, iw.?.delta.?.target_size);
            }
        }
    }
}

test "#14 appendix honours intern/put/retain for hashes the log does not carry" {
    const gpa = std.testing.allocator;
    const io = std.testing.io;
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();

    const file_size = try seedServeRepo(gpa, io, tmp.dir);
    var handle = try tmp.dir.openFile(io, repo_filename, .{});
    defer handle.close(io);
    var index = try LogIndexStore.openFromFile(gpa, io, handle, file_size);
    defer index.destroy();
    const s = index.store();

    // The serve base is contractually read-only — every write lands in the
    // per-request OverlayStore (ticket constraint 3), so these appendix arms never
    // fire in production. They exist so the store honours the whole ArtifactStore
    // interface instead of trapping on a write; exercise them directly here so that
    // completeness is not dead-untested surface.
    const persisted = s.count();
    try std.testing.expect(persisted > 0);

    // A hash the log cannot hold (all-0xAB digest, hex-named as the wire does).
    var d1: [digest_len]u8 = undefined;
    @memset(&d1, 0xAB);
    var h1_buf: [2 * digest_len]u8 = undefined;
    const h1 = digestToWireName(&d1, &h1_buf);
    try std.testing.expectEqual(@as(?repostate.Rid, null), s.lookup(h1));

    // intern mints an appendix phantom ABOVE the persisted rids: not held (no
    // content yet), count grows by one, identity round-trips, and it is idempotent.
    const r1 = try s.intern(h1);
    try std.testing.expect(r1 > persisted);
    try std.testing.expectEqual(r1, try s.intern(h1));
    try std.testing.expectEqual(@as(?repostate.Rid, r1), s.lookup(h1));
    try std.testing.expectEqualStrings(h1, s.hashOf(r1));
    try std.testing.expect(!s.held(r1));
    try std.testing.expectEqual(@as(?[]const u8, null), s.content(r1));
    try std.testing.expectEqual(persisted + 1, s.count());

    // put upgrades that phantom in place: held flips true, content reads back, and
    // the persisted count is unchanged (an off-log hash never touches the index).
    const body = "off-log body bytes";
    try std.testing.expectEqual(r1, try s.put(h1, body));
    try std.testing.expect(s.held(r1));
    try std.testing.expectEqualSlices(u8, body, s.content(r1).?);
    try std.testing.expectEqual(persisted + 1, s.count());

    // retain on a PERSISTED rid is the advisory no-op (sync §7.2 first-form-wins;
    // the log IS the retained form): it returns without error and ignores ingest.
    try s.retain(1, .{ .full = "ignored — a persisted rid keeps its log form" });

    // None of the appendix traffic left persisted-round memos resident.
    try std.testing.expectEqual(@as(usize, 0), index.residentBytes());
}

test "#14 index materialises only the container a round reads and releases it between rounds" {
    const gpa = std.testing.allocator;
    const io = std.testing.io;
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();

    const file_size = try seedServeRepo(gpa, io, tmp.dir);
    var handle = try tmp.dir.openFile(io, repo_filename, .{});
    defer handle.close(io);
    var index = try LogIndexStore.openFromFile(gpa, io, handle, file_size);
    defer index.destroy();
    const s = index.store();

    // Right after the prime the index holds ZERO container/plaintext bytes — the
    // whole point: it did not copy the store into RAM.
    try std.testing.expectEqual(@as(usize, 0), index.residentBytes());

    // Find one wire-bearing rid (a zlib/delta frame).
    var target: ?repostate.Rid = null;
    var rid: repostate.Rid = 1;
    while (rid <= s.count()) : (rid += 1) {
        if (s.wire(rid) != null) {
            target = rid;
            break;
        }
    }
    try std.testing.expect(target != null);

    // Reading it materialised only its container; drop the round's memos.
    try std.testing.expect(index.residentBytes() > 0);
    index.releaseTransient();
    try std.testing.expectEqual(@as(usize, 0), index.residentBytes());

    // A single round's residency is EXACTLY the one container it emitted — not
    // the whole store's containers (which the old snapshot held resident).
    const one = s.wire(target.?).?;
    try std.testing.expectEqual(one.container.len, index.residentBytes());
}

/// A test-only `storage.StorageFile` that reads faithfully until `fail_reads` is
/// flipped, then fails every `pread` — the instrument for a fault that strikes
/// AFTER `openReadOnly` has scanned and whole-frame-CRC'd the log, exactly the
/// post-scan bitrot / transient EIO the on-demand serve reads must not swallow
/// silently. Every other method delegates verbatim.
const FaultAfterScanStorage = struct {
    inner: storage.StorageFile,
    fail_reads: bool = false,

    fn storageFile(self: *FaultAfterScanStorage) 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: *FaultAfterScanStorage = @ptrCast(@alignCast(ctx));
        return self.inner.append(bytes);
    }
    fn preadImpl(ctx: *anyopaque, offset: u64, buf: []u8) storage.StorageError!usize {
        const self: *FaultAfterScanStorage = @ptrCast(@alignCast(ctx));
        if (self.fail_reads) return error.Io;
        return self.inner.pread(offset, buf);
    }
    fn syncImpl(ctx: *anyopaque) storage.StorageError!void {
        const self: *FaultAfterScanStorage = @ptrCast(@alignCast(ctx));
        return self.inner.sync();
    }
    fn sizeImpl(ctx: *anyopaque) storage.StorageError!u64 {
        const self: *FaultAfterScanStorage = @ptrCast(@alignCast(ctx));
        return self.inner.size();
    }
    fn truncateImpl(ctx: *anyopaque, len: u64) storage.StorageError!void {
        const self: *FaultAfterScanStorage = @ptrCast(@alignCast(ctx));
        return self.inner.truncate(len);
    }
};

test "#14 a post-scan read fault on a held rid is recorded, not silently swallowed" {
    const gpa = std.testing.allocator;
    const io = std.testing.io;
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();

    const file_size = try seedServeRepo(gpa, io, tmp.dir);
    var handle = try tmp.dir.openFile(io, repo_filename, .{});
    defer handle.close(io);
    var fs = storage.FileStorage.init(io, handle, file_size);
    var faulting = FaultAfterScanStorage{ .inner = fs.storageFile() };

    // Open succeeds: the scan CRCs the whole log BEFORE any fault is armed.
    var index = try LogIndexStore.open(gpa, faulting.storageFile());
    defer index.destroy();
    const s = index.store();

    // From here every on-demand container/plaintext read faults (post-scan EIO).
    faulting.fail_reads = true;

    // Walk the store as the §8.2 clone pager does: a held rid whose wire() AND
    // content() both return null is one the pager silently skips while still
    // reporting the sequence complete — a clone finishing MISSING an artifact.
    var silently_omitted = false;
    var rid: repostate.Rid = 1;
    while (rid <= s.count()) : (rid += 1) {
        if (!s.held(rid)) continue;
        if (s.wire(rid) == null and s.content(rid) == null) silently_omitted = true;
    }

    // The fault genuinely makes a held rid unemittable (the bug's precondition)...
    try std.testing.expect(silently_omitted);
    // ...and the store RECORDED it, so the serve layer can convert the round to an
    // error instead of completing a broken clone. A copy-based prime could not
    // reach this state: a read fault surfaced at prime, not silently at serve.
    try std.testing.expect(index.read_fault);
}

test "#14 serveAnswer converts a recorded read fault into a storage error (no silent short clone)" {
    const gpa = std.testing.allocator;
    const io = std.testing.io;
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();

    var hex_buf: [2 * digest_len]u8 = undefined;
    try i9SeedRepo(gpa, io, tmp.dir, 4, &hex_buf);

    const server = zit.sync.Server{
        .version = serve_mod.server_version,
        .self_servercode = serve_mod.server_servercode,
        .project_code = serve_mod.server_projectcode,
    };
    var file = try tmp.dir.openFile(io, repo_filename, .{});
    defer file.close(io);
    var ctx = ServeCtx{ .io = io, .root_dir = tmp.dir, .file = file, .server = &server };
    defer ctx.dropSnapshot();

    const clone_plain = try std.fmt.allocPrint(gpa, "pragma client-version 20000\nclone 3 1\n", .{});
    defer gpa.free(clone_plain);
    const clone_req = try zit.container.encode(gpa, clone_plain, .default);
    defer gpa.free(clone_req);

    // A healthy round builds the cached snapshot and completes normally.
    const ok = try serveAnswer(&ctx, gpa, clone_req);
    gpa.free(ok.body);

    // Simulate a post-scan read fault having struck THIS round — the test above
    // proves the store records exactly this. serveAnswer MUST refuse to hand back
    // a reply built while an artifact read faulted: otherwise the §8.2 pager's
    // silent skip ships a clone missing an artifact and reports it complete.
    ctx.snapshot.?.read_fault = true;
    try std.testing.expectError(error.ServerStorage, serveAnswer(&ctx, gpa, clone_req));

    // The flag is cleared with the round's memos, so an immediately retried round
    // over the now-healthy snapshot succeeds again (a transient fault self-heals).
    const recovered = try serveAnswer(&ctx, gpa, clone_req);
    gpa.free(recovered.body);
}

/// Build a full-form 3-check-in chain over an evolving file into `log`, plus a
/// live config frame; returns the tip check-in's 64-hex name. Every blob and
/// manifest lands as a FULL frame — pack's re-delta target.
fn packSeedFullStore(gpa: std.mem.Allocator, log: *logstore.LogStore, tip_out: *[64]u8) !void {
    try log.putConfig(.{ .hash_policy = 2, .default_branch = "trunk", .project_id = "pid", .project_code = "code-abc" });
    var root_name: [64]u8 = undefined;
    var mid_name: [64]u8 = undefined;
    const v1 = try packEvolvingBlob(gpa, 1);
    defer gpa.free(v1);
    const b1 = try packSeedBlob(log, v1);
    try packSeedCheckin(gpa, log, "2020-01-01T00:00:00", null, "f.txt", &b1, &root_name);
    const v2 = try packEvolvingBlob(gpa, 2);
    defer gpa.free(v2);
    const b2 = try packSeedBlob(log, v2);
    try packSeedCheckin(gpa, log, "2020-01-02T00:00:00", &root_name, "f.txt", &b2, &mid_name);
    const v3 = try packEvolvingBlob(gpa, 3);
    defer gpa.free(v3);
    const b3 = try packSeedBlob(log, v3);
    try packSeedCheckin(gpa, log, "2020-01-03T00:00:00", &mid_name, "f.txt", &b3, tip_out);
    try log.commit();
}

/// Store `content` under its sha3-256 name, delta-encoded against `base` when
/// one is given (the evolving blobs delta to a fraction, so the frame lands as a
/// `.delta`), and return its rid. Models a fossil-clone store that RETAINED
/// compact wire deltas rather than full frames.
fn packSeedDeltaBlob(log: *logstore.LogStore, content: []const u8, base: ?logstore.Rid) !logstore.Rid {
    var digest: [digest_len]u8 = undefined;
    std.crypto.hash.sha3.Sha3_256.hash(content, &digest, .{});
    return log.putArtifact(&digest, content, base);
}

/// Build a clone-shaped store whose successive blobs are RETAINED as `.delta`
/// frames (the compact wire form), under DISTINCT filenames so pack's
/// same-path-blob lineage hint finds NO base for them — exactly the case where
/// the old pack reconstructed every delta to a full frame and inflated the
/// store. `.zit` therefore holds real delta frames whose bases pack cannot
/// re-derive, so a correct pack must preserve them verbatim. Returns the tip
/// check-in's 64-hex name.
fn packSeedDeltaStore(gpa: std.mem.Allocator, log: *logstore.LogStore, tip_out: *[64]u8) !void {
    try log.putConfig(.{ .hash_policy = 2, .default_branch = "trunk", .project_id = "pid", .project_code = "code-abc" });
    var root_name: [64]u8 = undefined;
    var mid_name: [64]u8 = undefined;
    const v1 = try packEvolvingBlob(gpa, 1);
    defer gpa.free(v1);
    const r1 = try packSeedDeltaBlob(log, v1, null); // first version: full frame (base of the chain)
    var b1: [64]u8 = undefined;
    _ = zit.artifact.artifactName(.sha3_256, v1, &b1);
    try packSeedCheckin(gpa, log, "2020-01-01T00:00:00", null, "f1.txt", &b1, &root_name);
    const v2 = try packEvolvingBlob(gpa, 2);
    defer gpa.free(v2);
    const r2 = try packSeedDeltaBlob(log, v2, r1); // delta of v1
    var b2: [64]u8 = undefined;
    _ = zit.artifact.artifactName(.sha3_256, v2, &b2);
    try packSeedCheckin(gpa, log, "2020-01-02T00:00:00", &root_name, "f2.txt", &b2, &mid_name);
    const v3 = try packEvolvingBlob(gpa, 3);
    defer gpa.free(v3);
    _ = try packSeedDeltaBlob(log, v3, r2); // delta of v2
    var b3: [64]u8 = undefined;
    _ = zit.artifact.artifactName(.sha3_256, v3, &b3);
    try packSeedCheckin(gpa, log, "2020-01-03T00:00:00", &mid_name, "f3.txt", &b3, tip_out);
    try log.commit();
}

/// Seed a store holding a wire delta that pack CANNOT beat by re-deltification,
/// so ONLY the §6.1 monotonic-shrink guard keeps the store from regressing. `B`
/// is a one-line edit of `A`, stored as a COMPACT zit delta against `A` — the
/// smallest form pack could ever reproduce. But `max_base_candidates` unrelated
/// filler blobs sit between `A` and `B`, so pack's bounded backward candidate
/// window (§5.2) saturates on the fillers and never reaches `A`; with no lineage
/// hint either (bare blobs, no F-card path), pack's best re-encoding of `B` is a
/// full zlib frame STRICTLY LARGER than the stored A-based delta. The pre-guard
/// pack replaced the delta with that larger full frame (the +33% clone
/// regression in miniature); the guard keeps the existing smaller frame. Fills
/// A's and B's 64-hex names.
fn packSeedUnbeatableWireDelta(
    gpa: std.mem.Allocator,
    log: *logstore.LogStore,
    a_name: *[64]u8,
    b_name: *[64]u8,
) !void {
    try log.putConfig(.{ .hash_policy = 2, .default_branch = "trunk", .project_id = "pid", .project_code = "code-abc" });
    // A: B's ideal base, stored full (zlib).
    const a = try packEvolvingBlob(gpa, 1);
    defer gpa.free(a);
    var a_digest: [digest_len]u8 = undefined;
    std.crypto.hash.sha3.Sha3_256.hash(a, &a_digest, .{});
    const a_rid = try log.putZlib(&a_digest, a);
    _ = zit.artifact.artifactName(.sha3_256, a, a_name);
    // `max_base_candidates` unrelated fillers between A and B: they saturate B's
    // backward candidate window so A is out of reach when pack re-encodes B.
    var f: usize = 0;
    while (f < logstore.max_base_candidates) : (f += 1) {
        var fbuf: std.ArrayList(u8) = .empty;
        defer fbuf.deinit(gpa);
        var i: usize = 0;
        while (i < 200) : (i += 1) try fbuf.print(gpa, "filler {d} unrelated row {d}\n", .{ f, i });
        var fdig: [digest_len]u8 = undefined;
        std.crypto.hash.sha3.Sha3_256.hash(fbuf.items, &fdig, .{});
        _ = try log.putZlib(&fdig, fbuf.items);
    }
    // B: a one-line edit of A, stored as a COMPACT zit delta against A (the
    // smallest form pack could reproduce — and only against A, which pack's
    // window cannot reach here).
    const b = try packEvolvingBlob(gpa, 2);
    defer gpa.free(b);
    var b_digest: [digest_len]u8 = undefined;
    std.crypto.hash.sha3.Sha3_256.hash(b, &b_digest, .{});
    _ = zit.artifact.artifactName(.sha3_256, b, b_name);
    const raw = try zit.delta.create(gpa, a, b);
    defer gpa.free(raw);
    const container = try zit.container.encode(gpa, raw, .default);
    defer gpa.free(container);
    _ = try log.putDeltaContainer(&b_digest, a_rid, container);
    try log.commit();
}

/// Assert the §6.1 per-frame monotonicity: NO artifact's packed frame is larger
/// on disk than its source frame. Pack keeps the smaller of {existing frame, new
/// encoding} for every artifact, so a repack is non-worsening frame by frame.
/// This holds universally over THESE fixtures because their chains stay below the
/// read cap; the one path that can enlarge a frame — `copySrcFrame`'s §3.3
/// deep-chain fallback re-storing a kept delta as full zlib — needs a base at
/// `max_delta_chain` depth, which the seeds never build.
fn packExpectNoFrameLarger(gpa: std.mem.Allocator, src: *logstore.LogStore, dst: *logstore.LogStore) !void {
    const src_rids = try src.ridsInPersistedOrder(gpa);
    defer gpa.free(src_rids);
    for (src_rids) |sr| {
        const dr = dst.lookup(src.hashOf(sr)) orelse return error.PackedArtifactMissing;
        try std.testing.expect(dst.frameSpanOf(dr) <= src.frameSpanOf(sr));
    }
}

/// Count the `.delta` frames among a store's persisted artifacts.
fn packDeltaFrameCount(gpa: std.mem.Allocator, log: *logstore.LogStore) !usize {
    const rids = try log.ridsInPersistedOrder(gpa);
    defer gpa.free(rids);
    var n: usize = 0;
    for (rids) |r| {
        if (log.kindOf(r) == .delta) n += 1;
    }
    return n;
}

/// Assert that every artifact in `src` reconstructs byte-identically in `dst`
/// and the two artifact SETS are equal (no drop, no addition).
fn packExpectContentEqual(gpa: std.mem.Allocator, src: *logstore.LogStore, dst: *logstore.LogStore) !void {
    const src_rids = try src.ridsInPersistedOrder(gpa);
    defer gpa.free(src_rids);
    const dst_rids = try dst.ridsInPersistedOrder(gpa);
    defer gpa.free(dst_rids);
    try std.testing.expectEqual(src_rids.len, dst_rids.len);
    for (src_rids) |sr| {
        const digest = src.hashOf(sr);
        const dr = dst.lookup(digest) orelse return error.PackedArtifactMissing;
        const a = (try src.content(gpa, sr)).?;
        defer gpa.free(a);
        const b = (try dst.content(gpa, dr)).?;
        defer gpa.free(b);
        try std.testing.expectEqualSlices(u8, a, b);
    }
}

test "pack: the packed store's artifact set + every reconstructed content byte-equal the original (config preserved)" {
    const gpa = std.testing.allocator;
    const io = std.testing.io;
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();

    var tip: [64]u8 = undefined;
    {
        var wfs: storage.FileStorage = undefined;
        var wlog = try packOpenWritable(gpa, io, tmp.dir, &wfs);
        defer wfs.file.close(io);
        defer wlog.deinit(gpa);
        try packSeedFullStore(gpa, &wlog, &tip);
    }

    var sfs: storage.FileStorage = undefined;
    var src = try packOpenReadOnly(gpa, io, tmp.dir, &sfs);
    defer sfs.file.close(io);
    defer src.deinit(gpa);

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

    const stats = try packLog(gpa, &src, &dst);

    // No drop: the packed content-frame count equals the source's, and every
    // artifact reconstructs byte-for-byte.
    try std.testing.expectEqual(src.entries.items.len, stats.artifacts);
    try packExpectContentEqual(gpa, &src, &dst);

    // The live config survives as exactly one frame with identical fields.
    var soc = (try src.readConfig(gpa)).?;
    defer soc.deinit(gpa);
    var doc = (try dst.readConfig(gpa)).?;
    defer doc.deinit(gpa);
    try std.testing.expectEqual(soc.config.hash_policy, doc.config.hash_policy);
    try std.testing.expectEqualStrings(soc.config.default_branch, doc.config.default_branch);
    try std.testing.expectEqualStrings(soc.config.project_id, doc.config.project_id);
    try std.testing.expectEqualStrings(soc.config.project_code, doc.config.project_code);
}

test "pack re-deltas full-form frames (delta count rises, .zit shrinks) with byte-identical content" {
    const gpa = std.testing.allocator;
    const io = std.testing.io;
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();

    var tip: [64]u8 = undefined;
    {
        var wfs: storage.FileStorage = undefined;
        var wlog = try packOpenWritable(gpa, io, tmp.dir, &wfs);
        defer wfs.file.close(io);
        defer wlog.deinit(gpa);
        try packSeedFullStore(gpa, &wlog, &tip);
    }

    var sfs: storage.FileStorage = undefined;
    var src = try packOpenReadOnly(gpa, io, tmp.dir, &sfs);
    defer sfs.file.close(io);
    defer src.deinit(gpa);

    // The source is all full-form: not one delta frame.
    {
        const rids = try src.ridsInPersistedOrder(gpa);
        defer gpa.free(rids);
        for (rids) |r| try std.testing.expect(src.kindOf(r) != .delta);
    }

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

    const stats = try packLog(gpa, &src, &dst);

    // The two child blobs (v2 against v1, v3 against v2) re-delta against the
    // same path's parent blob, so the delta count rises and the store shrinks —
    // with content preserved exactly.
    try std.testing.expect(stats.deltas >= 2);
    try std.testing.expect(stats.bytes_after < stats.bytes_before);
    try packExpectContentEqual(gpa, &src, &dst);
}

test "pack preserves wire deltas — a store with delta frames packs to <= source size (regression for the 8.65x inflation bug)" {
    const gpa = std.testing.allocator;
    const io = std.testing.io;
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();

    var tip: [64]u8 = undefined;
    {
        var wfs: storage.FileStorage = undefined;
        var wlog = try packOpenWritable(gpa, io, tmp.dir, &wfs);
        defer wfs.file.close(io);
        defer wlog.deinit(gpa);
        try packSeedDeltaStore(gpa, &wlog, &tip);
    }

    var sfs: storage.FileStorage = undefined;
    var src = try packOpenReadOnly(gpa, io, tmp.dir, &sfs);
    defer sfs.file.close(io);
    defer src.deinit(gpa);

    // The source genuinely holds retained wire deltas (the compact form pack
    // must not discard) whose bases pack's lineage hint cannot re-derive.
    const src_deltas = try packDeltaFrameCount(gpa, &src);
    try std.testing.expect(src_deltas >= 2);

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

    const stats = try packLog(gpa, &src, &dst);

    // MONOTONIC: the packed store is never larger than the source. The old pack
    // reconstructed each delta to a full zlib frame and inflated the store; a
    // correct pack preserves the compact deltas verbatim.
    try std.testing.expect(stats.bytes_after <= stats.bytes_before);
    // Every source delta survives AS a delta frame — none reconstructed to full
    // (the optional re-delta may add further deltas, so the total only grows).
    {
        const rids = try src.ridsInPersistedOrder(gpa);
        defer gpa.free(rids);
        for (rids) |sr| {
            if (src.kindOf(sr) != .delta) continue;
            const dr = dst.lookup(src.hashOf(sr)).?;
            try std.testing.expectEqual(logstore.FrameKind.delta, dst.kindOf(dr));
        }
    }
    try std.testing.expect(try packDeltaFrameCount(gpa, &dst) >= src_deltas);
    // No drop, every reconstruction byte-exact.
    try packExpectContentEqual(gpa, &src, &dst);
}

test "pack copies a delta frame verbatim re-linked to its base's new rid, content byte-identical" {
    const gpa = std.testing.allocator;
    const io = std.testing.io;
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();

    var tip: [64]u8 = undefined;
    {
        var wfs: storage.FileStorage = undefined;
        var wlog = try packOpenWritable(gpa, io, tmp.dir, &wfs);
        defer wfs.file.close(io);
        defer wlog.deinit(gpa);
        try packSeedDeltaStore(gpa, &wlog, &tip);
    }

    var sfs: storage.FileStorage = undefined;
    var src = try packOpenReadOnly(gpa, io, tmp.dir, &sfs);
    defer sfs.file.close(io);
    defer src.deinit(gpa);

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

    _ = try packLog(gpa, &src, &dst);

    // For every delta frame in the source, the packed store holds it as a delta
    // whose stored container is byte-identical (verbatim re-serve) and whose base
    // names the SAME artifact (re-linked to the base's new rid). Content matches.
    const rids = try src.ridsInPersistedOrder(gpa);
    defer gpa.free(rids);
    var checked: usize = 0;
    for (rids) |sr| {
        if (src.kindOf(sr) != .delta) continue;
        const digest = src.hashOf(sr);
        const dr = dst.lookup(digest).?;
        try std.testing.expectEqual(logstore.FrameKind.delta, dst.kindOf(dr));

        var sw = (try src.wire(gpa, sr)).?;
        defer sw.deinit(gpa);
        var dw = (try dst.wire(gpa, dr)).?;
        defer dw.deinit(gpa);
        // Verbatim container bytes and the same base hash (re-link preserves the
        // logical base while its rid/offset differs in the packed generation).
        try std.testing.expectEqualSlices(u8, sw.delta.container, dw.delta.container);
        try std.testing.expectEqualSlices(u8, sw.delta.baseHash(), dw.delta.baseHash());

        const a = (try src.content(gpa, sr)).?;
        defer gpa.free(a);
        const b = (try dst.content(gpa, dr)).?;
        defer gpa.free(b);
        try std.testing.expectEqualSlices(u8, a, b);
        checked += 1;
    }
    try std.testing.expect(checked >= 2);
}

/// The rid naming the 64-hex `name` in `log`, or null when absent (test-only
/// name→digest→rid convenience over `lookup`).
fn packRidByName(log: *logstore.LogStore, name: *const [64]u8) ?logstore.Rid {
    const d = hexToDigest(name) orelse return null;
    return log.lookup(&d);
}

/// Seed a store modeling a peer-supplied (verbatim wire) delta whose sender
/// chose a POOR base. `A` (a large file) and `C` (an unrelated large blob) are
/// stored full; `B` (a one-line edit of `A`) is stored as a hand-crafted §4.1
/// delta container against `C` — a large, suboptimal delta the sender picked.
/// `A` precedes `B` chronologically and is B's ideal base, so pack's best-of-N
/// re-deltification (§6.1) must re-base `B` onto `A` rather than trust the
/// sender's `C`. Fills the 64-hex names of A/B/C.
fn packSeedSuboptimalWireDelta(
    gpa: std.mem.Allocator,
    log: *logstore.LogStore,
    a_name: *[64]u8,
    b_name: *[64]u8,
    c_name: *[64]u8,
) !void {
    try log.putConfig(.{ .hash_policy = 2, .default_branch = "trunk", .project_id = "pid", .project_code = "code-abc" });
    // A: the ideal base for B (near-identical predecessor), stored full.
    const a = try packEvolvingBlob(gpa, 1);
    defer gpa.free(a);
    a_name.* = try packSeedBlob(log, a);
    // C: an unrelated large blob (a poor base for B), stored full.
    var cbuf: std.ArrayList(u8) = .empty;
    defer cbuf.deinit(gpa);
    var i: usize = 0;
    while (i < 200) : (i += 1) try cbuf.print(gpa, "unrelated content row {d}\n", .{i});
    const c = cbuf.items;
    c_name.* = try packSeedBlob(log, c);
    const c_rid = packRidByName(log, c_name).?;
    // B: a one-line edit of A. delta(A,B) is tiny; delta(C,B) is large.
    const b = try packEvolvingBlob(gpa, 2);
    defer gpa.free(b);
    _ = zit.artifact.artifactName(.sha3_256, b, b_name);
    var b_digest: [digest_len]u8 = undefined;
    std.crypto.hash.sha3.Sha3_256.hash(b, &b_digest, .{});
    // Hand-craft the verbatim wire delta against the POOR base C — models a peer
    // that chose C as base, retained at rest by `putDeltaContainer`.
    const raw = try zit.delta.create(gpa, c, b);
    defer gpa.free(raw);
    const container = try zit.container.encode(gpa, raw, .default);
    defer gpa.free(container);
    _ = try log.putDeltaContainer(&b_digest, c_rid, container);
    try log.commit();
}

test "pack §6.1: a verbatim wire delta is re-deltified against the size-minimizing base, not the sender's" {
    const gpa = std.testing.allocator;
    const io = std.testing.io;
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();

    var a_name: [64]u8 = undefined;
    var b_name: [64]u8 = undefined;
    var c_name: [64]u8 = undefined;
    {
        var wfs: storage.FileStorage = undefined;
        var wlog = try packOpenWritable(gpa, io, tmp.dir, &wfs);
        defer wfs.file.close(io);
        defer wlog.deinit(gpa);
        try packSeedSuboptimalWireDelta(gpa, &wlog, &a_name, &b_name, &c_name);
    }

    var sfs: storage.FileStorage = undefined;
    var src = try packOpenReadOnly(gpa, io, tmp.dir, &sfs);
    defer sfs.file.close(io);
    defer src.deinit(gpa);

    // Precondition: the source really holds B as a delta whose base is C (the
    // sender's suboptimal choice) — the exact frame the old pack copied verbatim.
    {
        const sb = packRidByName(&src, &b_name).?;
        try std.testing.expectEqual(logstore.FrameKind.delta, src.kindOf(sb));
        var sw = (try src.wire(gpa, sb)).?;
        defer sw.deinit(gpa);
        var cbin: [digest_len]u8 = undefined;
        _ = try std.fmt.hexToBytes(&cbin, &c_name);
        try std.testing.expectEqualSlices(u8, &cbin, sw.delta.baseHash());
    }

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

    const stats = try packLog(gpa, &src, &dst);

    // After pack, B is still a delta, but its base is now A (the size-minimizing
    // candidate) — the sender's base C is no longer trusted at rest (§6.1/§3).
    const db = packRidByName(&dst, &b_name).?;
    try std.testing.expectEqual(logstore.FrameKind.delta, dst.kindOf(db));
    var dw = (try dst.wire(gpa, db)).?;
    defer dw.deinit(gpa);
    var abin: [digest_len]u8 = undefined;
    _ = try std.fmt.hexToBytes(&abin, &a_name);
    try std.testing.expectEqualSlices(u8, &abin, dw.delta.baseHash());

    // Every artifact still reconstructs byte-for-byte and the store shrank (a
    // tiny A-based delta replaced the large C-based one).
    try packExpectContentEqual(gpa, &src, &dst);
    try std.testing.expect(stats.bytes_after < stats.bytes_before);
    try std.testing.expectEqual(dst.size(), stats.bytes_after);
    try std.testing.expectEqual(src.size(), stats.bytes_before);
}

/// Seed the §6.2 same-path scenario: two versions of ONE tracked file whose ideal
/// base (the earlier version) sits OUTSIDE pack's bounded chronological window.
/// `v1` of "tracked.txt" is stored full, then `max_base_candidates` unrelated
/// filler blobs saturate the backward candidate window, then `v2` (a one-line
/// edit of `v1`) is stored full. With ONLY the chronological window (step §6.1)
/// pack cannot reach `v1` from `v2` — the fillers fill every slot — so `v2` stays
/// a full frame and the store does not shrink. The mlink-analog same-path walk
/// (§6.2) reads both check-in manifests, learns tracked.txt: [v1, v2], and offers
/// `v1` as `v2`'s priority candidate, so pack re-deltifies `v2` onto `v1`. Both
/// check-ins are FULL frames (pack's re-delta target); fills v1/v2's 64-hex names.
fn packSeedSamePathOutOfWindow(
    gpa: std.mem.Allocator,
    log: *logstore.LogStore,
    v1_name: *[64]u8,
    v2_name: *[64]u8,
) !void {
    try log.putConfig(.{ .hash_policy = 2, .default_branch = "trunk", .project_id = "pid", .project_code = "code-abc" });
    // v1 of tracked.txt (full), and the root check-in that names it.
    const v1 = try packEvolvingBlob(gpa, 1);
    defer gpa.free(v1);
    v1_name.* = try packSeedBlob(log, v1);
    var root_name: [64]u8 = undefined;
    try packSeedCheckin(gpa, log, "2020-01-01T00:00:00", null, "tracked.txt", v1_name, &root_name);
    // `max_base_candidates` unrelated fillers between v1 and v2: they saturate
    // v2's bounded backward window so v1 is out of chronological reach.
    var f: usize = 0;
    while (f < logstore.max_base_candidates) : (f += 1) {
        var fbuf: std.ArrayList(u8) = .empty;
        defer fbuf.deinit(gpa);
        var i: usize = 0;
        while (i < 200) : (i += 1) try fbuf.print(gpa, "samepath filler {d} unrelated row {d}\n", .{ f, i });
        _ = try packSeedBlob(log, fbuf.items);
    }
    // v2 of tracked.txt (a one-line edit of v1), stored FULL, and its check-in.
    const v2 = try packEvolvingBlob(gpa, 2);
    defer gpa.free(v2);
    v2_name.* = try packSeedBlob(log, v2);
    var tip_name: [64]u8 = undefined;
    try packSeedCheckin(gpa, log, "2020-01-02T00:00:00", &root_name, "tracked.txt", v2_name, &tip_name);
    try log.commit();
}

test "pack §6.2: a same-path predecessor beyond the chronological window becomes v2's base (store shrinks)" {
    const gpa = std.testing.allocator;
    const io = std.testing.io;
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();

    var v1_name: [64]u8 = undefined;
    var v2_name: [64]u8 = undefined;
    {
        var wfs: storage.FileStorage = undefined;
        var wlog = try packOpenWritable(gpa, io, tmp.dir, &wfs);
        defer wfs.file.close(io);
        defer wlog.deinit(gpa);
        try packSeedSamePathOutOfWindow(gpa, &wlog, &v1_name, &v2_name);
    }

    var sfs: storage.FileStorage = undefined;
    var src = try packOpenReadOnly(gpa, io, tmp.dir, &sfs);
    defer sfs.file.close(io);
    defer src.deinit(gpa);

    // Precondition: the source holds v2 as a NON-delta (full) frame — the case
    // step §6.1's chronological window alone cannot improve (v1 is out of reach).
    {
        const sv2 = packRidByName(&src, &v2_name).?;
        try std.testing.expect(src.kindOf(sv2) != .delta);
    }

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

    const stats = try packLog(gpa, &src, &dst);

    // After pack, v2 is a delta whose base is the same-path predecessor v1 — the
    // §6.2 mlink analog reached a base the chronological window could not.
    const dv2 = packRidByName(&dst, &v2_name).?;
    try std.testing.expectEqual(logstore.FrameKind.delta, dst.kindOf(dv2));
    var dw = (try dst.wire(gpa, dv2)).?;
    defer dw.deinit(gpa);
    var v1bin: [digest_len]u8 = undefined;
    _ = try std.fmt.hexToBytes(&v1bin, &v1_name);
    try std.testing.expectEqualSlices(u8, &v1bin, dw.delta.baseHash());

    // The store shrank (a tiny v1-based delta replaced the full v2 frame), every
    // artifact still reconstructs byte-for-byte, and no frame grew.
    try std.testing.expect(stats.bytes_after < stats.bytes_before);
    try packExpectContentEqual(gpa, &src, &dst);
    try packExpectNoFrameLarger(gpa, &src, &dst);
}

/// Seed the §6.2 same-path REVERT scenario: one tracked file whose content goes
/// A→B→A across three check-ins — the third restores a byte-identical earlier
/// blob. Because the store is content-addressed, the reverted version resolves to
/// the SAME blob rid as v1, so the pack manifest walk records that rid twice,
/// non-consecutively, for one path. Exercises `deriveCandidateHints`'s
/// distinct-rid compaction: a blob must never be offered as its own same-path
/// predecessor. Fills v1/v2's 64-hex names.
fn packSeedSamePathRevert(
    gpa: std.mem.Allocator,
    log: *logstore.LogStore,
    v1_name: *[64]u8,
    v2_name: *[64]u8,
) !void {
    try log.putConfig(.{ .hash_policy = 2, .default_branch = "trunk", .project_id = "pid", .project_code = "code-abc" });
    const v1 = try packEvolvingBlob(gpa, 1);
    defer gpa.free(v1);
    const v2 = try packEvolvingBlob(gpa, 2);
    defer gpa.free(v2);
    // v1 of tracked.txt (full) + its root check-in.
    v1_name.* = try packSeedBlob(log, v1);
    var c1: [64]u8 = undefined;
    try packSeedCheckin(gpa, log, "2020-01-01T00:00:00", null, "tracked.txt", v1_name, &c1);
    // v2 (a one-line edit) + its check-in.
    v2_name.* = try packSeedBlob(log, v2);
    var c2: [64]u8 = undefined;
    try packSeedCheckin(gpa, log, "2020-01-02T00:00:00", &c1, "tracked.txt", v2_name, &c2);
    // Revert: tracked.txt returns to v1's byte-identical content. The blob is
    // already stored (content-addressed), so this check-in re-references v1's rid
    // — the non-consecutive duplicate that recordPathBlob's consecutive-only dedup
    // lets through.
    var c3: [64]u8 = undefined;
    try packSeedCheckin(gpa, log, "2020-01-03T00:00:00", &c2, "tracked.txt", v1_name, &c3);
    try log.commit();
}

test "pack §6.2: a same-path revert (A→B→A) re-references one blob rid twice and packs cleanly (no self-predecessor panic)" {
    const gpa = std.testing.allocator;
    const io = std.testing.io;
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();

    var v1_name: [64]u8 = undefined;
    var v2_name: [64]u8 = undefined;
    {
        var wfs: storage.FileStorage = undefined;
        var wlog = try packOpenWritable(gpa, io, tmp.dir, &wfs);
        defer wfs.file.close(io);
        defer wlog.deinit(gpa);
        try packSeedSamePathRevert(gpa, &wlog, &v1_name, &v2_name);
    }

    var sfs: storage.FileStorage = undefined;
    var src = try packOpenReadOnly(gpa, io, tmp.dir, &sfs);
    defer sfs.file.close(io);
    defer src.deinit(gpa);

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

    // Before the distinct-rid compaction in deriveCandidateHints this panicked on
    // a self-predecessor (v1 sorted adjacent to its own duplicate ⇒ equal offsets
    // fail the strict offset-below assert). Now the pack completes and every
    // artifact reconstructs byte-for-byte with no frame enlarged.
    const stats = try packLog(gpa, &src, &dst);
    try packExpectContentEqual(gpa, &src, &dst);
    try packExpectNoFrameLarger(gpa, &src, &dst);
    try std.testing.expectEqual(dst.size(), stats.bytes_after);
    try std.testing.expectEqual(src.size(), stats.bytes_before);
}

test "pack §6.1 monotonic guard: a wire delta pack cannot beat is KEPT, not replaced by a larger frame (no regression)" {
    const gpa = std.testing.allocator;
    const io = std.testing.io;
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();

    var a_name: [64]u8 = undefined;
    var b_name: [64]u8 = undefined;
    {
        var wfs: storage.FileStorage = undefined;
        var wlog = try packOpenWritable(gpa, io, tmp.dir, &wfs);
        defer wfs.file.close(io);
        defer wlog.deinit(gpa);
        try packSeedUnbeatableWireDelta(gpa, &wlog, &a_name, &b_name);
    }

    var sfs: storage.FileStorage = undefined;
    var src = try packOpenReadOnly(gpa, io, tmp.dir, &sfs);
    defer sfs.file.close(io);
    defer src.deinit(gpa);

    // Precondition: the source holds B as a compact delta whose base is A.
    {
        const sb = packRidByName(&src, &b_name).?;
        try std.testing.expectEqual(logstore.FrameKind.delta, src.kindOf(sb));
        var sw = (try src.wire(gpa, sb)).?;
        defer sw.deinit(gpa);
        var abin: [digest_len]u8 = undefined;
        _ = try std.fmt.hexToBytes(&abin, &a_name);
        try std.testing.expectEqualSlices(u8, &abin, sw.delta.baseHash());
    }

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

    const stats = try packLog(gpa, &src, &dst);

    // (a) decode-all byte-identical pre/post.
    try packExpectContentEqual(gpa, &src, &dst);
    // (b) NO frame is larger post-pack than its pre-pack encoding.
    try packExpectNoFrameLarger(gpa, &src, &dst);
    // (c) the whole store did not regress on this wire-delta-heavy fixture.
    try std.testing.expect(stats.bytes_after <= stats.bytes_before);
    // Accounting exact: reported sizes are the real store sizes.
    try std.testing.expectEqual(src.size(), stats.bytes_before);
    try std.testing.expectEqual(dst.size(), stats.bytes_after);

    // B stayed a delta (the compact A-based frame was kept, re-linked to A's new
    // rid) — the guard refused to replace it with the larger full zlib pack's
    // window-only re-encoding would have produced.
    const db = packRidByName(&dst, &b_name).?;
    try std.testing.expectEqual(logstore.FrameKind.delta, dst.kindOf(db));
    var dw = (try dst.wire(gpa, db)).?;
    defer dw.deinit(gpa);
    var abin: [digest_len]u8 = undefined;
    _ = try std.fmt.hexToBytes(&abin, &a_name);
    try std.testing.expectEqualSlices(u8, &abin, dw.delta.baseHash());
    // The kept frame is no larger than the source's (equal — copied verbatim).
    try std.testing.expect(dst.frameSpanOf(db) <= src.frameSpanOf(packRidByName(&src, &b_name).?));
}

/// Seed a store exercising EVERY frame class pack must handle in one place: a
/// raw frame, a zlib frame, a locally-created delta, a verbatim wire delta, an
/// interned phantom (no frame on disk), and the config singleton. Fills the
/// content-bearing artifacts' 64-hex names plus the phantom's binary digest.
fn packSeedMixedStore(
    gpa: std.mem.Allocator,
    log: *logstore.LogStore,
    raw_name: *[64]u8,
    zlib_name: *[64]u8,
    delta_name: *[64]u8,
    wire_name: *[64]u8,
    phantom_digest: *[digest_len]u8,
) !void {
    try log.putConfig(.{ .hash_policy = 2, .default_branch = "trunk", .project_id = "pid", .project_code = "code-abc" });
    // raw: a unique blob large enough to be a real (if unrelated) base.
    var rbuf: std.ArrayList(u8) = .empty;
    defer rbuf.deinit(gpa);
    var i: usize = 0;
    while (i < 120) : (i += 1) try rbuf.print(gpa, "raw-only content row {d}\n", .{i});
    const rbytes = rbuf.items;
    raw_name.* = try packSeedBlob(log, rbytes); // `put` => .raw
    // zlib: a large evolving blob stored as a full zlib container.
    const z = try packEvolvingBlob(gpa, 1);
    defer gpa.free(z);
    var z_digest: [digest_len]u8 = undefined;
    std.crypto.hash.sha3.Sha3_256.hash(z, &z_digest, .{});
    const z_rid = try log.putZlib(&z_digest, z);
    _ = zit.artifact.artifactName(.sha3_256, z, zlib_name);
    // local delta: an edit of the zlib blob, delta-encoded against it.
    const d = try packEvolvingBlob(gpa, 2);
    defer gpa.free(d);
    var d_digest: [digest_len]u8 = undefined;
    std.crypto.hash.sha3.Sha3_256.hash(d, &d_digest, .{});
    _ = try log.putArtifact(&d_digest, d, z_rid);
    _ = zit.artifact.artifactName(.sha3_256, d, delta_name);
    // wire delta: a hand-crafted §4.1 delta container against the raw blob (a
    // poor base) — the verbatim-retained peer form.
    const w = try packEvolvingBlob(gpa, 3);
    defer gpa.free(w);
    var w_digest: [digest_len]u8 = undefined;
    std.crypto.hash.sha3.Sha3_256.hash(w, &w_digest, .{});
    _ = zit.artifact.artifactName(.sha3_256, w, wire_name);
    const r_rid = packRidByName(log, raw_name).?;
    const raw_delta = try zit.delta.create(gpa, rbytes, w);
    defer gpa.free(raw_delta);
    const container = try zit.container.encode(gpa, raw_delta, .default);
    defer gpa.free(container);
    _ = try log.putDeltaContainer(&w_digest, r_rid, container);
    // phantom: an interned forward reference that never reaches disk.
    std.crypto.hash.sha3.Sha3_256.hash("absent forward-referenced artifact", phantom_digest, .{});
    _ = try log.intern(phantom_digest);
    try log.commit();
}

test "pack §6.1/§3: a mixed raw/zlib/delta/wire-delta/phantom/config store round-trips, config+phantom preserved, accounting exact" {
    const gpa = std.testing.allocator;
    const io = std.testing.io;
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();

    var raw_name: [64]u8 = undefined;
    var zlib_name: [64]u8 = undefined;
    var delta_name: [64]u8 = undefined;
    var wire_name: [64]u8 = undefined;
    var phantom_digest: [digest_len]u8 = undefined;

    // A phantom never reaches disk, so we pack the in-memory writable store
    // directly (a reopen would drop it) — the case pack must tolerate without
    // choking or emitting a garbage frame.
    var wfs: storage.FileStorage = undefined;
    var wlog = try packOpenWritable(gpa, io, tmp.dir, &wfs);
    defer wfs.file.close(io);
    defer wlog.deinit(gpa);
    try packSeedMixedStore(gpa, &wlog, &raw_name, &zlib_name, &delta_name, &wire_name, &phantom_digest);

    // The seeded store really holds one frame of each class plus the phantom.
    try std.testing.expectEqual(logstore.FrameKind.raw, wlog.kindOf(packRidByName(&wlog, &raw_name).?));
    try std.testing.expectEqual(logstore.FrameKind.zlib, wlog.kindOf(packRidByName(&wlog, &zlib_name).?));
    try std.testing.expectEqual(logstore.FrameKind.delta, wlog.kindOf(packRidByName(&wlog, &delta_name).?));
    try std.testing.expectEqual(logstore.FrameKind.delta, wlog.kindOf(packRidByName(&wlog, &wire_name).?));
    try std.testing.expect(wlog.isPhantom(wlog.lookup(&phantom_digest).?));

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

    const stats = try packLog(gpa, &wlog, &dst);

    // Decode-all byte-identical pre/post, and the artifact SETS match (the
    // phantom is excluded from both, so counts still agree).
    try packExpectContentEqual(gpa, &wlog, &dst);
    // §6.1 per-frame monotonicity: no frame grew in the repack.
    try packExpectNoFrameLarger(gpa, &wlog, &dst);

    // Config preserved verbatim in fields.
    var soc = (try wlog.readConfig(gpa)).?;
    defer soc.deinit(gpa);
    var doc = (try dst.readConfig(gpa)).?;
    defer doc.deinit(gpa);
    try std.testing.expectEqual(soc.config.hash_policy, doc.config.hash_policy);
    try std.testing.expectEqualStrings(soc.config.default_branch, doc.config.default_branch);
    try std.testing.expectEqualStrings(soc.config.project_id, doc.config.project_id);
    try std.testing.expectEqualStrings(soc.config.project_code, doc.config.project_code);

    // Phantom preserved as a contentless forward reference: it carries no bytes,
    // so pack writes no frame for it (a phantom never reaches disk) and it is not
    // resurrected as a garbage artifact in the packed store.
    try std.testing.expect(dst.lookup(&phantom_digest) == null);

    // Accounting exact: the reported sizes are the actual store sizes and the
    // artifact tally is the source's real (non-phantom) frame count.
    try std.testing.expectEqual(wlog.size(), stats.bytes_before);
    try std.testing.expectEqual(dst.size(), stats.bytes_after);
    const src_rids = try wlog.ridsInPersistedOrder(gpa);
    defer gpa.free(src_rids);
    try std.testing.expectEqual(src_rids.len, stats.artifacts);
}

test "pack §7.7: packLog aborts on a source frame whose content does not name it (no packed frame written)" {
    const gpa = std.testing.allocator;
    const io = std.testing.io;
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();

    // A frame keyed by a name that is NOT sha3-256 of its bytes: `put` stores
    // bytes verbatim under any key, so this models a corrupt/lying source frame
    // that survives `openReadOnly` (its CRC + embedded-hash label are intact).
    const lying_name = [_]u8{0xab} ** digest_len;
    {
        var wfs: storage.FileStorage = undefined;
        var wlog = try packOpenWritable(gpa, io, tmp.dir, &wfs);
        defer wfs.file.close(io);
        defer wlog.deinit(gpa);
        _ = try wlog.put(&lying_name, "content these bytes do not name");
        try wlog.commit();
    }

    var sfs: storage.FileStorage = undefined;
    var src = try packOpenReadOnly(gpa, io, tmp.dir, &sfs);
    defer sfs.file.close(io);
    defer src.deinit(gpa);

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

    try std.testing.expectError(error.Corrupt, packLog(gpa, &src, &dst));
    try std.testing.expect(dst.lookup(&lying_name) == null); // no frame landed on the abort
}

test "pack is idempotent: pack∘pack is byte-identical (0 additional re-deltas)" {
    const gpa = std.testing.allocator;
    const io = std.testing.io;
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();

    var tip: [64]u8 = undefined;
    {
        var wfs: storage.FileStorage = undefined;
        var wlog = try packOpenWritable(gpa, io, tmp.dir, &wfs);
        defer wfs.file.close(io);
        defer wlog.deinit(gpa);
        try packSeedFullStore(gpa, &wlog, &tip);
    }

    // First pack: source (full) → L1.
    {
        var sfs: storage.FileStorage = undefined;
        var src = try packOpenReadOnly(gpa, io, tmp.dir, &sfs);
        defer sfs.file.close(io);
        defer src.deinit(gpa);
        var f1 = try tmp.dir.createFile(io, "pack1", .{ .read = true, .truncate = true });
        defer f1.close(io);
        var f1fs = storage.FileStorage.init(io, f1, 0);
        var dst = try logstore.LogStore.open(gpa, f1fs.storageFile(), src.hash_kind);
        defer dst.deinit(gpa);
        _ = try packLog(gpa, &src, &dst);
    }

    // Second pack: L1 → L2, byte-for-byte equal, and no further re-deltaing.
    var l1fs: storage.FileStorage = undefined;
    var l1 = try packOpenReadOnly2(gpa, io, tmp.dir, "pack1", &l1fs);
    defer l1fs.file.close(io);
    defer l1.deinit(gpa);
    var f2 = try tmp.dir.createFile(io, "pack2", .{ .read = true, .truncate = true });
    defer f2.close(io);
    var f2fs = storage.FileStorage.init(io, f2, 0);
    var l2 = try logstore.LogStore.open(gpa, f2fs.storageFile(), l1.hash_kind);
    defer l2.deinit(gpa);
    const stats2 = try packLog(gpa, &l1, &l2);

    try std.testing.expectEqual(l1.size(), l2.size());
    const a = try tmp.dir.readFileAlloc(io, "pack1", gpa, .limited(1 << 24));
    defer gpa.free(a);
    const b = try tmp.dir.readFileAlloc(io, "pack2", gpa, .limited(1 << 24));
    defer gpa.free(b);
    try std.testing.expectEqualSlices(u8, a, b); // byte-identical: pack∘pack == pack

    // An already-packed store still reports its deltas (they were re-created
    // identically), and packing it does not shrink further.
    try std.testing.expectEqual(l1.size(), stats2.bytes_after);
}

/// Reopen a named log file in `dir` read-only (a sibling to `packOpenReadOnly`
/// that takes an arbitrary name, for the multi-generation idempotence test).
fn packOpenReadOnly2(gpa: std.mem.Allocator, io: std.Io, dir: std.Io.Dir, name: []const u8, fs_out: *storage.FileStorage) !logstore.LogStore {
    const f = try dir.openFile(io, name, .{});
    const st = try f.stat(io);
    fs_out.* = storage.FileStorage.init(io, f, st.size);
    return logstore.LogStore.openReadOnly(gpa, fs_out.storageFile());
}

test "pack: packLog never writes the source (a crash before the rename leaves the original byte-identical)" {
    const gpa = std.testing.allocator;
    const io = std.testing.io;
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();

    var tip: [64]u8 = undefined;
    {
        var wfs: storage.FileStorage = undefined;
        var wlog = try packOpenWritable(gpa, io, tmp.dir, &wfs);
        defer wfs.file.close(io);
        defer wlog.deinit(gpa);
        try packSeedFullStore(gpa, &wlog, &tip);
    }

    const before = try tmp.dir.readFileAlloc(io, repo_filename, gpa, .limited(1 << 24));
    defer gpa.free(before);

    // Pack into a separate temp (the new generation). packLog only ever reads
    // the source; the atomic swap in `runPack` is the sole mutation of `.zit`.
    {
        var sfs: storage.FileStorage = undefined;
        var src = try packOpenReadOnly(gpa, io, tmp.dir, &sfs);
        defer sfs.file.close(io);
        defer src.deinit(gpa);
        var f = try tmp.dir.createFile(io, pack_tmp_filename, .{ .read = true, .truncate = true });
        defer f.close(io);
        var ffs = storage.FileStorage.init(io, f, 0);
        var dst = try logstore.LogStore.open(gpa, ffs.storageFile(), src.hash_kind);
        defer dst.deinit(gpa);
        _ = try packLog(gpa, &src, &dst);
    }

    // No rename happened (the "crash"): the original `.zit` is byte-for-byte
    // unchanged and still opens.
    const after = try tmp.dir.readFileAlloc(io, repo_filename, gpa, .limited(1 << 24));
    defer gpa.free(after);
    try std.testing.expectEqualSlices(u8, before, after);
}

/// The allocation-failure probe: pack a prebuilt source `.zit` into a fresh temp
/// destination, using the failing allocator throughout. Every OOM point must
/// unwind cleanly (no leak) and a fully-funded run must succeed.
fn packOomProbe(alloc: std.mem.Allocator, io: std.Io, dir: std.Io.Dir) !void {
    var sfs: storage.FileStorage = undefined;
    var src = try packOpenReadOnly(alloc, io, dir, &sfs);
    defer sfs.file.close(io);
    defer src.deinit(alloc);
    var f = try dir.createFile(io, "oomdst", .{ .read = true, .truncate = true });
    defer f.close(io);
    var ffs = storage.FileStorage.init(io, f, 0);
    var dst = try logstore.LogStore.open(alloc, ffs.storageFile(), src.hash_kind);
    defer dst.deinit(alloc);
    _ = try packLog(alloc, &src, &dst);
}

test "pack: checkAllAllocationFailures on packLog (every OOM path unwinds cleanly)" {
    const io = std.testing.io;
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    var tip: [64]u8 = undefined;
    {
        var wfs: storage.FileStorage = undefined;
        var wlog = try packOpenWritable(std.testing.allocator, io, tmp.dir, &wfs);
        defer wfs.file.close(io);
        defer wlog.deinit(std.testing.allocator);
        try packSeedFullStore(std.testing.allocator, &wlog, &tip);
    }
    try std.testing.checkAllAllocationFailures(std.testing.allocator, packOomProbe, .{ io, tmp.dir });
}

test "pack∘derive == from-scratch derive: the packed store derives the identical repo view" {
    const gpa = std.testing.allocator;
    const io = std.testing.io;
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();

    var tip: [64]u8 = undefined;
    {
        var wfs: storage.FileStorage = undefined;
        var wlog = try packOpenWritable(gpa, io, tmp.dir, &wfs);
        defer wfs.file.close(io);
        defer wlog.deinit(gpa);
        try packSeedFullStore(gpa, &wlog, &tip);
    }

    var sfs: storage.FileStorage = undefined;
    var src = try packOpenReadOnly(gpa, io, tmp.dir, &sfs);
    defer sfs.file.close(io);
    defer src.deinit(gpa);
    var src_view: zit.repoview.RepoView = undefined;
    try src_view.load(gpa, &src);
    defer src_view.deinit(gpa);

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

    var rfs: storage.FileStorage = undefined;
    var packed_log = try packOpenReadOnly2(gpa, io, tmp.dir, "packed", &rfs);
    defer rfs.file.close(io);
    defer packed_log.deinit(gpa);
    var packed_view: zit.repoview.RepoView = undefined;
    try packed_view.load(gpa, &packed_log);
    defer packed_view.deinit(gpa);

    // Same derived view: main branch, artifact count, and the leaf set + order
    // (§7.5) are identical — pack changed only the storage layout.
    try std.testing.expectEqualStrings(src_view.main_branch, packed_view.main_branch);
    try std.testing.expectEqual(src_view.store.store().count(), packed_view.store.store().count());
    var la = src_view.state.leaves();
    var lb = packed_view.state.leaves();
    while (la.next()) |ra| {
        const rb = lb.next() orelse return error.LeafCountMismatch;
        try std.testing.expectEqualStrings(src_view.hashOf(ra), packed_view.hashOf(rb));
    }
    try std.testing.expect(lb.next() == null);
}

test "renderPackStats: 'packed' when the store shrank, 'already packed' otherwise" {
    var buf: [128]u8 = undefined;
    {
        var w: std.Io.Writer = .fixed(&buf);
        renderPackStats(&w, .{ .artifacts = 1695, .deltas = 1373, .bytes_before = 9_680_000, .bytes_after = 920_000 });
        try std.testing.expectEqualStrings(
            "packed: 1695 artifacts, 1373 re-deltaed, 9680000 -> 920000 bytes\n",
            w.buffered(),
        );
    }
    {
        var w: std.Io.Writer = .fixed(&buf);
        renderPackStats(&w, .{ .artifacts = 3, .deltas = 2, .bytes_before = 4096, .bytes_after = 4096 });
        try std.testing.expectEqualStrings(
            "already packed: 3 artifacts, 2 re-deltaed, 4096 -> 4096 bytes\n",
            w.buffered(),
        );
    }
}

/// Render one `ApplyError` through `openApplyFail` into `buf`.
fn openFailText(buf: []u8, err: update.ApplyError) []const u8 {
    var w = Io.Writer.fixed(buf);
    var sink: [1]u8 = undefined;
    var wnull = Io.Writer.fixed(&sink);
    const s: cli.Streams = .{ .out = &wnull, .err = &w };
    _ = openApplyFail(s, err);
    return w.buffered();
}

// Same guard as the `zit update`, `zit merge` and `zit bisect` mappers.
// `Corrupt` alleges the stored bytes do not match the name they are stored
// under; `StateRecordInvalid` says only that an artifact was unusable as a
// state record. A clone of an intact fossil repository reaches the second
// without anything being damaged, so it must never be answered with a
// corruption claim.
test "open: the two read-path failures do not share a diagnostic" {
    var a: [256]u8 = undefined;
    var b: [256]u8 = undefined;
    const corrupt = openFailText(&a, error.Corrupt);
    const state = openFailText(&b, error.StateRecordInvalid);

    try std.testing.expect(!std.mem.eql(u8, corrupt, state));
    try std.testing.expect(std.mem.indexOf(u8, corrupt, "corrupt") != null);
    try std.testing.expect(std.mem.indexOf(u8, state, "corrupt") == null);
}

// ---------------------------------------------------------------------------
// Issue #9 — `zit serve` must answer one sync round in time bounded by the
// REQUEST, not by the served store. Two per-request costs made a clone
// O(rounds x store): counting held artifacts by forcing a full inflate of every
// artifact, and rebuilding the entire in-memory snapshot from the log on every
// request. The guards below pin both as SCALING properties measured with work
// counters (plaintext bytes materialised, snapshot builds performed) — never
// wall-clock thresholds, which flake and prove nothing about the mechanism.
// ---------------------------------------------------------------------------

/// Seed `ms` with `n` distinct wire-backed artifacts exactly as
/// `loadContentStore` does for a served repository: one §4.1 full container per
/// artifact, with NO plaintext materialised. `ms.plaintextBytesHeld()` is 0 on
/// return, so any later non-zero reading is inflation this store did not ask for.
fn i9SeedWireStore(gpa: std.mem.Allocator, ms: *repostate.MemStore, n: usize) !void {
    var i: usize = 0;
    while (i < n) : (i += 1) {
        var body_buf: [64]u8 = undefined;
        const body = try std.fmt.bufPrint(&body_buf, "issue nine artifact {d}\n", .{i});
        var name_buf: [64]u8 = undefined;
        const name = zit.artifact.artifactName(.sha3_256, body, &name_buf);
        const container = try zit.container.encode(gpa, body, .default);
        defer gpa.free(container);
        _ = try ms.putWire(name, container);
    }
    std.debug.assert(ms.plaintextBytesHeld() == 0);
}

test "serve #9: the held-artifact count serve uses never materialises plaintext (constant in store size)" {
    const gpa = std.testing.allocator;
    // Two store sizes: the work this count performs must not grow with n. The
    // count VALUE alone would pass before and after the fix — it is the
    // materialised-plaintext reading that observes the defect, so that is what
    // is asserted, at both sizes, as the scaling statement.
    for ([_]usize{ 8, 64 }) |n| {
        var ms = repostate.MemStore.init(gpa);
        defer ms.deinit();
        try i9SeedWireStore(gpa, &ms, n);

        try std.testing.expectEqual(n, ms.store().heldCount());
        try std.testing.expectEqual(@as(usize, 0), ms.plaintextBytesHeld());
    }
}

test "serve #9: a held artifact whose container does not inflate is still counted as servable" {
    const gpa = std.testing.allocator;
    var ms = repostate.MemStore.init(gpa);
    defer ms.deinit();

    // §4.1 shape (4-byte length prefix) with a payload that is not a valid zlib
    // stream: `held()==true`, `content()==null` — the case `held()`'s contract at
    // repostate.zig names. `respond` retransmits it verbatim and `persistStore`
    // counts it, so serve's own count must agree with them.
    const bogus = "\x00\x00\x00\x0anot-zlib-bytes";
    _ = try ms.putWire("f" ** 64, bogus);

    try std.testing.expectEqual(@as(usize, 1), ms.store().heldCount());
    try std.testing.expect(ms.store().held(1));
    try std.testing.expect(ms.store().content(1) == null);
}

/// Append one content frame (raw `digest` naming `bytes`) to the `.zit` in
/// `dir`, creating the repository on first call — a separate writer's committed
/// append, exactly what a `zit commit` alongside a live `zit serve` performs.
fn i9AppendFrame(gpa: std.mem.Allocator, io: std.Io, dir: std.Io.Dir, digest: []const u8, bytes: []const u8) !void {
    const hash_kind: u8 = @intFromEnum(zit.artifact.defaultPolicy(false, false));
    var f = dir.openFile(io, repo_filename, .{ .mode = .read_write }) catch
        try dir.createFile(io, repo_filename, .{ .read = true });
    defer f.close(io);
    const st = try f.stat(io);
    var fs = storage.FileStorage.init(io, f, st.size);
    var log = try logstore.LogStore.open(gpa, fs.storageFile(), hash_kind);
    defer log.deinit(gpa);
    _ = try log.put(digest, bytes);
    try log.commit();
}

/// A §4.1-framed read-only request naming `hash` in a `gimme` card.
fn i9GimmeRequest(gpa: std.mem.Allocator, card: []const u8, hash: []const u8) ![]u8 {
    const plain = try std.fmt.allocPrint(gpa, "pragma client-version 20000\n{s} {s}\n", .{ card, hash });
    defer gpa.free(plain);
    return zit.container.encode(gpa, plain, .default);
}

test "serve #9: repeated rounds over an unchanged log build the snapshot once (constant in round count)" {
    const gpa = std.testing.allocator;
    const io = std.testing.io;
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();

    const digest_a = "a" ** digest_len;
    try i9AppendFrame(gpa, io, tmp.dir, digest_a, "content a\n");
    var hex_buf: [2 * digest_len]u8 = undefined;
    const hex_a = digestToWireName(digest_a, &hex_buf);
    const request = try i9GimmeRequest(gpa, "gimme", hex_a);
    defer gpa.free(request);

    const server = zit.sync.Server{ .version = serve_mod.server_version };
    var file = try tmp.dir.openFile(io, repo_filename, .{});
    defer file.close(io);
    var ctx = ServeCtx{ .io = io, .root_dir = tmp.dir, .file = file, .server = &server };
    defer ctx.dropSnapshot();

    // Eight rounds, log unchanged: the snapshot is a function of the log, not of
    // the request, so the number of builds must not grow with the round count.
    for (0..8) |_| {
        const reply = try serveAnswer(&ctx, gpa, request);
        gpa.free(reply.body);
    }
    try std.testing.expectEqual(@as(u64, 1), ctx.loads);
}

test "serve #9: a responder's phantom interning cannot leak into the next round" {
    const gpa = std.testing.allocator;
    const io = std.testing.io;
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();

    const digest_a = "a" ** digest_len;
    try i9AppendFrame(gpa, io, tmp.dir, digest_a, "content a\n");
    var hex_buf: [2 * digest_len]u8 = undefined;
    const hex_a = digestToWireName(digest_a, &hex_buf);

    // §3.3 — an `igot` for a hash the server lacks makes `respond` intern a
    // phantom into whatever store it answers over. That mutation must not
    // survive into the reused snapshot.
    const unknown_hex = "c" ** (2 * digest_len);
    const igot_req = try i9GimmeRequest(gpa, "igot", unknown_hex);
    defer gpa.free(igot_req);
    const gimme_req = try i9GimmeRequest(gpa, "gimme", hex_a);
    defer gpa.free(gimme_req);

    const server = zit.sync.Server{ .version = serve_mod.server_version };
    var file = try tmp.dir.openFile(io, repo_filename, .{});
    defer file.close(io);

    // NOTE ON THIS GUARD'S ROLE: it is green against the pre-fix code (which
    // rebuilds per request and so cannot leak) — the round-count guard above is
    // the red one. This guard exists because the obvious WRONG way to satisfy
    // that guard — answer directly over one shared mutable snapshot — passes it
    // and truncates real clones, since the phantoms and memoisations `respond`
    // writes change what later rounds serve. Without this assertion that
    // implementation ships.
    var ctx = ServeCtx{ .io = io, .root_dir = tmp.dir, .file = file, .server = &server };
    defer ctx.dropSnapshot();

    const r1 = try serveAnswer(&ctx, gpa, igot_req);
    gpa.free(r1.body);
    const snap = ctx.snapshot.?.store();
    try std.testing.expectEqual(@as(usize, 1), snap.count()); // the one on-disk artifact, nothing more
    try std.testing.expect(snap.lookup(unknown_hex) == null); // the phantom did not land here

    const r2 = try serveAnswer(&ctx, gpa, gimme_req);
    defer gpa.free(r2.body);
    try std.testing.expectEqual(@as(u64, 1), ctx.loads); // still one build: the log never changed

    // The reply is byte-identical to the one a server with no prior round produces.
    var fresh_file = try tmp.dir.openFile(io, repo_filename, .{});
    defer fresh_file.close(io);
    var fresh_ctx = ServeCtx{ .io = io, .root_dir = tmp.dir, .file = fresh_file, .server = &server };
    defer fresh_ctx.dropSnapshot();
    const expected = try serveAnswer(&fresh_ctx, gpa, gimme_req);
    defer gpa.free(expected.body);
    try std.testing.expectEqualSlices(u8, expected.body, r2.body);
}

test "serve #9/§3.8: a reused snapshot still sees a concurrent commit's append" {
    const gpa = std.testing.allocator;
    const io = std.testing.io;
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();

    const digest_a = "a" ** digest_len;
    const digest_b = "b" ** digest_len;
    try i9AppendFrame(gpa, io, tmp.dir, digest_a, "content a\n");
    var hex_a_buf: [2 * digest_len]u8 = undefined;
    var hex_b_buf: [2 * digest_len]u8 = undefined;
    const hex_a = digestToWireName(digest_a, &hex_a_buf);
    const hex_b = digestToWireName(digest_b, &hex_b_buf);
    const req_a = try i9GimmeRequest(gpa, "gimme", hex_a);
    defer gpa.free(req_a);
    const req_b = try i9GimmeRequest(gpa, "gimme", hex_b);
    defer gpa.free(req_b);

    const server = zit.sync.Server{ .version = serve_mod.server_version };
    var file = try tmp.dir.openFile(io, repo_filename, .{});
    defer file.close(io);
    var ctx = ServeCtx{ .io = io, .root_dir = tmp.dir, .file = file, .server = &server };
    defer ctx.dropSnapshot();

    const r1 = try serveAnswer(&ctx, gpa, req_a);
    gpa.free(r1.body);
    try std.testing.expectEqual(@as(usize, 1), ctx.snapshot.?.store().count());

    // A SEPARATE writer appends while the serve handle stays open.
    try i9AppendFrame(gpa, io, tmp.dir, digest_b, "content b\n");

    const r2 = try serveAnswer(&ctx, gpa, req_b);
    gpa.free(r2.body);
    // The log grew, so the snapshot was rebuilt and the new artifact is servable:
    // a cache that never invalidates would still report one artifact and one load.
    try std.testing.expectEqual(@as(u64, 2), ctx.loads);
    try std.testing.expectEqual(@as(usize, 2), ctx.snapshot.?.store().count());
    try std.testing.expect(ctx.snapshot.?.store().lookup(hex_b) != null);
}

/// Seed `dir`'s `.zit` with `n` distinct compressible artifacts, returning the
/// wire (hex) name of the first. Compressible so the log retains a §4.1
/// container per frame — the shape a served repository actually has, and the
/// only shape in which "answering a round does not inflate the store" has
/// anything to say.
fn i9SeedRepo(gpa: std.mem.Allocator, io: std.Io, dir: std.Io.Dir, n: usize, first_hex: *[2 * digest_len]u8) !void {
    const hash_kind: u8 = @intFromEnum(zit.artifact.defaultPolicy(false, false));
    var f = dir.openFile(io, repo_filename, .{ .mode = .read_write }) catch
        try dir.createFile(io, repo_filename, .{ .read = true });
    defer f.close(io);
    const st = try f.stat(io);
    var fs = storage.FileStorage.init(io, f, st.size);
    var log = try logstore.LogStore.open(gpa, fs.storageFile(), hash_kind);
    defer log.deinit(gpa);
    var i: usize = 0;
    while (i < n) : (i += 1) {
        var body: [512]u8 = undefined;
        @memset(&body, 'a' + @as(u8, @intCast(i % 26)));
        var digest: [digest_len]u8 = undefined;
        @memset(&digest, 0);
        std.mem.writeInt(u64, digest[0..8], i + 1, .big);
        _ = try log.putZlib(&digest, &body);
        if (i == 0) _ = digestToWireName(&digest, first_hex);
    }
    try log.commit();
}

test "serve #9: answering a round inflates only what the round emits (constant in store size)" {
    const gpa = std.testing.allocator;
    const io = std.testing.io;

    // The same read round against two store sizes. The reply is identical, so
    // the plaintext the server inflates to produce it must be identical too: any
    // dependence on n is the server paying for artifacts the round never names.
    // This is the assertion the whole change exists to make true — the pre-fix
    // held-count walked `content()` over every rid, and so did inflate the store.
    var inflated: [2]usize = undefined;
    for ([_]usize{ 4, 40 }, 0..) |n, k| {
        var tmp = std.testing.tmpDir(.{});
        defer tmp.cleanup();
        var hex_buf: [2 * digest_len]u8 = undefined;
        try i9SeedRepo(gpa, io, tmp.dir, n, &hex_buf);
        const request = try i9GimmeRequest(gpa, "gimme", &hex_buf);
        defer gpa.free(request);

        const server = zit.sync.Server{
            .version = serve_mod.server_version,
            .self_servercode = serve_mod.server_servercode,
            .project_code = serve_mod.server_projectcode,
        };
        var file = try tmp.dir.openFile(io, repo_filename, .{});
        defer file.close(io);
        var ctx = ServeCtx{ .io = io, .root_dir = tmp.dir, .file = file, .server = &server };
        defer ctx.dropSnapshot();

        for (0..3) |_| {
            const reply = try serveAnswer(&ctx, gpa, request);
            gpa.free(reply.body);
            // Nothing the round materialised is still resident: the cached index
            // reads containers on demand and frees them between rounds, never
            // holding the repository's uncompressed history (or its containers).
            try std.testing.expectEqual(@as(usize, 0), ctx.snapshot.?.residentBytes());
        }
        const emitted = ctx.snapshot.?.materialized_bytes;

        // A CLONE round is the worst case for residency: it emits every artifact
        // and proves each one's declared size against the plaintext, so it
        // legitimately inflates the whole store. What must not survive it is the
        // plaintext — a snapshot that outlives the round would otherwise end up
        // holding the entire repository uncompressed, for the process's life.
        const clone_plain = try std.fmt.allocPrint(gpa, "pragma client-version 20000\nclone 3 1\n", .{});
        defer gpa.free(clone_plain);
        const clone_req = try zit.container.encode(gpa, clone_plain, .default);
        defer gpa.free(clone_req);
        const clone_reply = try serveAnswer(&ctx, gpa, clone_req);
        gpa.free(clone_reply.body);
        try std.testing.expectEqual(@as(usize, 0), ctx.snapshot.?.residentBytes());
        try std.testing.expect(ctx.snapshot.?.materialized_bytes >= emitted);
        try std.testing.expectEqual(@as(usize, n), ctx.snapshot.?.store().count());
        inflated[k] = emitted;
    }
    try std.testing.expectEqual(inflated[0], inflated[1]);
    // ... and the constant is the round's own emission, not the store's: a
    // 40-artifact store holds 20480 bytes of plaintext, so an equality that
    // happened to hold at "everything" would still be caught here.
    try std.testing.expect(inflated[1] < 4 * 512);
}

test "serve #13: the pushing round does not rebuild inside its deadline; maintain rebuilds between connections" {
    const gpa = std.testing.allocator;
    const io = std.testing.io;
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();

    const hash_kind: u8 = @intFromEnum(zit.artifact.defaultPolicy(false, false));
    {
        var f = try tmp.dir.createFile(io, repo_filename, .{ .read = true, .exclusive = true });
        defer f.close(io);
        var fs = storage.FileStorage.init(io, f, 0);
        var seed = try logstore.LogStore.open(gpa, fs.storageFile(), hash_kind);
        defer seed.deinit(gpa);
        try seed.commit();
    }

    const push_body = "pushed artifact bytes\n";
    var name_buf: [64]u8 = undefined;
    const push_hash = zit.artifact.artifactName(.sha3_256, push_body, &name_buf);
    const plain = try std.fmt.allocPrint(gpa, "pragma client-version 20000\nfile {s} {d}\n{s}", .{ push_hash, push_body.len, push_body });
    defer gpa.free(plain);
    const push_req = try zit.container.encode(gpa, plain, .default);
    defer gpa.free(push_req);
    const read_req = try i9GimmeRequest(gpa, "gimme", push_hash);
    defer gpa.free(read_req);

    const server = zit.sync.Server{ .version = serve_mod.server_version };
    var file = try tmp.dir.openFile(io, repo_filename, .{});
    defer file.close(io);
    var ctx = ServeCtx{ .io = io, .root_dir = tmp.dir, .file = file, .server = &server };
    defer ctx.dropSnapshot();

    const r0 = try serveAnswer(&ctx, gpa, read_req);
    gpa.free(r0.body);
    try std.testing.expectEqual(@as(u64, 1), ctx.loads);

    // The push makes the snapshot stale, but answering it must NOT pay the
    // O(store) rebuild: the reply is written inside a connection deadline, and
    // #13's failure is exactly the whole-repository rebuild landing inside that
    // budget. The rebuild is deferred to `maintain`, which the accept loop runs
    // after the connection closes — outside every deadline. So the pushing round
    // leaves `loads` untouched.
    const r1 = try serveAnswer(&ctx, gpa, push_req);
    gpa.free(r1.body);
    try std.testing.expectEqual(@as(u64, 1), ctx.loads); // no rebuild inside the round
    // The snapshot is momentarily stale — it predates the push — yet the push is
    // already durable on disk: `persistStore` ran inside the round.
    try std.testing.expect(ctx.snapshot != null);
    try std.testing.expect(ctx.snapshot.?.store().lookup(push_hash) == null);

    // The accept loop's post-connection maintenance hook does the O(store)
    // rebuild, between connections, with no connection open to charge it to.
    ServeCtx.maintain(&ctx, gpa);
    try std.testing.expectEqual(@as(u64, 2), ctx.loads);
    try std.testing.expect(ctx.snapshot.?.store().lookup(push_hash) != null);
    // The rebuilt snapshot describes the grown log exactly (the append-only stat
    // gate that `baseSnapshot` already carries).
    const st = try file.stat(io);
    try std.testing.expectEqual(st.size, ctx.snapshot_log_size);

    // A later read round is stat-gated to no rebuild, and still serves the push.
    const r2 = try serveAnswer(&ctx, gpa, read_req);
    defer gpa.free(r2.body);
    try std.testing.expectEqual(@as(u64, 2), ctx.loads);
}