Artifact [9441fee57a]

Artifact 9441fee57a5610bbe404d73c6877ba96ea05944575861aac651ef62293b651ae:

• File src/cmd/history.zig — part of check-in [df99743ca4] at 2026-09-16 22:11:23 on branch trunk — file index: built by the derive, at ingest

  fossil fills mlink as each check-in arrives and answers finfo by lookup. zit's derive computed the same row per check-in and dropped it; the index was rebuilt by a separate pass on the first PATH query, 6.8 min on pkgsrc and not resumable. The derive now hands each change set to an index builder as it links, stores it beside the checkpoint, and extends it over the tail on append; a cache hit computes nothing. The on-query build is the fallback and checkpoints every 8192 entries. Index format 4 records completeness so a partial checkpoint is never adopted whole. crosslink reports a tree's files instead of unknown. pkgsrc, fresh clone to first answer: 11.7 min -> 7.1 min + 1s. (user: dmestas, size: 194111) [annotate] [blame] [check-ins using]

//! 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,
    /// zit#188 — the rids the file index says changed PATH. When present the
    /// predicate is a set membership test and no manifest is reconstructed;
    /// when null the walk falls back to `logChangesPath`, which is the same
    /// answer computed the slow way. The index is refused whenever it cannot
    /// be trusted whole (stale key, changed shun set, or any check-in whose
    /// change set is UNKNOWN), so the fallback is the correctness backstop,
    /// never a second definition of the predicate.
    changed: ?*const std.AutoHashMapUnmanaged(repostate.Rid, void) = null,
    /// The check-ins whose change set the index could not know. They are
    /// resolved ONE AT A TIME by reconstruction, rather than refusing the
    /// whole index: a real corpus carries a handful (the SQLite mirror has
    /// exactly one, a 2009 check-in whose parent is genuinely absent), and
    /// discarding 32,000 answers over one of them is how an index ends up
    /// built, stored, and never used.
    unknown: ?*const std.AutoHashMapUnmanaged(repostate.Rid, void) = null,
};

/// #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);
    // zit#188 — the PATH predicate reads through the SAME plaintext memo, plus
    // a baseline memo and an answer memo of its own. It runs twice per node
    // (here, and the primary parent), so it, not the walk, was the cost.
    var probe: PathProbe = .{ .plaintext = &plaintext };
    defer probe.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| {
            const indexed = if (f.unknown) |u| !u.contains(rid) else true;
            const hit = if (f.changed) |set|
                (if (indexed) set.contains(rid) else try logChangesPath(gpa, f.repo, &probe, id, manifest, f.path))
            else
                try logChangesPath(gpa, f.repo, &probe, id, manifest, f.path);
            if (hit) {
                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,
    /// zit#188 — the PATH filter, answered from the file index. Null walks
    /// everything (B-log-1).
    filter: ?LogPathFilter,
    /// `--long` needs each row's own D/U/C cards, and derived state keeps the
    /// timestamp PARSED, not as the check-in wrote it. So a long row reads its
    /// own manifest — one read per EMITTED row, a page of them, rather than
    /// the per-visited-node read that made this verb slow. The bytes then come
    /// from the same place the manifest walk printed them, which is what keeps
    /// the line identical.
    long: bool,
) 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;
    var plaintext: logstore.PlaintextCache = .{};
    defer plaintext.deinit(gpa);
    var probe: PathProbe = .{ .plaintext = &plaintext };
    defer probe.deinit(gpa);
    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) {
            // A PATH filter cannot assert that a phantom changes PATH — its
            // effective set is unknowable — so it is skipped silently, which
            // is what the manifest walk does. Unfiltered, it is named.
            if (filter == null) {
                out.print("phantom {s}\n", .{repo.hashOf(rid)}) catch {};
                emitted += 1;
                if (limit) |n| if (emitted >= n) return;
            }
            continue;
        }
        // A manifest is read ONLY when something needs its bytes: a check-in
        // the index recorded as unknown, or an emitted row under `--long`.
        // Reading one per visited check-in is the cost this issue removed.
        var owned: ?Repository.OwnedManifest = null;
        defer if (owned) |*o| o.deinit(gpa);
        if (filter) |f| {
            // The index NARROWS; the B-log-2 predicate DECIDES. A check-in the
            // index does not name is skipped without reading anything; one it
            // does name is still confirmed by reconstruction, because the two
            // relations are not identical: `captureFileFacts` diffs at F-card
            // level, so a delta manifest that restates a path with the hash it
            // already had reads as a change there while the path's effective
            // content did not move. On the SQLite mirror that is one check-in
            // in 2,113 — an over-report, never an omission — and confirming
            // costs one reconstruction per CANDIDATE (a page of them) instead
            // of one per visited check-in (twelve thousand).
            //
            // The direction being trusted is stated plainly: a check-in the
            // index omits is taken as unchanged. The differential test pins
            // that against the reconstructing walk so a regression in
            // `captureFileFacts` cannot silently drop a file's history.
            const indexed = f.changed != null and (if (f.unknown) |u| !u.contains(rid) else true);
            if (indexed and !f.changed.?.contains(rid)) {
                try pushParents(gpa, repo, rid, &pending);
                continue;
            }
            owned = repo.manifestCached(gpa, rid, &plaintext) catch |e| switch (e) {
                error.OutOfMemory => return error.OutOfMemory,
                else => null,
            };
            const hit = if (owned) |*o|
                try logChangesPath(gpa, f.repo, &probe, repo.hashOf(rid), &o.manifest, f.path)
            else
                false;
            if (!hit) {
                try pushParents(gpa, repo, rid, &pending);
                continue;
            }
        }
        if (long and owned == null) {
            owned = repo.manifestCached(gpa, rid, &plaintext) catch |e| switch (e) {
                error.OutOfMemory => return error.OutOfMemory,
                else => null,
            };
        }
        if (long and owned != null) {
            logEmit(out, repo.hashOf(rid), &owned.?.manifest, true);
        } else {
            out.print("checkin {s}\n", .{repo.hashOf(rid)}) catch {};
        }
        emitted += 1;
        if (limit) |n| if (emitted >= n) return;

        try pushParents(gpa, repo, rid, &pending);
    }
}

