Skip to content
zit Artifact [6a0d16be6f]
Login

Artifact [6a0d16be6f]

Check-ins Using Download Hex

Artifact 6a0d16be6f7522e2f6f3a056252e42e05f3918228f694f75657f96c4a21e5135:

//! zit history / read-only inspection verbs (stat, log, leaves, timeline,
//! branch, tag, info, ls, cat, artifact, annotate) and the helpers and tests
//! private to them.
//!
//! Extracted verbatim from src/main.zig (release step S4). 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_history.runLog` ...); 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("zit").cli;
const merge3 = @import("../merge3.zig");
const cmd_sync = @import("sync.zig");
const annotate_mod = @import("../annotate.zig");
const logstore = @import("zit").logstore;
const storage = @import("zit").storage;
const storestat = @import("zit").storestat;
const Repository = @import("zit").Repository;

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 repo_filename = main.repo_filename;
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 digest_len = main.digest_len;
const timeline_hash_display = main.timeline_hash_display;
const firstCommentLine = main.firstCommentLine;
const renderBranchList = main.renderBranchList;
const sym_prefix = main.sym_prefix;
const lessByHash = main.lessByHash;
const mintControlArtifact = main.mintControlArtifact;
const packSeedBlob = main.packSeedBlob;
const packSeedCheckin = main.packSeedCheckin;
const packEvolvingBlob = main.packEvolvingBlob;
const packOpenWritable = main.packOpenWritable;
const packOpenReadOnly = main.packOpenReadOnly;
const stat_help = main.stat_help;
const log_help = main.log_help;
const leaves_help = main.leaves_help;
const timeline_help = main.timeline_help;
const branch_help = main.branch_help;
const tag_help = main.tag_help;
const info_help = main.info_help;
const ls_help = main.ls_help;
const grep_help = main.grep_help;
const cat_help = main.cat_help;
const artifact_help = main.artifact_help;
const annotate_help = main.annotate_help;

/// Which artifacts `zit stat --artifacts` lists. `all` is the unfiltered roll;
/// `full` and `delta` partition it by storage form, which is the one property
/// the frame walk already knows and so costs nothing extra to filter on.
const StatType = enum { all, full, delta };

/// `zit stat` — answer questions about the store itself: how large it is, how
/// much of that is artifact content, and which artifacts it holds.
///
/// The honesty constraint that shapes this verb: `content-bytes` is MEASURED,
/// by summing each artifact frame's payload length during the walk. It is
/// never the `.zit` file size, which additionally carries the store header, a
/// header per frame, and the padding that aligns each frame to 8 bytes.
/// Reporting the file size as content would look like a result and be a
/// fabrication, so the two travel as separate fields and the overhead that
/// separates them is reported alongside, making the accounting checkable:
/// `header-bytes + content-bytes + frame-overhead-bytes + config-bytes +
/// shun-bytes` is the file, exactly — the shun table's frames are part of the
/// store, so leaving them out of the report would leave an unlabelled gap on
/// every repository an operator has shunned anything in.
///
/// Two traversals, both bounded by the store rather than by the artifact
/// count. `openReadOnly` is the first: it whole-frame-CRC-verifies every frame
/// and builds the in-memory index, which is what `--artifacts` enumerates and
/// what lets `storestat.walk` trust the prefix bytes it reads. The summary path
/// then runs the walk, a prefix-only second pass. The open is not skippable on
/// the summary path: unverified frame prefixes would make the totals a report
/// about bytes nobody checked. What is avoided is the per-artifact shape — no
/// invocation, and no payload decode, per stored artifact. Read-only; the
/// store is not modified.
/// `zit users [--raw]` — emit the repository's user table (#124), one row per
/// line, TAB-separated, sorted by login. Machine-consumable like
/// `zit crosslink`: a consumer (quarry, hosting a mirrored fossil repository)
/// ingests it verbatim.
///
/// `pw` IS WITHHELD BY DEFAULT and printed only under `--raw`. The column is
/// the RFC §6.3 shared-secret hash, and fossil guards the whole user area
/// behind Admin on the wire (`configure.c`'s
/// `m &= ~(CONFIGSET_USER|CONFIGSET_SCRIBER)` for a non-admin client). A
/// default listing that spilled hashes would make them easier to lift from a
/// zit mirror than from the fossil repository it mirrors — so the operator
/// asks for them explicitly, and the flag's presence is the audit trail.
///
/// `cap` is passed through verbatim and never interpreted: the capability
/// vocabulary is fossil's, and a zit that parsed it would need changing every
/// time fossil adds a letter.
pub fn runUsers(
    gpa: std.mem.Allocator,
    io: std.Io,
    arena: std.mem.Allocator,
    verb_args: []const []const u8,
    streams: cli.Streams,
    amb: cli.Ambient,
) cli.Exit {
    var raw = false;
    for (verb_args) |tok| {
        if (std.mem.eql(u8, tok, "-h") or std.mem.eql(u8, tok, "--help")) {
            streams.out.writeAll(main.users_help) catch {};
            return .ok;
        } else if (std.mem.eql(u8, tok, "--raw")) {
            raw = true;
        } else if (tok.len > 1 and tok[0] == '-') {
            return cli.usageError(streams, "users", .unknown_option, tok);
        } else {
            return cli.usageError(streams, "users", .extra_argument, tok);
        }
    }

    const root = (cli.resolveRoot(arena, io, amb.repo_override) catch
        return stateOom(streams, "users")) orelse {
        main.reportNoRoot(streams, io, "users", amb.repo_override);
        return .failure;
    };
    var root_dir = root.dir;
    defer root_dir.close(io);

    var file = cli.openStoreRead(io, root_dir) catch return repoReadFail(streams, "users");
    defer file.close(io);
    const st = file.stat(io) catch return repoReadFail(streams, "users");
    var fs = storage.FileStorage.init(io, file, st.size);
    var store = cmd_sync.openLogAdoptingIndex(gpa, io, root_dir, fs.storageFile(), .refresh_on_scan) catch |e| switch (e) {
        error.OutOfMemory => return stateOom(streams, "users"),
        else => {
            streams.err.print("zit users: cannot read repository: {s}\n", .{reasonOf(e)}) catch {};
            return .failure;
        },
    };
    defer store.deinit(gpa);

    const table = store.userTable(gpa) catch return stateOom(streams, "users");
    defer gpa.free(table);
    for (table) |row| {
        if (raw) {
            streams.out.print("{s}\t{s}\t{s}\t{d}\t{s}\n", .{ row.login, row.cap, row.pw, row.mtime, row.info }) catch {};
        } else {
            streams.out.print("{s}\t{s}\t{d}\t{s}\n", .{ row.login, row.cap, row.mtime, row.info }) catch {};
        }
    }
    return .ok;
}

pub fn runStat(
    gpa: std.mem.Allocator,
    io: std.Io,
    arena: std.mem.Allocator,
    verb_args: []const []const u8,
    streams: cli.Streams,
    amb: cli.Ambient,
) cli.Exit {
    var list_artifacts = false;
    var want: StatType = .all;
    var typed = false;
    for (verb_args) |tok| {
        if (std.mem.eql(u8, tok, "-h") or std.mem.eql(u8, tok, "--help")) {
            streams.out.writeAll(stat_help) catch {};
            return .ok;
        } else if (std.mem.eql(u8, tok, "--artifacts")) {
            list_artifacts = true;
        } else if (std.mem.startsWith(u8, tok, "--type=")) {
            // A second --type is a mistake, not a refinement: silently keeping
            // the last one would answer a question the user did not ask.
            if (typed) return cli.usageError(streams, "stat", .repeated_option, "--type");
            const val = tok["--type=".len..];
            want = if (std.mem.eql(u8, val, "full"))
                .full
            else if (std.mem.eql(u8, val, "delta"))
                .delta
            else
                return cli.usageError(streams, "stat", .unknown_option, tok);
            typed = true;
        } else if (tok.len > 1 and tok[0] == '-') {
            return cli.usageError(streams, "stat", .unknown_option, tok);
        } else {
            return cli.usageError(streams, "stat", .extra_argument, tok);
        }
    }
    // `--type` selects among artifacts; without a listing there is nothing for
    // it to select, and silently ignoring it would hide a mistaken invocation.
    if (typed and !list_artifacts) {
        streams.err.writeAll("zit stat: --type needs --artifacts\n") catch {};
        return .usage;
    }

    // #85 — `-R` joins the read-only family here. `stat` was refused it for a
    // plumbing reason (its handler took no redirected repository), never a
    // policy one, and the cost of that accident was real: a repository's
    // project code was reachable ONLY by speaking the sync protocol to a server
    // you started yourself, so a consumer needing one constant had to spawn a
    // subprocess and a listening socket per repository.
    const root = (cli.resolveRoot(arena, io, amb.repo_override) catch
        return stateOom(streams, "stat")) orelse {
        main.reportNoRoot(streams, io, "stat", amb.repo_override);
        return .failure;
    };
    var root_dir = root.dir;
    defer root_dir.close(io);

    var file = cli.openStoreRead(io, root_dir) catch return repoReadFail(streams, "stat");
    defer file.close(io);
    const st = file.stat(io) catch return repoReadFail(streams, "stat");
    var fs = storage.FileStorage.init(io, file, st.size);
    // #121 — adopt the persisted scan index for the config/roll read; the
    // census below stays a MEASURED physical walk (`storestat.walk`), so
    // every printed number is still read from the frames, never the cache.
    var store = cmd_sync.openLogAdoptingIndex(gpa, io, root_dir, fs.storageFile(), .refresh_on_scan) catch |e| switch (e) {
        error.OutOfMemory => return stateOom(streams, "stat"),
        else => {
            streams.err.print("zit stat: cannot read repository: {s}\n", .{reasonOf(e)}) catch {};
            return .failure;
        },
    };
    defer store.deinit(gpa);

    if (list_artifacts) return statArtifacts(gpa, &store, want, streams);

    var fs2 = storage.FileStorage.init(io, file, st.size);
    const bd = storestat.walk(fs2.storageFile(), st.size) catch |e| {
        streams.err.print("zit stat: cannot read repository: {s}\n", .{reasonOf(e)}) catch {};
        return .failure;
    };
    // #85 — the repository's IDENTITY, printed first because it is what a
    // consumer most often came for. `project-code` is the RFC section 6.3 shared
    // secret's first input; before this it was obtainable only by speaking the
    // sync protocol to a server you started yourself, which cost a subprocess, a
    // listening socket and a startup stall per repository.
    //
    // Emitted unconditionally, with an empty value on a repository that has
    // none, so a parser reads one stable shape: a missing LINE and an empty
    // VALUE are different facts, and only one of them means "this repository
    // predates project codes".
    var identity_oc = store.readConfig(gpa) catch null;
    defer if (identity_oc) |*oc| oc.deinit(gpa);
    const project_code: []const u8 = if (identity_oc) |oc| oc.config.project_code else "";
    // The stored byte is opaque to storage (it round-trips whatever it was
    // given), so a value outside the enum is reported as the raw number rather
    // than guessed at: a repository written by something else is a fact worth
    // seeing, not one to normalise away.
    var policy_buf: [16]u8 = undefined;
    const hash_policy: []const u8 = if (identity_oc) |oc| blk: {
        const raw = oc.config.hash_policy;
        break :blk if (std.enums.fromInt(artifact.HashPolicy, raw)) |p|
            @tagName(p)
        else
            std.fmt.bufPrint(&policy_buf, "{d}", .{raw}) catch "";
    } else "";
    streams.out.print("project-code {s}\nhash-policy {s}\n", .{ project_code, hash_policy }) catch {};

    streams.out.print(
        \\store-bytes {d}
        \\header-bytes {d}
        \\content-bytes {d}
        \\frame-overhead-bytes {d}
        \\config-bytes {d}
        \\shun-bytes {d}
        \\shun-rows {d}
        \\user-bytes {d}
        \\user-frames {d}
        \\artifacts {d}
        \\full-form {d}
        \\full-form-bytes {d}
        \\deltas {d}
        \\delta-bytes {d}
        \\
    , .{
        st.size,          logstore.header_len, bd.contentBytes(), bd.contentOverhead(),
        bd.configBytes(), bd.shunBytes(),      bd.shunRows(),     bd.userBytes(),
        bd.userFrames(),  bd.artifacts(),      bd.fullForm(),     bd.fullFormBytes(),
        bd.deltas(),      bd.deltaBytes(),
    }) catch {};
    return .ok;
}

/// Emits one lowercase-hex artifact hash per line, in ascending hash order, so
/// two stores' rolls compare with `comm`. Reads the already-open store's index
/// — no further I/O, and no per-artifact process.
fn statArtifacts(
    gpa: std.mem.Allocator,
    store: *logstore.LogStore,
    want: StatType,
    streams: cli.Streams,
) cli.Exit {
    const rids = store.ridsInPersistedOrder(gpa) catch return stateOom(streams, "stat");
    defer gpa.free(rids);

    var hashes: std.ArrayListUnmanaged([]const u8) = .empty;
    defer hashes.deinit(gpa);
    hashes.ensureTotalCapacity(gpa, rids.len) catch return stateOom(streams, "stat");
    for (rids) |rid| {
        const kind = store.kindOf(rid);
        // The config record is repository metadata, not a stored artifact, so
        // it is absent from every listing and from the `artifacts` count.
        if (kind == .config) continue;
        const keep = switch (want) {
            .all => true,
            .delta => kind == .delta,
            .full => kind == .raw or kind == .zlib,
        };
        if (keep) hashes.appendAssumeCapacity(store.hashOf(rid));
    }
    // The listing is a filtered projection of the persisted rid set — never more
    // hashes than records walked.
    std.debug.assert(hashes.items.len <= rids.len);
    std.mem.sort([]const u8, hashes.items, {}, struct {
        fn lt(_: void, a: []const u8, b: []const u8) bool {
            return std.mem.order(u8, a, b) == .lt;
        }
    }.lt);
    for (hashes.items) |h| streams.out.print("{x}\n", .{h}) catch {};
    return .ok;
}

const LogWalkError = error{
    OutOfMemory,
    Storage,
    Corrupt,
    MissingRoot,
    InvalidRoot,
    InvalidTarget,
};

fn decodeFullId(id: []const u8, out: *[digest_len]u8) error{InvalidId}![]const u8 {
    if (!artifact.validateName(id)) return error.InvalidId;
    const decoded = out[0 .. id.len / 2];
    _ = std.fmt.hexToBytes(decoded, id) catch return error.InvalidId;
    // §3.6: `validateName` accepted a 40/64-hex token, so the decode is a whole
    // 20/32-byte digest that fits the fixed buffer — never a truncated prefix.
    std.debug.assert(decoded.len <= digest_len);
    return decoded;
}

/// The optional PATH restriction for a `zit log CHECKIN PATH` walk (B-log-2).
/// `repo` answers the one general-purpose §8 effective-set query (`fileAt`) the
/// change predicate consumes (B-log-5); `path` is the exact repo-relative name.
const LogPathFilter = struct {
    repo: *Repository,
    path: []const u8,
};

/// #93 — emit each check-in's own D/U/C cards beside its id. The manifest is
/// already parsed at the emit site to walk its P cards, so the facts are free
/// here and cost a subprocess per row anywhere else: a consumer drawing a file
/// history page (`log CHECKIN PATH`) otherwise runs `zit info` once per entry.
/// Same defect, and same fix, as `ls --long` (#96).
fn logEmit(out: *Io.Writer, id: []const u8, manifest: *const artifact.Manifest, long: bool) void {
    if (!long) {
        out.print("checkin {s}\n", .{id}) catch {};
        return;
    }
    // Comment last: it is the only field that may contain spaces, so every
    // earlier column stays splittable. A check-in with no C card prints an
    // empty trailing field rather than a short line, so the arity is fixed.
    const when = if (manifest.d) |d| d.timestamp else "-";
    const who = if (manifest.u) |u| u.login else "-";
    const what = if (manifest.c) |c| c.comment else "";
    out.print("checkin {s} {s} {s} {s}\n", .{ id, when, who, what }) catch {};
}

fn walkLog(
    gpa: std.mem.Allocator,
    log: *logstore.LogStore,
    root_id: []const u8,
    out: *Io.Writer,
    filter: ?LogPathFilter,
    limit: ?usize,
    long: bool,
) LogWalkError!void {
    var pending: std.ArrayList([]u8) = .empty;
    defer {
        for (pending.items) |id| gpa.free(id);
        pending.deinit(gpa);
    }
    var visited: std.StringHashMapUnmanaged(void) = .empty;
    defer visited.deinit(gpa);
    var visited_ids: std.ArrayList([]u8) = .empty;
    defer {
        for (visited_ids.items) |id| gpa.free(id);
        visited_ids.deinit(gpa);
    }
    // #84's memo, and this walk is the case it exists for. Check-in manifests
    // are routinely stored as deltas against their neighbours, so reconstructing
    // each one from `content` alone re-resolves the SAME base chain once per
    // check-in — work that grows with how much history sits behind the node
    // rather than with the node, which is why the per-check-in cost climbed with
    // repository size instead of staying flat. Every other read path in this
    // file already takes the memo; this one did not.
    var plaintext: logstore.PlaintextCache = .{};
    defer plaintext.deinit(gpa);
    // `-n N` caps the ENTRIES NAMED, not the nodes traversed: with a PATH filter
    // most visited check-ins print nothing, and capping traversal would make the
    // answer depend on how much unrelated history sat in between. Counting the
    // lines the user actually sees is the only reading that stays stable.
    var emitted: usize = 0;

    const root_copy = try gpa.dupe(u8, root_id);
    pending.append(gpa, root_copy) catch {
        gpa.free(root_copy);
        return error.OutOfMemory;
    };
    while (pending.pop()) |id| {
        // `-n N` (N ≥ 1; `parseLimitValue` maps 0 → null) returns the moment the
        // Nth line is named, so at every loop entry fewer than N lines are out.
        if (limit) |n| std.debug.assert(emitted < n);
        if (visited.contains(id)) {
            gpa.free(id);
            continue;
        }
        visited_ids.append(gpa, id) catch {
            gpa.free(id);
            return error.OutOfMemory;
        };
        visited.put(gpa, id, {}) catch {
            _ = visited_ids.pop();
            gpa.free(id);
            return error.OutOfMemory;
        };
        const is_root = visited_ids.items.len == 1;

        var hash_buf: [digest_len]u8 = undefined;
        const decoded = decodeFullId(id, &hash_buf) catch
            return if (is_root) error.InvalidRoot else error.InvalidTarget;
        // A decoded id is a full §3.6 artifact name (40/64 hex), never a prefix —
        // it is exactly what every `checkin`/`phantom` line below renders.
        std.debug.assert(artifact.validateName(id));
        const rid = log.lookup(decoded) orelse {
            if (is_root) return error.MissingRoot;
            // A phantom's effective set is unknowable, so the PATH filter cannot
            // assert it changes PATH — it is silently skipped (the unfiltered
            // walk still names it once).
            if (filter == null) {
                out.print("phantom {s}\n", .{id}) catch {};
                emitted += 1;
                if (limit) |n| if (emitted >= n) return;
            }
            continue;
        };
        const bytes = (try log.contentCached(gpa, rid, &plaintext)) orelse {
            if (is_root) return error.MissingRoot;
            if (filter == null) {
                out.print("phantom {s}\n", .{id}) catch {};
                emitted += 1;
                if (limit) |n| if (emitted >= n) return;
            }
            continue;
        };
        defer gpa.free(bytes);
        if (!artifact.verify(id, bytes)) return error.Corrupt;

        // Read path: the log-walk reads fossil-origin history, so it tolerates
        // the legacy F-card trailing separator.
        var loaded = artifact.parseForImport(gpa, bytes) catch |e| switch (e) {
            error.OutOfMemory => return error.OutOfMemory,
            else => return if (is_root) error.InvalidRoot else error.InvalidTarget,
        };
        defer loaded.deinit(gpa);
        const manifest = &loaded.manifest;
        if (manifest.kind != .manifest)
            return if (is_root) error.InvalidRoot else error.InvalidTarget;

        // With no PATH the walk names every reachable check-in (B-log-1); with
        // PATH it names only the check-ins whose effective content for PATH
        // differs from their primary parent's (B-log-2).
        if (filter) |f| {
            if (try logChangesPath(gpa, f.repo, id, manifest, f.path)) {
                logEmit(out, id, manifest, long);
                emitted += 1;
            }
        } else {
            logEmit(out, id, manifest, long);
            emitted += 1;
        }
        if (limit) |n| if (emitted >= n) return;

        // Push in reverse wire order so the LIFO stack visits the first P hash
        // first. Reversing only this node's appended segment preserves DFS when
        // older sibling work is already below it.
        const parent_start = pending.items.len;
        for (manifest.p) |pc| {
            var hashes = pc.hashes();
            while (hashes.next()) |parent| {
                const owned = try gpa.dupe(u8, parent);
                pending.append(gpa, owned) catch {
                    gpa.free(owned);
                    return error.OutOfMemory;
                };
            }
        }
        std.mem.reverse([]u8, pending.items[parent_start..]);
    }
}

