Artifact [feeab7fa31]

Artifact feeab7fa3123f8a9a1e6a43b6feee0e4d68479a6a19eb9374d57edbe9c600d4b:

• File src/sim/sync_sweep.zig — part of check-in [8446adcd98] at 2026-07-28 04:54:13 on branch sub-manifest — Sub-manifest: sync and repository state for trees

  An absent S reference now phantomizes like any other missing reference, its dependent is deferred and retried when the artifact arrives, and the sweep excludes sub-manifests so it still terminates. Phantoms survive a restart through a rebuild-on-open invariant resting on a durable seen-submanifest setting rather than on records the log never wrote. The transfer gate decides by artifact shape on all three send paths, so a leaf is refused wherever a tree manifest is, and the memo returns its own copy rather than a view into pages the store may free.

  The gate remembers what it withheld from a peer, which is what lets an exchange with a peer that refuses sub-manifests terminate at all. That record is now cleared only when the peer's announced version actually changes: a peer that re-announces the same version in every reply was clearing it every round, so the same artifacts were offered and refused forever.

  One requirement is not implemented: a server whose setting is off must send the avoid-submanifests pragma, and emitting it changes the card sequence every clone-reply test walks. That is a protocol-shape change worth its own review rather than a rider here; the pragma is parsed and latched on receipt already. (user: dmestas, size: 81984) [annotate] [blame] [check-ins using]

//! Two-phase deterministic simulation of the sync protocol (transport-simnet-
//! plan.md §A3) — the PRIMARY DST for sync. It drives the full round loop of an
//! A0 `Client`/`Server` pair (`buildRequest` → `respond` → `consumeReply`)
//! through the A1 `SimNet`, which injects the A2 scenario's drop/dup/corrupt/
//! clog faults, and asserts the sync RFC's convergence (§3.7), termination
//! (§3.7/§3.8), and private-content (§7.5) invariants.
//!
//! The seed is the whole universe: one `std.Random.DefaultPrng.init(seed)`
//! feeds the scenario generator AND the network, in a fixed draw order, so any
//! failing sweep replays exactly — dst.zig prints `reproduce with seed=N`.
//!
//! Runs are two-phase (safety → liveness), per the project DST discipline:
//!   Phase 1 (safety): bounded rounds under injected faults. Every round probes
//!   invariants that MUST always hold — no corrupt delivery is ever accepted as
//!   content, held content never regresses, the in-flight queue stays bounded,
//!   and a receive/exchange operation never aborts. A dropped/duplicated/
//!   corrupted packet may lose a round but can never break these.
//!   Phase 2 (liveness): the network is HEALED (all fault probabilities → 0) and
//!   a fresh sync session runs to completion within a BOUNDED convergence window
//!   (rounds, not wall time). The stores MUST converge to the direction-correct
//!   artifact sets, no phantom the peer holds intact may remain on the receiving
//!   side, and the client MUST reach `terminated`. Exceeding the window is a
//!   bug, never flakiness.
//!   Phase 2b (liveness under DEGRADATION — TEST-HARDENING §2 H7): the same
//!   convergence/termination oracles, but the wire is NOT healed — a fresh
//!   session runs under a SUSTAINED fixed drop/dup rate (no corrupt: a corrupted
//!   round is already a modeled lost round) and MUST still converge, within a
//!   WIDER window scaled from the loss rate (`degradedWindow`: expected rounds ≈
//!   clean-window / (1−p_loss)², times a safety factor — a recomputable bound,
//!   not a magic number). A protocol that converges on a perfect wire but
//!   livelocks under loss fails here; exceeding the window is a liveness bug,
//!   never flakiness. The `runDegraded` entry point drives it.
//!
//! Transport model (§4 / §A1): the sans-IO boundary is a whole request/reply
//! BODY. A `SimNet` corruption stands for a body a real transport's zlib
//! container decode (§4.1) would reject, so a `Delivery.corrupted` message is
//! treated as a lost round — it is NEVER handed to `respond`/`consumeReply`.
//! That is why "no corrupt content accepted" holds by construction and a
//! corrupted round degrades to a retry rather than a spurious abort.
//!
//! Crash-during-sync composition (TEST-HARDENING §2 H3) — the `runCrashResume`
//! entry point composes the two fault domains the disjoint `sync_sweep`/
//! `storage_sweep` streams never crossed: it backs the CLIENT store with a real
//! `LogStore` over a tearable `sim/FaultyFile` (the pairing the logstore fuzz
//! targets prove) instead of the pure `MemStore`, and — between phase-1 rounds,
//! with seeded probability — power-cuts that store mid-persist (`crashRestart`),
//! reopens it, reloads the recovered prefix, and RESUMES the exchange with a
//! fresh `Client` over the same durable store rather than aborting the run. The
//! composed oracle is the existing sweep's invariants PLUS the crash-boundary
//! bite: post-recovery client contents ⊆ the pre-crash held set (recovery never
//! fabricates or resurrects content the client never held), no corrupt frame is
//! ever served as truth across the boundary, and phase-2 liveness now heals BOTH
//! domains — net faults → 0 AND disk faults → 0 — before a fresh session must
//! converge within the bounded window. The client's identity space is the sync
//! wire's 64-hex name; the log keys frames by the 32-byte digest that name
//! decodes to (`max_hash_len`), hex-encoded back on reload — the single place
//! binary-log identity becomes wire identity, mirroring cmd/sync's
//! `loadContentStore`. ONE PRNG stream drives the WHOLE universe in a fixed draw
//! order: scenario generation, then per round the SimNet faults, the durable
//! persist's commit coin, and the crash coin — so a failing composed sweep
//! replays byte-identically from the shared `reproduce with seed=N` line.
//!
//! Attribution: the composed-fault (storage × network) simulation and the
//! liveness-under-faults discipline follow TigerBeetle's VOPR — one seeded
//! simulator driving every fault domain against a single deterministic replay.
//! The crash-tear storage model reuses this repo's `sim/FaultyFile`.

const std = @import("std");
const zit = @import("zit");
const sync = zit.sync;
const repostate = zit.repostate;
const logstore = zit.logstore;
/// The artifact layer's §5 tree-walk seam (draft-fossil-submanifests-00): the
/// sub-manifest convergence lane below drives the REAL walk, never a copy.
const artifact = zit.artifact;
const net = @import("net.zig");
const sync_gen = @import("sync_gen.zig");
const faulty_file = @import("faulty_file.zig");

const assert = std.debug.assert;

/// Closed error set: transport/protocol propagations that would signal a real
/// A0/A1/A2 bug if they ever fired here, plus the named invariant violations.
/// `FatalProtocolError`/`NotImplemented` do not appear — an abort is caught and
/// judged by the outer loop, and an unimplemented reply card is a server bug
/// surfaced as `UnexpectedNotImplemented`.
pub const Error = error{
    OutOfMemory,
    PayloadTooLarge,
    QueueFull,
    RequestTooLarge,
    ResponseTooLarge,
    // invariants
    CorruptContentAccepted,
    ContentRegressed,
    InFlightUnbounded,
    UnexpectedAbort,
    UnexpectedNotImplemented,
    RoundTickBudgetExceeded,
    StoresDiverged,
    ResidualPhantom,
    NoTermination,
    ConvergenceWindowExceeded,
    PrivateContentLeaked,
    PrivateStallsTermination,
    // H3 crash-during-sync composition invariants.
    /// A crash-reopen recovered a held hash the client did not hold before the
    /// crash — recovery fabricated or resurrected content (post ⊄ pre).
    RecoveryFabricated,
    /// Reopening the durable client store after a crash failed outright, or the
    /// recovered store served a corrupt frame at reload — a torn tail must
    /// recover to a clean prefix, never to an unreadable store.
    RecoveryOpenFailed,
    // draft-fossil-submanifests-00 §7 sub-manifest convergence invariants.
    /// §7.1 — the tree's closure never completed inside the `depth + 1` pull
    /// rounds the section bounds it by, or the cascade stalled with work left.
    TreeNeverConverged,
    /// §5.1 — the walked enumeration is not the generator's file list: a
    /// converged tree must reproduce it exactly, in order.
    TreeEnumerationDiverged,
    /// §5.2/§7.1 — the closure is incomplete but no phantom is observable, so
    /// nothing would ever be requested and the gap is invisible.
    TreePhantomNotVisible,
    /// §7.3/§9.4 — a deferred tree manifest is not recorded against the
    /// sub-manifest it waits on, or the record outlived the closure.
    TreeSEdgeNotRecorded,
    /// §7.4 — a memoized enumeration differs from the unmemoized one, or the
    /// memo failed to remove the repeated parse it exists to remove.
    TreeMemoDiverged,
    /// §9.2 — a negative-space sweep re-selects a SUBMANIFEST, so it never
    /// reaches a fixpoint.
    TreeSweepDidNotTerminate,
};

/// Phase-1 fault-window length, in rounds. Bounded: no unbounded run.
const phase1_rounds: usize = 32;
/// Phase-2 convergence window, in rounds. A clean bidirectional exchange
/// converges in a handful of rounds (discovery is single-hop — a reply walks
/// the whole store once), so this is generous slack, not a tuned minimum.
const phase2_window: usize = 64;
/// Total round budget for one seed: an explicit upper bound on the whole run.
const total_budget: usize = phase1_rounds + phase2_window;
/// §8.2 L1384 — the sweep server's per-reply clone batch cap, deliberately far
/// below the production default so a generated clone (1..max_server_only server
/// artifacts) routinely pages across MULTIPLE cfile batches. This makes the
/// primary sync DST exercise the §8.2 L1413 multi-batch continuation — the
/// client threading `clone_seqno NEXT` into successive `clone 3 NEXT` requests
/// to exhaustion — on every clone run, not just the single-batch path.
const clone_batch_max_sim: usize = 3;
/// Per-round tick bound while draining one request/reply exchange to
/// quiescence. `max_delay_ticks` + `max_clog_ticks` (both ≤ 64) bound one
/// message; this comfortably bounds the few messages of one round.
const max_round_ticks: usize = 8192;