/// Push `rid`'s parents for the LIFO stack, reversing only this node's own
/// appended segment so the first P hash is visited first while older sibling
/// work stays below it.
fn pushParents(
    gpa: std.mem.Allocator,
    repo: *Repository,
    rid: repostate.Rid,
    pending: *std.ArrayList(repostate.Rid),
) error{OutOfMemory}!void {
    const 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[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,
};

/// zit#188 — the memos the PATH predicate reads through. Two of them, because
/// the predicate repeats two different things:
///
///   * `plaintext` and `baseline` are the pair `annotate` already threads
///     through `fileAtMemo`. A check-in manifest is routinely stored as a
///     delta against its neighbour, so reconstructing one per `fileAt` call
///     re-resolves the SAME base chain per check-in, and a wide baseline is
///     re-parsed per check-in on top of that — the cost then grows with how
///     much history sits behind the node rather than with the node. The walk
///     built this memo for its own manifest reads and did not hand it to the
///     predicate, which is the part that runs twice per node.
///   * `content` memoizes the ANSWER by rid. Every check-in is asked for its
///     content twice — once as `here` when it is visited, once as `parent`
///     when its child is — so in a mostly-linear history this halves what is
///     left. It is bounded by the nodes visited, which the walk already holds
///     one heap-allocated id apiece for.
///
/// Neither changes what the predicate answers: both are pure memos over a
/// reconstruction that is a function of the store.
const PathProbe = struct {
    plaintext: *logstore.PlaintextCache,
    baseline: ?*Repository.BaselineMemo = null,
    content: std.AutoHashMapUnmanaged(repostate.Rid, PathContent) = .empty,

    fn deinit(self: *PathProbe, gpa: std.mem.Allocator) void {
        if (self.baseline) |b| b.deinit(gpa);
        self.content.deinit(gpa);
    }
};

/// 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,
    probe: *PathProbe,
    id: []const u8,
    path: []const u8,
) LogWalkError!PathContent {
    const rid = repo.lookup(id) orelse return .{ .present = false };
    // zit#188 — asked once as `here`, once as `parent` of the child below it.
    if (probe.content.get(rid)) |hit| return hit;
    const answer = try effectivePathContentAt(gpa, repo, probe, rid, path);
    try probe.content.put(gpa, rid, answer);
    return answer;
}