/// #98/`plink` — walk the check-in DAG from DERIVED ADJACENCY instead of
/// re-parsing every manifest, which is what fossil does and what this walk
/// used to not do.
///
/// Fossil parses a manifest exactly ONCE, when it crosslinks it, and records
/// the edge in a persistent table:
///
///     CREATE TABLE plink(pid, cid, isprim, mtime, baseid, UNIQUE(pid,cid));
///     CREATE INDEX plink_i2 ON plink(cid,pid);
///
/// Every later traversal is an indexed query over that table -- `timeline.c`'s
/// ancestor walk is a recursive CTE reading `plink.cid ... WHERE plink.pid=dx.id`
/// and touches no manifest at all. The tell is fossil's parsed-manifest cache:
/// `#define MX_MANIFEST_CACHE 6`, six entries, and its comment says it is there
/// for `rebuild`. Nothing on fossil's read path re-parses manifests, so nothing
/// needed a cache big enough to make that fast.
///
/// zit's derived state holds the same relation -- `parents_by_child`, carrying
/// the `is_primary` flag `plink.isprim` carries -- and since #98 the zero-parse
/// image carries it too, so this walk reads no artifact content whatever.
///
/// WHAT THIS GIVES UP, stated plainly: the manifest walk verified every
/// reachable artifact's name hash and re-validated its card grammar, so a
/// corrupt frame or a wrong-ID blob surfaced as a `log` failure. This does not.
/// That is fossil's behaviour too -- validation happens once at crosslink, and
/// `fossil test-integrity` is the verb that re-checks a repository, exactly as
/// `zit verify` is here. Paying a full-history re-verification on every history
/// listing was the anomaly, not the guarantee.
///
/// Order is preserved exactly: depth-first, parents pushed in reverse wire
/// order so the LIFO stack visits the first P hash first, first visit winning
/// across a diamond. A parent the store never received is `phantom`, the same
/// line the manifest walk printed for it.
fn walkLogDerived(
    gpa: std.mem.Allocator,
    repo: *Repository,
    root_id: []const u8,
    out: *Io.Writer,
    limit: ?usize,
) LogWalkError!void {
    const root_rid = repo.lookup(root_id) orelse return error.MissingRoot;
    // A crosslinked check-in has an event row; a rid interned only because some
    // P card named it does not. That is the derive's record of "we actually
    // received this", and it is what distinguishes a check-in from a phantom
    // here -- NOT the view's own store, whose `held` answers about the identity
    // space rather than about artifact bytes.
    if (repo.state().mtimeOf(root_rid) == null) return error.MissingRoot;

    var pending: std.ArrayList(repostate.Rid) = .empty;
    defer pending.deinit(gpa);
    var visited: std.AutoHashMapUnmanaged(repostate.Rid, void) = .empty;
    defer visited.deinit(gpa);

    try pending.append(gpa, root_rid);
    var emitted: usize = 0;
    while (pending.pop()) |rid| {
        if (limit) |n| std.debug.assert(emitted < n);
        if ((try visited.getOrPut(gpa, rid)).found_existing) continue;

        // A rid interned from a P card that never arrived. The manifest walk
        // reached the same verdict by failing to read its content.
        if (repo.state().mtimeOf(rid) == null) {
            out.print("phantom {s}\n", .{repo.hashOf(rid)}) catch {};
            emitted += 1;
            if (limit) |n| if (emitted >= n) return;
            continue;
        }
        out.print("checkin {s}\n", .{repo.hashOf(rid)}) catch {};
        emitted += 1;
        if (limit) |n| if (emitted >= n) return;

        // Reverse only this node's appended segment, so the LIFO stack takes
        // the first P hash first while older sibling work stays below it.
        const parent_start = pending.items.len;
        var it = repo.state().parentsOf(rid);
        while (it.next()) |p| try pending.append(gpa, p);
        std.mem.reverse(repostate.Rid, pending.items[parent_start..]);
    }
}

/// PATH's presence and effective content hash inside one check-in's file set.
/// The hash is a 40/64-char lowercase-hex F-card token (§4.7.6), copied into a
/// fixed buffer so the comparison outlives `fileAt`'s duped card.
const PathContent = struct {
    present: bool,
    hash: [2 * digest_len]u8 = undefined,
    hash_len: usize = 0,
};

/// The effective content of PATH in the check-in named `id` (artifact §8
/// reconstruction via `Repository.fileAt` — the one general-purpose query, so
/// no CLI-side re-derivation off raw F cards, B-log-5). `present` is false when
/// `id` is not interned (a phantom the derived view never crosslinked) or PATH
/// is not in its effective set (never added, or a delta delete marker §4.7.6).
fn effectivePathContent(
    gpa: std.mem.Allocator,
    repo: *Repository,
    id: []const u8,
    path: []const u8,
) LogWalkError!PathContent {
    const rid = repo.lookup(id) orelse return .{ .present = false };
    const card = (repo.fileAt(gpa, rid, path) catch |e| return switch (e) {
        error.OutOfMemory => error.OutOfMemory,
        error.CannotRead => error.Storage,
        // NotManifest/BaselineUnresolvable/BaselineIsDelta/Corrupt/NotARepo all
        // mean the reachable check-in cannot be reconstructed — the same class
        // the unfiltered walk reports as a corrupt/invalid target.
        else => error.Corrupt,
    }) orelse return .{ .present = false };
    defer Repository.freeFileCard(gpa, card);
    // The iterator behind `fileAt` suppresses delete markers, so a returned
    // card always carries a hash (§4.7.6).
    const h = card.hash.?;
    std.debug.assert(h.len <= 2 * digest_len);
    var out: PathContent = .{ .present = true, .hash_len = h.len };
    @memcpy(out.hash[0..h.len], h);
    return out;
}

/// Whether the check-in `id` changes PATH per the B-log change definition: its
/// effective content (or presence) for PATH differs from its PRIMARY parent's
/// (§3.2 first-listed P-card parent). A root has no parent, so presence alone is
/// a change; a rename surfaces at both names as a presence flip (no old-name
/// following, the B-merge-22 delete+add posture). `manifest` is the already-
/// parsed check-in the walk holds.
fn logChangesPath(
    gpa: std.mem.Allocator,
    repo: *Repository,
    id: []const u8,
    manifest: *const artifact.Manifest,
    path: []const u8,
) LogWalkError!bool {
    const here = try effectivePathContent(gpa, repo, id, path);

    // §3.2: the primary parent is the first hash of the first P card.
    var primary: ?[]const u8 = null;
    for (manifest.p) |pc| {
        var hashes = pc.hashes();
        if (hashes.next()) |first| {
            primary = first;
            break;
        }
    }
    const parent = if (primary) |pid|
        try effectivePathContent(gpa, repo, pid, path)
    else
        PathContent{ .present = false };

    if (here.present != parent.present) return true;
    if (!here.present) return false;
    return !std.mem.eql(u8, here.hash[0..here.hash_len], parent.hash[0..parent.hash_len]);
}

/// `zit log CHECKIN PATH` — the PATH-restricted ancestry walk (B-log-2). Opens
/// the `Repository` facade for the §8 effective-set query, then reuses the exact
/// unfiltered DFS (`walkLog`) so order and line layout match B-log-1, gating each
/// `checkin` line on the change predicate. `root_dir` is borrowed (the caller
/// closes it after this returns).
fn runLogPath(
    gpa: std.mem.Allocator,
    io: std.Io,
    streams: cli.Streams,
    root_dir: std.Io.Dir,
    root_id: []const u8,
    path: []const u8,
    limit: ?usize,
    long: bool,
) cli.Exit {
    var repo: Repository = undefined;
    repo.open(gpa, io, root_dir, .{}) catch |e| return repoOpenFail(streams, "log", e);
    defer repo.deinit(gpa);

    const log = repo.logStore(gpa) catch |e| return repoOpenFail(streams, "log", e);

    walkLog(gpa, log, root_id, streams.out, .{ .repo = &repo, .path = path }, limit, long) catch |e| {
        switch (e) {
            error.MissingRoot => streams.err.print("zit log: check-in '{s}' not found\n", .{root_id}) catch {},
            error.InvalidRoot => streams.err.print("zit log: check-in '{s}' is not a valid manifest\n", .{root_id}) catch {},
            error.InvalidTarget => streams.err.writeAll("zit log: reachable check-in is not a valid manifest\n") catch {},
            error.Corrupt => streams.err.writeAll("zit log: cannot read repository: repository is corrupt\n") catch {},
            error.Storage => streams.err.writeAll("zit log: cannot read repository: input/output error\n") catch {},
            error.OutOfMemory => streams.err.writeAll("zit log: out of memory\n") catch {},
        }
        return .failure;
    };
    return .ok;
}

pub fn runLog(
    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.parseLogArgs(verb_args)) {
        .ok => |p| p,
        .unknown_option => |tok| return cli.usageError(streams, "log", .unknown_option, tok),
        .extra_argument => |tok| return cli.usageError(streams, "log", .extra_argument, tok),
        .bad_limit => |tok| return cli.usageError(streams, "log", .bad_limit, tok),
    };
    if (parsed.help) {
        streams.out.writeAll(log_help) catch {};
        return .ok;
    }
    const name = parsed.checkin orelse
        return cli.usageError(streams, "log", .no_checkin, "");

    const root = (cli.resolveRoot(arena, io, amb.repo_override) catch {
        streams.err.writeAll("zit log: out of memory\n") catch {};
        return .failure;
    }) orelse {
        streams.err.writeAll(not_a_repo_msg) catch {};
        return .failure;
    };
    var root_dir = root.dir;
    defer root_dir.close(io);

    // A full lowercase-hex ID walks exactly as it always has, without deriving
    // repo state. Anything else — a prefix, a branch, a tag, a reserved form —
    // goes through the §8.3 resolver `zit info` uses, then walks by the
    // resolved full ID.
    const root_id: []const u8 = if (artifact.validateName(name)) name else blk: {
        var repo: Repository = undefined;
        repo.open(gpa, io, root_dir, .{}) catch |e| return repoOpenFail(streams, "log", e);
        defer repo.deinit(gpa);
        const rid = repo.resolve(name, .checkin) catch |e| {
            switch (e) {
                error.NotFound => streams.err.print("zit log: check-in '{s}' not found\n", .{name}) catch {},
                error.Ambiguous => streams.err.print("zit log: name '{s}' is ambiguous\n", .{name}) catch {},
                error.OutOfMemory => streams.err.writeAll("zit log: out of memory\n") catch {},
            }
            return .failure;
        };
        break :blk arena.dupe(u8, repo.hashOf(rid)) catch {
            streams.err.writeAll("zit log: out of memory\n") catch {};
            return .failure;
        };
    };

    // With a PATH the walk is restricted to the check-ins that change it
    // (B-log-2), a route that needs the `Repository` facade for §8 effective-set
    // queries; without one the walk is byte-identical to the shipped behavior
    // (B-log-1), so the no-PATH path below is left exactly as it was.
    if (parsed.path) |path| return runLogPath(gpa, io, streams, root_dir, root_id, path, parsed.limit, parsed.long);

    // The default listing prints an ID per check-in and reads nothing else, so
    // it takes the derived adjacency and touches no artifact content -- see
    // `walkLogDerived` for why that is the fossil-shaped answer. `--long` needs
    // the D/U/C cards, which only the manifest carries (the derive keeps the
    // comment and mtime but not the user), so it stays on the reading walk.
    if (!parsed.long) {
        var repo: Repository = undefined;
        if (repo.open(gpa, io, root_dir, .{})) |_| {
            defer repo.deinit(gpa);
            if (repo.lookup(root_id) != null) {
                walkLogDerived(gpa, &repo, root_id, streams.out, parsed.limit) catch |e| {
                    switch (e) {
                        error.MissingRoot => {
                            streams.err.print("zit log: check-in '{s}' not found\n", .{root_id}) catch {};
                            return .failure;
                        },
                        error.OutOfMemory => {
                            streams.err.writeAll("zit log: out of memory\n") catch {};
                            return .failure;
                        },
                        else => return .failure,
                    }
                };
                return .ok;
            }
        } else |_| {}
        // No usable derived state (an unreadable or absent cache, or a root the
        // derive never saw): fall through to the reading walk, which needs no
        // derived state at all.
    }

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

    walkLog(gpa, &log, root_id, streams.out, null, parsed.limit, parsed.long) catch |e| {
        switch (e) {
            error.MissingRoot => streams.err.print("zit log: check-in '{s}' not found\n", .{root_id}) catch {},
            error.InvalidRoot => streams.err.print("zit log: check-in '{s}' is not a valid manifest\n", .{root_id}) catch {},
            error.InvalidTarget => streams.err.writeAll("zit log: reachable check-in is not a valid manifest\n") catch {},
            error.Corrupt => streams.err.writeAll("zit log: cannot read repository: repository is corrupt\n") catch {},
            error.Storage => streams.err.writeAll("zit log: cannot read repository: input/output error\n") catch {},
            error.OutOfMemory => streams.err.writeAll("zit log: out of memory\n") catch {},
        }
        return .failure;
    };
    return .ok;
}

/// `zit leaves [--all]` — list the open leaves of the check-in DAG (repo-state
/// §7.1/§7.4/§7.5). Opens the repository through `Repository` and renders each
/// leaf's full artifact ID and branch, mtime-descending (§7.5 SHOULD order),
/// marking the leaf that matches the current checkout with "*". `--all`
/// additionally lists closed leaves — §7.4: closed status is derived, so a
/// closed leaf stays a leaf-table member. Enumeration is distinct from name
/// resolution (§7.7): the listing is exactly the leaf table filtered by the
/// §7.4 open predicate. An unborn repository has no leaves: empty stdout, exit 0.
pub fn runLeaves(
    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.parseLeavesArgs(verb_args)) {
        .ok => |p| p,
        .unknown_option => |tok| return cli.usageError(streams, "leaves", .unknown_option, tok),
        .extra_argument => |tok| return cli.usageError(streams, "leaves", .extra_argument, tok),
    };
    if (parsed.help) {
        streams.out.writeAll(leaves_help) catch {};
        return .ok;
    }

    var repo: Repository = undefined;
    var root_dir: std.Io.Dir = undefined;
    const rc = openRepo(gpa, io, arena, streams, "leaves", amb.repo_override, parsed.no_cache, true, &repo, &root_dir);
    if (rc != .ok) return rc;
    defer root_dir.close(io);
    defer repo.deinit(gpa);

    // Resolve the current checkout's version to a rid so its leaf is marked
    // (§7.5). Absence/unknown/eventless leaves nothing marked — the mark is a
    // checkout convenience, never the source of the leaf set.
    const cur = repo.currentCheckoutRid();
    return renderLeaves(arena, &repo, parsed.all, cur, streams);
}

/// One row of the leaf listing, gathered before the §7.5 mtime-descending sort.
/// `hash` borrows the view's identity space (stable for the view's lifetime).
const LeafRow = struct {
    rid: repostate.Rid,
    mtime: repostate.Mtime,
    hash: []const u8,
    open: bool,
};

/// §7.5 order: modification time descending. Ties break by artifact ID ascending
/// so the listing is a deterministic total order (IDs are unique per check-in).
fn leafRowLess(_: void, a: LeafRow, b: LeafRow) bool {
    if (a.mtime != b.mtime) return a.mtime > b.mtime;
    return std.mem.order(u8, a.hash, b.hash) == .lt;
}

/// Gathers, sorts (§7.5), and prints the leaf listing to stdout. A row is
/// included when `all` is set or the leaf is open (§7.4 open predicate). Each
/// line is `<mark><id>  <branch>`, `<mark>` = "* " for the current-checkout leaf
/// else "  "; under `all` a closed leaf appends "  (closed)". Bounded by the
/// leaf-table size.
fn renderLeaves(
    arena: std.mem.Allocator,
    repo: *Repository,
    all: bool,
    cur: ?repostate.Rid,
    streams: cli.Streams,
) cli.Exit {
    var rows: std.ArrayList(LeafRow) = .empty;
    defer rows.deinit(arena);

    const bound = repo.state().leaves().rids.len;
    var scanned: usize = 0;
    var it = repo.state().leaves();
    while (it.next()) |rid| {
        scanned += 1;
        std.debug.assert(scanned <= bound);
        const open = repo.state().isOpenLeaf(rid);
        if (!all and !open) continue;
        // A leaf-table rid is always a crosslinked check-in, so it has an event.
        const mtime = repo.state().mtimeOf(rid) orelse unreachable;
        rows.append(arena, .{
            .rid = rid,
            .mtime = mtime,
            .hash = repo.hashOf(rid),
            .open = open,
        }) catch {
            streams.err.writeAll("zit leaves: out of memory\n") catch {};
            return .failure;
        };
    }
    std.debug.assert(rows.items.len <= bound);

    std.mem.sort(LeafRow, rows.items, {}, leafRowLess);

    for (rows.items) |row| {
        std.debug.assert(repo.state().isLeaf(row.rid));
        std.debug.assert(all or row.open);
        const mark = if (cur != null and cur.? == row.rid) "* " else "  ";
        const branch = repo.branchOf(row.rid);
        streams.out.print("{s}{s}  {s}", .{ mark, row.hash, branch }) catch {};
        if (all and !row.open) streams.out.writeAll("  (closed)") catch {};
        streams.out.writeAll("\n") catch {};
    }
    return .ok;
}

