minerva 0.2.0

//! The fleet carriage of recorded membership (PRD 0028, sequencing step
//! four): a joiner carried through the whole shipped recipe, end to end.
//!
//! The unit pins in `crate::metis::tests::arrival` prove the machine's
//! doors; these exhibits prove the *recipe* --- the roster read from the
//! record just sealed, both trackers widening in one act, the joiner's
//! allocator opening at bottom, and a checkpoint re-stamped for a station
//! that never held one. R-35's two charter-waiting falsifiers are
//! discharged here: falsifier 6 (join-during-window) by the headline
//! exhibit's sealed join, and falsifier 5 (partition-heal wrong-family,
//! with the delayed cross-epoch replay as its strengthened form) by the
//! late-tail and replay exhibits. PRD 0028 falsifier 7 lands whole: the
//! attested fleet admits, the racing attestation moves no record byte,
//! and the round frozen by an attestation prices the arrival one
//! boundary. Falsifier 8's verdict is in `carry_departures` and pinned
//! at `crate::metis::tests::departure`: riding the record *relocates*
//! the S340 cure (the attestation re-keys over the family the new plane
//! opens with) rather than retiring it. The R-89 counterweight --- a
//! locally abandoned member stalls a tagged window at its own adopt door
//! --- is exhibited live, with its caller-side cure.

extern crate alloc;

use alloc::collections::BTreeMap;
use alloc::vec::Vec;

use core::num::NonZeroUsize;

use crate::metis::tests::support::dot as d;
use crate::metis::{ArrivalRound, SealRecord, VersionVector};

use super::super::fabric::Fabric;
use super::super::replica::Replica;
use super::super::{Note, act, assert_converged, crash, fleet_of, lineage_of, resync};
use super::{ROSTER, condense_at, horizon};

/// One write from every present member, drained, then a seal declared
/// from station 1 and drained: the generic boundary crossing the arrival
/// exhibits compose over.
fn seal_once(fabric: &mut Fabric, fleet: &mut BTreeMap<u32, Replica>) {
    let stations: Vec<u32> = fleet.keys().copied().collect();
    for &station in &stations {
        let _ = act(fabric, fleet, station, |replica, out| {
            replica.insert_visible(0, out)
        });
    }
    fabric.drain(fleet);
    let _ = act(fabric, fleet, 1, Replica::try_declare)
        .expect("the settled plane declares its boundary");
    fabric.drain(fleet);
}

/// Every named member's operator admits `joiners`, and the fabric carries
/// the endorsements until each admitting member's round is complete.
fn admit_at(
    fabric: &mut Fabric,
    fleet: &mut BTreeMap<u32, Replica>,
    members: &[u32],
    joiners: &[u32],
) {
    for &station in members {
        act(fabric, fleet, station, |replica, out| {
            replica.open_arrival(joiners, out)
        })
        .expect("an incumbent's admission opens over the newest seal");
    }
    fabric.drain(fleet);
    for &station in members {
        let round = fleet[&station]
            .epochs()
            .arrival()
            .expect("the admitting member holds its round");
        assert!(
            round.complete(),
            "station {station}: the round completes once every family word folds"
        );
    }
}

/// Installs the admitted joiner from an incumbent's re-stamped checkpoint
/// and the lineage proof that admits it, then repairs it into the live
/// fleet: the carriage recipe under test.
fn install_admitted_joiner(
    fabric: &mut Fabric,
    fleet: &mut BTreeMap<u32, Replica>,
    station: u32,
    depth: NonZeroUsize,
) {
    fabric.admit(station);
    let checkpoint = fleet[&1]
        .joiner_checkpoint_for(station)
        .expect("an incumbent at its seal re-stamps for the admitted station");
    let lineage = lineage_of(&fleet[&1]);
    let joiner = Replica::bootstrap(checkpoint, &ROSTER, depth, &lineage)
        .expect("the joiner bootstraps from the very proof that admits it");
    assert_eq!(
        joiner.epochs(),
        fleet[&1].epochs(),
        "the joiner's recognizer agrees with the incumbents'"
    );
    assert_eq!(joiner.effective_order(), fleet[&1].effective_order());
    let _ = fleet.insert(station, joiner);
    resync(fabric, fleet, station);
    fabric.drain(fleet);
}

