Artifact [b54f90e720]

Artifact b54f90e720976e4f22e9ac250e71f6edd069d13ffe041f25ce718c264e58ee2c:

• File src/cmd/sync.zig — part of check-in [2874978461] at 2026-07-23 10:59:20 on branch fix9-serve-stall — Serve: answer a sync round in time bounded by the request, not the store

  zit serve rebuilt its whole in-memory snapshot from the log on every request, and classified held-vs-phantom by calling content on every rid, forcing a full inflate of the entire repository per request. A clone cost O(rounds x store) of server work; past a few thousand artifacts one round outran the connection deadline and the server answered nothing at all.

  ArtifactStore grows heldCount, the aggregate of the existing non-inflating held predicate, and serve counts with it rather than re-deriving a worse predicate caller-side. ServeCtx caches the snapshot and rebuilds only when a stat shows the append-only log grew, so a commit made alongside a live server is still advertised on the next request.

  Reuse needs isolation: the responder interns phantoms and upgrades entries as it answers, and that would change what later rounds serve. A new OverlayStore answers each request over a per-request write overlay, reads falling through, writes confined to the overlay. The persist pass now takes an ArtifactStore.

  On 67,499 artifacts: unfixed, no round completes; fixed, one snapshot build then sub-millisecond rounds. (user: dmestas, size: 194351) [annotate] [blame] [check-ins using]

//! 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");
    };
    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.deinit();
    const responder = serve_mod.Responder{ .ctx = &ctx, .answerFn = serveAnswer };

    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;
    };
    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 {};

    serve_mod.serve(io, gpa, address, responder, .{}) catch |e| switch (e) {
        error.Listen => {
            streams.err.print("zit serve: cannot bind 127.0.0.1:{d}\n", .{parsed.port}) catch {};
            return .failure;
        },
        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` snapshots the
    /// CURRENT on-disk log from it per request (§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.
    file: File,
    server: *const zit.sync.Server,
    /// The most recent snapshot of the on-disk log, reused across requests. Built
    /// from the long-lived allocator `serve` threads into every answer, so it
    /// outlives the request that built it, and owned here: `deinit` frees it.
    /// Never handed to `respond` directly — see `serveAnswer`.
    snapshot: ?repostate.MemStore = 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,

    fn deinit(self: *ServeCtx) void {
        if (self.snapshot) |*s| s.deinit();
        self.snapshot = null;
        self.snapshot_log_size = 0;
    }

    /// §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.
    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 != null and st.size == self.snapshot_log_size) return self.snapshot.?.store();

        var fresh = repostate.MemStore.init(gpa);
        errdefer fresh.deinit();
        try readServedStore(gpa, self.io, self.file, &fresh);
        self.deinit();
        self.snapshot = fresh;
        self.snapshot_log_size = st.size;
        self.loads += 1;
        return self.snapshot.?.store();
    }
};

/// 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, so the
/// persist pass sees exactly what this request produced.
///
/// 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));
    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,
        };
        // The log grew by what was just made durable, so the snapshot no longer
        // describes it. Drop it rather than patch it: the next request rebuilds
        // from the authoritative bytes, and a push is rare next to a read round.
        ctx.deinit();
    }
    return reply;
}

/// 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;
        }
    }
}

/// §3.8 — read the CURRENT on-disk log from the open serve handle `file` into
/// `ms` for one request. The server is stateless per request, so its holdings
/// are re-read every time rather than snapshotted at startup: a fresh `stat`
/// fixes the committed end-of-log (a separate `zit commit` process, or this
/// server's own push persist, appends complete frames that grow the file), and
/// `openReadOnly` scans exactly that snapshot through a positional read — the
/// handle is never advanced, so re-reading it sees the new bytes. A concurrent
/// append caught mid-write leaves a torn tail, which the read-only scanner
/// rejects as corruption (never a torn artifact); the committed frames below it
/// are read consistently, and the request retries once the write completes. The
/// upshot: a commit made while `serve` runs is advertised on the next request,
/// with no restart (the Direction-A incremental-pull gap).
fn readServedStore(
    gpa: std.mem.Allocator,
    io: std.Io,
    file: File,
    ms: *repostate.MemStore,
) serve_mod.AnswerError!void {
    const st = file.stat(io) catch return error.ServerStorage;
    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,
        error.Storage, error.Corrupt => return error.ServerStorage,
    };
    defer log.deinit(gpa);
    try loadContentStore(gpa, &log, ms);
}

// ---------------------------------------------------------------------------
// `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, 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.deinit();
        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 ms1 = repostate.MemStore.init(gpa);
    defer ms1.deinit();
    try readServedStore(gpa, io, handle, &ms1);
    const n1 = contentCount(ms1.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 ms2 = repostate.MemStore.init(gpa);
    defer ms2.deinit();
    try readServedStore(gpa, io, handle, &ms2);
    try std.testing.expectEqual(n1 + 1, contentCount(ms2.store()));
}

/// 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(),
        );
    }
}

// ---------------------------------------------------------------------------
// 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.deinit();

    // 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.deinit();

    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.deinit();
    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.deinit();

    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);
}