/// `effectivePathContent` past the memo: the actual §8 reconstruction for one
/// interned rid, through the caller's baseline and plaintext memos.
fn effectivePathContentAt(
    gpa: std.mem.Allocator,
    repo: *Repository,
    probe: *PathProbe,
    rid: repostate.Rid,
    path: []const u8,
) LogWalkError!PathContent {
    const card = (repo.fileAtMemo(gpa, rid, path, &probe.baseline, probe.plaintext) 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,
    probe: *PathProbe,
    id: []const u8,
    manifest: *const artifact.Manifest,
    path: []const u8,
) LogWalkError!bool {
    const here = try effectivePathContent(gpa, repo, probe, 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, probe, 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#188 — set from `$ZIT_NO_FILE_INDEX` at the one environment seam in
/// `main`. The differential test drives the reconstructing answer through it.
pub var no_file_index: bool = false;

/// zit#192 evidence seam — how many check-ins the adopted file index recorded
/// as UNKNOWN (reconstructed per query rather than answered). On a tree-
/// manifest store this was the whole log, which is how an index that was
/// valid, adopted and never rebuilt still cost 39 minutes a query. Printed
/// by `main` under `ZIT_FILEINDEX_WALK`, beside the walk count. Null when
/// no index was adopted at all — printed as `none`, so a test asserting
/// zero unknowns cannot pass on a query that never consulted an index.
pub var file_index_unknown: ?usize = null;

/// zit#188 — fill `out` with the rids the file index says changed `path`, or
/// report false when the index cannot be trusted and the caller must fall back
/// to reconstructing the predicate.
///
/// Adopt-or-build, the shape every other `.zit-cache` member uses: a blob whose
/// key and shun fingerprint still validate is used as it stands; anything else
/// is rebuilt by one `repoview.eachChangeSet` pass and stored. The store is
/// advisory — a write that fails costs the next query a rebuild and nothing
/// else.
///
/// A check-in whose change set is UNKNOWN (a tree manifest, an absent or
/// shunned parent, an unresolvable baseline) is recorded in `unknown_out` and
/// reconstructed individually by the walk. Dropping such a check-in from the
/// index's answer would silently drop it from a file's history — a wrong
/// answer rather than a slow one — and refusing the whole index over one of
/// them makes the index dead weight: the SQLite mirror carries exactly one.
fn loadPathChanges(
    gpa: std.mem.Allocator,
    io: std.Io,
    root_dir: std.Io.Dir,
    repo: *Repository,
    log: *logstore.LogStore,
    path: []const u8,
    out: *std.AutoHashMapUnmanaged(repostate.Rid, void),
    unknown_out: *std.AutoHashMapUnmanaged(repostate.Rid, void),
) error{OutOfMemory}!bool {
    // A test seam, and the operator's escape hatch: set, the query is answered
    // by reconstruction alone, which is what the differential test compares
    // the indexed answer against. Read from the environment in `main` and set
    // here, never read in a runner: that is the design-seams rule, and reading
    // it through libc also made the build refuse on any target that does not
    // link libc (every Linux runner, while macOS links it always).
    if (no_file_index) return false;
    var blob: ?[]u8 = zit.repocache.tryLoadFileIndex(gpa, io, root_dir);
    defer if (blob) |b| gpa.free(b);
    var idx = if (blob) |b| zit.fileindex.parse(b, log) else null;
    const have: u64 = if (idx) |i| i.n_rids else 0;
    const want: u64 = @intCast(log.entryCount());

    // Three cases, and only the first two cost anything. An index that
    // already covers the log is used as it stands; one that covers a prefix
    // is EXTENDED over the appended tail; anything else is built once.
    // Rebuilding on every append is what made a commit cost a full pass.
    if (idx == null or have < want) {
        var builder = zit.fileindex.Builder.init(gpa);
        defer builder.deinit();
        if (idx) |*i| try builder.absorb(i);
        // The derive builds this index as it captures, so this is the fallback
        // for a store whose cache predates that. Walked in bounded chunks, each
        // checkpointed: a build a restart interrupts resumes from the last
        // chunk instead of from zero, so a forge that restarts more often than
        // the whole build takes still converges.
        // Known limit of this fallback, and the reason the derive-built index
        // is primary: coverage here is by entry count, so a phantom filled
        // after a build (its rid already below the count) is never revisited
        // by a later resume. The derive walks by frame offset and re-covers it.
        const chunk: u64 = 8192;
        var covered: u64 = have;
        while (covered < want) {
            const upto: u64 = @min(covered + chunk, want);
            try zit.repoview.eachChangeSet(gpa, log, @intCast(covered), @intCast(upto), builder.sink());
            const fresh = (try zit.fileindex.serialize(gpa, log, &builder, upto)) orelse return false;
            zit.repocache.storeFileIndex(io, root_dir, fresh);
            if (blob) |b| gpa.free(b);
            blob = fresh;
            covered = upto;
        }
        const final = blob orelse return false; // an empty log: nothing to index
        idx = zit.fileindex.parse(final, log);
    }
    const index = idx orelse return false;
    var u: usize = 0;
    while (u < index.unknownCount()) : (u += 1) {
        if (ridOfLogRid(repo, log, index.unknownAt(u))) |view_rid| try unknown_out.put(gpa, view_rid, {});
    }
    file_index_unknown = index.unknownCount();

    var rids: std.ArrayListUnmanaged(u32) = .empty;
    defer rids.deinit(gpa);
    try index.changedBy(path, &rids, gpa);
    for (rids.items) |r| {
        if (ridOfLogRid(repo, log, r)) |view_rid| try out.put(gpa, view_rid, {});
    }
    return true;
}

/// Translate a LOG rid (a frame index, which is what the file index stores)
/// into the DERIVED rid the walk uses. The two spaces are not the same — the
/// derived space interns phantoms a P card named and orders by first sight —
/// so comparing one to the other silently matches unrelated check-ins. The
/// artifact name is the identity both agree on.
fn ridOfLogRid(repo: *Repository, log: *logstore.LogStore, log_rid: u32) ?repostate.Rid {
    const digest = log.hashOf(@intCast(log_rid));
    const hex: []const u8 = if (digest.len == 32)
        &std.fmt.bytesToHex(digest[0..32].*, .lower)
    else if (digest.len == 20)
        &std.fmt.bytesToHex(digest[0..20].*, .lower)
    else
        return null;
    return repo.lookup(hex);
}

/// `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);

    // zit#188 — answer the change predicate from the file index when it can be
    // trusted whole. Building it costs one pass that reads each manifest once;
    // the walk that used to reconstruct one per visited check-in cost that much
    // per QUERY. Any refusal — no index, a stale one, or a check-in whose
    // change set is unknown — leaves `changed` null and the walk computes the
    // same answer the old way.
    var changed_set: std.AutoHashMapUnmanaged(repostate.Rid, void) = .empty;
    defer changed_set.deinit(gpa);
    var unknown_set: std.AutoHashMapUnmanaged(repostate.Rid, void) = .empty;
    defer unknown_set.deinit(gpa);
    const have_index = loadPathChanges(gpa, io, root_dir, &repo, log, path, &changed_set, &unknown_set) catch |e| switch (e) {
        error.OutOfMemory => return stateOom(streams, "log"),
    };

    const filter: LogPathFilter = .{
        .repo = &repo,
        .path = path,
        .changed = if (have_index) &changed_set else null,
        .unknown = if (have_index) &unknown_set else null,
    };

    // zit#188 — traverse the DERIVED adjacency, exactly as the unfiltered walk
    // has since #98: same depth-first order, same phantom line, and no
    // artifact content read for traversal. With the predicate answered from
    // the index, that leaves nothing per visited check-in at all — the reads
    // that remain are one per EMITTED row under `--long`. The manifest walk
    // stays as the fallback for a root the derived state cannot resolve.
    if (repo.lookup(root_id) != null and repo.state().mtimeOf(repo.lookup(root_id).?) != null) {
        walkLogDerived(gpa, &repo, root_id, streams.out, limit, filter, 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;
    }

    walkLog(gpa, log, root_id, streams.out, filter, 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, null, false) 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(gpa, 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(
    gpa: std.mem.Allocator,
    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, 0..) |row, idx| {
        std.debug.assert(repo.state().isLeaf(row.rid));
        std.debug.assert(all or row.open);
        // Fossil's leaf line (zit#170): number, time, name, comment, user and
        // tags, with the weekday, a twelve-character name, and *CURRENT*
        // where the checkout sits — the same marker timeline uses.
        const branch = repo.branchOf(row.rid);
        var tbuf: [32]u8 = undefined;
        var comment: []const u8 = "";
        var user: []const u8 = "";
        // The crosslinked state answers both, so the common path reads no
        // manifest at all. One read and parse per leaf — through whatever
        // delta chain it sits on — was the whole cost of this verb: sqlite's
        // 472 leaves spent a second here. Image backing and control-only rids
        // answer null and take the fallback.
        // gpa, not the arena: the repository opens its log with the allocator
        // it is handed and frees it with its own.
        var om: ?Repository.OwnedManifest = null;
        defer if (om) |*m| m.deinit(gpa);
        const from_state = blk: {
            const c = repo.state().commentOf(row.rid) orelse break :blk false;
            const u = repo.state().userOf(row.rid) orelse break :blk false;
            comment = main.firstCommentLine(c);
            user = u;
            break :blk true;
        };
        if (!from_state) {
            om = repo.manifest(gpa, row.rid) catch null;
            if (om) |m| {
                if (m.manifest.c) |c| comment = main.firstCommentLine(c.comment);
                if (m.manifest.u) |u| user = u.login;
            }
        }
        streams.out.print("({d}) {s} [{s}] {s}{s} (user: {s} tags: {s}{s})\n", .{
            idx + 1,
            cli.humanTime(row.mtime, &tbuf),
            cli.shortHash(row.hash),
            if (cur != null and cur.? == row.rid) "*CURRENT* " else "",
            comment,
            user,
            branch,
            if (all and !row.open) ", closed" else "",
        }) 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;
    }

    // The image carries everything a row renders — adjacency for the §3.2
    // merge mark since layout 2, comment and login since layout 6 — so the
    // zero-parse path answers; a full derive replayed every record for one
    // row (zit#181). Only `full`/`-v`/`-p` read manifests, through the facade.
    var repo: Repository = undefined;
    var root_dir: std.Io.Dir = undefined;
    const rc = openRepo(gpa, io, arena, streams, "timeline", amb.repo_override, false, true, &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 {
    // zit#188 — `-p PATH` runs the same predicate the log walk does, and paid
    // the same unmemoized reconstruction per row.
    var tl_plaintext: logstore.PlaintextCache = .{};
    defer tl_plaintext.deinit(gpa);
    var tl_probe: PathProbe = .{ .plaintext = &tl_plaintext };
    defer tl_probe.deinit(gpa);
    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;
    if (a.format == .html) {
        // zit#165: a static page — the same rows as -F medium, in a table.
        streams.out.writeAll("<!doctype html>\n<meta charset=\"utf-8\">\n<title>zit timeline</title>\n" ++
            "<style>body{font:14px/1.4 system-ui,sans-serif;margin:1.5em}table.timeline{border-collapse:collapse}" ++
            "table.timeline td{padding:.2em .6em;vertical-align:top;border-bottom:1px solid #ddd}" ++
            "td.h{font-family:ui-monospace,monospace}tr.current td{background:#fff6d6}</style>\n" ++
            "<table class=\"timeline\">\n<tr><th>check-in</th><th>when</th><th>branch</th><th>user</th><th>comment</th></tr>\n") catch {};
    }
    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;

        // A manifest is read only for what the crosslinked state cannot
        // answer: the file lists `--verbose`, `full`, and a path filter need.
        // Comment and user come from the state — the image carries both since
        // zit#172 — so the default format reads no artifact at all. This
        // condition used to be `a.format != .oneline`, written when oneline
        // was the default; zit#170 made `medium` the default and every row
        // began inflating a manifest for two strings the state already had.
        const need_manifest = a.path != null or a.verbose or a.format == .full;
        const state_comment = repo.state().commentOf(rid);
        const state_user = repo.state().userOf(rid);
        var maybe: ?Repository.OwnedManifest = null;
        defer if (maybe) |*m| m.deinit(gpa);
        if (need_manifest or state_comment == null or state_user == 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 (state_user) |u|
            u
        else 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, &tl_probe, 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 (state_comment) |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 {};
                }
            },
            .html => {
                var row: std.ArrayList(u8) = .empty;
                defer row.deinit(gpa);
                row.print(gpa, "<tr{s}><td class=\"h\">{s}</td><td>{s}</td><td>", .{ if (is_current) " class=\"current\"" else "", hash[0..@min(hash.len, 10)], iso }) catch return stateOom(streams, "timeline");
                zit.textdiff.htmlEscape(gpa, &row, branch) catch return stateOom(streams, "timeline");
                row.appendSlice(gpa, "</td><td>") catch return stateOom(streams, "timeline");
                zit.textdiff.htmlEscape(gpa, &row, user) catch return stateOom(streams, "timeline");
                row.appendSlice(gpa, "</td><td>") catch return stateOom(streams, "timeline");
                if (mark[0] == 'M') row.appendSlice(gpa, "<b>merge</b> ") catch return stateOom(streams, "timeline");
                zit.textdiff.htmlEscape(gpa, &row, comment) catch return stateOom(streams, "timeline");
                row.appendSlice(gpa, "</td></tr>\n") catch return stateOom(streams, "timeline");
                streams.out.writeAll(row.items) 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} {s} ===\n", .{ zit.repostate.weekdayName(zit.repostate.weekdayOf(mtime)), day }) catch {}; // weekday: zit#170
                }
                streams.out.print("{s} [{s}] {s}{s}{s} (user: {s} tags: {s}", .{ iso[11..19], cli.shortHash(hash), 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");
        }
    }
    if (a.format == .html) streams.out.writeAll("</table>\n") catch {};
    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);
            if (parsed.verbose) {
                // -v: where each branch is (zit#170). The bare list stays
                // Fossil's names, one per line.
                for (rows.items) |e| {
                    var tbuf: [32]u8 = undefined;
                    const is_cur = current_name != null and std.mem.eql(u8, current_name.?, e.name);
                    streams.out.print("{s} {s}  {s}  {s}{s}\n", .{ if (is_cur) "*" else " ", e.name, cli.shortHash(repo.hashOf(e.tip)), cli.humanTime(e.mtime, &tbuf), if (e.closed) "  (closed)" else "" }) catch {};
                }
                return .ok;
            }
            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 {};
    }
}

/// Largest `-` (stdin) name list `zit info` reads: 64 KiB is a name per
/// line for every check-in of a very large store, with room to spare.
const max_info_stdin: usize = 8 * 1024 * 1024;

/// `zit info [NAME...]` — resolve each 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`.
///
/// Many names share one store open (zit#177 step 2 — a forge rendering a
/// page asks about hundreds of check-ins). Blocks are separated by a blank
/// line so a consumer can split them; a single name prints exactly what it
/// always did. A name that fails to resolve gets its diagnostic and the batch
/// goes on; the exit is 1 if any name failed. `-F json` prints one object per
/// name in input order, so a failed name is a row (`"error"`), not a hole.
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(gpa, verb_args) catch return stateOom(streams, "info")) {
        .ok => |p| p,
        .unknown_option => |tok| return cli.usageError(streams, "info", .unknown_option, tok),
    };
    defer gpa.free(parsed.names);
    if (parsed.help) {
        streams.out.writeAll(info_help) catch {};
        return .ok;
    }
    // With no NAME, print Fossil's checkout block (the same one `status`
    // leads with) and the check-in count — Fossil's bare `info` (zit#170).
    if (parsed.names.len == 0) return infoCheckout(gpa, io, arena, streams, parsed.no_cache);

    // Expand a lone `-` into the names on stdin, one per line, in place.
    var names: std.ArrayList([]const u8) = .empty;
    defer names.deinit(gpa);
    for (parsed.names) |n| {
        if (!std.mem.eql(u8, n, "-")) {
            names.append(gpa, n) catch return stateOom(streams, "info");
            continue;
        }
        const stdin = std.Io.File.stdin();
        var read_buf: [64 * 1024]u8 = undefined;
        var reader = stdin.readerStreaming(io, &read_buf);
        const blob = reader.interface.allocRemaining(arena, .limited(max_info_stdin)) catch {
            streams.err.print("zit info: cannot read names from stdin, or more than {d} bytes were given\n", .{max_info_stdin}) catch {};
            return .failure;
        };
        var lines = std.mem.splitScalar(u8, blob, '\n');
        while (lines.next()) |raw| {
            const line = std.mem.trimEnd(u8, raw, "\r");
            if (line.len == 0) continue; // A trailing newline is not a name.
            names.append(gpa, line) catch return stateOom(streams, "info");
        }
    }
    if (names.items.len == 0) 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);

    var failed = false;
    for (names.items, 0..) |name, i| {
        // §8.3 full dispatch. §8.2's two MUST-NOTs: closed-ness and leaf-ness
        // of a candidate never filter the selection — that lives in `resolve`.
        const rid = repo.resolve(name, .checkin) catch |e| {
            const why: []const u8 = switch (e) {
                error.NotFound => "not found",
                error.Ambiguous => "ambiguous",
                error.OutOfMemory => return stateOom(streams, "info"),
            };
            failed = true;
            switch (parsed.format) {
                .text => switch (e) {
                    error.NotFound => streams.err.print("zit info: check-in '{s}' not found\n", .{name}) catch {},
                    else => streams.err.print("zit info: name '{s}' is ambiguous\n", .{name}) catch {},
                },
                .json => renderInfoJsonError(streams, name, why),
            }
            continue;
        };
        switch (parsed.format) {
            .text => {
                if (i > 0) streams.out.writeAll("\n") catch {};
                const r = renderInfo(gpa, arena, &repo, rid, streams);
                if (r != .ok) return r;
            },
            .json => {
                const r = renderInfoJson(gpa, arena, &repo, name, rid, streams);
                if (r != .ok) return r;
            },
        }
    }
    return if (failed) .failure else .ok;
}