/// H7 (TEST-HARDENING §2) degraded-network liveness. The GATE runs phase 2b at
/// this single small loss rate so it stays fast and deterministic; the deeper
/// H1 soak sweeps `soak_degraded_pcts`.
const degraded_gate_pct: u8 = 10;

/// H7 soak depth: the loss rates the out-of-band H1 soak sweeps phase 2b across
/// (via `runDegradedSweep`). The gate uses only `degraded_gate_pct`; this wider,
/// documented, configurable set is the soak's job — the gate stays at one small p.
pub const soak_degraded_pcts = [_]u8{ 10, 25, 40 };

/// H7 convergence-window safety factor over the recomputed expectation (see
/// `degradedWindow`). A clean session's `phase2_window` base is already ~6× the
/// real handful-of-rounds need, so this modest factor still leaves the bound many
/// σ above the negative-binomial mean at every `soak_degraded_pcts` rate —
/// exceeding it is a livelock bug, never flakiness.
const degraded_safety_factor: usize = 2;

const Endpoint = net.Endpoint;

/// Per-phase-1-round probability (percent) that the composed sweep power-cuts
/// the durable client store mid-persist and resumes. Chosen so a bounded run
/// (`phase1_rounds`) sees several crashes on average — enough to compose tears
/// with the network faults — without every round paying a reopen. The seed
/// still fixes exactly which rounds crash.
const crash_pct: u8 = 30;

/// The durable client store's crash-tear fault profile, engaged AFTER the
/// healthy initial seed lands (mirrors storage_sweep.run's post-open cfg). H3
/// composes TEARS (short/lost/torn writes) with the network faults; `crash_rot`
/// and `read_rot` stay 0 — latent bit rot on the read path is H8's dimension,
/// and keeping corruption off this store means a torn tail always recovers to a
/// clean prefix (`open`'s torn-tail truncation), never a `Corrupt` reopen.
const durable_faults: faulty_file.FaultCfg = .{
    .short_write_pct = 8,
    .lost_tail_sync_pct = 8,
    .crash_cut_at_durable_pct = 15,
    .sector_round = 512,
    .sector_round_pct = 25,
};

/// Debug-only defect injections for the prove-the-harness-can-fail discipline
/// (TEST-HARDENING §4.1). Never set on the gate path — the permanent
/// "oracle trips" test flips them to confirm the composed oracle fires.
const Inject = struct {
    /// Force a crash EVERY phase-1 round (bypass the seeded crash coin), so a
    /// defect that only bites at a crash boundary is reached deterministically.
    force_crash: bool = false,
    /// On reload, flip one byte of the first recovered held frame — model a
    /// recovery that ACCEPTS a torn frame (held hash, wrong bytes). The
    /// integrity oracle must then fire `CorruptContentAccepted`.
    torn_accept: bool = false,
};

/// The digest width the sync wire's 64-hex artifact name decodes to, and the
/// key width the client's durable `LogStore` frames by (`logstore.max_hash_len`).
const digest_len: usize = logstore.max_hash_len;

/// A durable, tearable backing for the CLIENT store: a `LogStore` over a
/// `FaultyFile`, kept as a SIDECAR to the live `Scenario.client` MemStore (which
/// stays the store the sans-IO protocol drives, so every existing helper works
/// unchanged). `persist` mirrors the live store into the log; a seeded crash
/// tears the unsynced tail, and `crashReopen` reloads the recovered prefix back
/// into a fresh `Scenario.client`. Must not move after `init` — the log borrows
/// `&self.faulty`.
const DurableClient = struct {
    faulty: faulty_file.FaultyFile,
    log: logstore.LogStore,

    /// Open an empty durable store on a HEALTHY disk (faults engage only after
    /// the initial seed is committed).
    fn init(self: *DurableClient, gpa: std.mem.Allocator, random: std.Random) Error!void {
        self.faulty = faulty_file.FaultyFile.init(gpa, random);
        self.log = logstore.LogStore.open(gpa, self.faulty.storageFile(), 0) catch |e| switch (e) {
            error.OutOfMemory => return error.OutOfMemory,
            error.Storage, error.Corrupt => return error.RecoveryOpenFailed,
        };
    }

    fn deinit(self: *DurableClient, gpa: std.mem.Allocator) void {
        self.log.deinit(gpa);
        self.faulty.deinit();
        self.* = undefined;
    }

    /// Durably seed the initial client content on the healthy disk, commit it,
    /// then arm the crash-tear faults. C₀ now sits below `durable_len`, so no
    /// later tear can drop it — only content received DURING sync can tear.
    fn seedFrom(self: *DurableClient, store: repostate.ArtifactStore) Error!void {
        try self.appendHeld(store);
        self.log.commit() catch return error.RecoveryOpenFailed; // healthy disk never fails here
        self.faulty.cfg = durable_faults;
    }

    /// Append every held (content-bearing) entry of `store` into the log,
    /// idempotently (content-addressed: a repeat of an already-stored digest
    /// appends nothing). A torn/short write rolls back in the log — the same
    /// content re-persists on a later round. Phantoms are not durable (§C) and
    /// re-derive from the peer, so they are not written.
    fn appendHeld(self: *DurableClient, store: repostate.ArtifactStore) Error!void {
        const n = store.count();
        var rid: repostate.Rid = 1;
        while (rid <= n) : (rid += 1) {
            const bytes = store.content(rid) orelse continue; // phantom: not durable
            const digest = wireNameToDigest(store.hashOf(rid));
            _ = self.log.put(&digest, bytes) catch |e| switch (e) {
                error.OutOfMemory => return error.OutOfMemory,
                error.Storage => {}, // torn/short write rolled back; re-persists later
            };
        }
    }

    /// fsync the log (best-effort: a lying fsync leaves the tail tearable, which
    /// is the whole point of the crash test).
    fn commit(self: *DurableClient) void {
        self.log.commit() catch {};
    }

    /// Power-cut, reopen, and reload the recovered prefix into a FRESH
    /// `scen.client`. The crash tears the unsynced tail; `open` recovers the
    /// clean committed prefix; the reload re-derives the live MemStore the
    /// protocol drives. `scen.client`'s address is stable, so `clientStore()`
    /// stays valid across the swap.
    fn crashReopen(self: *DurableClient, gpa: std.mem.Allocator, scen: *sync_gen.Scenario, inject: Inject) Error!void {
        self.faulty.crashRestart();
        self.log.deinit(gpa);
        self.log = logstore.LogStore.open(gpa, self.faulty.storageFile(), 0) catch |e| switch (e) {
            error.OutOfMemory => return error.OutOfMemory,
            error.Storage, error.Corrupt => return error.RecoveryOpenFailed,
        };
        scen.client.deinit();
        scen.client = repostate.MemStore.init(gpa);
        try loadInto(gpa, &self.log, &scen.client, inject);
    }

    /// Heal the disk for the phase-2 liveness window (net faults → 0 AND disk
    /// faults → 0).
    fn heal(self: *DurableClient) void {
        self.faulty.cfg = .{};
    }
};

/// Reload every recovered held frame of `log` into `ms`, keyed by the 64-hex
/// wire name its 32-byte digest decodes to. `ridsInPersistedOrder` skips
/// phantoms, so only content survives here. A reopened frame is CRC-validated by
/// `open`, so `content` must succeed; a `Corrupt`/`Storage` return would mean
/// recovery served an unreadable frame — a finding, surfaced as
/// `RecoveryOpenFailed`.
fn loadInto(gpa: std.mem.Allocator, log: *logstore.LogStore, ms: *repostate.MemStore, inject: Inject) Error!void {
    const rids = log.ridsInPersistedOrder(gpa) catch return error.OutOfMemory;
    defer gpa.free(rids);
    var injected = false;
    for (rids) |rid| {
        if (log.kindOf(rid) == .config) continue;
        const name = digestToWireName(log.hashOf(rid));
        const bytes = (log.content(gpa, rid) catch |e| switch (e) {
            error.OutOfMemory => return error.OutOfMemory,
            error.Storage, error.Corrupt => return error.RecoveryOpenFailed,
        }) orelse continue; // rids skip phantoms, so a null here cannot occur
        defer gpa.free(bytes);
        if (inject.torn_accept and !injected and bytes.len > 0) {
            bytes[0] +%= 1; // recovery "accepts" a torn frame: held hash, wrong bytes
            injected = true;
        }
        _ = ms.store().put(&name, bytes) catch return error.OutOfMemory;
    }
}

/// Decode a 64-hex sync wire name to its 32-byte log digest. Generator names are
/// well-formed lowercase hex by construction, so a decode failure is impossible.
fn wireNameToDigest(name: []const u8) [digest_len]u8 {
    assert(name.len == 2 * digest_len);
    var out: [digest_len]u8 = undefined;
    _ = std.fmt.hexToBytes(&out, name) catch unreachable;
    return out;
}

/// Encode a 32-byte log digest back to its 64-hex wire name.
fn digestToWireName(digest: []const u8) [2 * digest_len]u8 {
    assert(digest.len == digest_len);
    return std.fmt.bytesToHex(digest[0..digest_len].*, .lower);
}

/// The seeded phase-1 crash coin. Drawn every phase-1 round (durable path only),
/// so the stream stays fixed regardless of outcome.
fn drawCrash(random: std.Random) bool {
    return random.uintLessThan(u8, 100) < crash_pct;
}