/// `zit timeline` — the global chronological history of every check-in
/// (repo-state §2/§8.3): one line per check-in, event.mtime descending, ties by
/// full artifact hash so the listing is byte-reproducible. Read-only: opens the
/// repository, enumerates through the single `checkinsByMtime` query, renders,
/// and mutates nothing. No positional, `-h`/`--help` only (the house
/// `parseNoPositionalArgs` grammar). (contracts/timeline.md B-timeline-1..13.)
pub fn runTimeline(
    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.parseTimelineArgs(verb_args)) {
        .ok => |p| p,
        .unknown_option => |tok| return cli.usageError(streams, "timeline", .unknown_option, tok),
        .extra_argument => |tok| return cli.usageError(streams, "timeline", .extra_argument, tok),
        .bad_limit => |tok| return cli.usageError(streams, "timeline", .bad_limit, tok),
    };
    // --help wins over every effect (B-timeline-8).
    if (parsed.help) {
        streams.out.writeAll(timeline_help) catch {};
        return .ok;
    }

    // A cold `.full` verb: it reads adjacency (the §3.2 merge mark) and content
    // (the C-card comment), so it never takes the zero-parse image path.
    var repo: Repository = undefined;
    var root_dir: std.Io.Dir = undefined;
    const rc = openRepo(gpa, io, arena, streams, "timeline", amb.repo_override, false, false, &repo, &root_dir);
    if (rc != .ok) return rc;
    defer root_dir.close(io);
    defer repo.deinit(gpa);
    return renderTimeline(gpa, &repo, streams, parsed);
}

/// §3.2: a check-in that joins two or more lines of development — one carrying
/// at least one merge parent (a second-or-later P-card hash), i.e. more than one
/// distinct parent. The mainline (primary-only) check-in has exactly one.
fn timelineIsMerge(state: *const repostate.RepoState, rid: repostate.Rid) bool {
    return state.parentCount(rid) >= 2;
}

/// Renders the timeline to stdout: for each check-in in `checkinsByMtime` order,
/// one line `<mark><hash12>  <ISO>  <branch>  <first-comment-line>`. `<mark>` is
/// "M " for a merge check-in (§3.2) else "  ". Bounded by the enumeration size;
/// each row reads and frees exactly one manifest (no unbounded retention). An
/// empty repository prints nothing (B-timeline-7).
fn renderTimeline(gpa: std.mem.Allocator, repo: *Repository, streams: cli.Streams, a: cli.TimelineArgs) cli.Exit {
    const rids = repo.state().checkinsByMtime(gpa) catch return stateOom(streams, "timeline");
    defer gpa.free(rids);
    // `-c`: the checkout's branch, which needs a checkout (zit#163).
    var current_branch_buf: []const u8 = "";
    if (a.current_branch) {
        const cur = repo.current_hint orelse {
            streams.err.writeAll("zit timeline: -c needs a checkout; name the branch with -b instead\n") catch {};
            return .failure;
        };
        current_branch_buf = repo.branchOf(cur);
    }
    const want_branch: ?[]const u8 = if (a.current_branch) current_branch_buf else a.branch;

    var last_day: [10]u8 = undefined;
    var have_day = false;
    var matched: usize = 0;
    var shown: usize = 0;
    var idx: usize = 0;
    while (idx < rids.len) : (idx += 1) {
        const rid = if (a.reverse) rids[rids.len - 1 - idx] else rids[idx];
        if (a.limit) |n| if (n != 0 and shown >= n) break;
        const mtime = repo.state().mtimeOf(rid) orelse unreachable;
        const branch = repo.branchOf(rid); // §6.1, config main-branch fallback
        if (want_branch) |b| if (!std.mem.eql(u8, b, branch)) continue;

        // The manifest is read when a filter or format needs its cards
        // (user, path, files); the oneline default keeps its memoized comment.
        const need_manifest = a.user != null or a.path != null or a.format != .oneline or a.verbose;
        var maybe: ?Repository.OwnedManifest = null;
        defer if (maybe) |*m| m.deinit(gpa);
        if (need_manifest or repo.state().commentOf(rid) == null) {
            maybe = repo.manifest(gpa, rid) catch |e| switch (e) {
                error.OutOfMemory => return stateOom(streams, "timeline"),
                error.NotARepo, error.CannotRead, error.Corrupt => return repoReadFail(streams, "timeline"),
            };
        }
        const user: []const u8 = if (maybe) |m| (if (m.manifest.u) |u| u.login else "") else "";
        if (a.user) |u| if (!std.mem.eql(u8, u, user)) continue;
        if (a.path) |path| {
            const m = maybe orelse continue;
            const changes = logChangesPath(gpa, repo, repo.hashOf(rid), &m.manifest, path) catch |e| switch (e) {
                error.OutOfMemory => return stateOom(streams, "timeline"),
                else => false,
            };
            if (!changes) continue;
        }
        matched += 1;
        if (a.offset) |off| if (matched <= off) continue;
        shown += 1;

        var iso_buf: [19]u8 = undefined;
        const iso = cli.formatWriterTimestamp(@intCast(mtime / 1000), &iso_buf);
        const mark = if (timelineIsMerge(repo.state(), rid)) "M " else "  ";
        const hash = repo.hashOf(rid);
        const short = hash[0..@min(hash.len, timeline_hash_display)];
        const manifest_comment: []const u8 = if (repo.state().commentOf(rid)) |c|
            firstCommentLine(c)
        else if (maybe) |m|
            (if (m.manifest.c) |c| firstCommentLine(c.comment) else "")
        else
            "";
        const comment = repo.override(rid, .comment, manifest_comment);
        const is_current = if (repo.current_hint) |cur| cur == rid else false;

        switch (a.format) {
            .oneline => {
                if (a.weekday) {
                    const day = zit.repostate.weekdayName(zit.repostate.weekdayOf(mtime));
                    streams.out.print("{s}{s}  {s} {s}  {s}  {s}\n", .{ mark, short, day, iso, branch, comment }) catch {};
                } else {
                    streams.out.print("{s}{s}  {s}  {s}  {s}\n", .{ mark, short, iso, branch, comment }) catch {};
                }
            },
            .medium, .full => {
                // Fossil's shape: a date header once per day, then
                // `hh:mm:ss [hash] *CURRENT* comment (user: u tags: ...)`.
                const day = iso[0..10];
                if (!have_day or !std.mem.eql(u8, &last_day, day)) {
                    @memcpy(&last_day, day);
                    have_day = true;
                    streams.out.print("=== {s} ===\n", .{day}) catch {};
                }
                streams.out.print("{s} [{s}] {s}{s}{s} (user: {s} tags: {s}", .{ iso[11..19], hash[0..@min(hash.len, 10)], if (is_current) "*CURRENT* " else "", if (mark[0] == 'M') "*MERGE* " else "", comment, user, branch }) catch {};
                const tags = repo.state().tagsOf(gpa, rid) catch return stateOom(streams, "timeline");
                defer gpa.free(tags);
                for (tags) |t| {
                    if (t.tagtype != .singleton or !std.mem.startsWith(u8, t.name, sym_prefix)) continue;
                    streams.out.print(", {s}", .{t.name[sym_prefix.len..]}) catch {};
                }
                streams.out.writeAll(")\n") catch {};
            },
        }
        if (a.format == .full or a.verbose) {
            if (maybe) |m| timelineFiles(gpa, repo, rid, &m.manifest, streams) catch return stateOom(streams, "timeline");
        }
    }
    return .ok;
}

/// The files a check-in changes against its primary parent (zit#163), one
/// indented line each in Fossil's words.
fn timelineFiles(gpa: std.mem.Allocator, repo: *Repository, rid: repostate.Rid, manifest: *const artifact.Manifest, streams: cli.Streams) error{OutOfMemory}!void {
    var primary: ?repostate.Rid = null;
    for (manifest.p) |pc| {
        var hashes = pc.hashes();
        if (hashes.next()) |first| {
            primary = repo.lookup(first);
            break;
        }
    }
    var here = (repo.files(gpa, rid) catch return) orelse return;
    defer here.deinit(gpa);
    var parent_list: ?Repository.FileList = if (primary) |p| (repo.files(gpa, p) catch null) else null;
    defer if (parent_list) |*pl| pl.deinit(gpa);
    var hit = here.iterator();
    var pit_opt: ?Repository.EffectiveFiles = if (parent_list) |*pl| pl.iterator() else null;
    var pcur: ?artifact.FCard = if (pit_opt) |*pit| pit.next() else null;
    var hcur: ?artifact.FCard = hit.next();
    while (hcur != null or pcur != null) {
        if (pcur == null or (hcur != null and std.mem.order(u8, hcur.?.filename, pcur.?.filename) == .lt)) {
            streams.out.print("   ADDED {s}\n", .{hcur.?.filename}) catch {};
            hcur = hit.next();
        } else if (hcur == null or std.mem.order(u8, hcur.?.filename, pcur.?.filename) == .gt) {
            streams.out.print("   DELETED {s}\n", .{pcur.?.filename}) catch {};
            pcur = pit_opt.?.next();
        } else {
            if (!std.mem.eql(u8, hcur.?.hash orelse "", pcur.?.hash orelse "")) {
                streams.out.print("   EDITED {s}\n", .{hcur.?.filename}) catch {};
            }
            hcur = hit.next();
            pcur = pit_opt.?.next();
        }
    }
}

pub fn runBranch(
    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.parseBranchArgs(verb_args)) {
        .ok => |a| a,
        .unknown_option => |tok| return cli.usageError(streams, "branch", .unknown_option, tok),
        .extra_argument => |tok| return cli.usageError(streams, "branch", .extra_argument, tok),
    };
    if (parsed.help) {
        streams.out.writeAll(branch_help) catch {};
        return .ok;
    }

    var repo: Repository = undefined;
    var root_dir: std.Io.Dir = undefined;
    const rc = openRepo(gpa, io, arena, streams, "branch", amb.repo_override, parsed.no_cache, true, &repo, &root_dir);
    if (rc != .ok) return rc;
    defer root_dir.close(io);
    defer repo.deinit(gpa);

    // §6.3: the branch set is derived by grouping active branch tags — never
    // stored, never re-hardcoded here.
    const entries = repo.state().branches(gpa) catch return stateOom(streams, "branch");
    defer gpa.free(entries);

    // §6.1/§6.4: mark the checked-out check-in's branch, resolved through the
    // active-tag filter and the config main-branch. `currentCheckoutRid`
    // guarantees `cur` bears an event, so `branchOf`'s precondition holds.
    const cur = repo.currentCheckoutRid();
    const current_name: ?[]const u8 =
        if (cur) |r| repo.branchOf(r) else null;

    switch (parsed.action) {
        .list => {
            // zit#164: -c keeps closed branches only; -t orders by most
            // recent change; -r reverses whichever order is in force.
            var rows: std.ArrayList(repostate.RepoState.BranchEntry) = .empty;
            defer rows.deinit(gpa);
            for (entries) |e| {
                if (parsed.closed_only and !e.closed) continue;
                rows.append(gpa, e) catch return stateOom(streams, "branch");
            }
            if (parsed.by_time) std.mem.sort(repostate.RepoState.BranchEntry, rows.items, {}, struct {
                fn newer(_: void, a: repostate.RepoState.BranchEntry, b: repostate.RepoState.BranchEntry) bool {
                    return a.mtime > b.mtime;
                }
            }.newer);
            if (parsed.reverse) std.mem.reverse(repostate.RepoState.BranchEntry, rows.items);
            renderBranchList(streams.out, rows.items, current_name);
            return .ok;
        },
        .current => {
            const name = current_name orelse {
                streams.err.writeAll("zit branch current: no checkout here\n") catch {};
                return .failure;
            };
            streams.out.print("{s}\n", .{name}) catch {};
            return .ok;
        },
        .info => {
            const name = parsed.name orelse return cli.usageError(streams, "branch", .no_name, "");
            for (entries) |e| {
                if (!std.mem.eql(u8, e.name, name)) continue;
                var iso_buf: [19]u8 = undefined;
                const iso = cli.formatWriterTimestamp(@intCast(e.mtime / 1000), &iso_buf);
                const root_rid = repo.resolve(name, .branch) catch null;
                streams.out.print("branch:    {s}{s}\ntip:       {s} {s}\ncheck-ins: {d}\n", .{ e.name, if (e.closed) " (closed)" else "", repo.hashOf(e.tip), iso, e.count }) catch {};
                if (root_rid) |r| streams.out.print("root:      {s}\n", .{repo.hashOf(r)}) catch {};
                return .ok;
            }
            streams.err.print("zit branch: no branch named '{s}'\n", .{name}) catch {};
            return .failure;
        },
        .close, .reopen, .hide, .unhide => {
            const name = parsed.name orelse return cli.usageError(streams, "branch", .no_name, "");
            const tip_rid = repo.resolve(name, .checkin) catch {
                streams.err.print("zit branch: no branch named '{s}'\n", .{name}) catch {};
                return .failure;
            };
            const target_hash = arena.dupe(u8, repo.hashOf(tip_rid)) catch return stateOom(streams, "branch");
            const raw_name: []const u8 = switch (parsed.action) {
                .close, .reopen => "closed",
                else => "hidden",
            };
            const sign: artifact.TSign = switch (parsed.action) {
                .close, .hide => .add,
                else => .cancel,
            };
            const tags = cli.composeTagCards(arena, target_hash, .{ .sign = sign, .name = raw_name, .raw = true, .value = null }) catch return stateOom(streams, "branch");
            var cname_buf: [64]u8 = undefined;
            const control_name = mintControlArtifact(gpa, io, root_dir, amb, tags, &cname_buf) catch |e|
                return branchLogFail(streams, e);
            streams.out.print("branch {s} {s} at {s} ({s})\n", .{ @tagName(parsed.action), name, target_hash[0..12], control_name }) catch {};
            return .ok;
        },
        .new => {
            // A new branch is a check-in on it (repo-state §6.2), so this is
            // the commit path with --branch and --allow-empty, from the
            // checkout it is on. A BASIS other than the checkout needs an
            // update first; saying so beats minting from the wrong parent.
            const name = parsed.name orelse return cli.usageError(streams, "branch", .no_name, "");
            if (parsed.basis) |b| {
                const want = repo.resolve(b, .checkin) catch {
                    streams.err.print("zit branch new: '{s}' is not a check-in\n", .{b}) catch {};
                    return .failure;
                };
                if (cur == null or cur.? != want) {
                    streams.err.print("zit branch new: the checkout is not at '{s}'; run 'zit update {s}' first\n", .{ b, b }) catch {};
                    return .failure;
                }
            }
            const msg = std.fmt.allocPrint(arena, "Create new branch named \"{s}\"", .{name}) catch return stateOom(streams, "branch");
            return @import("checkout.zig").runCommitIn(gpa, io, arena, &.{ "-m", msg, "--branch", name, "--allow-empty" }, streams, amb, null);
        },
    }
}

fn branchLogFail(s: cli.Streams, err: anyerror) cli.Exit {
    s.err.print("zit branch: cannot write the control artifact: {s}\n", .{@errorName(err)}) catch {};
    return .failure;
}

/// `zit tag list [--raw] [CHECKIN]` (alias `ls`), `zit tag add [--raw]
/// [--propagate] NAME CHECKIN [VALUE]`, and `zit tag cancel [--raw] NAME
/// CHECKIN`.
///
/// `list` is read-only enumeration over the crosslinked repo-view (repo-state
/// §5.1/§5.2): without `--raw`, only singleton `sym-` tags are shown, with the
/// prefix stripped, so branch machinery, raw properties, and inherited names
/// cannot masquerade as one-check-in release names. With `--raw`, every active
/// tag is shown with its stored prefix. The views are pure presentation over
/// the `tagNames`/`tagNamesOfType`/`tagsOf` derivations.
///
/// `add` and `cancel` are the write paths (REGRESSION B6): each mints one
/// control artifact carrying a single T card.
pub fn runTag(
    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.parseTagArgs(verb_args)) {
        .ok => |p| p,
        .unknown_option => |tok| return cli.usageError(streams, "tag", .unknown_option, tok),
        .extra_argument => |tok| return cli.usageError(streams, "tag", .extra_argument, tok),
        .bad_subcommand => |tok| return cli.usageError(streams, "tag", .bad_subcommand, tok),
        .no_name => return cli.usageError(streams, "tag", .no_name, ""),
        .no_checkin => return cli.usageError(streams, "tag", .no_checkin, ""),
        .option_not_for_subcommand => |tok| return cli.usageError(streams, "tag", .option_not_for_subcommand, tok),
    };
    if (parsed.help) {
        streams.out.writeAll(tag_help) catch {};
        return .ok;
    }
    // The mutating arms own their own repository lifetime: the read view must
    // be CLOSED before the read-write append, so they cannot share the listing
    // path's `defer repo.deinit`.
    if (parsed.sub != .list) return runTagWrite(gpa, io, arena, parsed, streams, amb);

    var repo: Repository = undefined;
    var root_dir: std.Io.Dir = undefined;
    const rc = openRepo(gpa, io, arena, streams, "tag", amb.repo_override, parsed.no_cache, true, &repo, &root_dir);
    if (rc != .ok) return rc;
    defer root_dir.close(io);
    defer repo.deinit(gpa);

    if (parsed.checkin) |name| return runTagCheckin(gpa, &repo, name, parsed.raw, streams);

    // §5.1/§2: plain output is the singleton symbolic release namespace;
    // --raw retains the complete active tag-name view.
    const names = if (parsed.raw)
        repo.state().tagNames(gpa)
    else
        repo.state().tagNamesOfType(gpa, .singleton);
    const owned = names catch return stateOom(streams, "tag");
    defer gpa.free(owned);
    renderTagNames(streams.out, owned, parsed.raw);
    return .ok;
}