/// 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 infoCheckout(gpa: std.mem.Allocator, io: std.Io, arena: std.mem.Allocator, streams: cli.Streams, bypass_cache: bool) cli.Exit {
    const root = (cli.discoverRoot(arena, io) catch return stateOom(streams, "info")) orelse {
        streams.err.writeAll(not_a_repo_msg) catch {};
        return .failure;
    };
    var root_dir = root.dir;
    defer root_dir.close(io);
    var ck_buf: [zit.checkout.max_ck_payload + 1]u8 = undefined;
    const ck = (zit.checkout.read(io, root_dir, &ck_buf) catch null) orelse {
        streams.err.writeAll("zit info: not an open checkout; name a check-in\n") catch {};
        return .failure;
    };
    var report: std.ArrayList(u8) = .empty;
    defer report.deinit(gpa);
    if (ck.version.len == 0) {
        const branch = if (ck.branch.len == 0) "trunk" else ck.branch;
        report.print(gpa, "repository:   {s}/.zit\nlocal-root:   {s}/\ncheckout:     none yet\ntags:         {s}\ncheck-ins:    0\n", .{ root.abs, root.abs, branch }) catch return stateOom(streams, "info");
        streams.out.writeAll(report.items) catch {};
        return .ok;
    }
    @import("tracking.zig").statusHeader(gpa, io, arena, streams, root.abs, ck.version, &report);
    var repo: Repository = undefined;
    var repo_root: std.Io.Dir = undefined;
    const rc = main.openRepo(gpa, io, arena, streams, "info", null, bypass_cache, true, &repo, &repo_root);
    if (rc != .ok) return rc;
    defer repo_root.close(io);
    defer repo.deinit(gpa);
    const rids = repo.state().checkinsByMtime(gpa) catch return stateOom(streams, "info");
    defer gpa.free(rids);
    report.print(gpa, "check-ins:    {d}\n", .{rids.len}) catch return stateOom(streams, "info");
    streams.out.writeAll(report.items) catch {};
    return .ok;
}

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 leaf status: open, closed, or no.
    streams.out.print("leaf:    {s}\n", .{leafWord(repo, rid)}) 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. The state carries the cards (zit#177) — reading the manifest here
    // was the one parse that made `info NAME` cost more than `leaves` — and
    // REV is not a parent, so no edge is implied. A check-in with no Q card
    // prints nothing here — byte-identical to its pre-B-merge-41 baseline
    // (negative space). The value column widens past the fixed lines above
    // because `cherrypick:` overflows their 8-space field.
    if (repo.state().picksOf(gpa, rid) catch null) |picks| {
        defer gpa.free(picks);
        for (picks) |pk| {
            const label: []const u8 = switch (pk.sign) {
                .cherrypick => "cherrypick: ",
                .backout => "backout:    ",
            };
            streams.out.print("{s}{s}\n", .{ label, pk.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;
}

/// The `-F json` row for one resolved name: the same facts as the text block,
/// as one object on one line. Keys are the text block's labels; `parent` is
/// null for a root, `merge` and the pick lists are arrays (possibly empty),
/// and an override key is present only when the override is active — the
/// same negative space as the text form. Bounded by the parent count.
fn renderInfoJson(
    gpa: std.mem.Allocator,
    arena: std.mem.Allocator,
    repo: *Repository,
    name: []const u8,
    rid: repostate.Rid,
    streams: cli.Streams,
) cli.Exit {
    std.debug.assert(repo.state().mtimeOf(rid) != null);
    var aw: std.Io.Writer.Allocating = .init(gpa);
    defer aw.deinit();
    var jw: std.json.Stringify = .{ .writer = &aw.writer };
    // An Allocating writer fails only on OOM, so every error here is one.
    writeInfoJson(gpa, arena, repo, name, rid, &jw) catch return stateOom(streams, "info");
    aw.writer.writeByte('\n') catch return stateOom(streams, "info");
    streams.out.writeAll(aw.written()) catch {};
    return .ok;
}

fn writeInfoJson(
    gpa: std.mem.Allocator,
    arena: std.mem.Allocator,
    repo: *Repository,
    name: []const u8,
    rid: repostate.Rid,
    jw: *std.json.Stringify,
) !void {
    try jw.beginObject();
    try jw.objectField("name");
    try jw.write(name);
    try jw.objectField("hash");
    try jw.write(repo.hashOf(rid));
    try jw.objectField("branch");
    try jw.write(repo.branchOf(rid));
    try jw.objectField("leaf");
    try jw.write(leafWord(repo, rid));
    const primary = repo.state().primaryParentOf(rid);
    try jw.objectField("parent");
    if (primary) |p| try jw.write(repo.hashOf(p)) else try jw.write(null);
    try jw.objectField("merge");
    try jw.beginArray();
    const merges = try mergeParentsSorted(arena, repo, rid, primary);
    for (merges) |h| try jw.write(h);
    try jw.endArray();
    try jw.objectField("cherrypick");
    try writePicks(gpa, repo, rid, .cherrypick, jw);
    try jw.objectField("backout");
    try writePicks(gpa, repo, rid, .backout, jw);
    for (std.enums.values(zit.repoimage.DisplayOverride)) |kind| {
        if (repo.state().displayOverride(rid, kind)) |v| {
            try jw.objectField(zit.repoimage.override_names[@intFromEnum(kind)]);
            try jw.write(v);
        }
    }
    try jw.endObject();
}

fn writePicks(gpa: std.mem.Allocator, repo: *Repository, rid: repostate.Rid, sign: zit.artifact.QSign, jw: *std.json.Stringify) !void {
    try jw.beginArray();
    const picks = try repo.state().picksOf(gpa, rid);
    defer gpa.free(picks);
    for (picks) |pk| if (pk.sign == sign) try jw.write(pk.target);
    try jw.endArray();
}

fn renderInfoJsonError(streams: cli.Streams, name: []const u8, why: []const u8) void {
    var buf: [256]u8 = undefined;
    var w: std.Io.Writer = .fixed(&buf);
    var jw: std.json.Stringify = .{ .writer = &w };
    const ok = blk: {
        jw.beginObject() catch break :blk false;
        jw.objectField("name") catch break :blk false;
        jw.write(name) catch break :blk false;
        jw.objectField("error") catch break :blk false;
        jw.write(why) catch break :blk false;
        jw.endObject() catch break :blk false;
        w.writeByte('\n') catch break :blk false;
        break :blk true;
    };
    if (ok) {
        streams.out.writeAll(w.buffered()) catch {};
    } else {
        // A name too long for the fixed row still gets a row a consumer can
        // split on; the name itself is in the stderr diagnostic's place.
        streams.out.writeAll("{\"name\":null,\"error\":\"name too long\"}\n") 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}.
fn leafWord(repo: *Repository, rid: repostate.Rid) []const u8 {
    return if (repo.state().isOpenLeaf(rid))
        "open"
    else if (repo.state().isLeaf(rid))
        "closed"
    else
        "no";
}

/// 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 {
    const merges = mergeParentsSorted(arena, repo, rid, primary) catch {
        streams.err.writeAll("zit info: out of memory\n") catch {};
        return .failure;
    };
    for (merges) |hash| streams.out.print("merge:   {s}\n", .{hash}) catch {};
    return .ok;
}

/// The §3.3 merge parents of `rid` — every parent but the primary — as
/// artifact IDs sorted ascending, in an arena slice bounded by the parent count.
fn mergeParentsSorted(
    arena: std.mem.Allocator,
    repo: *Repository,
    rid: repostate.Rid,
    primary: ?repostate.Rid,
) error{OutOfMemory}![]const []const u8 {
    var merges: std.ArrayList([]const u8) = .empty;
    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;
        try merges.append(arena, repo.hashOf(parent));
    }
    // 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);
    return merges.items;
}

/// `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}\n", .{f.filename}) catch {}; // names, as Fossil lists them (zit#170)
            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 {};
            // zit#187 — a quarry attachment stub is a pointer, and the honest
            // description is what it points at, not its own hundred bytes.
            if (zit.attachstub.parse(bytes)) |st| {
                streams.out.print("attachment-stub: {s}, {d} bytes external (sha256 {s}{s}{s})\n", .{
                    st.name, st.size, st.sha256, if (st.mimetype.len != 0) ", " else "", st.mimetype,
                }) 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/S/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 {};
    // submanifests §3.6: a tree check-in carries its whole file set by S
    // cards and no F cards, so dropping this lane showed a manifest with no
    // files at all while `ls` on the same hash listed thousands (zit#194).
    for (m.s) |sc| s.out.print("S {s} {s}\n", .{ sc.filename, sc.hash }) 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;
}

/// Which of PATH's ancestors actually CHANGED it, read from the derived
/// cache's record body in one sequential pass (zit#172). Fossil answers this
/// from its mlink table; the equivalent here is the #109 changed-file list
/// every record carries. A record's list is a pure function of
/// content-addressed artifacts — the diff of a manifest against its primary
/// parent over their shared baseline — so even a cache older than the log is
/// CORRECT for the check-ins it covers; coverage, not freshness, is the only
/// question, and `covered` answers it. Null when there is no usable cache:
/// the caller then reads a manifest per ancestor, exactly as before. Annotate
/// asks over PATH's ancestry; grep (zit#178) asks over the whole store.
const AnnotateTouched = struct {
    /// Check-ins with a record whose file facts are known.
    covered: std.AutoHashMapUnmanaged(repostate.Rid, void) = .empty,
    /// The subset of `covered` whose record lists PATH as changed, with the
    /// hash the check-in gives PATH -- null when it deletes it (zit#176).
    touched: std.AutoHashMapUnmanaged(repostate.Rid, ?[]u8) = .empty,

    fn deinit(self: *AnnotateTouched, gpa: std.mem.Allocator) void {
        self.covered.deinit(gpa);
        var it = self.touched.valueIterator();
        while (it.next()) |h| if (h.*) |b| gpa.free(b);
        self.touched.deinit(gpa);
    }
};

fn annotateTouchedFromCache(
    gpa: std.mem.Allocator,
    io: std.Io,
    repo: *Repository,
    path: []const u8,
) ?AnnotateTouched {
    var raw = zit.repocache.loadForExtension(gpa, io, repo.root_dir) orelse return null;
    defer raw.deinit(gpa);
    var out: AnnotateTouched = .{};
    errdefer out.deinit(gpa);
    var it = zit.repoview.RecordIter.init(raw.body);
    while (true) {
        var rec = (it.next(gpa) catch {
            // A short or undecodable tail: keep what decoded — every kept
            // record is still individually correct — and let the uncovered
            // rids take the manifest fallback.
            break;
        }) orelse break;
        defer rec.deinit(gpa);
        if (rec.kind != .manifest or !rec.files_known) continue;
        const rid = repo.lookup(rec.hash) orelse continue;
        out.covered.put(gpa, rid, {}) catch return null;
        for (rec.files.items) |f| {
            if (std.mem.eql(u8, f.path, path)) {
                const hash: ?[]u8 = if (f.hash) |h| (gpa.dupe(u8, h) catch return null) else null;
                errdefer if (hash) |h| gpa.free(h);
                out.touched.put(gpa, rid, hash) catch return null;
                break;
            }
        }
    }
    return out;
}

/// 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,
    cache: *logstore.PlaintextCache,
    memo: *?*Repository.BaselineMemo,
    rid: repostate.Rid,
    path: []const u8,
) AnnotateReadError!?[]const u8 {
    const maybe_card = repo.fileAtMemo(gpa, rid, path, memo, cache) 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).
    return annotateContentByHash(gpa, arena, repo, cache, card.hash.?);
}

/// The blob `hash` names, duplicated into `arena`, or null when the store
/// lacks it. Through the plaintext memo: adjacent versions of PATH are deltas
/// over shared bases, and without the memo every version re-walks its whole
/// chain -- on SQLite's shell.c.in that walk, not the diff, was most of the
/// remaining time once the diff core went O(ND) (zit#176).
fn annotateContentByHash(
    gpa: std.mem.Allocator,
    arena: std.mem.Allocator,
    repo: *Repository,
    cache: *logstore.PlaintextCache,
    hash: []const u8,
) AnnotateReadError!?[]const u8 {
    const bytes = (repo.blobCached(gpa, hash, cache) 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,
    /// `blame`/`praise`: Fossil's user column in place of the line number.
    blame: bool,
) 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, blame);
}