/// H7 phase-2b convergence bound: the maximum rounds a fresh session may take to
/// converge under a SUSTAINED `p_loss`% drop wire, as an EXPLICIT function of the
/// loss rate — a bound someone can recompute, never a magic number (the
/// bounded-waits rule, TEST-HARDENING §4.3).
///
/// Derivation. A round makes protocol progress only if BOTH its request (client→
/// server) AND its reply (server→client) are delivered; each is dropped
/// independently with probability p = p_loss/100, so
///     P(round progresses) = (1 − p)².
/// Duplication cannot lower this — a non-dropped message still arrives, dup only
/// adds a redundant copy — so the drop-only expression is a conservative bound
/// even though the degraded wire also dups. A clean session converges within
/// `phase2_window` progress-rounds, so the rounds needed to accumulate that many
/// under loss is the negative-binomial mean
///     E[rounds] ≈ phase2_window / (1 − p)²  =  phase2_window · 100² / (100 − p)².
/// The window is that expectation × `degraded_safety_factor`, which covers the
/// distribution's upper tail. Worked at the gate p = 10: ⌈64·10000/8100⌉·2 =
/// ⌈79.0⌉·2 = 160 rounds; at 25% → 228; at 40% → 356.
fn degradedWindow(p_loss: u8) usize {
    assert(p_loss < 100);
    const q: usize = 100 - p_loss; // percent delivered per direction, ≥ 1
    const numerator = phase2_window * 100 * 100;
    const denom = q * q; // ≥ 1, so the ceil division below never divides by zero
    const expected = (numerator + denom - 1) / denom; // ⌈numerator / denom⌉
    return expected * degraded_safety_factor;
}

/// The fixed degraded wire phase 2b imposes on BOTH links: a sustained `p`% drop
/// plus `p`% duplication, NO corruption (a corrupted body is already a modeled
/// lost round, §A1) and NO clog. A small mean delay exercises the delivery-delay
/// path within the per-round tick bound. `p` is the drop/loss rate the window
/// bound is computed from; dup only adds idempotent redundancy.
fn degradedLink(p: u8) net.LinkCfg {
    return .{ .drop_pct = p, .dup_pct = p, .corrupt_pct = 0, .clog_pct = 0, .mean_delay_ticks = 2 };
}

// ----------------------------------------------------------------------------
// Public entry point
// ----------------------------------------------------------------------------

/// Run the two-phase sweep for one seed, asserting every invariant. Returns on
/// success; a violation returns the named error. dst.zig calls this per seed in
/// its `zig build dst` gate loop, so `--seed N`/`--seeds K` select depth.
pub fn runSeed(gpa: std.mem.Allocator, seed: u64) Error!void {
    _ = try simulate(gpa, seed, .{});
}

/// H3 (TEST-HARDENING §2) — the crash-during-sync composition: the same
/// two-phase sweep, but with the client store backed by a real `LogStore` over a
/// tearable `FaultyFile` and seeded mid-persist crash+resume between phase-1
/// rounds. dst.zig calls this per seed in the same gate loop as `runSeed`, off
/// the shared reproduce line.
pub fn runCrashResume(gpa: std.mem.Allocator, seed: u64) Error!void {
    _ = try simulateImpl(gpa, seed, .{}, true, .{});
}

/// H7 (TEST-HARDENING §2) — degraded-network liveness, the GATE case: a fresh
/// session must converge and terminate under a SUSTAINED `degraded_gate_pct`%
/// drop/dup wire (never healed) within `degradedWindow`. dst.zig calls this per
/// seed in the same gate loop as `runSeed`; the reproduce line is shared.
pub fn runDegraded(gpa: std.mem.Allocator, seed: u64) Error!void {
    _ = try simulateDegraded(gpa, seed, degraded_gate_pct, .{});
}

/// H7 soak: sweep phase 2b across the wider `soak_degraded_pcts` for one seed.
/// The out-of-band H1 soak calls this; the gate stays at the single small p so
/// it is fast and deterministic.
pub fn runDegradedSweep(gpa: std.mem.Allocator, seed: u64) Error!void {
    for (soak_degraded_pcts) |p| _ = try simulateDegraded(gpa, seed, p, .{});
}

/// The observable result of one run: the client's terminal state and a stable
/// digest of both converged stores — two runs of the same seed hash equal
/// (the reproducibility oracle).
const Outcome = struct { state: sync.ClientState, digest: u64 };

// ----------------------------------------------------------------------------
// The two-phase driver
// ----------------------------------------------------------------------------

/// Generate the scenario, then run phase 1 (faults) and phase 2 (healed),
/// probing safety every round and asserting liveness + convergence at the end.
/// The pure-`MemStore` two-phase sweep; `simulateImpl` adds the H3 durable
/// composition without perturbing this path's PRNG stream.
fn simulate(gpa: std.mem.Allocator, seed: u64, params: sync_gen.Params) Error!Outcome {
    return simulateImpl(gpa, seed, params, false, .{});
}

/// The two-phase driver. `durable` backs the client store with a `LogStore` over
/// a tearable `FaultyFile` and injects seeded mid-persist crash+resume between
/// phase-1 rounds (H3); `inject` carries the prove-the-harness-can-fail defect
/// switches. On the non-durable path NO extra bit is drawn, so the existing gate
/// stream is byte-identical.
fn simulateImpl(gpa: std.mem.Allocator, seed: u64, params: sync_gen.Params, durable: bool, inject: Inject) Error!Outcome {
    var prng = std.Random.DefaultPrng.init(seed);
    const random = prng.random();

    var scen = try sync_gen.generate(gpa, random, params);
    defer scen.deinit();

    // Scratch for the run's oracles: the content-integrity reference and the
    // initial held sets. One arena, freed at return.
    var arena_state = std.heap.ArenaAllocator.init(gpa);
    defer arena_state.deinit();
    const arena = arena_state.allocator();
    const ref = try buildReference(arena, &scen);
    const c0 = try heldHashes(arena, scen.clientStore());
    const s0 = try heldHashes(arena, scen.serverStore());

    // H3: a durable, tearable sidecar for the client store, seeded from the t0
    // client content on a healthy disk, then armed with crash-tear faults. It
    // stays a sidecar to the live `scen.client`, so every helper below drives
    // the same store it always did.
    var dc: DurableClient = undefined;
    var dc_ptr: ?*DurableClient = null;
    if (durable) {
        try dc.init(gpa, random);
        try dc.seedFrom(scen.clientStore());
        dc_ptr = &dc;
    }
    defer if (dc_ptr) |d| d.deinit(gpa);

    var netp = try net.SimNet.init(gpa, random);
    defer netp.deinit();
    netp.links[@intFromEnum(Endpoint.b)] = scen.link_to_b; // client → server
    netp.links[@intFromEnum(Endpoint.a)] = scen.link_to_a; // server → client

    // §6.4 — the server holds the same symmetric shared secret the client signs
    // with, so a named login verifies (a mismatch would be an in-band error
    // card, never a returned error).
    // §8.2/§9.1 — the server carries the clone reply's SERVERCODE/PROJECTCODE.
    // The project code matches the client's, so a clone's `push` card is a
    // well-formed three-token card the peer accepts (an empty field would abort
    // it); servercode is informational.
    const server = sync.Server{
        .version = scen.version,
        .credential = scen.credential,
        .self_servercode = scen.selfServercode(),
        .project_code = scen.projectCode(),
        .clone_batch_max = clone_batch_max_sim,
    };

    var req_buf: std.ArrayListUnmanaged(u8) = .empty;
    defer req_buf.deinit(gpa);
    var rep_buf: std.ArrayListUnmanaged(u8) = .empty;
    defer rep_buf.deinit(gpa);

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

    var prev_c: usize = c0.len;
    var prev_s: usize = s0.len;
    var phase: u8 = 1;
    var phase2_start: usize = 0;
    var round: usize = 0;
    while (round < total_budget) : (round += 1) {
        assert(round < total_budget); // explicit bound: no unbounded run

        // Phase boundary: heal the wire and start a fresh session for the clean
        // convergence run (a real re-invocation of `zit sync`; the durable
        // stores carry phase-1 progress, phantoms re-derive from the store).
        if (phase == 1 and round == phase1_rounds) {
            netp.heal();
            // H3 phase-2 heals BOTH domains: persist+commit the phase-1 progress
            // durably, heal the disk, then one clean crash+reopen so the healed
            // session starts from the fully-durable recovered store.
            if (dc_ptr) |d| {
                d.heal();
                try d.appendHeld(scen.clientStore());
                d.commit();
                try d.crashReopen(gpa, &scen, .{}); // committed: nothing to tear
                prev_c = heldCount(scen.clientStore());
            }
            phase = 2;
            phase2_start = round;
            client.deinit(gpa);
            client = newClient(&scen);
        }

        if (client.state == .terminated or client.state == .aborted) {
            if (phase == 2) break; // the healed session reached its terminal
            // A phase-1 session ended (possibly a packet-loss-induced early
            // stop): re-invoke a fresh session over the same durable stores.
            client.deinit(gpa);
            client = newClient(&scen);
        }

        try runRound(gpa, &netp, &client, &server, &scen, &req_buf, &rep_buf);

        // H3 crash-during-sync: persist what this round received into the durable
        // client store, then — with a seeded coin — power-cut mid-persist and
        // resume. Only phase 1 crashes; phase 2 is the clean convergence window.
        if (dc_ptr) |d| {
            if (phase == 1) {
                try d.appendHeld(scen.clientStore());
                if (random.boolean()) d.commit(); // leave an uncommitted tail to tear
                if (inject.force_crash or drawCrash(random)) {
                    const pre = try heldHashes(arena, scen.clientStore());
                    try d.crashReopen(gpa, &scen, inject);
                    const post = try heldHashes(arena, scen.clientStore());
                    // The crash-boundary bite: recovery holds no hash the client
                    // did not hold before the cut (post ⊆ pre — no fabrication)…
                    if (!subsetSorted(post, pre)) return error.RecoveryFabricated;
                    // …and serves no corrupt frame as content across the boundary.
                    try probeIntegrity(scen.clientStore(), ref);
                    // A crash legitimately drops the unsynced tail, so the
                    // monotonic held baseline resets to the recovered count (it
                    // stays monotonic WITHIN the resumed session).
                    prev_c = heldCount(scen.clientStore());
                    client.deinit(gpa);
                    client = newClient(&scen); // resume: fresh session, same store
                }
            }
        }

        try probeRoundSafety(&scen, ref, &netp, &client, &prev_c, &prev_s);
    }

    // Liveness: a terminal state must be reached within the phase-2 window.
    assert(phase == 2);
    if (client.state != .terminated and client.state != .aborted) return error.ConvergenceWindowExceeded;
    assert(round - phase2_start <= phase2_window);

    try checkTermination(&client, scen.operation);
    try checkConvergence(arena, &scen, c0, s0);
    return .{ .state = client.state, .digest = try outcomeDigest(arena, &scen, client.state) };
}