/// The `zit tag list CHECKIN` arm: resolve CHECKIN through the §8.3 procedure
/// and list tags visible on it. Plain output is restricted to singleton
/// symbolic release names; `raw` exposes every active tag with its stored
/// prefix and value. An unresolvable or ambiguous name is a failure (exit 1,
/// diagnostic to stderr); a resolved check-in with no matching tags lists
/// nothing (empty stdout, exit 0).
fn runTagCheckin(
    gpa: std.mem.Allocator,
    repo: *Repository,
    name: []const u8,
    raw: bool,
    streams: cli.Streams,
) cli.Exit {
    const rid = repo.resolve(name, .checkin) catch |e| {
        switch (e) {
            error.NotFound => streams.err.print("zit tag: check-in '{s}' not found\n", .{name}) catch {},
            error.Ambiguous => streams.err.print("zit tag: check-in '{s}' is ambiguous\n", .{name}) catch {},
            // Settling an equal-mtime tie by ancestry allocates (`MtimeMaxPick`).
            error.OutOfMemory => return stateOom(streams, "tag"),
        }
        return .failure;
    };
    const tags = repo.state().tagsOf(gpa, rid) catch return stateOom(streams, "tag");
    defer gpa.free(tags);
    renderTagList(streams.out, tags, raw);
    return .ok;
}

/// Tag-name prefixes fossil reserves for its own crosslinking of wiki pages,
/// tickets, and technotes. A T card in one of these namespaces asserts that
/// its target IS such an artifact, so minting one from the tag verb would
/// forge that association — fossil refuses them here for the same reason, and
/// matching the refusal keeps a zit-written repository readable by fossil
/// without surprises.
const reserved_tag_prefixes = [_][]const u8{ "wiki-", "tkt-", "event-" };

/// The `zit tag add` / `zit tag cancel` arm (REGRESSION B6): mint one control
/// artifact carrying a single `T <sign><name> <hash> [value]` card.
///
/// Mirrors `amend`'s shape, which is the point — both resolve a check-in, then
/// compose a T run, then mint. Resolution first, over a read view that is closed
/// before the read-write append (commit's durability pattern); the resolved hash
/// is duped onto `arena` so it outlives that view. A failure to resolve mints
/// NOTHING: every diagnostic below happens before the single fsync.
///
/// Default sign is SINGLETON (`+`), so a symbolic name pins to exactly this
/// check-in and is not inherited by its descendants — see `composeTagCards`
/// for why that sign is the whole correctness argument, and why `--propagate`
/// belongs to the raw namespace.
fn runTagWrite(
    gpa: std.mem.Allocator,
    io: std.Io,
    arena: std.mem.Allocator,
    parsed: cli.TagArgs,
    streams: cli.Streams,
    amb: cli.Ambient,
) cli.Exit {
    const name = parsed.name.?; // the parser's `no_name` arm guarantees it
    const checkin = parsed.checkin.?; // likewise `no_checkin`
    // An empty NAME names nothing (repo-state §2), rejected before any store is
    // opened — the same early check `commit --branch` makes.
    if (name.len == 0) {
        streams.err.writeAll("zit tag: NAME must not be empty\n") catch {};
        return .failure;
    }
    // An empty VALUE cannot be written: §4.7.16's value is `[ SP value ]`, so
    // "present but empty" has no spelling on the wire. Omitting VALUE is how
    // you say valueless.
    if (parsed.value) |v| {
        if (v.len == 0) {
            streams.err.writeAll("zit tag: VALUE must not be empty; omit it for a valueless tag\n") catch {};
            return .failure;
        }
    }
    for (reserved_tag_prefixes) |prefix| {
        if (std.mem.startsWith(u8, name, prefix)) {
            streams.err.print(
                "zit tag: '{s}' is reserved for fossil's own crosslinking; pick a name outside it\n",
                .{prefix},
            ) catch {};
            return .failure;
        }
    }

    var repo: Repository = undefined;
    var root_dir: std.Io.Dir = undefined;
    const rc = openRepo(gpa, io, arena, streams, "tag", amb.repo_override, parsed.no_cache, true, &repo, &root_dir);
    if (rc != .ok) return rc;
    defer root_dir.close(io);

    var target_hash: []const u8 = undefined;
    {
        defer repo.deinit(gpa);
        const rid = repo.resolve(checkin, .checkin) catch |e| {
            switch (e) {
                error.NotFound => streams.err.print("zit tag: check-in '{s}' not found\n", .{checkin}) catch {},
                error.Ambiguous => streams.err.print("zit tag: check-in '{s}' is ambiguous\n", .{checkin}) catch {},
                // Settling an equal-mtime tie by ancestry allocates.
                error.OutOfMemory => return stateOom(streams, "tag"),
            }
            return .failure;
        };
        target_hash = arena.dupe(u8, repo.hashOf(rid)) catch return stateOom(streams, "tag");
    }
    // The mint targets a resolved check-in by its full §3.6 artifact id.
    std.debug.assert(artifact.validateName(target_hash));

    const sign: artifact.TSign = switch (parsed.sub) {
        .cancel => .cancel,
        .add => if (parsed.propagate) .propagate else .add,
        .list => unreachable, // the caller routes `.list` elsewhere
    };
    const tags = cli.composeTagCards(arena, target_hash, .{
        .sign = sign,
        .name = name,
        .raw = parsed.raw,
        .value = parsed.value,
    }) catch return stateOom(streams, "tag");
    var cname_buf: [64]u8 = undefined;
    const control_name = mintControlArtifact(gpa, io, root_dir, amb, tags, &cname_buf) catch |e|
        return tagLogFail(streams, e);

    // One confirmation line naming the control artifact, as `amend` reports its
    // mint. Every diagnostic already went to stderr.
    streams.out.print("tag {s}\n", .{control_name}) catch {};
    return .ok;
}

/// Renders a `tag add` mint failure. Exhaustive over the widest closed set
/// (`logstore.OpenError`); the narrower put/commit sets coerce into it.
fn tagLogFail(s: cli.Streams, err: logstore.OpenError) cli.Exit {
    switch (err) {
        error.OutOfMemory => s.err.writeAll("zit tag: out of memory\n") catch {},
        error.Storage => s.err.writeAll("zit tag: cannot write the tag: input/output error\n") catch {},
        error.Corrupt => s.err.writeAll("zit tag: the repository is corrupt\n") catch {},
    }
    return .failure;
}

/// §5.1/§2 no-arg listing: one tag name per line, ascending. By default only
/// symbolic ('sym-') tags are listed — they are the user-facing names, and are
/// surfaced with the prefix stripped. The machinery tags (`branch`, `closed`)
/// and other reserved prefixes are not listed as bare names: branch identity
/// is already present via its stripped 'sym-' tag.
///
/// `raw` lists every tag name verbatim, prefixes intact. It is the read half
/// of `tag add --raw`: a property tag is deliberately outside the name space,
/// so without this the tag you just wrote would be invisible here. Pure;
/// bounded by `names.len`.
fn renderTagNames(out: *std.Io.Writer, names: []const []const u8, raw: bool) void {
    for (names) |name| {
        if (raw) {
            out.writeAll(name) catch {};
        } else {
            if (!std.mem.startsWith(u8, name, sym_prefix)) continue;
            out.writeAll(name[sym_prefix.len..]) catch {};
        }
        out.writeAll("\n") catch {};
    }
}

/// §5.1/§5.2 CHECKIN listing: active tags in raw-name order, as `tagsOf`
/// returns them. Plain output includes only singleton `sym-` tags, stripped to
/// their user-facing names. Raw output includes every active tag with its
/// stored prefix intact. A tag carrying a value payload (§5.2) prints
/// `<name> <value>`; a valueless tag prints `<name>` alone. Pure; bounded by
/// `tags.len`.
fn renderTagList(
    out: *std.Io.Writer,
    tags: []const repostate.RepoState.TagEntry,
    raw: bool,
) void {
    for (tags) |tag| {
        if (!raw and (tag.tagtype != .singleton or !std.mem.startsWith(u8, tag.name, sym_prefix))) {
            continue;
        }
        const shown = if (raw) tag.name else tag.name[sym_prefix.len..];
        out.writeAll(shown) catch {};
        if (tag.value) |v| {
            out.writeAll(" ") catch {};
            out.writeAll(v) catch {};
        }
        out.writeAll("\n") catch {};
    }
}

/// `zit info NAME` — resolve NAME through the repo-state §8 procedure and
/// describe the check-in it names: full artifact ID, branch (§6.1), leaf
/// status (§7), and parents (§3.2/§3.3). A thin renderer over the shared
/// resolver: it calls `resolve(name, .checkin)` exactly as `tag` does, never a
/// second resolver. §8.5: resolution is recency-based (§8.2 max-mtime), so a
/// name can name a closed or non-leaf check-in that `leaves` would not list —
/// name resolution is not open-leaf enumeration. Read-only; joins the
/// leaves/branch/tag family through `openRepo`.
pub fn runInfo(
    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.parseInfoArgs(verb_args)) {
        .ok => |p| p,
        .unknown_option => |tok| return cli.usageError(streams, "info", .unknown_option, tok),
        .extra_argument => |tok| return cli.usageError(streams, "info", .extra_argument, tok),
    };
    if (parsed.help) {
        streams.out.writeAll(info_help) catch {};
        return .ok;
    }
    // NAME is required; the arg is a symbolic name, not a check-in ID.
    const name = parsed.name orelse return cli.usageError(streams, "info", .no_name, "");

    var repo: Repository = undefined;
    var root_dir: std.Io.Dir = undefined;
    const rc = openRepo(gpa, io, arena, streams, "info", amb.repo_override, parsed.no_cache, true, &repo, &root_dir);
    if (rc != .ok) return rc;
    defer root_dir.close(io);
    defer repo.deinit(gpa);

    // §8.3 full dispatch. §8.2's two MUST-NOTs: closed-ness and leaf-ness of a
    // candidate never filter the selection — that lives entirely in `resolve`.
    const rid = repo.resolve(name, .checkin) catch |e| {
        switch (e) {
            error.NotFound => streams.err.print("zit info: check-in '{s}' not found\n", .{name}) catch {},
            error.Ambiguous => streams.err.print("zit info: name '{s}' is ambiguous\n", .{name}) catch {},
            error.OutOfMemory => return stateOom(streams, "info"),
        }
        return .failure;
    };
    return renderInfo(gpa, arena, &repo, rid, streams);
}

/// Renders the resolved check-in's facts to stdout only: its full artifact ID,
/// branch (§6.1, config-threaded main-branch as fallback), leaf status
/// (§7.1 membership + §7.4 open/closed), the primary parent (§3.2), and each
/// merge parent (§3.3). Values left-align at a fixed column. Data only — every
/// diagnostic already went to stderr. Bounded by the check-in's parent count.
fn renderInfo(
    gpa: std.mem.Allocator,
    arena: std.mem.Allocator,
    repo: *Repository,
    rid: repostate.Rid,
    streams: cli.Streams,
) cli.Exit {
    // `resolve`'s postcondition: a resolved rid always bears an event, so every
    // derived query below has a defined answer — no failure path to invent.
    std.debug.assert(repo.state().mtimeOf(rid) != null);

    // Line 1: the canonical full artifact ID (echoes the resolved id even when
    // NAME was a prefix or a symbolic name).
    streams.out.print("{s}\n", .{repo.hashOf(rid)}) catch {};

    // §6.1: the active-tag branch value, config main-branch as the fallback —
    // never the literal "trunk".
    streams.out.print("branch:  {s}\n", .{repo.branchOf(rid)}) catch {};

    // §7.1/§7.4: open (a leaf with no active closed tag), closed (a leaf that
    // carries one), or no (not a leaf at all). open/closed is defined only over
    // leaves, so a non-leaf's closed tag — if any — is not surfaced here. The
    // three arms are total over {isLeaf, hasActiveClosed}.
    const leaf_word: []const u8 = if (repo.state().isOpenLeaf(rid))
        "open"
    else if (repo.state().isLeaf(rid))
        "closed"
    else
        "no";
    streams.out.print("leaf:    {s}\n", .{leaf_word}) catch {};

    // §3.2: the primary parent line (omitted for a root).
    const primary = repo.state().primaryParentOf(rid);
    if (primary) |p| streams.out.print("parent:  {s}\n", .{repo.hashOf(p)}) catch {};

    const mp = renderMergeParents(arena, repo, rid, primary, streams);
    if (mp != .ok) return mp;

    // §4.7.14 cherrypick/backout provenance (B-merge-41): a Q card on this
    // check-in surfaces as a `cherrypick:`/`backout:` line naming the picked
    // REV. Derived through the shipped artifact parse path (repo.manifest →
    // artifact.parse → parseQCard), never a bespoke re-parse; repostate keeps
    // no Q edge (REV is not a parent), so the manifest is the only source. A
    // check-in with no Q card prints nothing here — byte-identical to its
    // pre-B-merge-41 baseline (negative space). A parse fault skips the line
    // rather than failing the already-rendered facts. Bounded by the (v1:
    // single) Q-card count; the value column widens past the fixed lines
    // above because `cherrypick:` overflows their 8-space field.
    if (repo.manifest(gpa, rid) catch null) |owned| {
        var om = owned;
        defer om.deinit(gpa);
        for (om.manifest.q) |q| {
            const label: []const u8 = switch (q.sign) {
                .cherrypick => "cherrypick: ",
                .backout => "backout:    ",
            };
            streams.out.print("{s}{s}\n", .{ label, q.target }) catch {};
        }
    }

    // Display overrides (repo-state §5.2, minted by `zit amend`): emit a
    // `comment:`/`user:`/`date:` line for each override ONLY when it is active, so
    // an un-amended check-in's info output stays byte-identical to its B-info
    // baseline (negative space). This is the one render path that surfaces a
    // check-in's overridden user and date. Values align at the same column as the
    // fixed lines above (8-space field, so `comment:`+1, `user:`/`date:`+4).
    // Read through `displayOverride` so this works on either backing (#98). The
    // image carries these three resolved; `activeTagValue` would refuse here.
    for (std.enums.values(zit.repoimage.DisplayOverride)) |kind| {
        const name = zit.repoimage.override_names[@intFromEnum(kind)];
        std.debug.assert(name.len >= 1 and name.len <= 7);
        if (repo.state().displayOverride(rid, kind)) |v| {
            const pad = "        "; // 8 spaces; the value column lands at index 9
            streams.out.print("{s}:{s}{s}\n", .{ name, pad[0 .. 8 - name.len], v }) catch {};
        }
    }
    return .ok;
}

/// Renders the §3.3 merge parents of `rid` — every parent but the primary —
/// one `merge:` line each, sorted by artifact ID ascending. Gathered into an
/// arena vector bounded by the parent count; OOM → exit 1.
fn renderMergeParents(
    arena: std.mem.Allocator,
    repo: *Repository,
    rid: repostate.Rid,
    primary: ?repostate.Rid,
    streams: cli.Streams,
) cli.Exit {
    var merges: std.ArrayList([]const u8) = .empty;
    defer merges.deinit(arena);
    const bound = repo.state().parentCount(rid);
    var scanned: usize = 0;
    var it = repo.state().parentsOf(rid);
    while (it.next()) |parent| {
        scanned += 1;
        std.debug.assert(scanned <= bound);
        if (primary != null and parent == primary.?) continue;
        merges.append(arena, repo.hashOf(parent)) catch {
            streams.err.writeAll("zit info: out of memory\n") catch {};
            return .failure;
        };
    }
    // Negative space: exactly the parents minus the single §3.2 primary — no
    // parent dropped or doubled.
    std.debug.assert(merges.items.len == bound - @as(usize, if (primary == null) 0 else 1));
    std.mem.sort([]const u8, merges.items, {}, lessByHash);
    for (merges.items) |hash| streams.out.print("merge:   {s}\n", .{hash}) catch {};
    return .ok;
}

/// `zit ls [VERSION]` — list the effective file set of a resolved check-in, one
/// `<blob-id> <filename>` line per file in canonical §4.5.3 order. Thin over the
/// `Repository` facade: resolve VERSION (or fall back to the current checkout),
/// then stream `repo.files`' iterator. All the §8.2 merge and §8.3 baseline
/// policy lives in the library; this only selects the check-in and renders.
pub fn runLs(
    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.parseLsArgs(verb_args)) {
        .ok => |p| p,
        .unknown_option => |tok| return cli.usageError(streams, "ls", .unknown_option, tok),
        .extra_argument => |tok| return cli.usageError(streams, "ls", .extra_argument, tok),
    };
    if (parsed.help) {
        streams.out.writeAll(ls_help) catch {};
        return .ok;
    }

    var repo: Repository = undefined;
    var root_dir: std.Io.Dir = undefined;
    const rc = openRepo(gpa, io, arena, streams, "ls", amb.repo_override, parsed.no_cache, true, &repo, &root_dir);
    if (rc != .ok) return rc;
    defer root_dir.close(io);
    defer repo.deinit(gpa);

    // An explicit VERSION resolves through the §8.3 name procedure; an omitted
    // VERSION lists the current checkout (§7.5). An unborn or absent checkout has
    // no check-in to list — a failure, not empty output.
    const rid = if (parsed.version) |name|
        repo.resolve(name, .checkin) catch |e| return lsResolveFail(streams, name, e)
    else
        repo.currentCheckoutRid() orelse return lsNoCheckout(streams);

    var list = (repo.files(gpa, rid) catch |e| return lsFilesFail(streams, e)) orelse
        return lsMissingContent(streams);
    defer list.deinit(gpa);

    // #96 — `--long` needs each blob's bytes for size and binary class, so it
    // opens the store; the default listing does not and must not start paying
    // for one. The memo is #84's: consecutive files in a tree routinely share
    // delta bases, so reconstructing without it re-resolves the same base once
    // per file.
    var log: ?*logstore.LogStore = null;
    var plaintext: logstore.PlaintextCache = .{};
    defer plaintext.deinit(gpa);
    if (parsed.long or parsed.size) {
        log = repo.logStore(gpa) catch |e| return repoOpenFail(streams, "ls", e);
    }

    // Data only: one line per effective file, in canonical filename order.
    var it = list.iterator();
    while (it.next()) |f| {
        std.debug.assert(f.hash != null); // the iterator suppresses delete markers
        if (parsed.under) |prefix| {
            if (!lsUnder(f.filename, prefix)) continue;
        }
        if (!parsed.long and !parsed.size) {
            streams.out.print("{s} {s}\n", .{ f.hash.?, f.filename }) catch {};
            continue;
        }
        // The permission token is the one fact a consumer cannot recover by
        // spending more: it lives in the MANIFEST's F card, and the blob it
        // names does not know it is executable or a symlink.
        const perm = if (f.perm) |p| p else "-";

        // `--size` reads the length out of the frame header instead of
        // reconstructing the blob to measure it. `absent` covers the same three
        // conditions `--long` reports it for — a sparse elision, a shun, or a
        // phantom — and a size of 0 there means nothing was measured, not that
        // the file is empty.
        if (parsed.size) {
            if (lsBlobSize(gpa, log.?, f.hash.?)) |n| {
                streams.out.print("{s} {d} {s} {s}\n", .{ f.hash.?, n, perm, f.filename }) catch {};
            } else {
                streams.out.print("{s} absent {s} {s}\n", .{ f.hash.?, perm, f.filename }) catch {};
            }
            continue;
        }
        var size: u64 = 0;
        var class: []const u8 = "text";
        if (lsBlobBytes(gpa, log.?, &plaintext, f.hash.?)) |bytes| {
            defer gpa.free(bytes);
            size = bytes.len;
            if (zit.textdiff.isBinary(bytes)) class = "binary";
        } else {
            // Elided by a sparse store, shunned, or a phantom. Say so rather
            // than reporting a zero-byte text file, which is a lie that looks
            // like data.
            class = "absent";
        }
        streams.out.print("{s} {d} {s} {s} {s}\n", .{ f.hash.?, size, perm, class, f.filename }) catch {};
    }
    return .ok;
}