/// The headline carriage: three incumbents admit station 4, the winner
/// carries the boundary, the seal widens the roster, and the joiner
/// bootstraps from a re-stamped checkpoint, mints from bottom, and
/// co-seals the next window.
///
/// R-35 falsifier 6 (join-during-window) is discharged at the sealed
/// join: the admitting window's record carries no station-4 entry, so the
/// joiner counted nothing toward the minima of the window that admitted
/// it --- membership starts at the successor generation, by construction
/// rather than by staging rule. The delayed cross-epoch replay (the
/// strengthened half of falsifier 5) rides the tail: a pre-admission
/// old-plane note replayed at the joiner reads against the retained
/// record of its own generation and teaches nothing.
#[test]
fn an_admitted_joiner_bootstraps_at_bottom_and_co_seals() {
    let depth = horizon(4);
    let mut fleet = fleet_of(&ROSTER, depth);
    let mut fabric = Fabric::new(0xA221_7001, &ROSTER, 8);
    seal_once(&mut fabric, &mut fleet);
    assert_converged(&fleet);

    admit_at(&mut fabric, &mut fleet, &ROSTER, &[4]);

    // Window traffic, and one pre-admission note kept for the replay tail.
    let mut replay = None;
    for station in ROSTER {
        let replica = fleet.get_mut(&station).expect("roster member");
        let mut outbox = Vec::new();
        let _ = replica.insert_visible(0, &mut outbox);
        let _ = replay.get_or_insert_with(|| {
            outbox
                .iter()
                .find(|note| matches!(note, Note::Old { .. }))
                .cloned()
                .expect("an insert emits old-addressed traffic")
        });
        fabric.post(station, outbox);
    }
    fabric.drain(&mut fleet);

    let _ = act(&mut fabric, &mut fleet, 1, Replica::try_declare)
        .expect("the complete round's declarer tags");
    fabric.drain(&mut fleet);

    for station in ROSTER {
        let replica = &fleet[&station];
        assert_eq!(replica.generation(), 3, "the admitting window sealed");
        let sealed = replica
            .epochs()
            .newest_sealed()
            .expect("the widening record is retained");
        let admission = sealed
            .admission()
            .expect("the record carries what its winner declared");
        assert!(admission.admits(4));
        assert_eq!(
            sealed.sealed_join().get(4),
            0,
            "station {station}: the joiner counted nothing toward the admitting window's minima"
        );
        assert!(
            replica.epochs().roster().any(|seat| seat == 4),
            "station {station}: the roster widened at the seal"
        );
        assert_eq!(
            replica.watermark(),
            VersionVector::new(),
            "station {station}: the meet ranges over the joiner's bottom slot from this seal on"
        );
        assert!(
            replica.epochs().arrival().is_none(),
            "station {station}: the seal consumed the round"
        );
    }

    install_admitted_joiner(&mut fabric, &mut fleet, 4, depth);

    // The allocator answer: the joiner's plane coordinate is bottom, so
    // its first mint is exactly `(4, 1)` --- a re-stamp that carried the
    // exporter's counter would open the sequence at a gap no floor could
    // ever close.
    let dot = act(&mut fabric, &mut fleet, 4, |replica, out| {
        replica.insert_visible(0, out)
    });
    assert_eq!(dot, d(4, 1), "the admitted joiner mints from bottom");
    fabric.drain(&mut fleet);

    let _ = act(&mut fabric, &mut fleet, 4, Replica::try_declare)
        .expect("the admitted joiner declares the next boundary");
    fabric.drain(&mut fleet);
    assert_converged(&fleet);
    for replica in fleet.values() {
        assert_eq!(
            replica.generation(),
            4,
            "the joiner co-sealed the next window"
        );
    }

    // The delayed cross-epoch replay, delivered straight to the joiner: a
    // generation-2 note from before the admission. The recognizer answers
    // from the retained record of that generation --- the per-generation
    // roster the lineage proved --- and the machine state does not move.
    let replay = replay.expect("captured pre-admission traffic");
    let before = fleet[&4].epochs().clone();
    let mut discard = Vec::new();
    fleet
        .get_mut(&4)
        .expect("the joiner is installed")
        .handle(&replay, &mut discard);
    assert_eq!(
        &before,
        fleet[&4].epochs(),
        "a pre-admission replay reads as a duplicate and teaches nothing"
    );
}

/// The R-87 question answered by exhibit: retirement evidence widens with
/// the roster. After the admitting seal the meet ranges over the joiner,
/// so an incumbent's condense waits for the joiner's acknowledgement ---
/// a tracker left narrow would have licensed the excision while the
/// joiner could still anchor through the removed dot, which is the
/// unrecoverable direction (R-41's strand). Widening merely stalls
/// reclamation until the word arrives: the safe direction, priced and
/// then released.
#[test]
fn retirement_evidence_widens_with_the_roster() {
    let depth = horizon(4);
    let mut fleet = fleet_of(&ROSTER, depth);
    let mut fabric = Fabric::new(0xA221_7002, &ROSTER, 0);
    seal_once(&mut fabric, &mut fleet);
    admit_at(&mut fabric, &mut fleet, &ROSTER, &[4]);
    seal_once(&mut fabric, &mut fleet);
    install_admitted_joiner(&mut fabric, &mut fleet, 4, depth);

    // Fresh element, everywhere --- the joiner included.
    let _ = act(&mut fabric, &mut fleet, 2, |replica, out| {
        replica.insert_visible(0, out)
    });
    fabric.drain(&mut fleet);

    // The joiner goes quiet; the incumbents delete and acknowledge.
    fabric.sever(&[4]);
    let deleted = act(&mut fabric, &mut fleet, 1, |replica, out| {
        replica.delete_visible(0, out)
    });
    assert!(deleted.is_some(), "the plane has an element to delete");
    fabric.drain(&mut fleet);
    assert_eq!(
        condense_at(&mut fleet, 1),
        0,
        "the widened meet waits on the joiner's acknowledgement"
    );

    // The word arrives; the meet completes; the excision fires.
    fabric.heal();
    fabric.drain(&mut fleet);
    assert_eq!(
        condense_at(&mut fleet, 1),
        1,
        "the complete widened meet licenses the excision"
    );
    for station in [2, 3, 4] {
        assert_eq!(condense_at(&mut fleet, station), 1);
    }
    assert_converged(&fleet);
}