/// H7 phase-2b defect injection (prove-the-harness-can-fail, TEST-HARDENING
/// §4.1). `window_override`, when set, replaces the computed convergence window;
/// the permanent oracle-trips test sets it BELOW the real convergence need to
/// confirm `ConvergenceWindowExceeded` fires. Never set on the gate path.
const DegradedInject = struct {
    window_override: ?usize = null,
};

/// Phase 2b (H7, TEST-HARDENING §2) — degraded-network liveness. Generate the
/// scenario, impose a SUSTAINED `p_loss`% drop/dup wire on BOTH links that is
/// NEVER healed (the bite: phase 2 proves convergence only at fault probability
/// zero), and drive ONE fresh session, probing the same per-round safety
/// invariants every round and asserting the phase-2 convergence (§3.7) and
/// termination (§3.7/§3.8) oracles at the end — within `degradedWindow(p_loss)`
/// (or `inject.window_override`). Exceeding the window is a livelock/liveness
/// bug (`ConvergenceWindowExceeded`), never flakiness — the phase-2 sentence,
/// extended to p > 0. Its own PRNG universe off the seed, so a failure replays
/// byte-identically from the shared reproduce line.
fn simulateDegraded(gpa: std.mem.Allocator, seed: u64, p_loss: u8, inject: DegradedInject) Error!Outcome {
    var prng = std.Random.DefaultPrng.init(seed);
    const random = prng.random();

    var scen = try sync_gen.generate(gpa, random, .{});
    defer scen.deinit();

    var arena_state = std.heap.ArenaAllocator.init(gpa);
    defer arena_state.deinit();
    const arena = arena_state.allocator();
    const ref = try buildReference(arena, &scen);
    const c0 = try heldHashes(arena, scen.clientStore());
    const s0 = try heldHashes(arena, scen.serverStore());

    var netp = try net.SimNet.init(gpa, random);
    defer netp.deinit();
    // Impose the sustained degraded wire on both directions — NEVER healed.
    netp.links[@intFromEnum(Endpoint.b)] = degradedLink(p_loss); // client → server
    netp.links[@intFromEnum(Endpoint.a)] = degradedLink(p_loss); // server → client

    const server = sync.Server{
        .version = scen.version,
        .credential = scen.credential,
        .self_servercode = scen.selfServercode(),
        .project_code = scen.projectCode(),
        .clone_batch_max = clone_batch_max_sim,
    };

    var req_buf: std.ArrayListUnmanaged(u8) = .empty;
    defer req_buf.deinit(gpa);
    var rep_buf: std.ArrayListUnmanaged(u8) = .empty;
    defer rep_buf.deinit(gpa);

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

    const window = inject.window_override orelse degradedWindow(p_loss);
    var prev_c: usize = c0.len;
    var prev_s: usize = s0.len;
    var round: usize = 0;
    while (round < window) : (round += 1) {
        assert(round < window); // explicit bound: no unbounded run
        // A terminal reached under sustained loss IS convergence for this
        // session (like phase 2, one fresh session runs to completion); a
        // dropped final reply stalls-and-retries rather than terminating early.
        if (client.state == .terminated or client.state == .aborted) break;
        try runRound(gpa, &netp, &client, &server, &scen, &req_buf, &rep_buf);
        try probeRoundSafety(&scen, ref, &netp, &client, &prev_c, &prev_s);
    }

    // Liveness under degradation: a terminal MUST be reached within the wider,
    // loss-scaled window. Exceeding it is a livelock bug, not flakiness.
    if (client.state != .terminated and client.state != .aborted) return error.ConvergenceWindowExceeded;
    try checkTermination(&client, scen.operation);
    try checkConvergence(arena, &scen, c0, s0);
    return .{ .state = client.state, .digest = try outcomeDigest(arena, &scen, client.state) };
}

/// The per-round safety block, probed EVERY round of both phases: no corrupt
/// content accepted, held content never regresses, the in-flight queue stays
/// bounded, and (non-clone) the client never spuriously aborts. `prev_*` carry
/// the monotonic held counts across rounds.
fn probeRoundSafety(
    scen: *sync_gen.Scenario,
    ref: Reference,
    netp: *const net.SimNet,
    client: *const sync.Client,
    prev_c: *usize,
    prev_s: *usize,
) Error!void {
    try probeIntegrity(scen.clientStore(), ref);
    try probeIntegrity(scen.serverStore(), ref);
    const ch = heldCount(scen.clientStore());
    const sh = heldCount(scen.serverStore());
    if (ch < prev_c.* or sh < prev_s.*) return error.ContentRegressed; // content never un-stored
    prev_c.* = ch;
    prev_s.* = sh;
    if (netp.inFlight() > net.SimNet.max_inflight) return error.InFlightUnbounded;
    try probeState(client, scen.operation);
}

/// §3.7/§3.8 termination: every operation, clone included, MUST reach
/// `terminated`. The server now answers §8 clone (paginated cfile batches with a
/// `clone_seqno` cursor to exhaustion), so a clone runs the round loop to a clean
/// §3.7 termination just like pull/push/sync — no operation aborts under
/// clean-or-filtered traffic.
fn checkTermination(client: *const sync.Client, op: sync.Operation) Error!void {
    switch (op) {
        .pull, .push, .sync, .clone => if (client.state != .terminated) return error.NoTermination,
    }
}

/// A fresh client for `scen`'s operation: state seeded from §3.8's initial-state
/// map, identity/version/credential from the scenario. The phantom set and send
/// buffer default empty and re-derive from the durable store on each session.
fn newClient(scen: *const sync_gen.Scenario) sync.Client {
    return .{
        .state = sync.initialState(scen.operation),
        .operation = scen.operation,
        .self_servercode = scen.selfServercode(),
        .project_code = scen.projectCode(),
        .version = scen.version,
        .credential = scen.credential,
    };
}

/// One synchronous request/reply exchange, drained to quiescence: emit the
/// request, drive the net, `respond` to each intact request delivery, and
/// `consumeReply` the first intact reply (later duplicates are ignored — the
/// server is idempotent, §3.8). A corrupted delivery is a transport reject (a
/// lost round). Bounded by `max_round_ticks`.
fn runRound(
    gpa: std.mem.Allocator,
    netp: *net.SimNet,
    client: *sync.Client,
    server: *const sync.Server,
    scen: *sync_gen.Scenario,
    req_buf: *std.ArrayListUnmanaged(u8),
    rep_buf: *std.ArrayListUnmanaged(u8),
) Error!void {
    assert(client.state == .exchange or client.state == .clone_round0);
    const client_store = scen.clientStore();
    const server_store = scen.serverStore();

    const req = try client.buildRequest(gpa, client_store, req_buf);
    try netp.enqueue(.b, req);

    var consumed = false;
    var ticks: usize = 0;
    while (netp.inFlight() > 0) {
        while (netp.deliver()) |d| {
            switch (d.dst) {
                .b => {
                    // Request at the server. A corrupt body a real transport
                    // would reject before decode never reaches `respond`.
                    if (d.corrupted) continue;
                    const reply = try server.respond(gpa, server_store, d.payload, rep_buf);
                    try netp.enqueue(.a, reply);
                },
                .a => {
                    // Reply at the client.
                    if (d.corrupted) continue; // transport reject: a lost round
                    if (consumed) continue; // idempotent dup: consume exactly one
                    if (client.state != .exchange and client.state != .clone_round0) continue;
                    if (client.consumeReply(gpa, client_store, d.payload)) |obs| {
                        consumed = true;
                        // §8.2 — carry the received clone sequence number into
                        // the next request's cursor so clone pages to exhaustion.
                        if (client.operation == .clone) client.clone_seqno = obs.clone_seqno_next;
                    } else |e| switch (e) {
                        error.OutOfMemory => return error.OutOfMemory,
                        // The server emits only full-content file/cfile cards; a
                        // NotImplemented (delta/uv form) reply would be a server
                        // bug, not wire noise.
                        error.NotImplemented => return error.UnexpectedNotImplemented,
                        // §3.8 r18 — the client aborted (`state == .aborted`).
                        // The outer loop judges whether that is legal for the op.
                        error.FatalProtocolError => consumed = true,
                    }
                },
            }
        }
        netp.tick();
        ticks += 1;
        if (ticks > max_round_ticks) return error.RoundTickBudgetExceeded;
    }
}

/// Legal-state + no-spurious-abort probe. Any operation reaching `aborted` under
/// clean-or-filtered traffic implicates `consumeReply` or the A1 corruption
/// filter, so it is a hard failure — clone included now that the server serves
/// §8 clone and the clone client runs to a clean termination.
fn probeState(client: *const sync.Client, op: sync.Operation) Error!void {
    _ = op;
    assert(client.state != .init); // negative space: never regresses to pre-start
    if (client.state == .aborted) return error.UnexpectedAbort;
}

// ----------------------------------------------------------------------------
// Safety oracle: content integrity
// ----------------------------------------------------------------------------

/// hash → the one true content for that hash. Every artifact is content-
/// addressed, so any bytes that appear under a hash must equal these.
const Reference = std.StringHashMapUnmanaged([]const u8);

/// Snapshot every held (hash, content) pair from both initial stores into
/// `arena`. Shared artifacts carry identical bytes on both sides (generator
/// contract), so the first sighting wins.
fn buildReference(arena: std.mem.Allocator, scen: *sync_gen.Scenario) Error!Reference {
    var ref: Reference = .empty;
    try indexReference(arena, scen.clientStore(), &ref);
    try indexReference(arena, scen.serverStore(), &ref);
    return ref;
}

fn indexReference(arena: std.mem.Allocator, store: repostate.ArtifactStore, ref: *Reference) Error!void {
    const n = store.count();
    var rid: repostate.Rid = 1;
    while (rid <= n) : (rid += 1) {
        if (store.content(rid)) |bytes| {
            const h = store.hashOf(rid);
            if (!ref.contains(h)) {
                try ref.put(arena, try arena.dupe(u8, h), try arena.dupe(u8, bytes));
            }
        }
    }
}