/// Whether `path` is at or beneath `prefix`, on a path-segment boundary so
/// `--under=src` never matches `src-other`. A prefix naming the file itself
/// matches, which is how a consumer asks about one entry.
fn lsUnder(path: []const u8, prefix: []const u8) bool {
    const p = std.mem.trimEnd(u8, prefix, "/");
    if (p.len == 0) return true; // `--under=/` is the whole tree
    if (!std.mem.startsWith(u8, path, p)) return false;
    if (path.len == p.len) return true;
    return path[p.len] == '/';
}

/// One blob's plaintext, or null when the store cannot produce it (a sparse
/// elision, a shun, or a phantom). Never fails the listing: a file whose bytes
/// are unavailable is reported as such, because `ls` answers about the
/// MANIFEST, and the manifest is intact whatever the store holds.
fn lsBlobBytes(
    gpa: std.mem.Allocator,
    log: *logstore.LogStore,
    cache: *logstore.PlaintextCache,
    hash: []const u8,
) ?[]u8 {
    // The F card carries the 64-hex artifact NAME (§4.7.1 wire form); the
    // store's index is keyed by the binary digest, so the hex resolver is the
    // right door. A full name can never be ambiguous, but the error is handled
    // rather than asserted — a listing must not crash on a damaged store.
    const rid = (log.resolveHexPrefix(hash) catch return null) orelse return null;
    return (log.contentCached(gpa, rid, cache) catch return null) orelse null;
}

/// One blob's plaintext SIZE, without reconstructing it (#96), or null when
/// the store cannot produce it. Same door and same null cases as
/// `lsBlobBytes`; the difference is that this never resolves a delta chain,
/// because every frame kind already carries its plaintext length.
fn lsBlobSize(gpa: std.mem.Allocator, log: *logstore.LogStore, hash: []const u8) ?u64 {
    const rid = (log.resolveHexPrefix(hash) catch return null) orelse return null;
    return (log.plaintextSize(gpa, rid) catch return null) orelse null;
}

/// Maps a `Repository.resolve` failure for `ls` to its diagnostic + exit 1.
fn lsResolveFail(s: cli.Streams, name: []const u8, e: Repository.ResolveError) cli.Exit {
    switch (e) {
        error.NotFound => s.err.print("zit ls: check-in '{s}' not found\n", .{name}) catch {},
        error.Ambiguous => s.err.print("zit ls: name '{s}' is ambiguous\n", .{name}) catch {},
        error.OutOfMemory => s.err.writeAll("zit ls: out of memory\n") catch {},
    }
    return .failure;
}

/// Maps a `Repository.files` failure to its diagnostic + exit 1. Closed over
/// `FilesError` so the switch stays exhaustive; the open/content phrasings match
/// the shared `repoOpenFail` text.
fn lsFilesFail(s: cli.Streams, e: Repository.FilesError) cli.Exit {
    switch (e) {
        error.OutOfMemory => s.err.writeAll("zit ls: out of memory\n") catch {},
        error.NotARepo => s.err.writeAll(not_a_repo_msg) catch {},
        error.CannotRead => s.err.writeAll("zit ls: cannot read repository: input/output error\n") catch {},
        error.Corrupt => s.err.writeAll("zit ls: cannot read repository: repository is corrupt\n") catch {},
        error.NotManifest => s.err.writeAll("zit ls: not a check-in\n") catch {},
        error.BaselineUnresolvable => s.err.writeAll("zit ls: cannot resolve the baseline of a delta check-in\n") catch {},
        error.BaselineIsDelta => s.err.writeAll("zit ls: the baseline of a delta check-in is itself a delta\n") catch {},
    }
    return .failure;
}

fn lsNoCheckout(s: cli.Streams) cli.Exit {
    s.err.writeAll("zit ls: no check-in to list (the working checkout is unborn); name a version\n") catch {};
    return .failure;
}

fn lsMissingContent(s: cli.Streams) cli.Exit {
    s.err.writeAll("zit ls: check-in content is missing\n") catch {};
    return .failure;
}

/// `zit cat FILE [VERSION]` — emit one file's uncompressed content at a resolved
/// check-in. `ls` narrowed to a single file (§8.4 single-name lookup): resolve
/// the version the same way `ls` does, look the file up in the effective set,
/// reconstruct its blob (delta chain → uncompressed, hash-verified §7.2), and
/// write the bytes verbatim — no framing, no trailing newline (content is
/// `[]const u8`, so embedded NUL and arbitrary binary bytes pass through).
pub fn runCat(
    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.parseCatArgs(verb_args)) {
        .ok => |p| p,
        .unknown_option => |tok| return cli.usageError(streams, "cat", .unknown_option, tok),
        .extra_argument => |tok| return cli.usageError(streams, "cat", .extra_argument, tok),
    };
    if (parsed.help) {
        streams.out.writeAll(cat_help) catch {};
        return .ok;
    }
    const file = parsed.file orelse return cli.usageError(streams, "cat", .no_file, "");

    var repo: Repository = undefined;
    var root_dir: std.Io.Dir = undefined;
    const rc = openRepo(gpa, io, arena, streams, "cat", amb.repo_override, parsed.no_cache, true, &repo, &root_dir);
    if (rc != .ok) return rc;
    defer root_dir.close(io);
    defer repo.deinit(gpa);

    // Version resolution mirrors `ls`: an explicit VERSION through the §8.3 name
    // procedure, an omitted one the current checkout (§7.5); an unborn/absent
    // checkout has no check-in to read.
    const rid = if (parsed.version) |name|
        repo.resolve(name, .checkin) catch |e| return catResolveFail(streams, name, e)
    else
        repo.currentCheckoutRid() orelse return catNoCheckout(streams);

    // §8.4 single-name lookup: null ⇒ the file is not in this check-in's
    // effective set (absent, or a delete marker suppresses it).
    const found = (repo.fileAt(gpa, rid, file) catch |e| return catFail(streams, e)) orelse
        return catNotFound(streams, file);
    defer Repository.freeFileCard(gpa, found);
    std.debug.assert(found.hash != null); // fileAt yields only hash-bearing cards (§4.7.6)

    // Reconstruct the blob and verify it hashes to its F-card id (§7.2). A null
    // blob is a stored-content gap for a named-but-absent file.
    const bytes = (repo.blob(gpa, found.hash.?) catch |e| return catFail(streams, e)) orelse
        return catMissingContent(streams, file);
    defer gpa.free(bytes);
    streams.out.writeAll(bytes) catch {};
    return .ok;
}

/// Maps a `Repository.resolve` failure for `cat` to its diagnostic + exit 1.
fn catResolveFail(s: cli.Streams, name: []const u8, e: Repository.ResolveError) cli.Exit {
    switch (e) {
        error.NotFound => s.err.print("zit cat: check-in '{s}' not found\n", .{name}) catch {},
        error.Ambiguous => s.err.print("zit cat: name '{s}' is ambiguous\n", .{name}) catch {},
        error.OutOfMemory => s.err.writeAll("zit cat: out of memory\n") catch {},
    }
    return .failure;
}

/// Maps a `Repository.fileAt` / `blob` failure to its diagnostic + exit 1.
/// Switches the full `FilesError` set exhaustively; `blob`'s narrower `OpenError`
/// coerces into it, so one renderer covers both surfaces. The open/content
/// phrasings match the shared `repoOpenFail` text.
fn catFail(s: cli.Streams, e: Repository.FilesError) cli.Exit {
    switch (e) {
        error.OutOfMemory => s.err.writeAll("zit cat: out of memory\n") catch {},
        error.NotARepo => s.err.writeAll(not_a_repo_msg) catch {},
        error.CannotRead => s.err.writeAll("zit cat: cannot read repository: input/output error\n") catch {},
        error.Corrupt => s.err.writeAll("zit cat: cannot read repository: repository is corrupt\n") catch {},
        error.NotManifest => s.err.writeAll("zit cat: not a check-in\n") catch {},
        error.BaselineUnresolvable => s.err.writeAll("zit cat: cannot resolve the baseline of a delta check-in\n") catch {},
        error.BaselineIsDelta => s.err.writeAll("zit cat: the baseline of a delta check-in is itself a delta\n") catch {},
    }
    return .failure;
}

fn catNoCheckout(s: cli.Streams) cli.Exit {
    s.err.writeAll("zit cat: no check-in to read (the working checkout is unborn); name a version\n") catch {};
    return .failure;
}

fn catNotFound(s: cli.Streams, file: []const u8) cli.Exit {
    s.err.print("zit cat: file '{s}' is not in the check-in\n", .{file}) catch {};
    return .failure;
}

fn catMissingContent(s: cli.Streams, file: []const u8) cli.Exit {
    s.err.print("zit cat: content for '{s}' is missing\n", .{file}) catch {};
    return .failure;
}

/// `zit artifact HASH [--raw]` — zit's plumbing window (contracts/artifact.md).
/// A strictly read-only describe/dump of ONE stored artifact addressed by its
/// hash. HASH covers every artifact kind (file blobs included), so it resolves
/// through the artifact store's prefix lookup, not the symbolic-name procedure
/// `info`/`ls`/`cat` use. No flag → the whatis summary; `--raw` → the exact
/// stored bytes. All card interpretation flows through the one `artifact`
/// parser — never a CLI-side re-parse.
pub fn runArtifact(
    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.parseArtifactArgs(verb_args)) {
        .ok => |p| p,
        .unknown_option => |tok| return cli.usageError(streams, "artifact", .unknown_option, tok),
        .extra_argument => |tok| return cli.usageError(streams, "artifact", .extra_argument, tok),
    };
    if (parsed.help) {
        streams.out.writeAll(artifact_help) catch {};
        return .ok;
    }
    const hash_arg = parsed.hash orelse return cli.usageError(streams, "artifact", .no_hash, "");

    var repo: Repository = undefined;
    var root_dir: std.Io.Dir = undefined;
    // The image path (#141): everything below reads the CONTENT LOG and
    // nothing else -- `resolveArtifactPrefix` and `blob` both go through
    // `logStore`, and no query here touches the derived view. A `.full` derive
    // was work whose product this verb never read.
    const rc = openRepo(gpa, io, arena, streams, "artifact", amb.repo_override, false, true, &repo, &root_dir);
    if (rc != .ok) return rc;
    defer root_dir.close(io);
    defer repo.deinit(gpa);

    // Canonicalize HASH: lowercase, hex-only, length band 4..64 (§3.2/§3.6). A
    // syntactically invalid HASH is the SAME not-found family as an unknown one
    // (B-artifact-8) — no distinct error surface.
    var canon_buf: [2 * logstore.max_hash_len]u8 = undefined;
    const canon = canonicalizeHexPrefix(hash_arg, &canon_buf) orelse return artifactNotFound(streams, hash_arg);

    var full_buf: [2 * logstore.max_hash_len]u8 = undefined;
    const full = (repo.resolveArtifactPrefix(gpa, canon, &full_buf) catch |e| switch (e) {
        error.Ambiguous => return artifactAmbiguous(streams, hash_arg), // B-artifact-7
        error.OutOfMemory => return artifactOom(streams),
        error.NotARepo, error.CannotRead, error.Corrupt => |oe| return repoOpenFail(streams, "artifact", oe),
    }) orelse return artifactNotFound(streams, hash_arg); // B-artifact-8

    // §7.2-verified stored bytes. `resolveArtifactPrefix` named a real stored
    // artifact, so a null here would be a store race, not user input.
    const bytes = (repo.blob(gpa, full) catch |e| return repoOpenFail(streams, "artifact", e)) orelse
        return artifactNotFound(streams, hash_arg);
    defer gpa.free(bytes);
    // B-artifact-4 round-trip invariant: the emitted bytes hash back to HASH.
    std.debug.assert(artifact.verify(full, bytes));

    if (parsed.raw) {
        // B-artifact-4: verbatim, no trailing newline, binary-safe (mirrors cat).
        // Never parses or verifies Z.
        streams.out.writeAll(bytes) catch {};
        return .ok;
    }
    return renderArtifactSummary(streams, gpa, full, bytes);
}

/// Canonicalize a HASH argument into `out`: lowercase every byte, accept only
/// hex digits and only a 4..64-length token (§3.2 length band, §3.6 case-
/// insensitive prefix). Returns null for anything outside that — the caller
/// folds null into the not-found refusal, so a bad HASH never has its own
/// error surface (B-artifact-8). No allocation.
fn canonicalizeHexPrefix(arg: []const u8, out: *[2 * logstore.max_hash_len]u8) ?[]const u8 {
    if (arg.len < 4 or arg.len > out.len) return null;
    for (arg, 0..) |c, i| {
        const lc = std.ascii.toLower(c);
        if (!((lc >= '0' and lc <= '9') or (lc >= 'a' and lc <= 'f'))) return null;
        out[i] = lc;
    }
    return out[0..arg.len];
}

/// Render the whatis summary of a resolved artifact to stdout (data-only). Tries
/// the Z-tolerant `parseWindow`: on success the artifact is structural (its type,
/// size, one line per card, and the §6.2 Z verdict); on any framing/grammar/
/// classification fault it is an opaque blob (B-artifact-3) — type file, size,
/// and the NUL binary verdict, never an error. Exit 0 either way.
fn renderArtifactSummary(streams: cli.Streams, gpa: std.mem.Allocator, full: []const u8, bytes: []const u8) cli.Exit {
    streams.out.print("{s}\n", .{full}) catch {};

    var m = artifact.parseWindow(gpa, bytes) catch |e| switch (e) {
        error.OutOfMemory => return artifactOom(streams),
        else => {
            // A blob is a first-class answer (B-artifact-3), not a failure.
            streams.out.writeAll("type: file\n") catch {};
            streams.out.print("size: {d} bytes\n", .{bytes.len}) catch {};
            streams.out.print("binary: {s}\n", .{if (merge3.isBinary(bytes)) "yes" else "no"}) catch {};
            return .ok;
        },
    };
    defer m.deinit();

    streams.out.print("type: {s}\n", .{m.kind.diagnosticName()}) catch {};
    streams.out.print("size: {d} bytes\n", .{bytes.len}) catch {};
    // #72 — PRESENCE, never validity: reporting "valid" would mean shelling
    // to gpg from a read verb and inheriting its keyring's answer, which
    // makes an offline or key-less reader print alarming things about
    // artifacts that are fine. Presence is what the store knows; validity is
    // the operator's own `gpg --verify` on `artifact --raw` output.
    if (artifact.isSignatureWrapped(bytes)) {
        streams.out.writeAll("signature: clearsigned (cards read from the stripped body)\n") catch {};
    }
    renderCardLines(streams, m);

    // B-artifact-5: the Z verdict is REPORTED, not enforced — a mismatch stays
    // exit 0 (this is a window, not a gate). Emitted only when a Z card is
    // present; `parseWindow` succeeded so the framed content is recoverable.
    if (m.z != null) {
        const content = artifact.frameNoZ(bytes).?; // non-null: parseWindow framed it
        const verdict = artifact.verifyZCard(content);
        std.debug.assert(verdict != .absent); // a present Z card is structurally last
        streams.out.print("Z: {s}\n", .{if (verdict == .match) "verified" else "checksum mismatch"}) catch {};
    }
    return .ok;
}

/// One line per card of a structural artifact, in ascending card-letter order
/// (A..W; Z is the separate verdict line the caller prints). §4.4 makes wire
/// order non-decreasing, so lane order == stored order — this reads decoded
/// struct fields, never re-walking the bytes. Repeatable lanes (F/J/M/P/Q/T)
/// render in retained wire order, one line per card. Best-effort writes.
fn renderCardLines(s: cli.Streams, m: artifact.Manifest) void {
    if (m.a) |a| {
        s.out.print("A {s} {s}", .{ a.name, a.target }) catch {};
        if (a.source) |src| s.out.print(" {s}", .{src}) catch {};
        s.out.writeAll("\n") catch {};
    }
    if (m.b) |b| s.out.print("B {s}\n", .{b.hash}) catch {};
    if (m.c) |c| s.out.print("C {s}\n", .{c.comment}) catch {};
    if (m.d) |d| s.out.print("D {s}\n", .{d.timestamp}) catch {};
    if (m.e) |e| s.out.print("E {s} {s}\n", .{ e.timestamp, e.technote_id }) catch {};
    for (m.f) |f| {
        s.out.print("F {s}", .{f.filename}) catch {};
        if (f.hash) |h| s.out.print(" {s}", .{h}) catch {};
        if (f.perm) |p| s.out.print(" {s}", .{p}) catch {};
        if (f.old_name) |o| s.out.print(" {s}", .{o}) catch {};
        s.out.writeAll("\n") catch {};
    }
    if (m.g) |g| s.out.print("G {s}\n", .{g.hash}) catch {};
    if (m.h) |h| s.out.print("H {s}\n", .{h.title}) catch {};
    if (m.i) |i| s.out.print("I {s}\n", .{i.hash}) catch {};
    for (m.j) |j| s.out.print("J {s} {s}\n", .{ j.field, j.value }) catch {};
    if (m.k) |k| s.out.print("K {s}\n", .{k.ticket_id}) catch {};
    if (m.l) |l| s.out.print("L {s}\n", .{l.title}) catch {};
    for (m.m) |mc| s.out.print("M {s}\n", .{mc.hash}) catch {};
    if (m.n) |n| s.out.print("N {s}\n", .{n.mimetype}) catch {};
    for (m.p) |p| {
        s.out.writeAll("P") catch {};
        var it = p.hashes();
        while (it.next()) |hh| s.out.print(" {s}", .{hh}) catch {};
        s.out.writeAll("\n") catch {};
    }
    for (m.q) |q| {
        const sign_byte: u8 = if (q.sign == .cherrypick) '+' else '-';
        s.out.print("Q {c}{s}", .{ sign_byte, q.target }) catch {};
        if (q.baseline) |base| s.out.print(" {s}", .{base}) catch {};
        s.out.writeAll("\n") catch {};
    }
    if (m.r) |r| s.out.print("R {s}\n", .{r.checksum}) catch {};
    for (m.t) |t| {
        const tgt = t.target orelse "*"; // §4.7.16 self-referential tag
        s.out.print("T {c}{s} {s}", .{ t.sign.byte(), t.name, tgt }) catch {};
        if (t.value) |v| s.out.print(" {s}", .{v}) catch {};
        s.out.writeAll("\n") catch {};
    }
    if (m.u) |u| s.out.print("U {s}\n", .{u.login}) catch {};
    if (m.w) |w| s.out.print("W {d} bytes\n", .{w.len}) catch {};
}