/// PRD 0028 falsifier 7, first half, composed with falsifier 8's carry:
/// a fleet that has attested a departure still admits. The departed
/// member's window slots are substituted, not awaited; the arrival round
/// ranges over the surviving family and completes without it; and the
/// widening seal carries the attestation across at the new base's own
/// compacted prefix, re-keyed over the family the new plane opens with
/// --- the schedule that panics without the `Departed::refounded` re-key,
/// so this exhibit is that cure's fleet-grade kill.
#[test]
fn an_attested_fleet_admits_and_the_carry_crosses_the_widening_seal() {
    let depth = horizon(4);
    let mut fleet = fleet_of(&ROSTER, depth);
    let mut fabric = Fabric::new(0xA221_7003, &ROSTER, 0);
    seal_once(&mut fabric, &mut fleet);
    assert_converged(&fleet);

    // Station 3 falls silent; the survivors evict it.
    fabric.sever(&[3]);
    for station in [1, 2] {
        let _ = act(&mut fabric, &mut fleet, station, |replica, out| {
            replica.abandon(3, out)
        });
        let _ = act(&mut fabric, &mut fleet, station, |replica, out| {
            replica.open_departure(3, out)
        })
        .expect("a survivor proposes the departed member's prefix");
    }
    fabric.drain(&mut fleet);
    for station in [1, 2] {
        assert!(
            fleet[&station].attested(3).is_some(),
            "station {station}: the departure round sealed"
        );
    }

    // The attested fleet admits: the round ranges over the survivors.
    admit_at(&mut fabric, &mut fleet, &[1, 2], &[4]);
    for station in [1, 2] {
        let round = fleet[&station].epochs().arrival().expect("a held round");
        assert!(
            round.family().eq([1u32, 2]),
            "station {station}: the family is the surviving members, no substitution"
        );
    }

    for station in [1, 2] {
        let _ = act(&mut fabric, &mut fleet, station, |replica, out| {
            replica.insert_visible(0, out)
        });
    }
    fabric.drain(&mut fleet);
    let _ =
        act(&mut fabric, &mut fleet, 1, Replica::try_declare).expect("the surviving declarer tags");
    fabric.drain(&mut fleet);

    // The widening seal crossed with the attestation carried: the exact
    // schedule that refuses `FamilyMismatch` at the tracker rebuild
    // without the re-key.
    let frames: Vec<Vec<u8>> = [1u32, 2]
        .iter()
        .map(|station| {
            let replica = &fleet[station];
            assert_eq!(replica.generation(), 3);
            SealRecord::from_sealed(
                replica
                    .epochs()
                    .newest_sealed()
                    .expect("the widening record is retained"),
            )
            .to_bytes()
        })
        .collect();
    assert_eq!(frames[0], frames[1], "one record, byte-identical");
    for station in [1, 2] {
        let replica = &fleet[&station];
        assert!(replica.epochs().roster().any(|seat| seat == 4));
        assert_eq!(
            replica.attested(3),
            Some(1),
            "station {station}: the attestation re-declared over the widened family, at the \
             compacted prefix the new base carries for the evictee"
        );
    }

    // The restore keeps the provenance: an incumbent that crashes after
    // the widening seal reconstructs the attestation from its checkpoint,
    // and the not-yet-bootstrapped joiner's word is coverage, never
    // possession --- derived from the newest record's own admission, the
    // one set of members whose first seal is still ahead of them. A
    // restore that refabricated the joiner's possession would let the
    // incumbents abandon every actual holder while an offline joiner's
    // entry serves the bound (the fourth review round's P1).
    crash(&mut fleet, &ROSTER, depth, 1);
    assert_eq!(
        fleet[&1].attested_proposal(3, 4),
        Some(0),
        "the restored attestation claims no possession for the unbootstrapped joiner"
    );
    assert_eq!(
        fleet[&1].attested_proposal(3, 2),
        Some(1),
        "the restored attestation carries a holder's base prefix"
    );
    resync(&mut fabric, &fleet, 1);
    fabric.drain(&mut fleet);

    // The departed member never returns; the widened fleet completes
    // without it.
    let _ = fleet.remove(&3);
    install_admitted_joiner(&mut fabric, &mut fleet, 4, depth);
    let _ = act(&mut fabric, &mut fleet, 4, |replica, out| {
        replica.insert_visible(0, out)
    });
    fabric.drain(&mut fleet);
    let _ = act(&mut fabric, &mut fleet, 2, Replica::try_declare)
        .expect("the widened fleet declares its next boundary");
    fabric.drain(&mut fleet);
    assert_converged(&fleet);
}