/// No corrupt/unknown content was ever interned as truth: every held entry's
/// bytes equal the content-addressed reference for its hash. A held hash absent
/// from the reference (a fabricated artifact) is equally a violation.
fn probeIntegrity(store: repostate.ArtifactStore, ref: Reference) Error!void {
    const n = store.count();
    var rid: repostate.Rid = 1;
    while (rid <= n) : (rid += 1) {
        if (store.content(rid)) |bytes| {
            const want = ref.get(store.hashOf(rid)) orelse return error.CorruptContentAccepted;
            if (!std.mem.eql(u8, bytes, want)) return error.CorruptContentAccepted;
        }
    }
}

// ----------------------------------------------------------------------------
// Convergence oracle (§3.7)
// ----------------------------------------------------------------------------

/// §3.7 — assert the healed run converged, per operation direction:
///   sync : both stores hold C₀ ∪ S₀, and neither retains a phantom.
///   pull : the client (receiver) holds C₀ ∪ S₀ with no phantom; the server's
///          content is unchanged (it keeps dangling phantoms for client-only
///          content it can never pull — expected, one-directional).
///   push : the server (receiver) holds C₀ ∪ S₀ with no phantom; the client
///          keeps at least its own content and no phantom.
/// §3.7(a): no residual phantom whose content the peer holds intact remains on
/// the receiving side(s) — asserted via `noPhantom`.
fn checkConvergence(
    arena: std.mem.Allocator,
    scen: *sync_gen.Scenario,
    c0: []const []const u8,
    s0: []const []const u8,
) Error!void {
    const client_now = try heldHashes(arena, scen.clientStore());
    const server_now = try heldHashes(arena, scen.serverStore());
    const uni = try unionSorted(arena, c0, s0);
    switch (scen.operation) {
        .sync => {
            if (!eqlSorted(client_now, uni) or !eqlSorted(server_now, uni)) return error.StoresDiverged;
            if (!noPhantom(scen.clientStore()) or !noPhantom(scen.serverStore())) return error.ResidualPhantom;
        },
        .pull => {
            if (!eqlSorted(client_now, uni)) return error.StoresDiverged;
            if (!eqlSorted(server_now, s0)) return error.StoresDiverged;
            if (!noPhantom(scen.clientStore())) return error.ResidualPhantom;
        },
        .push => {
            if (!eqlSorted(server_now, uni)) return error.StoresDiverged;
            if (!subsetSorted(c0, client_now)) return error.StoresDiverged;
            if (!noPhantom(scen.serverStore()) or !noPhantom(scen.clientStore())) return error.ResidualPhantom;
        },
        .clone => {
            // §8 — the client (which starts empty, generator contract) fetches
            // the server's whole repository over paginated cfile batches, so it
            // ends holding C₀ ∪ S₀ (= S₀) with no phantom; the server is
            // read-only during clone and ends unchanged at S₀.
            if (!eqlSorted(client_now, uni)) return error.StoresDiverged;
            if (!eqlSorted(server_now, s0)) return error.StoresDiverged;
            if (!noPhantom(scen.clientStore()) or !noPhantom(scen.serverStore())) return error.ResidualPhantom;
        },
    }
}

/// A stable digest of both stores' held sets plus the terminal state — the
/// reproducibility oracle.
fn outcomeDigest(arena: std.mem.Allocator, scen: *sync_gen.Scenario, state: sync.ClientState) Error!u64 {
    var h = std.hash.Wyhash.init(0);
    h.update(&.{@intFromEnum(state)});
    for (try heldHashes(arena, scen.clientStore())) |x| h.update(x);
    h.update("|");
    for (try heldHashes(arena, scen.serverStore())) |x| h.update(x);
    return h.final();
}

// ----------------------------------------------------------------------------
// Held-set helpers (sorted []const u8 sets)
// ----------------------------------------------------------------------------

fn heldCount(store: repostate.ArtifactStore) usize {
    var count: usize = 0;
    const n = store.count();
    var rid: repostate.Rid = 1;
    while (rid <= n) : (rid += 1) {
        if (store.content(rid) != null) count += 1;
    }
    return count;
}

/// True iff no rid in `store` is a phantom (content-absent). §3.3 r24 — a
/// phantom is exactly `content(rid) == null`.
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;
}

/// The held hashes of `store`, duped into `arena` and sorted, so sets compare
/// across runs that assign rids differently.
fn heldHashes(arena: std.mem.Allocator, store: repostate.ArtifactStore) Error![][]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(arena, try arena.dupe(u8, store.hashOf(rid)));
    }
    const slice = try list.toOwnedSlice(arena);
    std.mem.sort([]const u8, slice, {}, lessHash);
    return slice;
}

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

/// Two sorted, duplicate-free sets are equal iff same length and elementwise.
fn eqlSorted(a: []const []const u8, b: []const []const u8) bool {
    if (a.len != b.len) return false;
    for (a, b) |x, y| if (!std.mem.eql(u8, x, y)) return false;
    return true;
}

/// Every element of sorted `sub` appears in sorted `sup` (two-pointer merge).
fn subsetSorted(sub: []const []const u8, sup: []const []const u8) bool {
    var i: usize = 0;
    for (sup) |s| {
        if (i == sub.len) break;
        if (std.mem.eql(u8, sub[i], s)) i += 1;
    }
    return i == sub.len;
}

/// Merge two sorted, duplicate-free sets into their sorted union in `arena`.
fn unionSorted(arena: std.mem.Allocator, a: []const []const u8, b: []const []const u8) Error![][]const u8 {
    var list: std.ArrayListUnmanaged([]const u8) = .empty;
    var i: usize = 0;
    var j: usize = 0;
    while (i < a.len or j < b.len) {
        if (j == b.len) {
            try list.append(arena, a[i]);
            i += 1;
        } else if (i == a.len) {
            try list.append(arena, b[j]);
            j += 1;
        } else switch (std.mem.order(u8, a[i], b[j])) {
            .lt => {
                try list.append(arena, a[i]);
                i += 1;
            },
            .gt => {
                try list.append(arena, b[j]);
                j += 1;
            },
            .eq => {
                try list.append(arena, a[i]);
                i += 1;
                j += 1;
            },
        }
    }
    return list.toOwnedSlice(arena);
}

// ----------------------------------------------------------------------------
// §7.5 private-content oracle
// ----------------------------------------------------------------------------

/// §7.5 — feed a client a server reply carrying a private artifact (a `private`
/// card then a full-content `file`) and report the receive-side outcome:
/// whether it was discarded, whether the store now holds it, and whether the
/// round still reached `terminated`. `granted` is the receiver's Private
/// capability (§7.5 r4/r5).
const PrivateOutcome = struct { discarded: bool, held: bool, terminated: bool };

fn drivePrivateReply(gpa: std.mem.Allocator, granted: bool) Error!PrivateOutcome {
    var store = repostate.MemStore.init(gpa);
    defer store.deinit();
    var client = sync.Client{ .state = .exchange, .operation = .sync, .private_granted = granted };
    defer client.deinit(gpa);

    const hash = "aa00aa00aa00aa00aa00aa00aa00aa00aa00aa00aa00aa00aa00aa00aa00aa00";
    const content = "private-artifact-bytes";
    var reply: std.ArrayListUnmanaged(u8) = .empty;
    defer reply.deinit(gpa);
    try reply.appendSlice(gpa, "private\n"); // §7.5 L2359 — mark the next file private
    const header = try std.fmt.allocPrint(gpa, "file {s} {d}\n", .{ hash, content.len });
    defer gpa.free(header);
    try reply.appendSlice(gpa, header);
    try reply.appendSlice(gpa, content);

    const obs = client.consumeReply(gpa, store.store(), reply.items) catch |e| switch (e) {
        error.OutOfMemory => return error.OutOfMemory,
        // A well-formed reply must parse; either error would itself be the §7.5
        // leak/failure this oracle is here to catch.
        error.FatalProtocolError, error.NotImplemented => return error.PrivateContentLeaked,
    };
    const rid = store.store().lookup(hash);
    const held = rid != null and store.store().content(rid.?) != null;
    return .{
        .discarded = obs.private_content_discarded,
        .held = held,
        .terminated = client.state == .terminated,
    };
}

// ----------------------------------------------------------------------------
// Tests — the invariants ARE the oracle: they encode §3.7 convergence/
// termination and §7.5 privacy. Seeds are literals; the seed reproduces a run.
// ----------------------------------------------------------------------------

const testing = std.testing;

/// A fixed smoke set: a handful of seeds through the full two-phase sweep.
/// Small mix of perfect-wire and heavy-fault scenarios (default `Params`), so
/// safety-under-faults AND liveness-after-heal are both exercised.
const smoke_seeds = [_]u64{ 1, 2, 3, 4, 5, 6, 7, 8 };

test "sync_sweep §3.7: the smoke sweep converges and terminates over every seed" {
    for (smoke_seeds) |seed| try runSeed(testing.allocator, seed);
}

test "sync_sweep §3.7: a forced perfect wire converges and terminates" {
    // No faults in either phase: the pure convergence path must still hold.
    for (smoke_seeds) |seed| _ = try simulate(testing.allocator, seed, .{ .perfect_wire_pct = 100 });
}

test "sync_sweep §3.7: a heavy-fault scenario converges once the wire heals" {
    const heavy = sync_gen.Params{
        .perfect_wire_pct = 0,
        .max_drop_pct = 40,
        .max_dup_pct = 40,
        .max_corrupt_pct = 25,
        .max_clog_pct = 25,
        .max_mean_delay = 8,
    };
    for (smoke_seeds) |seed| _ = try simulate(testing.allocator, seed, heavy);
}

test "sync_sweep: the same seed reproduces an identical run" {
    for (smoke_seeds) |seed| {
        const a = try simulate(testing.allocator, seed, .{});
        const b = try simulate(testing.allocator, seed, .{});
        try testing.expectEqual(a.state, b.state);
        try testing.expectEqual(a.digest, b.digest);
    }
}