fn artifactNotFound(s: cli.Streams, arg: []const u8) cli.Exit {
    s.err.print("zit artifact: no artifact matches '{s}'\n", .{arg}) catch {};
    return .failure;
}

fn artifactAmbiguous(s: cli.Streams, arg: []const u8) cli.Exit {
    s.err.print("zit artifact: hash '{s}' is ambiguous\n", .{arg}) catch {};
    return .failure;
}

fn artifactOom(s: cli.Streams) cli.Exit {
    s.err.writeAll("zit artifact: out of memory\n") catch {};
    return .failure;
}

/// The narrow read-failure set the annotate content path collapses the
/// `Repository` file/blob errors into, so the renderer switches once.
const AnnotateReadError = error{ OutOfMemory, Storage, Corrupt };

/// PATH's effective content in the check-in `rid` (artifact §8 via
/// `Repository.fileAt` + `blob`), duplicated into `arena`, or null when PATH is
/// absent there (never added, or a delta delete). Transient `gpa` allocations
/// (the F card, the verified blob) are freed before returning; only the arena
/// copy survives, owned by the caller's arena.
fn annotateContentAt(
    gpa: std.mem.Allocator,
    arena: std.mem.Allocator,
    repo: *Repository,
    rid: repostate.Rid,
    path: []const u8,
) AnnotateReadError!?[]const u8 {
    const maybe_card = repo.fileAt(gpa, rid, path) catch |e| return switch (e) {
        error.OutOfMemory => error.OutOfMemory,
        error.CannotRead => error.Storage,
        // NotARepo/Corrupt/NotManifest/BaselineUnresolvable/BaselineIsDelta all
        // mean this reachable check-in cannot be reconstructed — corrupt.
        else => error.Corrupt,
    };
    const card = maybe_card orelse return null;
    defer Repository.freeFileCard(gpa, card);
    // The effective-set iterator suppresses delete markers, so a returned card
    // always carries a hash (§4.7.6).
    const bytes = (repo.blob(gpa, card.hash.?) catch |e| return switch (e) {
        error.OutOfMemory => error.OutOfMemory,
        error.CannotRead => error.Storage,
        error.NotARepo, error.Corrupt => error.Corrupt,
    }) orelse return null;
    defer gpa.free(bytes);
    return try arena.dupe(u8, bytes);
}

/// One check-in in PATH's ancestry, in the processing order the repostate
/// topological linearization fixes (parent before child; B-annotate-5). `hash`
/// borrows the view's identity space (stable for the view's lifetime).
const AnnotateEntry = struct {
    rid: repostate.Rid,
    mtime: repostate.Mtime,
    hash: []const u8,
};

fn annotateReadFail(streams: cli.Streams, e: AnnotateReadError) cli.Exit {
    switch (e) {
        error.OutOfMemory => streams.err.writeAll("zit annotate: out of memory\n") catch {},
        error.Storage => streams.err.writeAll("zit annotate: cannot read repository: input/output error\n") catch {},
        error.Corrupt => streams.err.writeAll("zit annotate: cannot read repository: repository is corrupt\n") catch {},
    }
    return .failure;
}

/// The B-annotate-10 refusal: PATH is not in the current check-out's managed set.
/// Distinguishes an untracked-but-present working file from a nonexistent one so
/// the diagnostic tells the user which mistake they made. Exit 1, stdout empty.
fn annotateNotTracked(io: std.Io, root_dir: std.Io.Dir, path: []const u8, streams: cli.Streams) cli.Exit {
    const present = blk: {
        var f = root_dir.openFile(io, path, .{}) catch break :blk false;
        f.close(io);
        break :blk true;
    };
    if (present)
        streams.err.print("zit annotate: '{s}' is not tracked at the current check-out\n", .{path}) catch {}
    else
        streams.err.print("zit annotate: '{s}': no such tracked file\n", .{path}) catch {};
    return .failure;
}

/// `zit annotate PATH` — parse (one repo-relative positional via the shared
/// one-positional grammar), open the repository read-only, and render the
/// annotation. `-h`/`--help` wins; a missing PATH / unknown option / second
/// positional is a usage error (exit 2). (B-annotate-12/-13/-16.)
pub fn runAnnotate(
    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.parseAnnotateArgs(verb_args)) {
        .ok => |p| p,
        .unknown_option => |tok| return cli.usageError(streams, "annotate", .unknown_option, tok),
        .extra_argument => |tok| return cli.usageError(streams, "annotate", .extra_argument, tok),
    };
    if (parsed.help) {
        streams.out.writeAll(annotate_help) catch {};
        return .ok;
    }
    const path = parsed.path orelse return cli.usageError(streams, "annotate", .no_file, "");

    var repo: Repository = undefined;
    var root_dir: std.Io.Dir = undefined;
    // A cold read: it walks ancestry (§3.3) and reconstructs effective sets, so
    // it takes the full derive, never the zero-parse image path.
    const rc = openRepo(gpa, io, arena, streams, "annotate", amb.repo_override, false, false, &repo, &root_dir);
    if (rc != .ok) return rc;
    defer root_dir.close(io);
    defer repo.deinit(gpa);

    // VERSION resolves exactly as `cat`'s (the §8.3 name procedure); omitted
    // means the current checkout, so a bare repository (-R with no working
    // tree) names its target explicitly (issue #67).
    const at: ?repostate.Rid = if (parsed.version) |name|
        repo.resolve(name, .checkin) catch |e| return annotateResolveFail(streams, name, e)
    else
        null;

    return annotateFile(gpa, io, &repo, root_dir, path, at, parsed.full, parsed.limit, parsed.log, streams);
}

/// Maps a `Repository.resolve` failure for `annotate` to its diagnostic + exit 1.
fn annotateResolveFail(s: cli.Streams, name: []const u8, e: Repository.ResolveError) cli.Exit {
    switch (e) {
        error.NotFound => s.err.print("zit annotate: check-in '{s}' not found\n", .{name}) catch {},
        error.Ambiguous => s.err.print("zit annotate: name '{s}' is ambiguous\n", .{name}) catch {},
        error.OutOfMemory => s.err.writeAll("zit annotate: out of memory\n") catch {},
    }
    return .failure;
}

/// Gather PATH's version chain over the current check-out's full ancestry,
/// attribute each target line through the pure engine, and render. A scratch
/// arena owns every derived array and every materialized version body; the
/// repository content channel stays on `gpa` so it is freed by `repo.deinit`.
fn annotateFile(
    gpa: std.mem.Allocator,
    io: std.Io,
    repo: *Repository,
    root_dir: std.Io.Dir,
    path: []const u8,
    at: ?repostate.Rid,
    full_ids: bool,
    limit: ?usize,
    log_versions: bool,
    streams: cli.Streams,
) cli.Exit {
    var arena_state = std.heap.ArenaAllocator.init(gpa);
    defer arena_state.deinit();
    const arena = arena_state.allocator();
    const cur = at orelse repo.currentCheckoutRid() orelse
        return annotateNotTracked(io, root_dir, path, streams);
    const all_rids = repo.ancestryLinearized(gpa, cur) catch return stateOom(streams, "annotate");
    defer gpa.free(all_rids);
    // zit#164 -n N: look back N check-ins. The linearization is oldest first
    // with the target last, so the window is its tail; a parent outside it
    // reads as the file's origin.
    const ordered_rids = if (limit) |n| all_rids[all_rids.len - @min(n, all_rids.len) ..] else all_rids;
    if (log_versions) {
        // zit#164 -l: the check-ins consulted, newest first, before the lines.
        var k = ordered_rids.len;
        while (k > 0) {
            k -= 1;
            const rid = ordered_rids[k];
            var iso: [19]u8 = undefined;
            const mt = repo.state().mtimeOf(rid) orelse 0;
            const full = repo.hashOf(rid);
            streams.out.print("version {s} {s}\n", .{ if (full_ids) full else full[0..@min(full.len, 10)], cli.formatWriterTimestamp(@intCast(mt / 1000), &iso) }) catch {};
        }
        streams.out.writeAll("---\n") catch {};
    }
    var entries: std.ArrayList(AnnotateEntry) = .empty;
    for (ordered_rids) |rid| {
        entries.append(arena, .{
            .rid = rid,
            .mtime = repo.state().mtimeOf(rid) orelse 0,
            .hash = repo.hashOf(rid),
        }) catch return stateOom(streams, "annotate");
    }

    // rid → node index in processing order, so parent links resolve to indices.
    var index_of: std.AutoHashMapUnmanaged(repostate.Rid, u32) = .empty;
    for (entries.items, 0..) |e, i| index_of.put(arena, e.rid, @intCast(i)) catch return stateOom(streams, "annotate");

    // Build the pure engine's node view: content + §3.2 primary/merge parents.
    const nodes = arena.alloc(annotate_mod.Node, entries.items.len) catch return stateOom(streams, "annotate");
    for (entries.items, 0..) |e, i| {
        const content = annotateContentAt(gpa, arena, repo, e.rid, path) catch |err|
            return annotateReadFail(streams, err);

        const primary_rid = repo.state().primaryParentOf(e.rid);
        const primary: ?u32 = if (primary_rid) |pr| index_of.get(pr) else null;

        var merges: std.ArrayList(u32) = .empty;
        var it = repo.state().parentsOf(e.rid);
        while (it.next()) |p| {
            if (primary_rid != null and p == primary_rid.?) continue; // primary handled separately
            if (index_of.get(p)) |idx| merges.append(arena, idx) catch return stateOom(streams, "annotate");
        }
        nodes[i] = .{ .content = content, .primary = primary, .merges = merges.items };
    }

    const cur_index = index_of.get(cur).?; // cur is self-inclusive in its ancestry
    const target = nodes[cur_index].content orelse
        return annotateNotTracked(io, root_dir, path, streams);

    // Provenance is undefined over binary content (B-annotate-9, the merge3 NUL
    // predicate) — refuse before emitting anything.
    if (merge3.isBinary(target)) {
        streams.err.print("zit annotate: '{s}': cannot annotate binary content\n", .{path}) catch {};
        return .failure;
    }

    const origins = annotate_mod.attribute(arena, nodes, cur_index) catch return stateOom(streams, "annotate");

    // Render one output line per file line: the origin's 10-char artifact-hash
    // prefix, a space, the ISO date (UTC, from the origin's event mtime), a space,
    // then the file line verbatim (its own trailing newline carries the break).
    const lines = merge3.splitLines(arena, target) catch return stateOom(streams, "annotate");
    std.debug.assert(lines.len == origins.len);
    for (lines, origins) |line, origin| {
        const oe = entries.items[origin];
        const full = repo.hashOf(oe.rid);
        // --full: the whole artifact ID, safe as an index key; the default
        // 10-character prefix stays for eyeballs (issue #67).
        const shown = if (full_ids) full else full[0..@min(full.len, 10)];
        var iso_buf: [19]u8 = undefined;
        const ts = cli.formatWriterTimestamp(@intCast(oe.mtime / 1000), &iso_buf);
        const iso_date = ts[0..10]; // YYYY-MM-DD
        streams.out.print("{s} {s} {s}", .{ shown, iso_date, merge3.lineBytes(target, line) }) catch {};
    }
    return .ok;
}

/// Seed a `depth`-deep linear check-in chain into `log` (index 0 = the deepest
/// ancestor, index depth-1 = the tip), filling `names_out` with each check-in's
/// 64-hex name. The deepest check-in's sole P parent is a check-in NOT stored
/// here — a phantom. Used to witness `walkLog`'s windowed early exit: the
/// phantom line is emitted only if the walk descends past the deepest ancestor.
fn logSeedPhantomChain(
    gpa: std.mem.Allocator,
    log: *logstore.LogStore,
    comptime depth: usize,
    names_out: *[depth][64]u8,
) !void {
    try log.putConfig(.{ .hash_policy = 2, .default_branch = "trunk", .project_id = "pid", .project_code = "code-abc" });
    // A valid 64-hex name for content deliberately absent from the store.
    var phantom: [64]u8 = undefined;
    _ = zit.artifact.artifactName(.sha3_256, "unstored-parent", &phantom);
    var i: usize = 0;
    while (i < depth) : (i += 1) {
        const v = try packEvolvingBlob(gpa, @intCast(i + 1));
        defer gpa.free(v);
        const b = try packSeedBlob(log, v);
        const parent: []const u8 = if (i == 0) &phantom else &names_out[i - 1];
        var ts_buf: [32]u8 = undefined;
        const ts = try std.fmt.bufPrint(&ts_buf, "2020-01-{d:0>2}T00:00:00", .{i + 1});
        try packSeedCheckin(gpa, log, ts, parent, "f.txt", &b, &names_out[i]);
    }
    try log.commit();
}

/// Count the lines of `text` that begin with `prefix` (test-only line tally).
fn logCountLinePrefix(text: []const u8, prefix: []const u8) usize {
    var n: usize = 0;
    var it = std.mem.splitScalar(u8, text, '\n');
    while (it.next()) |line| {
        if (std.mem.startsWith(u8, line, prefix)) n += 1;
    }
    return n;
}

test "log walk: `zit log CHECKIN` prints every reachable ancestor in P-card DFS order then the missing parent as a phantom" {
    const gpa = std.testing.allocator;
    const io = std.testing.io;
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();

    const depth = 6;
    var names: [depth][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 logSeedPhantomChain(gpa, &wlog, depth, &names);
    }
    var sfs: storage.FileStorage = undefined;
    var log = try packOpenReadOnly(gpa, io, tmp.dir, &sfs);
    defer sfs.file.close(io);
    defer log.deinit(gpa);

    const tip: []const u8 = &names[depth - 1];

    // The unbounded walk prints all `depth` check-ins tip->root in P-card DFS
    // order, then the deepest ancestor's phantom parent — exactly once each.
    var buf: [4096]u8 = undefined;
    var w = Io.Writer.fixed(&buf);
    try walkLog(gpa, &log, tip, &w, null, null, false);
    const out = w.buffered();
    try std.testing.expectEqual(@as(usize, depth), logCountLinePrefix(out, "checkin "));
    try std.testing.expectEqual(@as(usize, 1), logCountLinePrefix(out, "phantom "));

    var ebuf: [4096]u8 = undefined;
    var stream = Io.Writer.fixed(&ebuf);
    var d: usize = depth;
    while (d > 0) : (d -= 1) stream.print("checkin {s}\n", .{names[d - 1][0..]}) catch unreachable;
    const want_checkins = stream.buffered();
    try std.testing.expectEqualStrings(want_checkins, out[0..want_checkins.len]);
}

// ---------------------------------------------------------------------------
// frames (#66): the incremental-consumer read seam. Emits every artifact
// appended after a cursor as whole raw bytes, self-framed, with a trailer
// carrying the new cursor and a prefix-validation token. The token is the
// store's own LogKey — the O(1) freshness proof the derived-cache tier
// already trusts — so appends keep a cursor alive and a pack/shun rewrite
// (a new generation, new layout) kills it detectably.
// ---------------------------------------------------------------------------

const frames_help = main.frames_help;

/// Decode a `--since` cursor: the decimal rid count a previous trailer
/// emitted. Null on any malformation — the caller folds it into the
/// cursor-invalid refusal (a forged cursor is indistinguishable from a dead
/// one, and both have the same remedy: re-read from zero).
fn framesParseCursor(text: []const u8) ?u64 {
    return std.fmt.parseInt(u64, text, 10) catch null;
}

/// Decode a `--verify` token: the four LogKey fields a previous trailer
/// emitted, dash-joined decimal. Null on any malformation.
fn framesParseToken(text: []const u8) ?logstore.LogStore.LogKey {
    var it = std.mem.splitScalar(u8, text, '-');
    const end = std.fmt.parseInt(u64, it.next() orelse return null, 10) catch return null;
    const frame_count = std.fmt.parseInt(u64, it.next() orelse return null, 10) catch return null;
    const last_frame_offset = std.fmt.parseInt(u64, it.next() orelse return null, 10) catch return null;
    const last_header_crc = std.fmt.parseInt(u32, it.next() orelse return null, 10) catch return null;
    if (it.next() != null) return null; // no residue
    return .{
        .end = end,
        .frame_count = frame_count,
        .last_frame_offset = last_frame_offset,
        .last_header_crc = last_header_crc,
    };
}

// --- frames: verifying a batch across workers (#84) -------------------------
//
// Reconstruction stays on ONE thread and keeps the plaintext memo, because the
// memo is what made the sequential stream fast: consecutive artifacts chain
// through the same base, so fanning reconstruction out would either share a
// locked cache or redo base work. Verification has neither property — an
// artifact's hash depends on its own bytes and nothing else — and it is the
// larger half of the remaining cost (#84 profile: 59% verify, 38% reconstruct;
// deleting the verify call took a libfossil stream from 2.25s to 0.88s user).
//
// So the split is: the main thread reconstructs a bounded batch, workers hash
// it, the main thread emits it in order. Workers touch no store, no cache, and
// no allocator — only `verify(name, bytes)` over a slot they alone own. There
// is no queue, no lock, and no atomic: each worker walks a fixed STRIDE of the
// batch, so the writes land in disjoint slots and `join` supplies the ordering.

/// One artifact waiting to be verified and then emitted. `bytes` is owned and
/// freed by the emitter; the name is stored inline because the fill loop keeps
/// no per-slot buffer of its own.
const FrameJob = struct {
    hex_buf: [2 * logstore.max_hash_len]u8 = undefined,
    hex_len: usize = 0,
    bytes: []u8 = &.{},
    ok: bool = false,

    fn hex(self: *const FrameJob) []const u8 {
        return self.hex_buf[0..self.hex_len];
    }
};

/// Batch bounds (TigerStyle: both fixed; the byte budget is the one that binds
/// on a corpus with large artifacts). The count keeps per-batch thread spawns
/// amortised — 17,449 artifacts is ~69 batches, well under a percent of the run.
const frames_batch_max: usize = 256;
const frames_batch_bytes: usize = 16 << 20;

/// Upper bound on verify workers. Past #84's Amdahl ceiling (61% parallel, so
/// 2.6x however many cores exist) more threads buy nothing, and a fixed bound
/// keeps the thread array a comptime-sized stack value.
const frames_max_workers: usize = 8;