/// PRD 0028 falsifier 7, second half, at fleet grade: a departure
/// attestation applied at one honest replica while the tagged window is
/// open --- and never at another --- moves no byte of either record. The
/// admission rides the winner and the join is value-forced (R-86), so
/// the attested fold's substituted slot and the unattested folds' spoken
/// word agree, and the records agree bit for bit. The unit pin drives
/// the machine; this drives the whole recipe, with the departing member
/// live and honest in every round.
#[test]
fn a_racing_attestation_cannot_split_the_widening_record() {
    let depth = horizon(4);
    let mut fleet = fleet_of(&ROSTER, depth);
    let mut fabric = Fabric::new(0xA221_7004, &ROSTER, 0);
    seal_once(&mut fabric, &mut fleet);
    admit_at(&mut fabric, &mut fleet, &ROSTER, &[4]);

    // Station 3 is being evicted but is still live and honest: its
    // spoken words are in every round. The survivors' proposals are
    // minted now and held in flight; nothing of station 3's is minted
    // after the proposals, so the agreed bound and its spoken counter
    // name one coordinate (R-86's forcing, arranged honestly).
    for station in [1, 2] {
        let _ = act(&mut fabric, &mut fleet, station, |replica, out| {
            replica.open_departure(3, out)
        })
        .expect("a survivor proposes");
    }

    for station in [1, 2] {
        let _ = act(&mut fabric, &mut fleet, station, |replica, out| {
            replica.insert_visible(0, out)
        });
    }
    // Everything but the departure proposals and the adoption reports
    // flows: the fleet reaches open, adopted, unsealed windows.
    fabric.drain_selected(&mut fleet, |_, note| {
        !matches!(note, Note::Depart { .. } | Note::Adoption { .. })
    });
    let _ = act(&mut fabric, &mut fleet, 1, Replica::try_declare)
        .expect("the complete round's declarer tags");
    fabric.drain_selected(&mut fleet, |_, note| {
        !matches!(note, Note::Depart { .. } | Note::Adoption { .. })
    });
    for station in ROSTER {
        assert!(
            fleet[&station].adopted(),
            "station {station}: the window is open and adopted"
        );
    }

    // The race: station 2's proposal reaches station 1 mid-window, so 1
    // attests and its seal substitutes station 3's slots. Stations 2 and
    // 3 never see an attestation this generation.
    fabric.drain_selected(&mut fleet, |dst, note| {
        dst == 1 && matches!(note, Note::Depart { .. })
    });
    assert!(
        fleet[&1].attested(3).is_some(),
        "the attestation landed mid-window at station 1"
    );
    assert!(fleet[&2].attested(3).is_none());

    // Release the adoption reports: every window seals. Only then may the
    // leftover proposal travel, reaching station 2 after its seal, where
    // a sealed generation's proposal is void.
    fabric.drain_selected(&mut fleet, |_, note| matches!(note, Note::Adoption { .. }));
    fabric.drain(&mut fleet);
    let frames: Vec<Vec<u8>> = ROSTER
        .iter()
        .map(|station| {
            let replica = &fleet[station];
            assert_eq!(replica.generation(), 3);
            SealRecord::from_sealed(
                replica
                    .epochs()
                    .newest_sealed()
                    .expect("the widening record is retained"),
            )
            .to_bytes()
        })
        .collect();
    assert_eq!(
        frames[0], frames[1],
        "the attested and unattested folds seal one record"
    );
    assert_eq!(frames[1], frames[2]);
    assert!(
        fleet[&2].attested(3).is_none(),
        "station 2 sealed without ever applying the attestation"
    );
    for station in ROSTER {
        let sealed_admission = fleet[&station]
            .epochs()
            .newest_sealed()
            .and_then(|sealed| sealed.admission().cloned())
            .expect("the admission rode the winner at every member");
        assert!(sealed_admission.admits(4));
    }
}