test "sync_sweep §8.2 L1384-1413: a clone whose repository spans multiple cfile batches converges in full" {
    const gpa = testing.allocator;
    // The sweep server pages clone at `clone_batch_max_sim`. Find seeds whose
    // generated clone holds MORE server artifacts than one batch carries — the
    // multi-batch path — then drive the full two-phase sweep on that same seed.
    // `simulate` regenerates the identical scenario (deterministic from the
    // seed), so a classified multi-batch seed runs the very run classified.
    var multi_batch_clones: usize = 0;
    var seed: u64 = 0;
    while (seed < 96) : (seed += 1) {
        var prng = std.Random.DefaultPrng.init(seed);
        var scen = try sync_gen.generate(gpa, prng.random(), .{});
        const is_multi = scen.operation == .clone and scen.composition.serverHeld() > clone_batch_max_sim;
        scen.deinit();
        if (!is_multi) continue;
        multi_batch_clones += 1;
        // §8.2 L1413 / §3.7(e) — the whole repository clones across successive
        // batches and reaches a clean terminal; a truncating client would leave
        // the store short of S₀ and fail `checkConvergence`.
        _ = try simulate(gpa, seed, .{});
    }
    // The case must actually be covered: at least one multi-batch clone in range.
    try testing.expect(multi_batch_clones >= 1);
}

test "sync_sweep §7.5(a)(b): private content is not leaked and does not stall termination" {
    // (a) L2383-2385 — without the Private capability the artifact is discarded,
    // never stored, and (b) the discard sets no §3.7 continuation flag, so the
    // round still terminates.
    const denied = try drivePrivateReply(testing.allocator, false);
    try testing.expect(denied.discarded);
    try testing.expect(!denied.held); // §7.5(a): not interned — no leak into the store
    try testing.expect(denied.terminated); // §7.5(b): discard did not stall §3.7 termination

    // Negative space (L2381-2385) — with the capability the SAME reply IS stored
    // and is not discarded, proving the gate is a real decision, not always-drop.
    const allowed = try drivePrivateReply(testing.allocator, true);
    try testing.expect(!allowed.discarded);
    try testing.expect(allowed.held);
    try testing.expect(allowed.terminated);
}

// ----------------------------------------------------------------------------
// H3 (TEST-HARDENING §2) — crash-during-sync composition (storage × network).
// The composed sweep survives every existing invariant AND the crash-boundary
// bite; the seed still replays a run exactly.
// ----------------------------------------------------------------------------

test "sync_sweep H3: the composed storage×network sweep converges across crashes" {
    // The client store is a real LogStore over a tearable FaultyFile, power-cut
    // mid-persist between phase-1 rounds; every seed must still converge and
    // terminate once BOTH domains heal.
    for (smoke_seeds) |seed| try runCrashResume(testing.allocator, seed);
}

test "sync_sweep H3: the same seed reproduces an identical composed run" {
    // One PRNG universe: generation, network faults, the persist commit coin, and
    // the crash coin all draw from it in a fixed order, so a composed run is
    // byte-reproducible from its seed.
    for (smoke_seeds) |seed| {
        const a = try simulateImpl(testing.allocator, seed, .{}, true, .{});
        const b = try simulateImpl(testing.allocator, seed, .{}, true, .{});
        try testing.expectEqual(a.state, b.state);
        try testing.expectEqual(a.digest, b.digest);
    }
}

test "sync_sweep H3: the oracle catches a recovery that accepts a torn frame" {
    // Prove-the-harness-can-fail (TEST-HARDENING §4.1): under the debug switches,
    // every phase-1 round crashes and the reload flips one byte of a recovered
    // held frame — modelling a recovery that ACCEPTS a torn frame (held hash,
    // wrong bytes). The composed oracle MUST fire; a green run here would mean the
    // integrity/convergence check is vacuous. Scan a few seeds so at least one has
    // client content to corrupt at a crash boundary.
    var caught = false;
    var seed: u64 = 1;
    while (seed <= 12 and !caught) : (seed += 1) {
        if (simulateImpl(testing.allocator, seed, .{}, true, .{ .force_crash = true, .torn_accept = true })) |_| {
            // Converged despite the injected torn frame: the oracle missed it.
        } else |_| {
            caught = true; // an Error fired — the oracle is live
        }
    }
    try testing.expect(caught);
}

// ----------------------------------------------------------------------------
// H7 (TEST-HARDENING §2) — liveness under a DEGRADED (not healed) network.
// Phase 2 proves convergence only at fault probability zero; these extend that
// sentence to p > 0: a fresh session must still converge and terminate under a
// sustained drop/dup wire, within a bound recomputed from the loss rate.
// ----------------------------------------------------------------------------

test "sync_sweep H7: a fresh session converges under a sustained degraded wire" {
    // The gate case: every seed converges and terminates under a NEVER-healed
    // `degraded_gate_pct`% drop/dup wire, within the loss-scaled window.
    for (smoke_seeds) |seed| try runDegraded(testing.allocator, seed);
}

test "sync_sweep H7: convergence holds across the soak loss-rate sweep" {
    // The wider soak depth: every seed still converges at each of
    // `soak_degraded_pcts` (10/25/40%), each with its own recomputed window.
    for (smoke_seeds) |seed| try runDegradedSweep(testing.allocator, seed);
}

test "sync_sweep H7: the same seed reproduces an identical degraded run" {
    // One PRNG universe (generation + the sustained-fault schedule) off the
    // seed, so a degraded run is byte-reproducible: same seed ⇒ same verdict.
    for (smoke_seeds) |seed| {
        const a = try simulateDegraded(testing.allocator, seed, degraded_gate_pct, .{});
        const b = try simulateDegraded(testing.allocator, seed, degraded_gate_pct, .{});
        try testing.expectEqual(a.state, b.state);
        try testing.expectEqual(a.digest, b.digest);
    }
}

test "sync_sweep H7: the liveness oracle trips when the window is below the convergence need" {
    // Prove-the-harness-can-fail (TEST-HARDENING §4.1): pin the window FAR below
    // any non-trivial scenario's real convergence need under a 40% drop wire and
    // confirm `ConvergenceWindowExceeded` fires. A green run here would mean the
    // liveness bound is vacuous. Scan a few seeds so at least one has enough to
    // exchange that a single round cannot finish it.
    var caught = false;
    var seed: u64 = 1;
    while (seed <= 12 and !caught) : (seed += 1) {
        if (simulateDegraded(testing.allocator, seed, 40, .{ .window_override = 1 })) |_| {
            // Converged inside one round: nothing to prove on this seed.
        } else |e| {
            try testing.expectEqual(Error.ConvergenceWindowExceeded, e);
            caught = true; // the liveness oracle is live
        }
    }
    try testing.expect(caught);
}

// ---------------------------------------------------------------------------
// draft-fossil-submanifests-00 §7 — sub-manifest convergence sweep.
//
// A second, independent lane over the same seeded-universe discipline as the
// protocol sweep above: one PRNG builds a whole sub-manifest tree, a "client"
// starts holding only the tree manifest, and the harness drives the §7.1
// phantom / §7.3 deferral / §9.4 dephantomization cascade round by round until
// the tree's closure is complete.
//
// The universe is synthetic at the §5 SEAM, not at the byte level: nodes are
// `artifact.SubManifestView` values driven through the REAL `artifact.TreeWalk`
// (slice 2's §5.1 walk, unmodified). Card syntax and vector fidelity are the
// artifact layer's own sweeps; what this lane owns is the §7 question those
// cannot ask — does an incomplete closure CONVERGE, and does it ever leak a
// partial file set on the way?
//
// The store is a real `repostate.MemStore`, so "phantom" and "held" mean here
// exactly what they mean on the receive path: an interned rid with no content
// versus one with content. Delivered nodes carry a one-byte stand-in payload —
// the sweep asserts nothing about bytes, and using real artifact bytes would
// re-test the parser instead of the cascade.
// ---------------------------------------------------------------------------

/// Upper bound on generated file entries. Small on purpose: the property is
/// structural (does the cascade close?), and a big tree costs rounds without
/// widening the shape space.
const tree_max_files: usize = 24;
/// Upper bound on members per generated group, at every height.
const tree_max_fanout: usize = 4;
/// Ceiling on generated interior levels, well inside §5.1's depth bound of 16.
const tree_max_levels: usize = 4;

/// One generated sub-manifest: its name, its key (§5.1's first-filename
/// identity), and the view the resolver hands back once it is held.
const TreeNode = struct {
    hash: []const u8,
    key: []const u8,
    view: artifact.SubManifestView,
};

/// A whole generated tree: the manifest's S run, every sub-manifest it reaches,
/// and the enumeration the §5.1 walk must reproduce exactly.
const TreeUniverse = struct {
    root_hash: []const u8,
    root: []const artifact.SCard,
    nodes: []const TreeNode,
    files: []const []const u8,
    /// Interior levels below the manifest — §7.1's d, so the convergence bound
    /// is `depth + 1` pull rounds.
    depth: usize,

    fn byHash(self: *const TreeUniverse, hash: []const u8) ?*const TreeNode {
        for (self.nodes) |*n| {
            if (std.mem.eql(u8, n.hash, hash)) return n;
        }
        return null;
    }
};

/// Deterministic 64-hex synthetic artifact name for the `n`-th generated node.
/// Distinct per n and never equal to a real hash — the sweep never hashes bytes.
fn treeName(arena: std.mem.Allocator, n: usize) Error![]const u8 {
    const buf = arena.alloc(u8, 64) catch return error.OutOfMemory;
    _ = std.fmt.bufPrint(buf, "{x:0>64}", .{n + 1}) catch unreachable;
    return buf;
}