/// Verify every `stride`-th job from `start`. Pure and allocation-free: the
/// only writes are to `jobs[i].ok`, and the strides partition the batch, so no
/// two threads ever touch one slot.
fn framesVerifyStride(jobs: []FrameJob, start: usize, stride: usize) void {
    std.debug.assert(stride >= 1);
    std.debug.assert(start < stride);
    var i = start;
    while (i < jobs.len) : (i += stride) {
        jobs[i].ok = zit.artifact.verify(jobs[i].hex(), jobs[i].bytes);
    }
}

/// Hash a whole batch, using `workers` threads when that is worth it. Falls
/// back in-thread for a small batch, a single-core host, or a spawn that
/// fails — a stride whose thread never started is walked by the caller's own
/// thread, so a partial spawn can never leave a slot unverified, which would
/// read as a false corruption report.
fn framesVerifyBatch(jobs: []FrameJob, workers: usize) void {
    std.debug.assert(workers >= 1 and workers <= frames_max_workers);
    if (workers == 1 or jobs.len < 2) {
        framesVerifyStride(jobs, 0, 1);
        return;
    }
    const stride = @min(workers, jobs.len);
    var threads: [frames_max_workers - 1]std.Thread = undefined;
    var spawned: usize = 0;
    while (spawned + 1 < stride) : (spawned += 1) {
        threads[spawned] = std.Thread.spawn(.{}, framesVerifyStride, .{ jobs, spawned + 1, stride }) catch break;
    }
    // Stride 0 is always ours, and so is any stride whose spawn did not happen.
    var s: usize = 0;
    while (s < stride) : (s += 1) {
        if (s == 0 or s > spawned) framesVerifyStride(jobs, s, stride);
    }
    for (threads[0..spawned]) |t| t.join();
}

/// Verify a filled batch, then write it out in index order. Returns false once
/// an artifact fails to hash to its name, having already named it on stderr —
/// the caller frees the batch and exits. Emission order is the batch's own
/// order, which is persisted order, so the worker count never reaches the
/// stream.
fn framesFlushBatch(streams: cli.Streams, jobs: []FrameJob, workers: usize) bool {
    framesVerifyBatch(jobs, workers);
    for (jobs) |*job| {
        if (!job.ok) {
            streams.err.print("zit frames: stored artifact {s} does not hash to its name\n", .{job.hex()}) catch {};
            return false;
        }
        streams.out.print("artifact {s} {d}\n", .{ job.hex(), job.bytes.len }) catch {};
        streams.out.writeAll(job.bytes) catch {};
        streams.out.writeAll("\n") catch {};
    }
    return true;
}

/// How many verify workers this host should use: one per core, bounded, and
/// exactly 1 when the count cannot be determined — an unknown machine takes
/// the sequential path rather than a guess.
fn framesWorkerCount() usize {
    const cpus = std.Thread.getCpuCount() catch return 1;
    return @max(1, @min(cpus, frames_max_workers));
}

pub fn runFrames(
    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.parseFramesArgs(verb_args)) {
        .ok => |p| p,
        .unknown_option => |tok| return cli.usageError(streams, "frames", .unknown_option, tok),
        .extra_argument => |tok| return cli.usageError(streams, "frames", .extra_argument, tok),
        .empty_value => |tok| return cli.usageError(streams, "frames", .unknown_option, tok),
    };
    if (parsed.help) {
        streams.out.writeAll(frames_help) catch {};
        return .ok;
    }

    var repo: Repository = undefined;
    var root_dir: std.Io.Dir = undefined;
    const rc = openRepo(gpa, io, arena, streams, "frames", amb.repo_override, false, true, &repo, &root_dir);
    if (rc != .ok) return rc;
    defer root_dir.close(io);
    defer repo.deinit(gpa);

    const log = repo.logStore(gpa) catch |e| return repoOpenFail(streams, "frames", e);
    // PERSISTED (frame) order, not dense rid order: `intern` mints a rid for
    // a phantom before the frame that fills it lands, so a rid-ordered cursor
    // would pass a hole and never revisit it. Frame order is append order --
    // the same clock the LogKey prefix token describes.
    const persisted = log.ridsInPersistedOrder(gpa) catch return stateOom(streams, "frames");
    defer gpa.free(persisted);
    const count: u64 = persisted.len;

    // The consumed prefix: 0 (bootstrap) or the previous trailer's cursor.
    var since: u64 = 0;
    if (parsed.since) |text| {
        since = framesParseCursor(text) orelse return framesCursorInvalid(streams, "the cursor is not one this store issued");
        // A cursor past the store's end can only come from a rewritten
        // (shrunk) store or a foreign one -- dead either way.
        if (since > count) return framesCursorInvalid(streams, "the cursor names artifacts this store does not hold");
    }
    // Prefix validation: the token is the LogKey of the consumed prefix. A
    // rewrite (pack / shun reclaim) relocates or re-CRCs frames and fails
    // this check; plain appends never touch it.
    if (parsed.verify) |text| {
        const key = framesParseToken(text) orelse return framesCursorInvalid(streams, "the token is not one this store issued");
        if (!logstore.LogStore.validatePrefix(log.file, key)) {
            return framesCursorInvalid(streams, "the store was rewritten since the cursor was issued");
        }
    }

    // Emit rids since+1..count: whole reconstructed artifacts (the delta
    // chain and container layout stay private), each verified against its
    // name before a byte leaves — byte-identical to `artifact HASH --raw`.
    // Phantoms hold no bytes and shun-masked names read as absent; both are
    // skipped, and the cursor still advances over them (rid order is the
    // cursor's clock, presence is per-artifact).
    // A full-history stream is the read `contentCached` exists for. Nine
    // artifacts in ten are deltas in an imported fossil corpus (#84 measured
    // 89% on libfossil, 91% on sqlite), and consecutive artifacts in persisted
    // order routinely chain through the same base — so the uncached reader
    // reconstructed each base once per referencing delta, which is what made a
    // linear stream cost superlinear work. `memoize_undeltaed` stays OFF: this
    // sweep reads each artifact exactly once, so memoizing top-level raw/zlib
    // plaintext would duplicate bytes nothing asks for again. Bases are
    // memoized regardless of their own frame kind, which is the whole win.
    var plaintext: logstore.PlaintextCache = .{};
    defer plaintext.deinit(gpa);

    // Reconstruct into a bounded batch, hash the batch across workers, then
    // emit it in persisted order. Order is the batch's own index order, so the
    // stream is byte-identical to the single-threaded read whatever the worker
    // count — a property the #84 e2e pins by hashing the whole stream.
    const jobs = gpa.alloc(FrameJob, frames_batch_max) catch return stateOom(streams, "frames");
    defer gpa.free(jobs);
    const workers = framesWorkerCount();

    var emitted: u64 = 0;
    var filled: usize = 0;
    var batch_bytes: usize = 0;
    // Whatever exits this scope — a read fault, a verification failure, or the
    // clean end — the batch still holds owned plaintext that has not been
    // emitted. One place frees it.
    defer for (jobs[0..filled]) |job| gpa.free(job.bytes);

    for (persisted[@intCast(since)..]) |r| {
        const bytes = (log.contentCached(gpa, r, &plaintext) catch |e| return framesReadFail(streams, e)) orelse continue;
        // --structural (#108): emit only what the crosslink itself would link.
        // The gate is the SAME parse (`artifact.parseForImport`), applied to
        // the reconstructed bytes -- never a cache lookup, so the filtered
        // stream cannot disagree with the log. Skipped frames are neither
        // hashed nor emitted; the cursor advances over them regardless, so a
        // later poll -- with or without the flag -- resumes identically.
        if (parsed.structural) {
            // Reject-fast pre-screen. A structural artifact's first bytes are
            // either a card line, `<LETTER><SP>` (artifact-format section
            // 4.2), or a signature wrapper, `-----BEGIN ...` (section 4.1.1)
            // -- the sqlite corpus holds 1,087 clearsigned manifests, which is
            // how the first draft of this screen (card shape only) was caught
            // by the stream-equality check. The screen only ever REJECTS;
            // every accept still takes the full parse below, so nothing the
            // crosslink would refuse can be admitted. What it saves is the
            // whole-bytes Z-checksum a parse attempt pays before rejecting a
            // content blob.
            const card_shaped = bytes.len >= 2 and bytes[0] >= 'A' and bytes[0] <= 'Z' and bytes[1] == ' ';
            const wrapper_shaped = std.mem.startsWith(u8, bytes, "-----BEGIN ");
            if (!card_shaped and !wrapper_shaped) {
                gpa.free(bytes);
                continue;
            }
            if (artifact.parseForImport(gpa, bytes)) |im_const| {
                var im = im_const;
                im.deinit(gpa);
            } else |e| switch (e) {
                error.OutOfMemory => {
                    gpa.free(bytes);
                    return stateOom(streams, "frames");
                },
                else => { // not a structural artifact: a content blob
                    gpa.free(bytes);
                    continue;
                },
            }
        }
        jobs[filled] = .{ .bytes = bytes };
        const hex = std.fmt.bufPrint(&jobs[filled].hex_buf, "{x}", .{log.hashOf(r)}) catch unreachable;
        jobs[filled].hex_len = hex.len;
        filled += 1;
        batch_bytes += bytes.len;

        if (filled == frames_batch_max or batch_bytes >= frames_batch_bytes) {
            if (!framesFlushBatch(streams, jobs[0..filled], workers)) return .failure;
            emitted += filled;
            for (jobs[0..filled]) |job| gpa.free(job.bytes);
            filled = 0;
            batch_bytes = 0;
        }
    }
    if (filled != 0) {
        if (!framesFlushBatch(streams, jobs[0..filled], workers)) return .failure;
        emitted += filled;
        for (jobs[0..filled]) |job| gpa.free(job.bytes);
        filled = 0;
    }

    // The trailer: the new cursor and its token, emitted even when nothing
    // new exists so a bootstrap poll always learns its position.
    const key = if (count == 0)
        (log.logKey() catch |e| return framesKeyFail(streams, e))
    else
        (log.logKeyAt(persisted[persisted.len - 1]) catch |e| return framesKeyFail(streams, e));
    streams.out.print("cursor {d} {d}-{d}-{d}-{d}\n", .{
        count, key.end, key.frame_count, key.last_frame_offset, key.last_header_crc,
    }) catch {};

    if (emitted == 0) return .no_new;
    return .ok;
}

/// The #66 cursor-invalid refusal: a distinct exit code (4) so a consumer
/// distinguishes "re-read from zero" from every operational failure. The
/// message states the mechanism, never "corrupt".
fn framesCursorInvalid(s: cli.Streams, reason: []const u8) cli.Exit {
    s.err.print("zit frames: cursor invalid: {s}; re-read from --since=0\n", .{reason}) catch {};
    return .cursor_invalid;
}

fn framesReadFail(s: cli.Streams, e: logstore.ContentError) cli.Exit {
    switch (e) {
        error.OutOfMemory => s.err.writeAll("zit frames: out of memory\n") catch {},
        error.Storage => s.err.writeAll("zit frames: cannot read the repository\n") catch {},
        error.Corrupt => s.err.writeAll("zit frames: the repository is unreadable\n") catch {},
    }
    return .failure;
}

fn framesKeyFail(s: cli.Streams, e: logstore.LogStore.KeyError) cli.Exit {
    switch (e) {
        error.Storage => s.err.writeAll("zit frames: cannot read the repository\n") catch {},
        error.Corrupt => s.err.writeAll("zit frames: the repository is unreadable\n") catch {},
    }
    return .failure;
}

test "frames #84: strided verification agrees with the sequential answer at every worker count" {
    // The concurrency's whole safety argument is that the strides PARTITION
    // the batch: each slot is written by exactly one thread, and `join`
    // orders those writes before the emitter reads them. This drives the
    // partition at every legal worker count over a batch carrying both good
    // and corrupt names, and demands the identical verdict vector each time —
    // a stride that skipped a slot, double-covered one, or lost a spawn would
    // show up as a `false` where the sequential run says `true`.
    const gpa = std.testing.allocator;
    const count = 37; // deliberately coprime with every worker count below

    var jobs = try gpa.alloc(FrameJob, count);
    defer gpa.free(jobs);
    var bodies = try gpa.alloc([]u8, count);
    defer {
        for (bodies) |b| gpa.free(b);
        gpa.free(bodies);
    }

    // Every third artifact is given a name it does not hash to, so the
    // expected verdict vector is a non-trivial pattern rather than all-true.
    for (0..count) |i| {
        bodies[i] = try std.fmt.allocPrint(gpa, "artifact body number {d}\n", .{i});
        jobs[i] = .{ .bytes = bodies[i] };
        var name_buf: [64]u8 = undefined;
        const name = zit.artifact.artifactName(.sha3_256, bodies[i], &name_buf);
        @memcpy(jobs[i].hex_buf[0..name.len], name);
        jobs[i].hex_len = name.len;
        if (i % 3 == 0) jobs[i].hex_buf[0] = if (name[0] == 'a') 'b' else 'a'; // corrupt the name
    }

    var expected: [count]bool = undefined;
    framesVerifyBatch(jobs, 1);
    for (jobs, 0..) |job, i| expected[i] = job.ok;
    // The pattern must be mixed, or this test would pass on a no-op.
    var trues: usize = 0;
    for (expected) |e| {
        if (e) trues += 1;
    }
    try std.testing.expect(trues > 0 and trues < count);

    var workers: usize = 1;
    while (workers <= frames_max_workers) : (workers += 1) {
        for (jobs) |*job| job.ok = false;
        framesVerifyBatch(jobs, workers);
        for (jobs, 0..) |job, i| try std.testing.expectEqual(expected[i], job.ok);
    }

    // A batch smaller than the worker count takes the same partition path.
    for (jobs[0..1]) |*job| job.ok = false;
    framesVerifyBatch(jobs[0..1], frames_max_workers);
    try std.testing.expectEqual(expected[0], jobs[0].ok);
}

/// `zit grep PATTERN PATH...` (#92) — search a file's HISTORY for a literal.
///
/// The gap this closes is that zit had no content search at all: not `git log
/// -S`'s pickaxe, and not `fossil grep`'s file-scoped walk. Working-tree search
/// is deliberately NOT here — ripgrep does that faster than zit would, and the
/// bytes are on disk for it to read. What only zit can do is search content
/// that is not on disk at all, because it lives as delta chains inside `.zit`.
///
/// Semantics are `fossil grep`'s: walk the named file's historic versions
/// newest-first and report each DISTINCT version whose content matches. A run
/// of check-ins that left the file untouched reports once, not once per
/// check-in, because they share one blob.
pub fn runGrep(
    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.parseGrepArgs(verb_args)) {
        .ok => |p| p,
        .unknown_option => |tok| return cli.usageError(streams, "grep", .unknown_option, tok),
        .extra_argument => |tok| return cli.usageError(streams, "grep", .extra_argument, tok),
        .bad_limit => |tok| return cli.usageError(streams, "grep", .bad_limit, tok),
        .no_name => return cli.usageError(streams, "grep", .no_name, ""),
        .no_paths => return cli.usageError(streams, "grep", .no_paths, ""),
        .both_pickaxes => {
            streams.err.writeAll("zit grep: -S and -G ask different questions; give one\n") catch {};
            return .usage;
        },
    };
    if (parsed.help) {
        streams.out.writeAll(grep_help) catch {};
        return .ok;
    }
    const pattern = parsed.pattern.?;
    if (pattern.len == 0) {
        streams.err.writeAll("zit grep: PATTERN must not be empty\n") catch {};
        return .failure;
    }

    var repo: Repository = undefined;
    var root_dir: std.Io.Dir = undefined;
    const rc = openRepo(gpa, io, arena, streams, "grep", amb.repo_override, parsed.no_cache, false, &repo, &root_dir);
    if (rc != .ok) return rc;
    defer root_dir.close(io);
    defer repo.deinit(gpa);

    const log = repo.logStore(gpa) catch |e| return repoOpenFail(streams, "grep", e);
    if (parsed.pickaxe or parsed.diff_match) return runPickaxe(gpa, arena, &repo, log, parsed, streams);
    // #84's memo: successive check-ins share delta bases, and this walk reads
    // one blob per file per version. Without it each read re-resolves the same
    // base chain.
    var plaintext: logstore.PlaintextCache = .{};
    defer plaintext.deinit(gpa);

    // Newest-first and totally ordered, which is the order `fossil grep`
    // reports and the order an operator reads history in.
    const rids = repo.state().checkinsByMtime(gpa) catch return stateOom(streams, "grep");
    defer gpa.free(rids);

    var matched: usize = 0;
    var unreadable: usize = 0;
    for (parsed.paths[0..parsed.path_count]) |path| {
        // The blob reported on the previous (newer) check-in, so an unchanged
        // run of check-ins is one version rather than many.
        var seen_hash: ?[]const u8 = null;
        for (rids) |rid| {
            if (parsed.limit) |n| if (n != 0 and matched >= n) break;
            var list = (repo.files(gpa, rid) catch continue) orelse continue;
            defer list.deinit(gpa);

            var hash: ?[]const u8 = null;
            var it = list.iterator();
            while (it.next()) |f| {
                if (std.mem.eql(u8, f.filename, path)) {
                    hash = f.hash;
                    break;
                }
            }
            const h = hash orelse continue; // absent at this check-in
            if (seen_hash) |prev| {
                if (std.mem.eql(u8, prev, h)) continue; // same version, already judged
            }
            // Dupe: `list` owns the card bytes and is freed at the loop's end.
            seen_hash = arena.dupe(u8, h) catch return stateOom(streams, "grep");

            const bytes = grepBlobBytes(gpa, log, &plaintext, h) orelse {
                // Elided by a sparse store, shunned, or a phantom. Counted and
                // reported at the end rather than skipped in silence: a search
                // that quietly examined less than it claimed is the failure
                // mode that looks exactly like "no matches".
                unreadable += 1;
                continue;
            };
            defer gpa.free(bytes);
            if (zit.textdiff.isBinary(bytes)) continue; // a NUL-bearing blob has no lines
            if (grepReport(streams.out, repo.hashOf(rid), path, bytes, pattern, GrepMode.fromArgs(parsed))) matched += 1;
        }
    }

    if (unreadable != 0) {
        streams.err.print(
            "zit grep: {d} version(s) could not be read (sparse, shunned, or phantom) and were not searched\n",
            .{unreadable},
        ) catch {};
    }
    // `0` found something, `1` found nothing — the shell convention every grep
    // follows, so this composes in a conditional.
    return if (matched != 0) .ok else .failure;
}