/// PRD 0028 falsifier 7, third half: a departure attestation landing
/// while an arrival round is open freezes that round permanently --- the
/// delayed word is refused at the folding door, the tag license never
/// fires, and the window seals plain. The arrival re-agrees over the new
/// base with the surviving family and costs exactly one boundary: no
/// stall, no partial admission, no round surviving its base.
#[test]
fn a_racing_attestation_freezes_the_round_and_the_arrival_pays_one_boundary() {
    let depth = horizon(4);
    let mut fleet = fleet_of(&ROSTER, depth);
    let mut fabric = Fabric::new(0xA221_7005, &ROSTER, 0);
    seal_once(&mut fabric, &mut fleet);

    // Station 3 endorses --- its word is minted and in flight --- and the
    // survivors open their rounds. 3's word is withheld while their
    // departure round completes, so the attestation lands first.
    act(&mut fabric, &mut fleet, 3, |replica, out| {
        replica.open_arrival(&[4], out)
    })
    .expect("the departing member's own admission is honest");
    for station in [1, 2] {
        act(&mut fabric, &mut fleet, station, |replica, out| {
            replica.open_arrival(&[4], out)
        })
        .expect("an incumbent admits");
    }
    fabric.drain_selected(&mut fleet, |_, note| {
        !matches!(note, Note::Endorse { by: 3, .. })
    });
    for station in [1, 2] {
        let _ = act(&mut fabric, &mut fleet, station, |replica, out| {
            replica.abandon(3, out)
        });
        let _ = act(&mut fabric, &mut fleet, station, |replica, out| {
            replica.open_departure(3, out)
        })
        .expect("a survivor proposes");
    }
    fabric.drain_selected(&mut fleet, |dst, note| {
        dst != 3 && matches!(note, Note::Depart { .. })
    });
    for station in [1, 2] {
        assert!(fleet[&station].attested(3).is_some());
    }

    // The delayed word arrives and is refused: the freeze is enforced.
    fabric.drain_selected(&mut fleet, |_, note| matches!(note, Note::Endorse { .. }));
    for station in [1, 2] {
        let round = fleet[&station]
            .epochs()
            .arrival()
            .expect("the frozen round");
        assert!(
            !round.complete() && round.pending().eq([3u32]),
            "station {station}: the frozen slot never fills"
        );
    }

    // The window seals plain; the arrival cost itself the boundary.
    for station in [1, 2] {
        let _ = act(&mut fabric, &mut fleet, station, |replica, out| {
            replica.insert_visible(0, out)
        });
    }
    fabric.drain_selected(&mut fleet, |dst, _| dst != 3);
    let _ = act(&mut fabric, &mut fleet, 1, Replica::try_declare)
        .expect("the fleet declares plain past the frozen round");
    fabric.drain_selected(&mut fleet, |dst, _| dst != 3);
    for station in [1, 2] {
        let replica = &fleet[&station];
        assert_eq!(replica.generation(), 3);
        assert!(
            replica
                .epochs()
                .newest_sealed()
                .is_some_and(|sealed| sealed.admission().is_none()),
            "station {station}: the plain seal carries no admission"
        );
        assert!(
            replica.epochs().arrival().is_none(),
            "station {station}: the plain seal superseded the frozen round"
        );
    }

    // The re-agreement over the new base: the surviving family completes
    // without the attested member, the winner tags, and the admission
    // rides one boundary later than it hoped.
    fabric.sever(&[3]);
    let _ = fleet.remove(&3);
    admit_at(&mut fabric, &mut fleet, &[1, 2], &[4]);
    for station in [1, 2] {
        let _ = act(&mut fabric, &mut fleet, station, |replica, out| {
            replica.insert_visible(0, out)
        });
    }
    fabric.drain(&mut fleet);
    let _ = act(&mut fabric, &mut fleet, 2, Replica::try_declare)
        .expect("the surviving declarer tags over the new base");
    fabric.drain(&mut fleet);
    for station in [1, 2] {
        let replica = &fleet[&station];
        assert_eq!(replica.generation(), 4);
        assert!(
            replica
                .epochs()
                .newest_sealed()
                .is_some_and(|sealed| sealed.admission().is_some()),
            "station {station}: the re-agreed admission rode the next winner"
        );
        assert!(replica.epochs().roster().any(|seat| seat == 4));
    }
    assert_converged(&fleet);
}

/// The R-89 counterweight, live: a member excluded from the round's
/// family by a *local* abandon --- but neither attested nor substituted
/// --- is still required to adopt, and a lawfully tagged winner waits at
/// that member's own adopt door until its operator endorses the
/// boundary. The N-of-N veto surfacing at a new door, priced as the cost
/// of abandoning a live member, and cured caller-side: the late
/// `open_arrival` folds the parked endorsements, assents, and the fleet
/// seals wide.
#[test]
fn an_unassenting_member_stalls_the_window_until_its_operator_admits() {
    let depth = horizon(4);
    let mut fleet = fleet_of(&ROSTER, depth);
    let mut fabric = Fabric::new(0xA221_7006, &ROSTER, 0);
    seal_once(&mut fabric, &mut fleet);

    // Stations 1 and 2 abandon 3 locally (heterogeneous operator policy;
    // 3 is honest and live) and admit station 4 over their two-member
    // surviving family.
    for station in [1, 2] {
        let _ = act(&mut fabric, &mut fleet, station, |replica, out| {
            replica.abandon(3, out)
        });
    }
    admit_at(&mut fabric, &mut fleet, &[1, 2], &[4]);

    for station in ROSTER {
        let _ = act(&mut fabric, &mut fleet, station, |replica, out| {
            replica.insert_visible(0, out)
        });
    }
    fabric.drain(&mut fleet);
    let _ = act(&mut fabric, &mut fleet, 1, Replica::try_declare)
        .expect("the two-member round's declarer tags");
    fabric.drain(&mut fleet);

    // The stall: station 3 confirmed the winner but holds no round, so
    // its adopt door refuses assent; it is not attested, so no
    // substitution covers its slot, and nobody seals.
    for station in ROSTER {
        assert_eq!(
            fleet[&station].generation(),
            2,
            "station {station}: the window waits on the unassenting member"
        );
    }
    assert!(fleet[&1].adopted() && fleet[&2].adopted());
    assert!(
        !fleet[&3].adopted(),
        "the abandoned member stalls at its own adopt door"
    );

    // The caller-side cure: 3's operator endorses the boundary. The
    // parked endorsements fold, the round completes, the assent stands,
    // and the fleet seals the admission.
    act(&mut fabric, &mut fleet, 3, |replica, out| {
        replica.open_arrival(&[4], out)
    })
    .expect("the stalled member's operator endorses");
    fabric.drain(&mut fleet);
    for station in ROSTER {
        let replica = &fleet[&station];
        assert_eq!(replica.generation(), 3, "station {station}: sealed wide");
        let sealed = replica
            .epochs()
            .newest_sealed()
            .expect("the widening record is retained");
        assert!(
            sealed
                .admission()
                .is_some_and(|admission| admission.admits(4))
        );
        assert!(
            sealed.sealed_join().get(3) > 0,
            "station {station}: the cured member's own word is in the join, unsubstituted"
        );
        assert!(replica.epochs().roster().any(|seat| seat == 4));
    }
    assert_converged(&fleet);
}