/// Build one tree from the seeded stream: sorted file names partitioned into
/// leaf groups, then levels of interior groups, then the manifest's S run.
///
/// Every §5.1 structural precondition is met by construction, so a failed walk
/// in this sweep is always a cascade bug and never a malformed fixture: groups
/// are contiguous slices of the sorted name space (so each level's keys
/// strictly ascend), and a node's key is the first file name in its subtree
/// (so first-filename identity holds at every S card).
fn generateTree(arena: std.mem.Allocator, random: std.Random) Error!TreeUniverse {
    const n_files = 2 + random.uintLessThan(usize, tree_max_files - 1);
    const files = arena.alloc([]const u8, n_files) catch return error.OutOfMemory;
    for (files, 0..) |*f, i| {
        const buf = arena.alloc(u8, 8) catch return error.OutOfMemory;
        f.* = std.fmt.bufPrint(buf, "f{d:0>5}", .{i}) catch unreachable;
    }

    var next_name: usize = 0;
    var nodes: std.ArrayListUnmanaged(TreeNode) = .empty;

    // Height 0: contiguous groups of file entries become leaf sub-manifests.
    var level: std.ArrayListUnmanaged(TreeNode) = .empty;
    var i: usize = 0;
    while (i < files.len) {
        const take = @min(files.len - i, 1 + random.uintLessThan(usize, tree_max_fanout));
        const cards = arena.alloc(artifact.FCard, take) catch return error.OutOfMemory;
        for (cards, files[i .. i + take]) |*c, name| c.* = .{
            .filename = name,
            .hash = "0" ** 64,
            .perm = null,
            .old_name = null,
            .mode = .regular,
        };
        const node: TreeNode = .{
            .hash = try treeName(arena, next_name),
            .key = files[i],
            .view = .{ .files = cards, .subs = &.{} },
        };
        next_name += 1;
        level.append(arena, node) catch return error.OutOfMemory;
        nodes.append(arena, node) catch return error.OutOfMemory;
        i += take;
    }

    // Heights 1..: contiguous groups of the level below become interior nodes.
    // Stops on a non-reducing partition (every group a single child), which
    // would only lengthen the spine without changing what the cascade must do.
    var depth: usize = 1;
    while (level.items.len > 1 and depth < tree_max_levels) {
        var up: std.ArrayListUnmanaged(TreeNode) = .empty;
        // Rewind marker: a non-reducing partition is DISCARDED below, so the
        // nodes it built must leave `nodes` too — an unreachable node would
        // make the §7.4 distinct-artifact count a lie.
        const nodes_before = nodes.items.len;
        var j: usize = 0;
        while (j < level.items.len) {
            const take = @min(level.items.len - j, 1 + random.uintLessThan(usize, tree_max_fanout));
            const cards = arena.alloc(artifact.SCard, take) catch return error.OutOfMemory;
            for (cards, level.items[j .. j + take]) |*c, child| c.* = .{ .filename = child.key, .hash = child.hash };
            const node: TreeNode = .{
                .hash = try treeName(arena, next_name),
                .key = level.items[j].key,
                .view = .{ .files = &.{}, .subs = cards },
            };
            next_name += 1;
            up.append(arena, node) catch return error.OutOfMemory;
            nodes.append(arena, node) catch return error.OutOfMemory;
            j += take;
        }
        if (up.items.len == level.items.len) {
            nodes.shrinkRetainingCapacity(nodes_before); // non-reducing: stop here
            break;
        }
        level = up;
        depth += 1;
    }

    const root = arena.alloc(artifact.SCard, level.items.len) catch return error.OutOfMemory;
    for (root, level.items) |*c, child| c.* = .{ .filename = child.key, .hash = child.hash };

    return .{
        .root_hash = try treeName(arena, next_name),
        .root = root,
        .nodes = nodes.items,
        .files = files,
        .depth = depth,
    };
}

/// §5.2 resolver over the CLIENT's holdings: a node resolves only once it has
/// actually been delivered. Absent and wrong-type collapse to null exactly as
/// the repository-side resolver does.
const TreeClient = struct {
    univ: *const TreeUniverse,
    store: repostate.ArtifactStore,
    /// Inner-seam call count — the parse count §7.4 says a memo removes.
    calls: usize = 0,

    fn resolve(ctx: *anyopaque, hash: []const u8) ?artifact.SubManifestView {
        const self: *TreeClient = @ptrCast(@alignCast(ctx));
        self.calls += 1;
        const rid = self.store.lookup(hash) orelse return null;
        if (!self.store.held(rid)) return null; // §7.1: a phantom is unresolvable
        const node = self.univ.byHash(hash) orelse return null;
        return node.view;
    }

    fn seam(self: *TreeClient) artifact.SubResolver {
        return .{ .ctx = self, .resolveFn = TreeClient.resolve };
    }
};

/// Run the §5.1 walk and return the enumeration, or null on ANY §5 fault.
///
/// The null is where §5.2's "MUST NOT present a partially resolved tree as a
/// complete file set" is enforced for this harness: a fault discards the whole
/// accumulator inside this function, so no caller can observe a prefix.
fn treeEnumerate(
    arena: std.mem.Allocator,
    subs: []const artifact.SCard,
    resolver: artifact.SubResolver,
) Error!?[][]const u8 {
    var out: std.ArrayListUnmanaged([]const u8) = .empty;
    var walk = artifact.TreeWalk.init(subs, resolver);
    while (walk.next() catch return null) |f| {
        out.append(arena, f.filename) catch return error.OutOfMemory;
    }
    return out.items;
}

const TreeOutcome = struct { rounds: usize, nodes: usize, digest: u64 };

pub fn runTreeSeed(gpa: std.mem.Allocator, seed: u64) Error!void {
    _ = try treeSimulate(gpa, seed, true);
}

fn treeOutcome(gpa: std.mem.Allocator, seed: u64) Error!TreeOutcome {
    return treeSimulate(gpa, seed, true);
}

/// The §9.4 negative control: the same run with the THIRD edge kind disarmed.
/// A cascade that walks only baselines and delta sources never rediscovers the
/// tree manifest, so the run must fail to converge rather than quietly succeed.
fn runTreeSeedWithoutSEdge(gpa: std.mem.Allocator, seed: u64) Error!void {
    _ = try treeSimulate(gpa, seed, false);
}

/// One seeded convergence run.
///
/// `traverse_s_edge` selects whether the dephantomization path walks the S
/// reference record (§9.4). The whole loop is driven off that record: the tree
/// manifest is re-attempted ONLY when an arrival hands it back, which is
/// exactly what §9.4 requires an implementation to make possible and exactly
/// what a two-edge-kind cascade cannot do.
fn treeSimulate(gpa: std.mem.Allocator, seed: u64, traverse_s_edge: bool) Error!TreeOutcome {
    var scratch = std.heap.ArenaAllocator.init(gpa);
    defer scratch.deinit();
    const arena = scratch.allocator();

    var prng = std.Random.DefaultPrng.init(seed);
    const random = prng.random();
    const univ = try generateTree(arena, random);

    var ms = repostate.MemStore.init(gpa);
    defer ms.deinit();
    const store = ms.store();
    var td = repostate.TreeDeferrals.init(gpa, store);
    defer td.deinit();

    var client = TreeClient{ .univ = &univ, .store = store };

    // The tree manifest is stored first; §7.1 mints a phantom for each of its
    // absent S references, which is what the first pull round then requests.
    _ = store.put(univ.root_hash, "M") catch return error.OutOfMemory;
    var root_refs: std.ArrayListUnmanaged([]const u8) = .empty;
    for (univ.root) |c| root_refs.append(arena, c.hash) catch return error.OutOfMemory;
    const absent0 = td.phantomize(root_refs.items) catch return error.OutOfMemory;
    assert(absent0 == univ.root.len); // nothing is held yet

    // §7.3 — the gap is recorded keyed by dependent and missing hash, so the
    // manifest is rediscoverable the moment any of them arrives.
    for (root_refs.items) |h| td.deferProcessing(univ.root_hash, h) catch return error.OutOfMemory;

    // §7.1 — "a tree of depth d converges in at most d+1 pull rounds".
    const budget = univ.depth + 1;
    var rounds: usize = 0;
    var pending: std.ArrayListUnmanaged([]const u8) = .empty;
    pending.append(arena, univ.root_hash) catch return error.OutOfMemory;
    var converged = false;

    while (rounds <= budget) : (rounds += 1) {
        // Process every dependent the cascade currently owes work on.
        for (pending.items) |dep| {
            if (!std.mem.eql(u8, dep, univ.root_hash)) continue; // only the manifest carries crosslink work (§7.3)
            if (try treeEnumerate(arena, univ.root, client.seam())) |names| {
                if (names.len != univ.files.len) return error.TreeEnumerationDiverged;
                for (names, univ.files) |got, want| {
                    if (!std.mem.eql(u8, got, want)) return error.TreeEnumerationDiverged;
                }
                converged = true;
            } else {
                // §5.2 hard failure: the closure is incomplete, so a phantom
                // MUST be observable and the dependent MUST still be deferred.
                if (!treeAnyPhantom(store)) return error.TreePhantomNotVisible;
                if (!td.isDeferred(dep)) return error.TreeSEdgeNotRecorded;
            }
        }
        if (converged) break;

        // One pull round: every phantom is requested and the peer holds it all.
        // Snapshot the rid range first — delivering a node phantomizes its own
        // children (§7.1), which appends to the store mid-round.
        pending.clearRetainingCapacity();
        const before = store.count();
        var rid: repostate.Rid = 1;
        var delivered: usize = 0;
        while (rid <= before) : (rid += 1) {
            if (store.held(rid)) continue;
            const hash = arena.dupe(u8, store.hashOf(rid)) catch return error.OutOfMemory;
            const node = univ.byHash(hash) orelse continue;
            _ = store.put(hash, "S") catch return error.OutOfMemory;
            delivered += 1;
            // §7.1 — the arriving node's own S references become known now.
            var child_refs: std.ArrayListUnmanaged([]const u8) = .empty;
            for (node.view.subs) |c| child_refs.append(arena, c.hash) catch return error.OutOfMemory;
            _ = td.phantomize(child_refs.items) catch return error.OutOfMemory;
            // §9.4 — the dephantomization path traverses the S-reference record.
            if (traverse_s_edge) {
                var retry: std.ArrayListUnmanaged(repostate.Rid) = .empty;
                defer retry.deinit(gpa);
                td.onArrival(hash, &retry) catch return error.OutOfMemory;
                for (retry.items) |dep_rid| {
                    const dep = arena.dupe(u8, store.hashOf(dep_rid)) catch return error.OutOfMemory;
                    pending.append(arena, dep) catch return error.OutOfMemory;
                }
            }
        }
        if (delivered == 0) return error.TreeNeverConverged; // nothing left to deliver, still incomplete

        // §7.3 — a rediscovered dependent whose closure is still incomplete
        // re-records its gap against whatever is still missing, so the next
        // arrival rediscovers it again. Without this the cascade stops one
        // level short, which is the §9.4 failure in miniature.
        for (pending.items) |dep| {
            var p: repostate.Rid = 1;
            const total = store.count();
            while (p <= total) : (p += 1) {
                if (store.held(p)) continue;
                const missing = arena.dupe(u8, store.hashOf(p)) catch return error.OutOfMemory;
                td.deferProcessing(dep, missing) catch return error.OutOfMemory;
            }
        }
    }
    if (!converged) return error.TreeNeverConverged;

    // Convergence oracles.
    if (rounds > budget) return error.TreeNeverConverged; // §7.1's d+1 bound
    if (treeAnyPhantom(store)) return error.TreePhantomNotVisible; // closure complete ⇒ no gap left
    if (td.gapCount() != 0 or td.deferredCount() != 0) return error.TreeSEdgeNotRecorded;

    try treeCheckMemo(arena, &univ, &client);
    try treeCheckSweepTerminates(arena, &univ);

    var digest: u64 = std.hash.Wyhash.hash(seed, "submanifest-tree");
    digest ^= @as(u64, rounds) *% 0x9E37_79B9_7F4A_7C15;
    digest ^= @as(u64, univ.nodes.len) *% 0xBF58_476D_1CE4_E5B9;
    for (univ.files) |f| digest = std.hash.Wyhash.hash(digest, f);
    return .{ .rounds = rounds, .nodes = univ.nodes.len, .digest = digest };
}