/// #92 slice 2 — the pickaxe: report the check-ins where the NUMBER OF
/// OCCURRENCES of PATTERN changed, which is `git log -S`.
///
/// The distinction from slice 1 is the question, not the mechanism. `zit grep`
/// answers "which versions contain this", and a string that sat unchanged
/// through two hundred check-ins is reported for each version that held it.
/// The pickaxe answers "when did this appear, and when did it go" — for that
/// same string, exactly two lines.
///
/// THE COST THE ISSUE FEARED IS AVOIDABLE. #92 supposed this "needs a diff per
/// revision rather than a match per revision". It does not: a path whose blob
/// hash is IDENTICAL in a check-in and its parent cannot have changed the
/// count, so the walk compares F-card hashes first and reads blobs only for the
/// paths that actually differ. On ordinary history that is a handful of files
/// per check-in rather than the whole tree, and it is the same observation that
/// lets slice 1 share one blob across a run of untouching check-ins.
///
/// MERGES are compared against the PRIMARY parent only, so content arriving
/// through a merge is attributed to the merge check-in. `git log -S` does the
/// same under its default history simplification, and the alternative — a
/// check-in counted against every parent — reports one appearance twice.
fn runPickaxe(
    gpa: std.mem.Allocator,
    arena: std.mem.Allocator,
    repo: *Repository,
    log: *logstore.LogStore,
    parsed: cli.GrepArgs,
    streams: cli.Streams,
) cli.Exit {
    const pattern = parsed.pattern.?;
    var plaintext: logstore.PlaintextCache = .{};
    defer plaintext.deinit(gpa);

    const rids = repo.state().checkinsByMtime(gpa) catch return stateOom(streams, "grep");
    defer gpa.free(rids);

    var matched: usize = 0;
    var unreadable: usize = 0;
    for (rids) |rid| {
        if (parsed.limit) |n| if (n != 0 and matched >= n) break;

        // The parent's path -> blob-hash map, which is what makes "did this
        // path change" a string comparison instead of a blob read.
        var parent_of: std.StringHashMapUnmanaged([]const u8) = .empty;
        defer parent_of.deinit(gpa);
        var parent_list: ?Repository.FileList = null;
        defer if (parent_list) |*pl| pl.deinit(gpa);
        if (repo.state().primaryParentOf(rid)) |prid| {
            if (repo.files(gpa, prid) catch null) |pl| {
                parent_list = pl;
                var pit = parent_list.?.iterator();
                while (pit.next()) |f| {
                    const h = f.hash orelse continue;
                    parent_of.put(gpa, f.filename, h) catch return stateOom(streams, "grep");
                }
            }
        }

        var list = (repo.files(gpa, rid) catch continue) orelse continue;
        defer list.deinit(gpa);

        // Signed, because the two directions are the whole point: a check-in
        // that adds two occurrences and removes two has not changed the count
        // and is not a hit, exactly as `git log -S` treats it. Under `-G` this
        // instead counts the changed LINES that mention the pattern, which is
        // why that case is unsigned and reported with a different marker.
        var delta: isize = 0;
        var hit_path: ?[]const u8 = null;
        var it = list.iterator();
        while (it.next()) |f| {
            const h = f.hash orelse continue;
            if (!pathWanted(parsed, f.filename)) continue;
            const before = parent_of.get(f.filename);
            if (before) |b| if (std.mem.eql(u8, b, h)) continue; // untouched: neither answer can move
            if (parsed.diff_match) {
                const n = diffMentions(gpa, log, &plaintext, before, h, pattern) orelse {
                    unreadable += 1;
                    continue;
                };
                if (n != 0) {
                    delta += @intCast(n);
                    if (hit_path == null) hit_path = arena.dupe(u8, f.filename) catch null;
                }
                continue;
            }
            const now = pickaxeCount(gpa, log, &plaintext, h, pattern) orelse {
                unreadable += 1;
                continue;
            };
            const was = if (before) |b|
                pickaxeCount(gpa, log, &plaintext, b, pattern) orelse blk: {
                    unreadable += 1;
                    break :blk 0;
                }
            else
                0;
            if (now != was) {
                delta += @as(isize, @intCast(now)) - @as(isize, @intCast(was));
                if (hit_path == null) hit_path = arena.dupe(u8, f.filename) catch null;
            }
        }
        // A path the parent had and this check-in does not: its occurrences are
        // gone, and a deletion is the half of "when did it go" that a walk over
        // only the child's files would miss entirely.
        var pit = parent_of.iterator();
        while (pit.next()) |e| {
            const name = e.key_ptr.*;
            if (!pathWanted(parsed, name)) continue;
            if (listHolds(&list, name)) continue;
            if (parsed.diff_match) {
                // The file is gone, so every line it had is a removed line.
                const n = diffMentions(gpa, log, &plaintext, e.value_ptr.*, null, pattern) orelse {
                    unreadable += 1;
                    continue;
                };
                if (n != 0) {
                    delta += @intCast(n);
                    if (hit_path == null) hit_path = arena.dupe(u8, name) catch null;
                }
                continue;
            }
            const was = pickaxeCount(gpa, log, &plaintext, e.value_ptr.*, pattern) orelse {
                unreadable += 1;
                continue;
            };
            if (was != 0) {
                delta -= @as(isize, @intCast(was));
                if (hit_path == null) hit_path = arena.dupe(u8, name) catch null;
            }
        }

        if (delta == 0) continue;
        const id = repo.hashOf(rid);
        // Data only, one line per check-in: id, signed delta, and the first
        // path that moved. #71's machine-verb shape, because quarry indexes
        // this repository and a search it cannot parse is a search it cannot use.
        streams.out.print("{s} {s}{d} {s}\n", .{
            id[0..@min(id.len, 12)],
            if (parsed.diff_match) "~" else if (delta > 0) "+" else "-",
            @abs(delta),
            hit_path orelse "",
        }) catch {};
        matched += 1;
    }

    if (unreadable != 0) {
        streams.err.print(
            "zit grep: {d} version(s) could not be read (sparse, shunned, or phantom) and were not searched\n",
            .{unreadable},
        ) catch {};
    }
    return if (matched != 0) .ok else .failure;
}

/// Under `-S`, PATH operands narrow the walk rather than drive it; with none
/// given every path is wanted.
fn pathWanted(parsed: cli.GrepArgs, name: []const u8) bool {
    if (parsed.path_count == 0) return true;
    for (parsed.paths[0..parsed.path_count]) |p| {
        if (std.mem.eql(u8, p, name)) return true;
        // A directory operand covers its subtree, on the segment boundary so
        // `src` never matches `src-vendor`.
        if (name.len > p.len and name[p.len] == '/' and std.mem.startsWith(u8, name, p)) return true;
    }
    return false;
}

fn listHolds(list: anytype, name: []const u8) bool {
    var it = list.iterator();
    while (it.next()) |f| {
        if (std.mem.eql(u8, f.filename, name)) return true;
    }
    return false;
}

/// `-G`: how many CHANGED lines between two versions of one path mention
/// `pattern`. Null when either blob is unreadable. A null `to` means the path
/// was deleted, so every line it had is a removed line.
///
/// This is where `-G` earns its existence. Renaming `x = SECRET` to
/// `y = SECRET` leaves the occurrence count at one, so `-S` reports nothing —
/// correctly, by its own definition — while the line plainly changed. Asking
/// the diff rather than the counts is the only way to see that, and it is why
/// this half is the expensive one: it runs the diff engine per changed path
/// instead of two substring scans.
fn diffMentions(
    gpa: std.mem.Allocator,
    log: *logstore.LogStore,
    cache: *logstore.PlaintextCache,
    from: ?[]const u8,
    to: ?[]const u8,
    pattern: []const u8,
) ?usize {
    const a = if (from) |h| grepBlobBytes(gpa, log, cache, h) orelse return null else null;
    defer if (a) |bytes| gpa.free(bytes);
    const b = if (to) |h| grepBlobBytes(gpa, log, cache, h) orelse return null else null;
    defer if (b) |bytes| gpa.free(bytes);

    const a_bytes: []const u8 = a orelse "";
    const b_bytes: []const u8 = b orelse "";
    // A NUL-bearing version has no lines to change, matching the rule the rest
    // of this verb applies.
    if (zit.textdiff.isBinary(a_bytes) or zit.textdiff.isBinary(b_bytes)) return 0;

    var buf: std.ArrayList(u8) = .empty;
    defer buf.deinit(gpa);
    // Context 0: only the lines that actually changed are candidates, so no
    // unchanged neighbour can be mistaken for one.
    zit.textdiff.unified(gpa, a_bytes, b_bytes, .{ .context = 0 }, &buf) catch return null;

    var n: usize = 0;
    var rest: []const u8 = buf.items;
    while (rest.len != 0) {
        const nl = std.mem.indexOfScalar(u8, rest, '\n') orelse rest.len;
        const line = rest[0..nl];
        if (nl == rest.len) rest = "" else rest = rest[nl + 1 ..];
        if (line.len == 0) continue;
        // `+`/`-` are changed lines; `+++`/`---` are the file headers and
        // carry the PATH, which would match a pattern naming a filename and
        // report every check-in that touched it.
        if (line[0] != '+' and line[0] != '-') continue;
        if (std.mem.startsWith(u8, line, "+++") or std.mem.startsWith(u8, line, "---")) continue;
        if (std.mem.indexOf(u8, line[1..], pattern) != null) n += 1;
    }
    return n;
}

/// Non-overlapping occurrences of `pattern` in one blob, or null when the store
/// cannot produce it. A binary blob counts zero rather than being skipped: for
/// the pickaxe "it is not there" and "we did not look" must not be the same
/// answer, and a NUL-bearing blob has no lines to report anyway.
fn pickaxeCount(
    gpa: std.mem.Allocator,
    log: *logstore.LogStore,
    cache: *logstore.PlaintextCache,
    hash: []const u8,
    pattern: []const u8,
) ?usize {
    const bytes = grepBlobBytes(gpa, log, cache, hash) orelse return null;
    defer gpa.free(bytes);
    if (zit.textdiff.isBinary(bytes)) return 0;
    var n: usize = 0;
    var at: usize = 0;
    while (std.mem.indexOfPos(u8, bytes, at, pattern)) |i| {
        n += 1;
        at = i + pattern.len;
    }
    return n;
}

/// Print every matching line of one version. Returns whether any matched, so
/// the caller counts VERSIONS rather than lines.
/// How grep selects and prints lines (zit#164 -i -c -l -v --once).
const GrepMode = struct {
    ignore_case: bool = false,
    count_only: bool = false,
    files_only: bool = false,
    invert: bool = false,
    once: bool = false,

    fn fromArgs(a: cli.GrepArgs) GrepMode {
        return .{ .ignore_case = a.ignore_case, .count_only = a.count_only, .files_only = a.files_only, .invert = a.invert, .once = a.once };
    }
};

fn grepLineMatches(line: []const u8, pattern: []const u8, ignore_case: bool) bool {
    if (!ignore_case) return std.mem.indexOf(u8, line, pattern) != null;
    if (pattern.len == 0) return true;
    if (line.len < pattern.len) return false;
    var i: usize = 0;
    while (i + pattern.len <= line.len) : (i += 1) {
        if (std.ascii.eqlIgnoreCase(line[i..][0..pattern.len], pattern)) return true;
    }
    return false;
}

fn grepReport(out: *std.Io.Writer, checkin: []const u8, path: []const u8, bytes: []const u8, pattern: []const u8, mode: GrepMode) bool {
    var hits: usize = 0;
    var lineno: usize = 1;
    var rest = bytes;
    while (rest.len != 0) {
        const nl = std.mem.indexOfScalar(u8, rest, '\n') orelse rest.len;
        const line = rest[0..nl];
        if (grepLineMatches(line, pattern, mode.ignore_case) != mode.invert) {
            hits += 1;
            if (!mode.count_only and !mode.files_only) {
                out.print("{s} {s}:{d}: {s}\n", .{ checkin[0..@min(checkin.len, 12)], path, lineno, line }) catch {};
            }
            if (mode.once or mode.files_only) break;
        }
        if (nl == rest.len) break;
        rest = rest[nl + 1 ..];
        lineno += 1;
    }
    if (hits != 0 and mode.files_only) {
        out.print("{s} {s}\n", .{ checkin[0..@min(checkin.len, 12)], path }) catch {};
    } else if (mode.count_only) {
        out.print("{s} {s}:{d}\n", .{ checkin[0..@min(checkin.len, 12)], path, hits }) catch {};
    }
    return hits != 0;
}

/// One blob's plaintext, or null when the store cannot produce it.
fn grepBlobBytes(
    gpa: std.mem.Allocator,
    log: *logstore.LogStore,
    cache: *logstore.PlaintextCache,
    hash: []const u8,
) ?[]u8 {
    const rid = (log.resolveHexPrefix(hash) catch return null) orelse return null;
    return (log.contentCached(gpa, rid, cache) catch return null) orelse null;
}

// ---------------------------------------------------------------------------
// zit crosslink (#109): stream the derived per-check-in records.
// ---------------------------------------------------------------------------

/// The stream-format version `runCrosslink` emits as its first line. Bump it
/// when a line's meaning changes; consumers refuse versions they do not know.
const crosslink_stream_version = "crosslink 1";

pub fn runCrosslink(
    gpa: std.mem.Allocator,
    io: std.Io,
    arena: std.mem.Allocator,
    verb_args: []const []const u8,
    streams: cli.Streams,
    amb: cli.Ambient,
) cli.Exit {
    var bypass = false;
    for (verb_args) |tok| {
        if (std.mem.eql(u8, tok, "-h") or std.mem.eql(u8, tok, "--help")) {
            streams.out.writeAll(main.crosslink_help) catch {};
            return .ok;
        } else if (std.mem.eql(u8, tok, "--no-cache")) {
            bypass = true;
        } else if (tok.len > 1 and tok[0] == '-') {
            return cli.usageError(streams, "crosslink", .unknown_option, tok);
        } else {
            return cli.usageError(streams, "crosslink", .extra_argument, tok);
        }
    }

    // The open heals a stale cache (the #107 extension path), so the
    // cache-served branch below almost always finds a body covering the whole
    // log. `--no-cache` opens with the bypass, so nothing below can be
    // answered by a sidecar — the audit path this verb's byte-equality is
    // pinned against.
    var repo: Repository = undefined;
    var root_dir: std.Io.Dir = undefined;
    const rc = openRepo(gpa, io, arena, streams, "crosslink", amb.repo_override, bypass, !bypass, &repo, &root_dir);
    if (rc != .ok) return rc;
    defer root_dir.close(io);
    defer repo.deinit(gpa);
    const log = repo.logStore(gpa) catch |e| return repoOpenFail(streams, "crosslink", e);

    // Serve from the cache's record body when it provably covers the whole
    // log — same key + shun-fingerprint discipline as the extension path —
    // else derive the records fresh from the log. Both paths emit through one
    // encoder, and the e2e pins them byte-identical.
    var derived: std.ArrayList(u8) = .empty;
    defer derived.deinit(gpa);
    var raw: ?zit.repocache.RawCache = null;
    defer if (raw) |*rw| rw.deinit(gpa);

    var body: []const u8 = undefined;
    served_from_cache: {
        if (!bypass) {
            if (zit.repocache.loadForExtension(gpa, io, repo.root_dir)) |raw_const| {
                raw = raw_const;
                const key = log.logKey() catch break :served_from_cache;
                if (raw.?.shun_fp == log.shunFingerprint() and std.meta.eql(raw.?.key, key)) {
                    body = raw.?.body;
                    break :served_from_cache;
                }
                raw.?.deinit(gpa);
                raw = null;
            }
        }
        zit.repoview.captureRecords(gpa, log, 0, std.math.maxInt(u64), &derived) catch |e| switch (e) {
            error.OutOfMemory => return stateOom(streams, "crosslink"),
            else => {
                streams.err.writeAll("zit crosslink: the store could not be read\n") catch {};
                return .failure;
            },
        };
        body = derived.items;
    }

    return crosslinkEmit(gpa, streams, body);
}

/// Emit one record body as the versioned text stream. Free-text tokens (user,
/// tag names and values, paths) are card-escaped exactly as an artifact would
/// carry them, so every emitted line is newline-safe and a consumer reuses
/// the artifact-format decoder it already has. The comment is length-framed
/// raw bytes, the one field escaping cannot serve (it is free prose).
fn crosslinkEmit(gpa: std.mem.Allocator, streams: cli.Streams, body: []const u8) cli.Exit {
    var scratch: std.ArrayList(u8) = .empty;
    defer scratch.deinit(gpa);

    streams.out.print("{s}\n", .{crosslink_stream_version}) catch return .failure;
    var it = zit.repoview.RecordIter.init(body);
    while (it.next(gpa) catch {
        streams.err.writeAll("zit crosslink: the record body is corrupt; re-run with --no-cache\n") catch {};
        return .failure;
    }) |rec_const| {
        var rec = rec_const;
        defer rec.deinit(gpa);
        if (rec.kind != .manifest) continue; // control artifacts: not check-ins

        streams.out.print("checkin {s} {d}", .{ rec.hash, rec.mtime }) catch return .failure;
        for (rec.parents.items) |parent| streams.out.print(" {s}", .{parent}) catch return .failure;
        streams.out.writeAll("\n") catch return .failure;
        if (rec.user.len != 0) {
            streams.out.print("user {s}\n", .{escaped(gpa, &scratch, rec.user) catch return .failure}) catch return .failure;
        }
        streams.out.print("comment {d}\n", .{rec.comment.len}) catch return .failure;
        streams.out.writeAll(rec.comment) catch return .failure;
        streams.out.writeAll("\n") catch return .failure;
        for (rec.tags.items) |t| {
            const sigil: u8 = switch (t.type) {
                .singleton => '+',
                .cancel => '-',
                .propagate => '*',
            };
            streams.out.print("tag {c}{s}", .{ sigil, escaped(gpa, &scratch, t.name) catch return .failure }) catch return .failure;
            if (t.target) |target| {
                streams.out.print(" {s}", .{target}) catch return .failure;
            } else {
                streams.out.writeAll(" *") catch return .failure;
            }
            if (t.value) |v| {
                streams.out.print(" {s}", .{escaped(gpa, &scratch, v) catch return .failure}) catch return .failure;
            }
            streams.out.writeAll("\n") catch return .failure;
        }
        if (rec.files_known) {
            for (rec.files.items) |f| {
                streams.out.print("file {c} {s}\n", .{ @intFromEnum(f.disp), escaped(gpa, &scratch, f.path) catch return .failure }) catch return .failure;
            }
        } else {
            streams.out.writeAll("files unknown\n") catch return .failure;
        }
        streams.out.writeAll("end\n") catch return .failure;
    }
    return .ok;
}

/// Card-escape `token` into `scratch` (grown as needed) and return the
/// escaped slice, valid until the next call.
fn escaped(gpa: std.mem.Allocator, scratch: *std.ArrayList(u8), token: []const u8) error{OutOfMemory}![]const u8 {
    try scratch.resize(gpa, token.len * 2); // §4.3: an escaped byte doubles, nothing shrinks
    return artifact.encodeEscapes(token, scratch.items);
}