Artifact [cc818b370a]

Artifact cc818b370aa62f64e93a634892388854c947b66b8258abcdd3c89f78d0c42acf:

• File src/sync_driver_test.zig — part of check-in [6682409ff0] at 2026-08-06 13:21:13 on branch sub-manifest — sync: bound stalled rounds, not total rounds

  A total-round cap multiplied against the per-round batch size into an accidental ceiling on repository size (~1M artifacts), which a healthy pkgsrc-scale store tripped while converging every round. Guard now counts consecutive rounds that moved no bytes; announcement is not delivery, so a peer announcing phantoms forever still trips it.

  Also: import-git reports sub-manifest memo hit rate and whether the retention budget latched (#35). (user: dmestas, size: 29666) [annotate] [blame] [check-ins using]

//! Unit tests for the sync round driver (`sync_driver.runSync`).
//!
//! The transport is an IN-PROCESS fake: its `exchange` runs the sans-IO
//! `sync.Server.respond` over a server-side `MemStore`, so a full round loop
//! runs through `runSync` exactly as the A3 sweep (`sim/sync_sweep.zig`) drives
//! `buildRequest → respond → consumeReply` — but here through the production
//! driver, with no simulated network. Expected values are the sync RFC's
//! §3.7 convergence / §3.8 termination invariants (mirroring the sweep's
//! `checkConvergence` / `checkTermination`), never observed from fossil.

const std = @import("std");
const zit = @import("zit");
const sync = zit.sync;
const repostate = zit.repostate;
const sync_driver = @import("sync_driver.zig");

const testing = std.testing;

/// A generous convergence bound: a clean bidirectional exchange resolves in a
/// handful of rounds (discovery is single-hop — a reply walks the peer store
/// once), so this is slack, not a tuned minimum. Mirrors the sweep's window.
const round_bound: usize = 64;

/// §9.1 — the client's own 16-hex repository identifiers, emitted in the
/// pull/push operation cards. Any well-formed pair works: the stateless server
/// echoes the peer's codes and never gates on a configured project code.
const servercode = "0123456789abcdef";
const projectcode = "fedcba9876543210";

/// A fresh sync-operation client carrying the identifiers above.
fn newClient() sync.Client {
    return .{
        .state = .exchange,
        .operation = .sync,
        .self_servercode = servercode,
        .project_code = projectcode,
    };
}

// ---------------------------------------------------------------------------
// In-process fakes satisfying the driver's `exchange(gpa, request) -> ![]u8`
// ---------------------------------------------------------------------------

/// The convergence fake: `exchange` answers each request with the sans-IO
/// `Server.respond` over a durable server store. `respond` writes into a reused
/// buffer, so the returned slice is duped into a fresh caller-owned allocation
/// (the driver frees it) — matching the real transport's "caller owns" contract.
const ServerTransport = struct {
    server: sync.Server,
    store: repostate.ArtifactStore,
    reply_buf: std.ArrayListUnmanaged(u8) = .empty,

    pub const Error = sync.Server.RespondError; // {OutOfMemory, ResponseTooLarge}

    fn deinit(self: *ServerTransport, gpa: std.mem.Allocator) void {
        self.reply_buf.deinit(gpa);
    }

    pub fn exchange(self: *ServerTransport, gpa: std.mem.Allocator, request: []const u8) Error![]u8 {
        const reply = try self.server.respond(gpa, self.store, request, &self.reply_buf);
        return gpa.dupe(u8, reply);
    }
};

/// The round-limit fake: `exchange` announces a fresh phantom (`igot HASH`) on
/// every call, so §3.7(a) versioned-progress continuation holds every round and
/// the client never terminates — the driver's stall guard must catch it.
///
/// This is precisely the shape the guard has to recognise, and the reason it
/// cannot simply ask "did anything change": something changes every round here
/// (a new phantom), yet no content EVER arrives. Announcement is not delivery.
const NeverEndsTransport = struct {
    counter: usize = 0,

    pub const Error = error{OutOfMemory};

    pub fn exchange(self: *NeverEndsTransport, gpa: std.mem.Allocator, request: []const u8) Error![]u8 {
        _ = request;
        self.counter += 1;
        // A distinct 64-hex hash per round → a brand-new client phantom each
        // round → §3.7(a) never stops (the content is never delivered).
        return std.fmt.allocPrint(gpa, "igot {x:0>64}\n", .{self.counter});
    }
};

/// The many-rounds fake: a peer with more to send than fits in one round, which
/// delivers steadily and terminates. Each round announces one new artifact and
/// delivers the one announced last round, so every round after the first moves
/// bytes; after `announce_rounds` it stops announcing, drains the last delivery,
/// and lets the phantom table empty so §3.7(a) stops continuing.
///
/// This is the shape a large real repository has: converging perfectly, one
/// bounded batch at a time, for as many rounds as its size requires. A guard
/// that counted total rounds would abort it partway through and blame the peer
/// for failing to converge — the exact failure a 1,166,689-artifact store hit
/// against the old flat cap of 1024 rounds.
const ManyRoundsTransport = struct {
    announce_rounds: usize,
    round: usize = 0,

    pub const Error = error{OutOfMemory};

    pub fn exchange(self: *ManyRoundsTransport, gpa: std.mem.Allocator, request: []const u8) Error![]u8 {
        _ = request;
        self.round += 1;
        var out: std.ArrayListUnmanaged(u8) = .empty;
        errdefer out.deinit(gpa);
        if (self.round <= self.announce_rounds) {
            const h = hashN(self.round);
            try out.appendSlice(gpa, "igot ");
            try out.appendSlice(gpa, &h);
            try out.append(gpa, '\n');
        }
        // Deliver what the previous round announced (§3.3 content delivery).
        if (self.round > 1 and self.round <= self.announce_rounds + 1) {
            const prev = hashN(self.round - 1);
            try out.appendSlice(gpa, "file ");
            try out.appendSlice(gpa, &prev);
            try out.appendSlice(gpa, " 7\ncontent");
        }
        return out.toOwnedSlice(gpa);
    }
};

// ---------------------------------------------------------------------------
// Store seeding + held-set oracle (arbitrary but distinct 64-hex hashes: the
// protocol treats a hash as an opaque token at this layer, as the sweep does)
// ---------------------------------------------------------------------------

/// A distinct lowercase-64-hex hash for a small id — no crypto needed; the
/// stores only require both sides to agree on the hash→content mapping.
fn hashN(id: usize) [64]u8 {
    var buf: [64]u8 = undefined;
    @memset(&buf, 'a');
    const hexd = "0123456789abcdef";
    buf[0] = hexd[id & 0xf];
    buf[1] = hexd[(id >> 4) & 0xf];
    return buf;
}

/// The sorted set of held (content-bearing) hashes in `store`, duped into `a`.
fn heldHashes(a: std.mem.Allocator, store: repostate.ArtifactStore) ![][]const u8 {
    var list: std.ArrayListUnmanaged([]const u8) = .empty;
    const n = store.count();
    var rid: repostate.Rid = 1;
    while (rid <= n) : (rid += 1) {
        if (store.content(rid)) |_| try list.append(a, try a.dupe(u8, store.hashOf(rid)));
    }
    const slice = try list.toOwnedSlice(a);
    std.mem.sort([]const u8, slice, {}, lessHash);
    return slice;
}

fn lessHash(_: void, x: []const u8, y: []const u8) bool {
    return std.mem.order(u8, x, y) == .lt;
}

/// §3.3 r24 — a phantom is exactly a content-absent rid. True iff none remain.
fn noPhantom(store: repostate.ArtifactStore) bool {
    const n = store.count();
    var rid: repostate.Rid = 1;
    while (rid <= n) : (rid += 1) {
        if (store.content(rid) == null) return false;
    }
    return true;
}

/// Build a client/server store pair: `shared` artifacts on both, `client_only`
/// held by the client and interned as a server-side phantom, `server_only` held
/// by the server and interned as a client-side phantom. This is the sweep's
/// three-way composition — the general convergence shape, not a special case.
fn seedStores(
    client_store: repostate.ArtifactStore,
    server_store: repostate.ArtifactStore,
    shared: usize,
    client_only: usize,
    server_only: usize,
) !void {
    var id: usize = 0;
    var k: usize = 0;
    while (k < shared) : (k += 1) {
        const h = hashN(id);
        _ = try client_store.put(&h, "shared-content");
        _ = try server_store.put(&h, "shared-content");
        id += 1;
    }
    k = 0;
    while (k < client_only) : (k += 1) {
        const h = hashN(id);
        _ = try client_store.put(&h, "client-only-content");
        _ = try server_store.intern(&h); // server knows it only as a phantom
        id += 1;
    }
    k = 0;
    while (k < server_only) : (k += 1) {
        const h = hashN(id);
        _ = try server_store.put(&h, "server-only-content");
        _ = try client_store.intern(&h); // client knows it only as a phantom
        id += 1;
    }
}

/// A fresh clone-operation client: begins in `clone_round0` (§3.8), with no
/// credential or codes — the first clone round sends no login card (§8.3) and
/// the clone operation emits no push/pull operation card (§8.2).
fn newCloneClient() sync.Client {
    return .{ .state = .clone_round0, .operation = .clone };
}

/// A fresh pull-operation client carrying the §9.1 identifiers: begins mid-
/// exchange (`initialState(.pull)`), announces its holdings via igot, and
/// obtains but never uploads content (§3.3 r24). The receive-only counterpart of
/// `newClient` (which is bidirectional sync).
fn newPullClient() sync.Client {
    return .{
        .state = .exchange,
        .operation = .pull,
        .self_servercode = servercode,
        .project_code = projectcode,
    };
}

// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------

test "sync_driver §8.2: a clone spanning multiple cfile batches pulls every artifact and terminates on clone_seqno 0" {
    const gpa = testing.allocator;
    var client_ms = repostate.MemStore.init(gpa);
    defer client_ms.deinit();
    var server_ms = repostate.MemStore.init(gpa);
    defer server_ms.deinit();

    // A server holding more artifacts than one small batch carries: with a
    // batch cap of 2, five artifacts page across three cfile batches (§8.2
    // L1384), so the client MUST thread the `clone_seqno NEXT` cursor into
    // follow-up `clone 3 NEXT` requests to pull past the first batch.
    const total: usize = 5;
    var id: usize = 0;
    while (id < total) : (id += 1) {
        const h = hashN(id);
        _ = try server_ms.store().put(&h, "clone-artifact-content");
    }

    var fake = ServerTransport{
        .server = .{ .version = .{ .version = 0 }, .self_servercode = "s", .project_code = "p", .clone_batch_max = 2 },
        .store = server_ms.store(),
    };
    defer fake.deinit(gpa);

    var client = newCloneClient();
    defer client.deinit(gpa);

    const stats = try sync_driver.runSync(gpa, &client, client_ms.store(), &fake, round_bound);

    // §3.8 — the clone reached the clean terminal, well within the round bound.
    // Three cfile batches force at least three rounds: a client truncating at
    // batch one (the pre-fix bug) would stop at round two with content missing.
    try testing.expectEqual(sync.ClientState.terminated, client.state);
    try testing.expect(stats.rounds >= 3 and stats.rounds <= round_bound);

    // §8.2 — the whole repository cloned: every server artifact is now held by
    // the client (not just the first batch), and no phantom remains.
    var arena = std.heap.ArenaAllocator.init(gpa);
    defer arena.deinit();
    const a = arena.allocator();
    const held = try heldHashes(a, client_ms.store());
    try testing.expectEqual(total, held.len);
    try testing.expect(noPhantom(client_ms.store()));
    try testing.expectEqualDeep(try heldHashes(a, server_ms.store()), held);
}

test "sync_driver §3.7: runSync converges the client store against a peer and terminates" {
    const gpa = testing.allocator;
    var client_ms = repostate.MemStore.init(gpa);
    defer client_ms.deinit();
    var server_ms = repostate.MemStore.init(gpa);
    defer server_ms.deinit();
    // A mix with both directions and shared ancestry, plus phantoms on each
    // side — the general reconcile, exercising pull, push, and dephantomize.
    try seedStores(client_ms.store(), server_ms.store(), 3, 2, 2);

    var fake = ServerTransport{ .server = .{ .version = .{ .version = 0 } }, .store = server_ms.store() };
    defer fake.deinit(gpa);

    var client = newClient();
    defer client.deinit(gpa);

    const stats = try sync_driver.runSync(gpa, &client, client_ms.store(), &fake, round_bound);

    // §3.8 — the client reached the clean terminal within the round bound.
    try testing.expectEqual(sync.ClientState.terminated, client.state);
    try testing.expect(stats.rounds >= 1 and stats.rounds <= round_bound);

    // §3.7(a) — the client store now holds the union of intact content and
    // retains no phantom the peer could have satisfied.
    var arena = std.heap.ArenaAllocator.init(gpa);
    defer arena.deinit();
    const a = arena.allocator();
    const held = try heldHashes(a, client_ms.store());
    try testing.expectEqual(@as(usize, 3 + 2 + 2), held.len); // C₀ ∪ S₀
    try testing.expect(noPhantom(client_ms.store()));
    // Sync is symmetric: the server (also a receiver) converged to the union too.
    try testing.expect(noPhantom(server_ms.store()));
    try testing.expectEqual(@as(usize, 7), (try heldHashes(a, server_ms.store())).len);
}

test "sync_driver §3.7: an already-converged pair terminates in a single round with no phantom churn" {
    const gpa = testing.allocator;
    var client_ms = repostate.MemStore.init(gpa);
    defer client_ms.deinit();
    var server_ms = repostate.MemStore.init(gpa);
    defer server_ms.deinit();
    try seedStores(client_ms.store(), server_ms.store(), 4, 0, 0); // identical stores

    var fake = ServerTransport{ .server = .{ .version = .{ .version = 0 } }, .store = server_ms.store() };
    defer fake.deinit(gpa);
    var client = newClient();
    defer client.deinit(gpa);

    const stats = try sync_driver.runSync(gpa, &client, client_ms.store(), &fake, round_bound);
    try testing.expectEqual(sync.ClientState.terminated, client.state);
    // Nothing to reconcile: one round observes no continuation condition (§3.7).
    try testing.expectEqual(@as(usize, 1), stats.rounds);
    try testing.expect(noPhantom(client_ms.store()));
}

test "sync_driver §3.2/§3.3: pull fetches the artifacts a subset repo lacks and converges" {
    const gpa = testing.allocator;
    var client_ms = repostate.MemStore.init(gpa);
    defer client_ms.deinit();
    var server_ms = repostate.MemStore.init(gpa);
    defer server_ms.deinit();

    // The client holds a strict SUBSET of the server: three artifacts on both,
    // two more only on the server. The client is unaware of the server-only pair
    // until the peer announces them (no pre-seeded phantom) — exactly the state
    // `loadRepoHoldings` leaves a repository in before a real pull.
    const shared: usize = 3;
    const server_only: usize = 2;
    var id: usize = 0;
    while (id < shared) : (id += 1) {
        const h = hashN(id);
        _ = try client_ms.store().put(&h, "shared-content");
        _ = try server_ms.store().put(&h, "shared-content");
    }
    while (id < shared + server_only) : (id += 1) {
        const h = hashN(id);
        _ = try server_ms.store().put(&h, "server-only-content");
    }

    var fake = ServerTransport{ .server = .{ .version = .{ .version = 0 } }, .store = server_ms.store() };
    defer fake.deinit(gpa);
    var client = newPullClient();
    defer client.deinit(gpa);

    const stats = try sync_driver.runSync(gpa, &client, client_ms.store(), &fake, round_bound);

    // §3.8 — the pull reached the clean terminal within the round bound.
    try testing.expectEqual(sync.ClientState.terminated, client.state);
    try testing.expect(stats.rounds >= 1 and stats.rounds <= round_bound);

    // §3.7(a) — the client converged to the server's full set: every server
    // artifact is now held by the client and no phantom remains.
    var arena = std.heap.ArenaAllocator.init(gpa);
    defer arena.deinit();
    const a = arena.allocator();
    const held = try heldHashes(a, client_ms.store());
    try testing.expectEqual(shared + server_only, held.len);
    try testing.expect(noPhantom(client_ms.store()));
    try testing.expectEqualDeep(try heldHashes(a, server_ms.store()), held);
}

test "sync_driver §3.7: a pull against an already-current repo receives nothing and terminates in one round" {
    const gpa = testing.allocator;
    var client_ms = repostate.MemStore.init(gpa);
    defer client_ms.deinit();
    var server_ms = repostate.MemStore.init(gpa);
    defer server_ms.deinit();
    // Identical stores: the client already holds everything the server does.
    var id: usize = 0;
    while (id < 4) : (id += 1) {
        const h = hashN(id);
        _ = try client_ms.store().put(&h, "content");
        _ = try server_ms.store().put(&h, "content");
    }

    var arena = std.heap.ArenaAllocator.init(gpa);
    defer arena.deinit();
    const a = arena.allocator();
    const before = (try heldHashes(a, client_ms.store())).len;

    var fake = ServerTransport{ .server = .{ .version = .{ .version = 0 } }, .store = server_ms.store() };
    defer fake.deinit(gpa);
    var client = newPullClient();
    defer client.deinit(gpa);

    const stats = try sync_driver.runSync(gpa, &client, client_ms.store(), &fake, round_bound);
    try testing.expectEqual(sync.ClientState.terminated, client.state);
    // Nothing to fetch: one round observes no continuation condition (§3.7), and
    // the held set is unchanged — the "already up to date" case `runPull` reports.
    try testing.expectEqual(@as(usize, 1), stats.rounds);
    try testing.expectEqual(before, (try heldHashes(a, client_ms.store())).len);
    try testing.expect(noPhantom(client_ms.store()));
}

test "sync_driver: a peer that announces forever without delivering hits the stall guard, no leak" {
    const gpa = testing.allocator;
    var client_ms = repostate.MemStore.init(gpa);
    defer client_ms.deinit();
    var fake = NeverEndsTransport{};
    var client = newClient();
    defer client.deinit(gpa);

    const max_stalled: usize = 4;
    const err = sync_driver.runSync(gpa, &client, client_ms.store(), &fake, max_stalled);
    try testing.expectError(error.RoundLimitExceeded, err);
    // Every round stalls (a phantom is created, but no content is delivered),
    // so the consecutive count never resets and the guard fires at the bound:
    // one exchange per permitted round.
    try testing.expectEqual(max_stalled, fake.counter);
    // The client is still mid-exchange (never a terminal): the driver refused
    // to spin, it did not fabricate a stop.
    try testing.expectEqual(sync.ClientState.exchange, client.state);
}

test "sync_driver: a transfer needing many more rounds than the stall budget still converges" {
    const gpa = testing.allocator;
    var client_ms = repostate.MemStore.init(gpa);
    defer client_ms.deinit();

    // Far more rounds than the budget: under a total-round cap this is the
    // failure mode, since the cap would have to exceed the transfer's length.
    const announce_rounds: usize = 200;
    const max_stalled: usize = 4;
    var fake = ManyRoundsTransport{ .announce_rounds = announce_rounds };
    var client = newClient();
    defer client.deinit(gpa);

    const stats = try sync_driver.runSync(gpa, &client, client_ms.store(), &fake, max_stalled);

    try testing.expectEqual(sync.ClientState.terminated, client.state);
    // The point of the test: it ran an order of magnitude past the bound,
    // because the bound counts stalls and this peer never stalled twice.
    try testing.expect(stats.rounds > announce_rounds);
    try testing.expect(stats.stalled_peak <= max_stalled);
    // And it actually converged rather than merely surviving: every announced
    // artifact was delivered and stored.
    var arena = std.heap.ArenaAllocator.init(gpa);
    defer arena.deinit();
    try testing.expectEqual(announce_rounds, (try heldHashes(arena.allocator(), client_ms.store())).len);
}

test "sync §3.2/§3.3: a sync run both creates phantoms from peer igot and uploads held content" {
    const gpa = testing.allocator;
    var client_ms = repostate.MemStore.init(gpa);
    defer client_ms.deinit();
    var server_ms = repostate.MemStore.init(gpa);
    defer server_ms.deinit();
    // A both-directions gap: two artifacts only the client holds (it must UPLOAD
    // them) and two only the server holds (it must RECEIVE them). A `.sync`
    // client drives both on ONE §3.7 session — not pull-then-push over two.
    const client_only: usize = 2;
    const server_only: usize = 2;
    try seedStores(client_ms.store(), server_ms.store(), 1, client_only, server_only);

    var arena = std.heap.ArenaAllocator.init(gpa);
    defer arena.deinit();
    const a = arena.allocator();
    const held_before = (try heldHashes(a, client_ms.store())).len;

    var fake = ServerTransport{ .server = .{ .version = .{ .version = 0 } }, .store = server_ms.store() };
    defer fake.deinit(gpa);
    var client = newClient(); // operation = .sync
    defer client.deinit(gpa);

    _ = try sync_driver.runSync(gpa, &client, client_ms.store(), &fake, round_bound);

    // §3.2/§3.3 — one session both uploaded the client-only content (§3.2
    // both-directions, counted by `content_sent`) and dephantomized the
    // server-only content into the client store (§3.3 held-set growth). Both
    // strictly positive: sync is bidirectional.
    const received = (try heldHashes(a, client_ms.store())).len - held_before;
    try testing.expect(client.content_sent > 0);
    try testing.expect(received > 0);
    try testing.expectEqual(client_only, client.content_sent);
    try testing.expectEqual(server_only, received);
}

test "sync §3.7: sync drives to a clean protocol terminal in bounded rounds" {
    const gpa = testing.allocator;
    var client_ms = repostate.MemStore.init(gpa);
    defer client_ms.deinit();
    var server_ms = repostate.MemStore.init(gpa);
    defer server_ms.deinit();
    try seedStores(client_ms.store(), server_ms.store(), 2, 1, 1);

    var fake = ServerTransport{ .server = .{ .version = .{ .version = 0 } }, .store = server_ms.store() };
    defer fake.deinit(gpa);
    var client = newClient(); // operation = .sync
    defer client.deinit(gpa);

    const stats = try sync_driver.runSync(gpa, &client, client_ms.store(), &fake, round_bound);
    // §3.7/§3.8 — a bidirectional sync always reaches the clean terminal (never
    // `aborted`, never spinning) within a bounded number of rounds.
    try testing.expectEqual(sync.ClientState.terminated, client.state);
    try testing.expect(stats.rounds >= 1 and stats.rounds <= round_bound);
}

/// One full convergence run over `gpa`, no assertions — the body under
/// `checkAllAllocationFailures`, which drives every allocation-failure point and
/// checks each returns `error.OutOfMemory` leak-free.
fn convergeOnce(gpa: std.mem.Allocator) !void {
    var client_ms = repostate.MemStore.init(gpa);
    defer client_ms.deinit();
    var server_ms = repostate.MemStore.init(gpa);
    defer server_ms.deinit();
    try seedStores(client_ms.store(), server_ms.store(), 2, 1, 1);

    var fake = ServerTransport{ .server = .{ .version = .{ .version = 0 } }, .store = server_ms.store() };
    defer fake.deinit(gpa);
    var client = newClient();
    defer client.deinit(gpa);

    _ = try sync_driver.runSync(gpa, &client, client_ms.store(), &fake, round_bound);
}

test "sync_driver OOM: every allocation-failure point along a full run is handled leak-free" {
    try testing.checkAllAllocationFailures(testing.allocator, convergeOnce, .{});
}

// ---------------------------------------------------------------------------
// draft-fossil-submanifests-00 §7.2 — the peer-version gate, driven end to end
// ---------------------------------------------------------------------------

/// The tree manifest the gate tests upload: a check-in manifest carrying an S
/// run, which is §7.2's second shape ("a manifest carrying an S card").
const gate_tree = "C t\nD 2026-07-27T12:00:00.000\nS a.txt " ++ ("1" ** 64) ++ "\nU u\n";
/// A flat check-in manifest — everything that predates this draft — carried
/// alongside so every assertion can say what the gate left alone.
const gate_flat = "C m\nD 2026-07-27T12:00:00.000\nF a.txt " ++ ("2" ** 64) ++ "\nU u\n";

/// One §7.2 upload run through the PRODUCTION driver, with nothing hand-seeded:
/// the client learns the peer's version only from the wire, exactly as `zit
/// push`/`zit sync` do. `client_version` is the pragma the CLI would announce.
/// Returns the client after the run (caller deinits).
const GateRun = struct {
    refused: usize,
    peer_version: u32,
    server_holds_tree: bool,
    server_holds_flat: bool,
    /// §3.8 — the state the run ended in. A refusal must still reach the CLEAN
    /// terminal: a peer that keeps re-`gimme`ing what this side keeps refusing
    /// is the shape that spins to `RoundLimitExceeded`.
    state: sync.ClientState,
};

fn runGateUpload(gpa: std.mem.Allocator, client_version: sync.VersionInfo) !GateRun {
    var client_ms = repostate.MemStore.init(gpa);
    defer client_ms.deinit();
    var server_ms = repostate.MemStore.init(gpa);
    defer server_ms.deinit();

    var tree_name: [64]u8 = undefined;
    var flat_name: [64]u8 = undefined;
    _ = zit.artifact.artifactName(.sha3_256, gate_tree, &tree_name);
    _ = zit.artifact.artifactName(.sha3_256, gate_flat, &flat_name);
    _ = try client_ms.store().put(&tree_name, gate_tree);
    _ = try client_ms.store().put(&flat_name, gate_flat);

    // The server is a plain zit `serve`: a build identity carrying DATE/TIME so
    // §10.1 r11/r12 CAN answer, and §8.1's emission setting off (there is no
    // operator surface for it). What it announces is `announcedVersion`'s job.
    var fake = ServerTransport{
        .server = .{
            .version = .{ .version = 22800, .datetime = .{ .date = 20260715, .time = 0 } },
            .self_servercode = "s",
            .project_code = projectcode,
        },
        .store = server_ms.store(),
    };
    defer fake.deinit(gpa);

    var client = sync.Client{
        .state = .exchange,
        .operation = .sync,
        .self_servercode = servercode,
        .project_code = projectcode,
        .version = client_version,
        // §8.1's durable indicator: this repository holds a tree, so its send
        // gate is armed even though the emission setting is off.
        .seen_submanifest = true,
        // .peer is DELIBERATELY not seeded — the whole point of the run.
    };
    defer client.deinit(gpa);

    _ = try sync_driver.runSync(gpa, &client, client_ms.store(), &fake, round_bound);

    return .{
        .refused = client.submanifests_refused,
        .peer_version = client.peer.version,
        .state = client.state,
        .server_holds_tree = if (server_ms.store().lookup(&tree_name)) |r| server_ms.store().content(r) != null else false,
        .server_holds_flat = if (server_ms.store().lookup(&flat_name)) |r| server_ms.store().content(r) != null else false,
    };
}

test "sync_driver submanifests §7.2/§10.1: an armed client uploads a tree because the peer's announcement reached it" {
    // THE LIVELOCK THIS PINS. §7.2 gates every upload on "the remote peer's
    // announced version", and a client learns that version from exactly one
    // place: the `pragma server-version` reply. §10.1 r11/r12 make that reply
    // conditional on the request's client-version carrying DATE and TIME. An
    // undated request therefore leaves `Client.peer.version` at 0 — below every
    // threshold — for the whole session, so an armed client refuses its own
    // trees to a server that would have announced support if only it had been
    // asked: trees pullable, never pushable, and `submanifests_refused` fails
    // every run of the feature's happy path.
    //
    // Nothing here is hand-seeded. The client's `.peer` is left at its default
    // and the announcement travels the real request/reply loop through the
    // production driver, so the test cannot pass on unwired code.
    const gpa = testing.allocator;

    const dated = try runGateUpload(gpa, .{ .version = 22700, .datetime = .{ .date = 20260715, .time = 0 } });
    // The server's announcement arrived and was recorded (§10.1 r13)…
    try testing.expectEqual(sync.submanifest_min_peer_version, dated.peer_version);
    try testing.expectEqual(sync.ClientState.terminated, dated.state);
    // …so the gate opened and the tree was uploaded, not withheld.
    try testing.expectEqual(@as(usize, 0), dated.refused);
    try testing.expect(dated.server_holds_tree);
    try testing.expect(dated.server_holds_flat);

    // Drop DATE/TIME from the client-version pragma — the CLI's behaviour before
    // this was wired — and the server stays silent about its version (§10.1
    // r12), so the peer record never leaves 0 and the tree is refused to a peer
    // that has, from the client's side of the wire, announced nothing at all.
    const undated = try runGateUpload(gpa, .{ .version = 22700 });
    try testing.expectEqual(@as(u32, 0), undated.peer_version);
    try testing.expectEqual(@as(usize, 1), undated.refused);
    // …and the refusal SETTLES: the peer re-gimmes the tree it never received,
    // and a client that answered by re-queueing it would spin to the driver's
    // round limit instead of stopping with one counted refusal.
    try testing.expectEqual(sync.ClientState.terminated, undated.state);
    try testing.expect(!undated.server_holds_tree);
    // …and the refusal is discriminating: the flat manifest still lands.
    try testing.expect(undated.server_holds_flat);
}