/// True while any interned rid still carries no content — §7.1's phantom,
/// visible to the harness exactly as it is to the §3.3 request composer.
fn treeAnyPhantom(store: repostate.ArtifactStore) bool {
    var rid: repostate.Rid = 1;
    const total = store.count();
    while (rid <= total) : (rid += 1) {
        if (!store.held(rid)) return true;
    }
    return false;
}

/// §7.4 — the memo must be BOTH transparent and effective: two rebuilds over
/// the same converged tree (the shared-subtree case the section is about) must
/// enumerate identically to the unmemoized walk, and must reach the inner seam
/// once per DISTINCT sub-manifest rather than once per visit.
///
/// The equality half is what makes the memo safe to enable by default: a cached
/// answer that ever differed from a fresh one would show up here on every seed.
fn treeCheckMemo(arena: std.mem.Allocator, univ: *const TreeUniverse, client: *TreeClient) Error!void {
    const cold = (try treeEnumerate(arena, univ.root, client.seam())) orelse return error.TreeEnumerationDiverged;

    const before = client.calls;
    var memo = repostate.SubManifestMemo.init(arena, client.seam());
    defer memo.deinit();

    const first = (try treeEnumerate(arena, univ.root, memo.resolver())) orelse return error.TreeEnumerationDiverged;
    const second = (try treeEnumerate(arena, univ.root, memo.resolver())) orelse return error.TreeEnumerationDiverged;
    if (first.len != cold.len or second.len != cold.len) return error.TreeMemoDiverged;
    for (cold, first, second) |want, a, b| {
        if (!std.mem.eql(u8, want, a) or !std.mem.eql(u8, want, b)) return error.TreeMemoDiverged;
    }
    // Each distinct sub-manifest reached the inner seam exactly once across the
    // two rebuilds; the second rebuild is served entirely from the memo.
    if (client.calls - before != univ.nodes.len) return error.TreeMemoDiverged;
    if (memo.misses != univ.nodes.len or memo.hits != univ.nodes.len) return error.TreeMemoDiverged;
    if (memo.degraded) return error.TreeMemoDiverged;
}

/// §9.2 — a MODEL of the negative-space crosslink sweep, run to a fixpoint.
///
/// §9.2 describes a concrete algorithm: "An implementation that discovers
/// crosslink work by negative space -- selecting every structural artifact for
/// which no derived row exists -- MUST exclude SUBMANIFEST artifacts from the
/// sweep, or record an explicit completion marker for them; otherwise it
/// re-selects and re-parses them on every pass, forever."
///
/// So this runs that algorithm, rather than asserting the admission rule to
/// itself. Each pass selects the artifacts with no derived row, "processes"
/// them, and counts the parses; a manifest processed this way gains a derived
/// row, and a SUBMANIFEST — §7.3: "has no direct crosslink effect ... carries no
/// state-bearing card" — gains none, which is exactly why negative space alone
/// re-selects it forever. The oracle is that the loop reaches a pass selecting
/// nothing, and the parse total is bounded by the artifact count. Both are
/// falsifiable: `apply_exclusion = false` makes the same model diverge, which
/// the negative control below asserts.
///
/// Returns the total parses the sweep performed.
fn treeRunNegativeSpaceSweep(
    arena: std.mem.Allocator,
    univ: *const TreeUniverse,
    /// Whether the sweep applies §9.2's admission rule. False is the
    /// unconforming implementation the section is written against.
    apply_exclusion: bool,
) Error!usize {
    // The universe under sweep: the tree manifest plus every sub-manifest it
    // reaches — the artifacts a crosslink pass over this check-in would see.
    const total = univ.nodes.len + 1;
    const kinds = arena.alloc(artifact.ArtifactType, total) catch return error.OutOfMemory;
    const has_row = arena.alloc(bool, total) catch return error.OutOfMemory;
    kinds[0] = .manifest;
    for (kinds[1..]) |*k| k.* = .submanifest;
    for (has_row) |*r| r.* = false;

    // Bounded: a conforming sweep needs two passes (one that crosslinks the
    // manifest, one that finds nothing). Anything beyond `total + 2` is the
    // "forever" §9.2 names, and the bound is what turns it into a finite
    // failure instead of a hung test.
    const max_passes = total + 2;
    var parses: usize = 0;
    var pass: usize = 0;
    while (pass < max_passes) : (pass += 1) {
        var selected: usize = 0;
        for (kinds, has_row) |kind, *row| {
            if (row.*) continue; // a derived row exists: negative space skips it
            if (apply_exclusion and repostate.sweepDisposition(kind) == .complete_on_arrival) {
                continue; // §9.2 — excluded from the sweep
            }
            selected += 1;
            parses += 1; // the re-parse §9.2 says an unconforming sweep repeats
            // §7.3 — only an artifact with a direct crosslink effect leaves a
            // derived row behind, which is what lets the fixpoint be reached.
            if (repostate.sweepDisposition(kind) == .crosslink) row.* = true;
        }
        if (selected == 0) return parses; // fixpoint
    }
    return error.TreeSweepDidNotTerminate;
}

/// §9.2 — the sweep terminates, and terminates having parsed each artifact at
/// most once. The parse bound is the second half of the section's claim: a
/// sweep that terminated only because the pass bound saved it would still be
/// re-parsing, so counting the parses is what makes "forever" observable.
fn treeCheckSweepTerminates(arena: std.mem.Allocator, univ: *const TreeUniverse) Error!void {
    const parses = try treeRunNegativeSpaceSweep(arena, univ, true);
    // One parse: the tree manifest. Every sub-manifest is excluded and never
    // parsed at all, which is the cost reduction §9.2 exists to protect.
    if (parses != 1) return error.TreeSweepDidNotTerminate;
    // §9.2 narrows the sweep to the sub-manifests; it does not exempt the
    // check-in, which must still be crosslinked exactly once.
    if (repostate.sweepDisposition(.manifest) != .crosslink) return error.TreeSweepDidNotTerminate;
}

test "sync_sweep submanifests §7.1: a tree converges across pull rounds over every smoke seed" {
    var seed: u64 = 1;
    while (seed <= 64) : (seed += 1) try runTreeSeed(std.testing.allocator, seed);
}

test "sync_sweep submanifests §7.1: the same seed reproduces an identical tree run" {
    const a = try treeOutcome(std.testing.allocator, 0xC0FFEE);
    const b = try treeOutcome(std.testing.allocator, 0xC0FFEE);
    try std.testing.expectEqual(a.rounds, b.rounds);
    try std.testing.expectEqual(a.nodes, b.nodes);
    try std.testing.expectEqual(a.digest, b.digest);
}

test "sync_sweep submanifests §9.2: the sweep oracle trips when SUBMANIFEST is not excluded" {
    // The negative control that makes the §9.2 lane evidence rather than a
    // restatement: run the SAME model sweep without the exclusion and it must
    // fail to reach a fixpoint — "it re-selects and re-parses them on every
    // pass, forever." A sweep model that could not diverge would prove nothing
    // about a sweep that can.
    var scratch = std.heap.ArenaAllocator.init(std.testing.allocator);
    defer scratch.deinit();
    const arena = scratch.allocator();
    var prng = std.Random.DefaultPrng.init(11);
    const univ = try generateTree(arena, prng.random());
    try std.testing.expect(univ.nodes.len > 0);

    try std.testing.expectError(
        error.TreeSweepDidNotTerminate,
        treeRunNegativeSpaceSweep(arena, &univ, false),
    );
    // With the exclusion applied, the same universe reaches a fixpoint having
    // parsed the check-in manifest once and no sub-manifest at all.
    try std.testing.expectEqual(@as(usize, 1), try treeRunNegativeSpaceSweep(arena, &univ, true));
}

test "sync_sweep submanifests §9.4: the oracle trips when the S edge is not traversed" {
    // §9.4: "a cascade that walks only the two older edge kinds leaves such a
    // check-in permanently unprocessed." Disarming the third edge must be
    // caught, not silently tolerated — otherwise the sweep proves nothing.
    try std.testing.expectError(
        error.TreeNeverConverged,
        runTreeSeedWithoutSEdge(std.testing.allocator, 7),
    );
}