/// 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,
    blame: bool,
) cli.Exit {
    var arena_state = std.heap.ArenaAllocator.init(gpa);
    defer arena_state.deinit();
    const arena = arena_state.allocator();
    var plaintext: logstore.PlaintextCache = .{};
    defer plaintext.deinit(gpa);
    // Consecutive versions of PATH sit in delta check-ins over one baseline;
    // the memo parses that baseline once instead of once per version.
    var baseline: ?*Repository.BaselineMemo = null;
    defer if (baseline) |b| b.deinit(gpa);
    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");

    // Which ancestors changed PATH, from the cache's record body (zit#172).
    // Reading a manifest per ancestor made annotate proportional to the
    // store: SQLite's history is ~36k check-ins of which ~200 touch a given
    // file, and fossil (whose mlink table is the same precomputation) answers
    // in milliseconds where this took minutes. With the set in hand, an
    // ancestor that did not change PATH inherits its primary parent's
    // content outright — the record's diff IS against the primary parent, so
    // "not listed" means "byte-equal there" — a changed ancestor reads the blob
    // the record names (zit#176), and only an uncovered one opens its manifest.
    var touched_opt = annotateTouchedFromCache(gpa, io, repo, path);
    defer if (touched_opt) |*t| t.deinit(gpa);

    // 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 known: ?struct { c: ?[]const u8 } = blk: {
            const t = &(touched_opt orelse break :blk null);
            if (!t.covered.contains(e.rid)) break :blk null; // no record: open the manifest
            if (t.touched.get(e.rid)) |hash| {
                // Changed here: the record names the blob (a delete names none).
                const h = hash orelse break :blk .{ .c = null };
                const c = annotateContentByHash(gpa, arena, repo, &plaintext, h) catch |err|
                    return annotateReadFail(streams, err);
                break :blk .{ .c = c };
            }
            const pr = repo.state().primaryParentOf(e.rid) orelse break :blk null;
            const pi = index_of.get(pr) orelse break :blk null; // outside -n window
            std.debug.assert(pi < i); // parents precede children in the linearization
            break :blk .{ .c = nodes[pi].content };
        };
        const content = if (known) |k|
            k.c
        else
            annotateContentAt(gpa, arena, repo, &plaintext, &baseline, 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);
    // blame/praise: the committing user of every origin, read once each.
    const users = arena.alloc([]const u8, entries.items.len) catch return stateOom(streams, "annotate");
    for (entries.items, 0..) |oe, i| {
        users[i] = "";
        if (!blame) continue;
        if (repo.manifest(gpa, oe.rid) catch null) |om_const| {
            var om = om_const;
            defer om.deinit(gpa);
            if (om.manifest.u) |u| users[i] = arena.dupe(u8, u.login) catch return stateOom(streams, "annotate");
        }
    }
    for (lines, origins, 0..) |line, origin, n| {
        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 cli.shortHash(full);
        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
        // Fossil's columns (zit#170): the line number for annotate, the user
        // for blame and praise.
        if (blame) {
            streams.out.print("{s} {s} {s:>13}: {s}", .{ shown, iso_date, users[origin], merge3.lineBytes(target, line) }) catch {};
        } else {
            streams.out.print("{s} {s} {d:>5}: {s}", .{ shown, iso_date, n + 1, 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| {
        // zit#178: the cache's records name the check-ins that changed PATH
        // against their primary parent, so a manifest is read only there — the
        // same scan annotate makes — instead of once per check-in in the
        // store. Without a cache every check-in is read, as before.
        var touched = annotateTouchedFromCache(gpa, io, &repo, path);
        defer if (touched) |*t| t.deinit(gpa);
        // `fossil grep`'s shape: each distinct version once, under the OLDEST
        // check-in that introduced it, versions ordered by that check-in's
        // time newest-first. A merge that carries a branch's version to trunk,
        // or a revert, introduces nothing. Walking oldest-first and keeping
        // the first sighting of each blob yields exactly that list, reversed.
        var versions: std.ArrayList(struct { rid: repostate.Rid, hash: []const u8 }) = .empty;
        defer versions.deinit(gpa);
        var seen = std.StringHashMapUnmanaged(void).empty;
        defer seen.deinit(gpa);
        var i: usize = rids.len;
        while (i > 0) {
            i -= 1;
            const rid = rids[i];
            if (touched) |*t| {
                if (t.covered.contains(rid) and !t.touched.contains(rid)) continue;
            }
            const card = (repo.fileAt(gpa, rid, path) catch continue) orelse continue; // absent here
            defer Repository.freeFileCard(gpa, card);
            // Dupe: the card is freed at the loop's end.
            const h = arena.dupe(u8, card.hash.?) catch return stateOom(streams, "grep"); // §4.7.6: no delete markers
            const gop = seen.getOrPut(gpa, h) catch return stateOom(streams, "grep");
            if (gop.found_existing) continue;
            versions.append(gpa, .{ .rid = rid, .hash = h }) catch return stateOom(streams, "grep");
        }

        var v = versions.items.len;
        while (v > 0) {
            v -= 1;
            if (parsed.limit) |n| if (n != 0 and matched >= n) break;
            const rid = versions.items[v].rid;
            const h = versions.items[v].hash;
            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);
}