/// A peer's word spoken before this operator's decision waits parked,
/// and the late `open_arrival` folds it in the same act --- so the last
/// operator to decide can open and immediately declare the tag, with no
/// intervening delivery. Two mechanisms pinned together because each is
/// only observable through the other: the `NoRound` park (a dropped word
/// would leave the late round incomplete forever, no peer re-speaks an
/// endorsement outside repair) and the open door's full drive (a bare
/// poll would leave the parked words waiting for the next unrelated
/// delivery, and the tag license reads the round *now*).
#[test]
fn a_late_operators_open_folds_the_waiting_words_in_one_act() {
    let depth = horizon(4);
    let mut fleet = fleet_of(&ROSTER, depth);
    let mut fabric = Fabric::new(0xA221_7009, &ROSTER, 0);
    seal_once(&mut fabric, &mut fleet);

    for station in [1, 2] {
        act(&mut fabric, &mut fleet, station, |replica, out| {
            replica.open_arrival(&[4], out)
        })
        .expect("an early operator admits");
    }
    fabric.drain(&mut fleet);
    assert!(
        fleet[&3].parked_len() >= 2,
        "the early words wait parked for this operator's own decision"
    );

    for station in ROSTER {
        let _ = act(&mut fabric, &mut fleet, station, |replica, out| {
            replica.insert_visible(0, out)
        });
    }
    fabric.drain(&mut fleet);

    // The late operator's one act: open, fold the waiting words, and the
    // round is complete before anything else is delivered.
    act(&mut fabric, &mut fleet, 3, |replica, out| {
        replica.open_arrival(&[4], out)
    })
    .expect("the late operator admits");
    assert!(
        fleet[&3]
            .epochs()
            .arrival()
            .is_some_and(ArrivalRound::complete),
        "the open folded the parked words in the same act"
    );
    let _ = act(&mut fabric, &mut fleet, 3, Replica::try_declare)
        .expect("the late operator declares immediately");
    fabric.drain(&mut fleet);
    for station in ROSTER {
        let replica = &fleet[&station];
        assert_eq!(replica.generation(), 3);
        assert!(
            replica
                .epochs()
                .newest_sealed()
                .is_some_and(|sealed| sealed.admission().is_some()),
            "station {station}: the late opener's declaration carried the tag"
        );
    }
    assert_converged(&fleet);
}

/// The checkpoint door never exports past a racing attestation (the
/// sixth review round's P1): an attestation lands mid-generation without
/// disturbing any at-seal field, so an export after it would omit the
/// departure --- and a bootstrap from that checkpoint would restore the
/// evicted identity as active while every peer fences it, because an
/// epoch roster never removes a seat. The door refuses the attested
/// target and any export whose live attestation state outran the
/// journal's; the next seal re-journals the attestations and export
/// resumes.
#[test]
fn a_checkpoint_never_exports_past_a_racing_attestation() {
    let depth = horizon(4);
    let mut fleet = fleet_of(&ROSTER, depth);
    let mut fabric = Fabric::new(0xA221_700B, &ROSTER, 0);
    seal_once(&mut fabric, &mut fleet);
    admit_at(&mut fabric, &mut fleet, &ROSTER, &[4]);
    for station in ROSTER {
        let _ = act(&mut fabric, &mut fleet, station, |replica, out| {
            replica.insert_visible(0, out)
        });
    }
    fabric.drain(&mut fleet);
    let _ = act(&mut fabric, &mut fleet, 1, Replica::try_declare)
        .expect("the complete round's declarer tags");
    fabric.drain(&mut fleet);
    assert!(
        fleet[&1].joiner_checkpoint_for(4).is_some(),
        "at the widening seal the joiner's checkpoint exports"
    );

    // The fleet evicts the never-bootstrapped joiner before anyone
    // exports: the departure round completes over the incumbents, and
    // the attestation lands mid-generation.
    for station in ROSTER {
        let _ = act(&mut fabric, &mut fleet, station, |replica, out| {
            replica.open_departure(4, out)
        })
        .expect("a survivor proposes the joiner's empty prefix");
    }
    fabric.drain(&mut fleet);
    assert_eq!(
        fleet[&1].attested(4),
        Some(0),
        "the joiner minted nothing, so the agreed bound is bottom"
    );
    assert!(
        fleet[&1].joiner_checkpoint_for(4).is_none(),
        "no checkpoint is ever re-stamped for an attested identity"
    );
    assert!(
        fleet[&1].joiner_checkpoint_for(2).is_none(),
        "no checkpoint exports while the live attestations outrun the journal's"
    );

    // The next seal re-journals the attestation, and export resumes.
    for station in ROSTER {
        let _ = act(&mut fabric, &mut fleet, station, |replica, out| {
            replica.insert_visible(0, out)
        });
    }
    fabric.drain(&mut fleet);
    let _ = act(&mut fabric, &mut fleet, 1, Replica::try_declare)
        .expect("the narrowed fleet declares past the eviction");
    fabric.drain(&mut fleet);
    assert!(
        fleet[&1].joiner_checkpoint_for(2).is_some(),
        "the seal re-journaled the attestation and the door reopened"
    );
    assert!(
        fleet[&1].joiner_checkpoint_for(4).is_none(),
        "the evicted identity stays refused, forever"
    );
    assert_converged(&fleet);
}

/// The admit door is idempotent in the journal, not only in the machine:
/// only the open that creates the round journals its mint, and an
/// operator's retry during a stalled round re-emits the already-durable
/// word without appending it --- the bounded-record rule at the one door
/// this slice added.
#[test]
fn a_reopened_round_journals_its_endorsement_once() {
    let depth = horizon(4);
    let mut fleet = fleet_of(&ROSTER, depth);
    let mut fabric = Fabric::new(0xA221_700A, &ROSTER, 0);
    seal_once(&mut fabric, &mut fleet);

    act(&mut fabric, &mut fleet, 1, |replica, out| {
        replica.open_arrival(&[4], out)
    })
    .expect("the operator admits");
    let journaled = fleet[&1].journal_notes_len();
    for _ in 0..3 {
        act(&mut fabric, &mut fleet, 1, |replica, out| {
            replica.open_arrival(&[4], out)
        })
        .expect("the identical retry absorbs");
    }
    assert_eq!(
        fleet[&1].journal_notes_len(),
        journaled,
        "a retried open re-emits the durable word without re-journaling it"
    );
}

/// R-35 falsifier 5's partition half at fleet grade: an incumbent that
/// misses the admitting window's tail seals late into the widened family
/// by re-deriving it from the record it seals --- no membership message
/// exists to miss, so there is no wrong-family read to make. The delayed
/// adoption reports arrive, the laggard seals the identical widening
/// record, and its rebuilt trackers range over the joiner it learned
/// from the record alone.
#[test]
fn a_partition_healed_incumbent_rederives_the_widened_family() {
    let depth = horizon(4);
    let mut fleet = fleet_of(&ROSTER, depth);
    let mut fabric = Fabric::new(0xA221_7007, &ROSTER, 0);
    seal_once(&mut fabric, &mut fleet);
    admit_at(&mut fabric, &mut fleet, &ROSTER, &[4]);
    for station in ROSTER {
        let _ = act(&mut fabric, &mut fleet, station, |replica, out| {
            replica.insert_visible(0, out)
        });
    }
    fabric.drain(&mut fleet);
    let _ = act(&mut fabric, &mut fleet, 1, Replica::try_declare)
        .expect("the complete round's declarer tags");
    // Station 3 receives everything except the adoption reports, so it
    // adopts but cannot seal: the partition begins at the window's tail.
    fabric.drain_selected(&mut fleet, |dst, note| {
        dst != 3 || !matches!(note, Note::Adoption { .. })
    });
    assert_eq!(fleet[&1].generation(), 3, "the connected members sealed");
    assert_eq!(
        fleet[&3].generation(),
        2,
        "the partitioned member holds the open window"
    );

    // The tail arrives; the laggard completes the round from the same
    // agreed inputs and seals the identical record.
    fabric.drain(&mut fleet);
    assert_eq!(fleet[&3].generation(), 3, "the healed member sealed");
    assert!(fleet[&3].epochs().roster().any(|seat| seat == 4));
    assert_eq!(
        SealRecord::from_sealed(fleet[&3].epochs().newest_sealed().expect("retained")).to_bytes(),
        SealRecord::from_sealed(fleet[&1].epochs().newest_sealed().expect("retained")).to_bytes(),
        "the healed member's widening record is byte-identical"
    );
    assert_eq!(
        fleet[&3].watermark(),
        VersionVector::new(),
        "the healed member's meet ranges over the joiner it learned from the record"
    );
    assert_converged(&fleet);
}

/// PRD 0028 falsifier 4's fleet half: a declarer that tagged and crashed
/// re-derives the same tagged content under the same dot from its
/// journal --- the round from the journaled endorsements, the
/// declaration from the journaled mint --- so a crash cannot turn a
/// tagged declaration into self-equivocation.
#[test]
fn a_crashed_declarer_rederives_its_tag() {
    let depth = horizon(4);
    let mut fleet = fleet_of(&ROSTER, depth);
    let mut fabric = Fabric::new(0xA221_7008, &ROSTER, 0);
    seal_once(&mut fabric, &mut fleet);
    admit_at(&mut fabric, &mut fleet, &ROSTER, &[4]);
    for station in ROSTER {
        let _ = act(&mut fabric, &mut fleet, station, |replica, out| {
            replica.insert_visible(0, out)
        });
    }
    fabric.drain(&mut fleet);
    let _ = act(&mut fabric, &mut fleet, 1, Replica::try_declare)
        .expect("the complete round's declarer tags");
    let spoken: Vec<_> = fleet[&1]
        .epochs()
        .candidates()
        .filter(|declaration| declaration.dot().station() == 1)
        .map(|declaration| {
            (
                declaration.dot(),
                declaration.admission().cloned(),
                declaration.admission_commitment().copied(),
            )
        })
        .collect();
    assert_eq!(spoken.len(), 1, "one tagged candidate stands");
    assert!(
        spoken[0].1.is_some(),
        "the declaration carries the boundary"
    );

    crash(&mut fleet, &ROSTER, depth, 1);
    let rederived: Vec<_> = fleet[&1]
        .epochs()
        .candidates()
        .filter(|declaration| declaration.dot().station() == 1)
        .map(|declaration| {
            (
                declaration.dot(),
                declaration.admission().cloned(),
                declaration.admission_commitment().copied(),
            )
        })
        .collect();
    assert_eq!(
        spoken, rederived,
        "the restored declarer re-derives the same tagged content for the same dot"
    );
    assert!(
        fleet[&1]
            .epochs()
            .arrival()
            .is_some_and(ArrivalRound::complete),
        "the journaled endorsements rebuild the complete round"
    );

    resync(&mut fabric, &fleet, 1);
    fabric.drain(&mut fleet);
    for station in ROSTER {
        assert_eq!(fleet[&station].generation(), 3);
        assert!(
            fleet[&station]
                .epochs()
                .newest_sealed()
                .is_some_and(|sealed| sealed.admission().is_some()),
            "station {station}: the admission survived the declarer's crash"
        );
    }
    assert_converged(&fleet);
}
/// The defect this slice's exhibits caught in the *shipped* S339 carry,
/// pinned as its regression: no arrival machinery is involved. An
/// attested departure used to re-declare at bottom across every
/// boundary, but the re-foundation compacts the evictee's surviving
/// elements to `1..=held` in the new base --- so the *second*
/// post-attestation seal substituted an understated coordinate into its
/// join and wedged every honest member at the consignment door
/// (`BeyondSeal` over a base dot the join did not cover). The cure is
/// `Departed::refounded` carrying the base's compacted prefix: real
/// testimony, since every survivor holds those dots through the base it
/// agrees on. This exhibit is the minimal reproduction: evict a member
/// whose element survives, then cross two boundaries.
#[test]
fn an_evictees_survivors_cross_the_second_boundary() {
    let depth = horizon(4);
    let mut fleet = fleet_of(&ROSTER, depth);
    let mut fabric = Fabric::new(0xDEAD_0001, &ROSTER, 0);
    // Every member writes (station 3's element will survive compaction),
    // then the fleet seals once.
    seal_once(&mut fabric, &mut fleet);
    // 3 falls silent; survivors evict it.
    fabric.sever(&[3]);
    for station in [1, 2] {
        let _ = act(&mut fabric, &mut fleet, station, |replica, out| {
            replica.abandon(3, out)
        });
        let _ = act(&mut fabric, &mut fleet, station, |replica, out| {
            replica.open_departure(3, out)
        })
        .expect("a survivor proposes");
    }
    fabric.drain(&mut fleet);
    assert!(fleet[&1].attested(3).is_some());
    // First post-attestation boundary.
    for station in [1, 2] {
        let _ = act(&mut fabric, &mut fleet, station, |replica, out| {
            replica.insert_visible(0, out)
        });
    }
    fabric.drain(&mut fleet);
    let _ = act(&mut fabric, &mut fleet, 1, Replica::try_declare).expect("declares");
    fabric.drain(&mut fleet);
    assert_eq!(fleet[&1].generation(), 3, "first post-attestation seal");
    assert_eq!(
        fleet[&1].attested(3),
        Some(1),
        "the carry re-declared at the evictee's compacted base prefix"
    );
    // A restart between the boundaries: the restore path re-derives the
    // same coordinate from the restored base, so the crash changes
    // nothing about the second seal.
    crash(&mut fleet, &ROSTER, depth, 1);
    assert_eq!(
        fleet[&1].attested(3),
        Some(1),
        "the restored attestation speaks the restored base's own prefix"
    );
    resync(&mut fabric, &fleet, 1);
    fabric.drain(&mut fleet);
    // Second boundary: the shipped-carry hazard. A carry or restore at
    // bottom understates the evictee's coordinate and wedges exactly
    // here, at every honest member's consignment door.
    for station in [1, 2] {
        let _ = act(&mut fabric, &mut fleet, station, |replica, out| {
            replica.insert_visible(0, out)
        });
    }
    fabric.drain(&mut fleet);
    let _ = act(&mut fabric, &mut fleet, 1, Replica::try_declare).expect("declares again");
    fabric.drain(&mut fleet);
    assert_eq!(fleet[&1].generation(), 4, "second post-attestation seal");
}