minerva 0.2.0

//! The departure round: the agreed, enforced abandonment bound (S339).
//!
//! S336 shipped the family narrowing and measured where it stops: the epoch
//! rounds keep ranging over the whole roster, because a locally computed
//! bound is one honest survivors disagree about, and a round narrowed on one
//! seals a record the consignment door rejects one door after the lineage
//! advanced. R-84 named what closes it --- a bound both *carried*, so every
//! survivor's record agrees, and *enforced*, so nothing can arrive above it
//! --- and chartered it as a second protocol object with its own round.
//!
//! This is that object. The suite has three halves: the round's own doors,
//! the tracker door that consumes its attestation, and the epoch rounds the
//! attestation completes. The end-to-end companion is
//! `fleet::byzantine::attesting_the_silent_members_departure_carries_the_fleet_through_the_seal`;
//! the argument is `docs/metis-membership-departure.adoc`.

extern crate alloc;

use core::num::NonZeroUsize;

use crate::kairos::Kairos;
use crate::metis::{
    AbandonRefusal, Cut, Departed, Departure, DepartureRefusal, Dot, DotSet, EpochRefusal, Epochs,
    SealedEpoch, Stability, VersionVector, Vouched,
};

/// Builds an identity literal. Panics on the non-dot counter zero (R-91).
#[track_caller]
fn d(station: u32, counter: u64) -> Dot {
    Dot::from_parts(station, counter).expect("test literal names the non-dot counter zero")
}

const ROSTER: [u32; 3] = [1, 2, 3];
const SURVIVORS: [u32; 2] = [1, 2];
/// The member that departs.
const GONE: u32 = 3;
/// The declaring station's dot, one above the base cut's ceiling for it.
const DECLARATION: Dot = match Dot::from_parts(1, 2) {
    Ok(dot) => dot,
    Err(_) => unreachable!(),
};

fn horizon() -> NonZeroUsize {
    NonZeroUsize::new(2).expect("positive")
}

fn vector(pairs: &[(u32, u64)]) -> VersionVector {
    let mut vector = VersionVector::new();
    for &(station, counter) in pairs {
        vector.observe(station, counter);
    }
    vector
}

/// A have-set holding exactly `dots` of `station`.
fn holding(station: u32, dots: &[u64]) -> DotSet {
    let mut held = DotSet::new();
    for &dot in dots {
        let _ = held.insert(d(station, dot));
    }
    held
}

/// A have-set holding the gap-free prefix `1..=counter` of `station`.
fn prefix(station: u32, counter: u64) -> DotSet {
    holding(station, &(1..=counter).collect::<alloc::vec::Vec<_>>())
}

/// The round every survivor runs, driven to its seal from proposals the
/// caller supplies: `(station, prefix)` pairs, the first of which is this
/// replica's own.
fn attested(proposals: &[(u32, u64)]) -> Departed {
    let mut round = Departure::opened(
        GONE,
        proposals[0].0,
        proposals.iter().map(|&(station, _)| station),
    )
    .expect("a surviving family");
    let (_, own_prefix) = proposals[0];
    let _ = round
        .propose(&prefix(GONE, own_prefix), &DotSet::new())
        .expect("a gap-free holding");
    for &(station, counter) in &proposals[1..] {
        round
            .fold(&Vouched::trust(station, counter))
            .expect("a survivor's proposal");
    }
    round.try_seal().expect("every survivor proposed").clone()
}

// -- The round's own doors ---------------------------------------------------

/// The bound is the survivors' *join*, and it is the only number the round
/// ever hands out.
///
/// A meet would write off traffic a survivor holds; a local read would be a
/// different number at every survivor. The join is the one value that both
/// bounds every survivor's holding and is reachable by every survivor
/// through the repair lane.
#[test]
fn the_round_seals_at_the_join_of_the_survivors_proposals() {
    let mut round = Departure::opened(GONE, 1, SURVIVORS).expect("a surviving family");
    assert_eq!(round.propose(&prefix(GONE, 4), &DotSet::new()), Ok(4));
    assert!(
        round.try_seal().is_none(),
        "one survivor's word is not the family's"
    );
    round
        .fold(&Vouched::trust(2, 7))
        .expect("a survivor's proposal");
    let departed = round.try_seal().expect("every survivor proposed").clone();
    assert_eq!(departed.station(), GONE);
    assert_eq!(departed.bound(), 7, "the join, not this replica's own 4");
    assert_eq!(round.bound(), Some(7));
}

/// The proposal is a promise, and the fence is what makes it true.
///
/// Between the doors the fence sits at this replica's own proposal, so it
/// refuses dots that are perfectly lawful and will be admitted once the
/// round seals higher. That is a hold. Above the *sealed* bound the refusal
/// is terminal, and only then is it evidence.
#[test]
fn the_fence_holds_at_the_proposal_and_becomes_a_verdict_at_the_bound() {
    let mut round = Departure::opened(GONE, 1, SURVIVORS).expect("a surviving family");
    assert_eq!(round.admits(d(GONE, 9)), Ok(()), "nothing promised yet");

    let _ = round
        .propose(&prefix(GONE, 4), &DotSet::new())
        .expect("gap-free");
    let held = round
        .admits(d(GONE, 5))
        .expect_err("above this replica's own promise");
    assert_eq!(held.fence, 4);
    assert!(
        !held.is_resurgence(),
        "a provisional refusal is a hold: the round may still seal above it"
    );

    round.fold(&Vouched::trust(2, 7)).expect("a proposal");
    let _ = round.try_seal().expect("complete");
    assert_eq!(
        round.admits(d(GONE, 5)),
        Ok(()),
        "the fence rises to the agreed bound, so the held dot is admitted"
    );
    let refused = round
        .admits(d(GONE, 8))
        .expect_err("above the agreed bound");
    assert_eq!(refused.fence, 7);
    assert!(
        refused.is_resurgence(),
        "above the sealed bound the refusal is the resurgence verdict"
    );
}

/// A survivor that has already folded successor-plane traffic from the
/// departing station cannot propose at all.
///
/// The successor-plane twin of `Ragged`, and it closes the ordering the fence
/// alone cannot: a member adopts, mints in the plane the epoch founds, and a
/// survivor folds that dot *before* anyone moves to evict it. No bound this
/// round agrees covers it --- the round is about the old plane --- and
/// refounding to bottom underneath it would leave the dot above the
/// coordinate the next generation's record names.
///
/// Recoverable, and the cure is to wait one boundary: after the seal that
/// plane is the next generation's old plane, and a round opened there
/// proposes a prefix covering the dot.
#[test]
fn a_survivor_holding_successor_traffic_cannot_propose() {
    let mut round = Departure::opened(GONE, 1, SURVIVORS).expect("a surviving family");
    assert_eq!(
        round.propose(&prefix(GONE, 4), &holding(GONE, &[1])),
        Err(DepartureRefusal::SuccessorHeld {
            station: GONE,
            held: 1,
        })
    );
    assert_eq!(round.bound(), None, "and the round is untouched");
    assert_eq!(round.admits(d(GONE, 9)), Ok(()), "having fenced nothing");
    assert_eq!(
        round.propose(&prefix(GONE, 4), &holding(1, &[1, 2])),
        Ok(4),
        "another station's successor traffic is not this station's"
    );
}

/// The successor plane's fence is bottom from the moment the round **opens**,
/// not from the moment it seals.
///
/// A bound is a counter in one plane's dot space, and the departing station's
/// coordinate in the plane the epoch founds is bottom whatever the round
/// agrees about the old one. Wait for the seal to say so and the window
/// between is a hole: a survivor admits a low successor-plane dot against the
/// old-plane fence, the round seals, the attestation crosses the boundary at
/// bottom, and that already-held dot now sits above the coordinate the next
/// generation's record names --- `BeyondSeal`, one generation later, which is
/// the wedge this whole object exists to prevent.
///
/// The phase decides only what the refusal *means*.
#[test]
fn the_successor_planes_fence_is_bottom_from_the_open() {
    let mut round = Departure::opened(GONE, 1, SURVIVORS).expect("a surviving family");
    let _ = round
        .propose(&prefix(GONE, 4), &DotSet::new())
        .expect("gap-free");
    assert_eq!(
        round.admits(d(GONE, 3)),
        Ok(()),
        "the old plane admits below the promise"
    );
    let held = round
        .admits_successor(d(GONE, 3))
        .expect_err("and the successor plane admits nothing at all");
    assert_eq!(held.fence, 0);
    assert!(
        !held.is_resurgence(),
        "a hold while the round is open: abandoning the eviction admits it"
    );
    assert_eq!(
        round.admits_successor(d(1, u64::MAX)),
        Ok(()),
        "and it speaks for one station, like its old-plane twin"
    );

    round.fold(&Vouched::trust(2, 7)).expect("a proposal");
    let _ = round.try_seal().expect("complete");
    assert!(
        round.admits(d(GONE, 7)).is_ok(),
        "the old plane still admits up to the agreed bound"
    );
    let refused = round
        .admits_successor(d(GONE, 1))
        .expect_err("the successor plane never does");
    assert!(
        refused.is_resurgence(),
        "and once sealed the refusal is a verdict"
    );
}

/// The fence judges one coordinate and refuses to be read as an admission
/// policy.
#[test]
fn the_fence_speaks_only_for_the_departing_station() {
    let mut round = Departure::opened(GONE, 1, SURVIVORS).expect("a surviving family");
    let _ = round
        .propose(&prefix(GONE, 1), &DotSet::new())
        .expect("gap-free");
    for station in SURVIVORS {
        assert_eq!(round.admits(d(station, u64::MAX)), Ok(()));
    }
    assert!(round.admits(d(GONE, 2)).is_err());
}

/// **The causality-gap evictee** (R-35's third pre-registered falsifier). A
/// survivor holding a dot above its own gap-free floor cannot propose at
/// all.
///
/// The hole is what makes the eviction unsafe in *both* directions: a bound
/// below it cannot be enforced, because this survivor already holds a dot
/// above it and no fence can un-hold one; a bound at or above it cannot be
/// completed, because the missing dot may exist nowhere the survivors can
/// reach. So the round refuses before it starts, and says which dot. The
/// cure is redelivery, and the refusal is recoverable by construction: the
/// tracker, the rounds, and the lineage are all untouched.
#[test]
fn a_ragged_holding_cannot_be_proposed_and_redelivery_cures_it() {
    let mut round = Departure::opened(GONE, 1, SURVIVORS).expect("a surviving family");
    let ragged = holding(GONE, &[1, 2, 4]);
    assert_eq!(
        round.propose(&ragged, &DotSet::new()),
        Err(DepartureRefusal::Ragged {
            station: GONE,
            floor: 2,
            held: 4,
        })
    );
    assert_eq!(
        round.proposals().collect::<alloc::vec::Vec<_>>(),
        [(1, None), (2, None)],
        "a refused proposal leaves the round untouched"
    );
    assert_eq!(round.admits(d(GONE, 4)), Ok(()), "and fences nothing");

    let mut cured = ragged;
    let _ = cured.insert(d(GONE, 3));
    assert_eq!(round.propose(&cured, &DotSet::new()), Ok(4));
}

/// A survivor that admits traffic above its own promise is refused, loudly.
///
/// The one duty the caller carries that the round can still check: it cannot
/// see the intake, but it can see a second proposal that contradicts the
/// first. No later agreement repairs it, because the bound the other
/// survivors are computing was already premised on the promise.
#[test]
fn raising_a_promised_fence_is_refused() {
    let mut round = Departure::opened(GONE, 1, SURVIVORS).expect("a surviving family");
    assert_eq!(round.propose(&prefix(GONE, 4), &DotSet::new()), Ok(4));
    assert_eq!(
        round.propose(&prefix(GONE, 4), &DotSet::new()),
        Ok(4),
        "a repeat of the same proposal absorbs"
    );
    assert_eq!(
        round.propose(&prefix(GONE, 6), &DotSet::new()),
        Err(DepartureRefusal::FenceBroken {
            station: GONE,
            promised: 4,
            held: 6,
        })
    );
    assert_eq!(round.bound(), None);
}

/// The round reads the survivors and never the departing member's own
/// claim, at every door.
#[test]
fn the_departing_station_has_no_voice_in_its_own_departure() {
    assert_eq!(
        Departure::opened(GONE, 1, ROSTER),
        Err(DepartureRefusal::SelfDeparture { station: GONE })
    );
    assert_eq!(
        Departure::opened(GONE, 1, []),
        Err(DepartureRefusal::EmptyFamily { station: GONE })
    );
    assert_eq!(
        Departure::opened(GONE, GONE, SURVIVORS),
        Err(DepartureRefusal::UnknownStation { station: GONE }),
        "and it cannot run the round either"
    );
    assert_eq!(
        Departure::opened(GONE, 4, SURVIVORS),
        Err(DepartureRefusal::UnknownStation { station: 4 }),
        "nor can anyone outside the family"
    );
    let mut round = Departure::opened(GONE, 1, SURVIVORS).expect("a surviving family");
    assert_eq!(
        round.fold(&Vouched::trust(GONE, u64::MAX)),
        Err(DepartureRefusal::UnknownStation { station: GONE })
    );
    assert_eq!(
        round.fold(&Vouched::trust(4, 1)),
        Err(DepartureRefusal::UnknownStation { station: 4 })
    );
}

/// Proposals fold by max, so the round is order-robust: duplicated,
/// reordered, and stale proposals absorb and the bound never regresses.
#[test]
fn proposals_absorb_and_the_bound_never_regresses() {
    let mut round = Departure::opened(GONE, 1, SURVIVORS).expect("a surviving family");
    let _ = round
        .propose(&prefix(GONE, 4), &DotSet::new())
        .expect("gap-free");
    for claim in [7, 2, 7] {
        round.fold(&Vouched::trust(2, claim)).expect("a proposal");
    }
    assert_eq!(
        round.try_seal().map(Departed::bound),
        Some(7),
        "the stale restatement of 2 cannot lower the slot"
    );
    round
        .fold(&Vouched::trust(2, 9))
        .expect("a proposal after the seal absorbs");
    assert_eq!(
        round.bound(),
        Some(7),
        "and the sealed attestation is latched"
    );
}

/// **A sealed round proves this replica fenced.** Its own slot is filled by
/// its own door and never from the wire, and the seal waits for it.
///
/// The promise and the number are one act, so a round driven to a `Departed`
/// entirely by reports would hand this replica an attestation it made no
/// promise behind --- and its peers would be sealing over a bound nothing here
/// is holding to. One guard carries it: the local slot refuses reports, so
/// the completeness check cannot pass until the door that fills that slot has
/// run, and both doors that fill it set the fence in the same act.
///
/// It is not equivocation detection, which needs cross-member evidence this
/// crate structurally cannot hold. It is the local guard that a promise is
/// made where it can be kept.
#[test]
fn a_sealed_round_proves_this_replica_fenced() {
    let mut round = Departure::opened(GONE, 1, SURVIVORS).expect("a surviving family");
    assert_eq!(
        round.fold(&Vouched::trust(1, 4)),
        Err(DepartureRefusal::Impersonated { station: 1 }),
        "the local slot is the one no report may fill"
    );
    round
        .fold(&Vouched::trust(2, 3))
        .expect("a peer's proposal");
    assert!(
        round.try_seal().is_none(),
        "and every other slot being full is still not a promise"
    );
    assert_eq!(
        round.admits(d(GONE, u64::MAX)),
        Ok(()),
        "nothing is fenced yet"
    );

    let _ = round
        .propose(&prefix(GONE, 2), &DotSet::new())
        .expect("gap-free");
    assert_eq!(
        round.try_seal().map(Departed::bound),
        Some(3),
        "the bound is the family's, with this replica's own promise in it"
    );
}

/// The recorded promise, not the rebuilt one: `reinstate` re-installs a
/// proposal this station durably made.
///
/// A replica must fence durably *before* emitting a proposal, because peers
/// price their bound on it the moment they hear it. Re-deriving the number
/// from rebuilt state at restart is the one way a crash could lower a fence
/// somebody else is already relying on, so the replay door takes the recorded
/// value and refuses one below a promise in force.
#[test]
fn a_restart_reinstates_the_promise_it_recorded() {
    let mut restarted = Departure::opened(GONE, 1, SURVIVORS).expect("a surviving family");
    restarted.reinstate(4).expect("a recorded proposal");
    assert!(
        restarted.admits(d(GONE, 5)).is_err(),
        "the fence is back where it was promised, not where the rebuilt \
         holding happens to sit"
    );
    restarted
        .reinstate(4)
        .expect("idempotent at the same value");
    assert_eq!(
        restarted.reinstate(2),
        Err(DepartureRefusal::FenceBroken {
            station: GONE,
            promised: 4,
            held: 2,
        }),
        "and a record below the live fence is refused, never absorbed"
    );
    assert_eq!(
        restarted.reinstate(6),
        Err(DepartureRefusal::FenceBroken {
            station: GONE,
            promised: 4,
            held: 6,
        }),
        "and so is one above it: the local proposal is latched, and a replay \
         that could raise a fence is a replay that could unfence traffic"
    );
    restarted
        .fold(&Vouched::trust(2, 6))
        .expect("a peer's proposal");
    assert_eq!(
        restarted.try_seal().map(Departed::bound),
        Some(6),
        "a reinstated promise completes the round exactly as a fresh one does"
    );
    restarted
        .reinstate(4)
        .expect("the recorded proposal still sits inside the agreed bound");
    assert!(
        restarted.admits(d(GONE, 5)).is_ok() && restarted.admits(d(GONE, 7)).is_err(),
        "and replaying it does not move the sealed fence in either direction"
    );
    assert_eq!(
        restarted.reinstate(7),
        Err(DepartureRefusal::FenceBroken {
            station: GONE,
            promised: 6,
            held: 7,
        }),
        "a record above the agreed bound loses to the family's agreement"
    );
}

/// An attestation must **cover** the family it narrows, and coverage is the
/// condition rather than equality.
///
/// What a consumer needs is that every member whose shadow will have to
/// consign the record actually fenced. A round missing one of them agrees a
/// bound that member never promised anything about, and is refused. A round
/// that also ranged over a member since departed has one promise to spare,
/// and is not --- which is what lets an attestation outlive a later
/// departure instead of being invalidated by it, in either order.
#[test]
fn an_attestation_must_cover_the_family_it_narrows() {
    let narrow = attested(&[(1, 4)]);
    assert_eq!(
        narrow.family().collect::<alloc::vec::Vec<_>>(),
        [1],
        "the attestation says whose agreement it is"
    );
    let mut short = settled(&ROSTER, &base());
    assert_eq!(
        short.abandon_attested(&narrow),
        Err(AbandonRefusal::FamilyMismatch { station: GONE }),
        "station 2 never proposed, and this tracker still meets over it"
    );
    assert!(!short.is_abandoned(GONE) && short.attested(GONE).is_none());

    // Two departures: 3 agreed by {1,2}, then 2 agreed by {1} alone.
    let wide = attested(&[(1, 1), (2, 1)]);
    let mut second = Departure::opened(2, 1, [1]).expect("a surviving family");
    let _ = second
        .propose(&prefix(2, 1), &DotSet::new())
        .expect("gap-free");
    let second = second.try_seal().expect("complete").clone();

    let mut tracker = settled(&ROSTER, &base());
    for departed in [&wide, &second] {
        let _ = tracker
            .abandon_attested(departed)
            .expect("each covers the family it narrows when it is agreed");
    }
    assert_eq!(tracker.attested(GONE), Some(1));
    assert_eq!(tracker.attested(2), Some(1));
    assert!(
        tracker.abandon_attested(&wide).is_ok(),
        "and re-declaring the earlier one absorbs, though this tracker has \
         since narrowed past the family that agreed it: coverage survives a \
         later departure, where equality would not"
    );

    let mut backwards = settled(&ROSTER, &base());
    assert_eq!(
        backwards.abandon_attested(&second),
        Err(AbandonRefusal::FamilyMismatch { station: 2 }),
        "the later departure applied first is refused, and rightly: station 3 \
         is still in this tracker's family and promised nothing about it"
    );
}

// -- The tracker door --------------------------------------------------------

/// A settled tracker over `roster`, every member vouching for `cut`.
fn settled(roster: &[u32], cut: &Cut) -> Stability {
    let mut stability = Stability::new(roster.iter().copied());
    for &station in roster {
        stability.report_cut(station, cut).expect("on roster");
    }
    stability
}

fn base() -> Cut {
    Cut::from_witnessed(vector(&[(1, 1), (2, 1), (3, 1)]))
}

/// The cut the survivors reach once the declaration is delivered.
fn delivered() -> Cut {
    Cut::from_witnessed(vector(&[(1, 2), (2, 1), (3, 1)]))
}

fn rank() -> Kairos {
    Kairos::new(2, 0, 1, 0u16)
}

/// Both doors narrow identically; only the attested one hands a coordinate
/// onward.
#[test]
fn only_the_attested_door_supplies_a_coordinate() {
    let mut bare = settled(&ROSTER, &base());
    let mut attesting = bare.clone();
    let departed = attested(&[(1, 1), (2, 1)]);

    let by_hand = bare.abandon(GONE).expect("a family remains");
    let by_round = attesting
        .abandon_attested(&departed)
        .expect("a family remains");
    assert_eq!(
        by_hand, by_round,
        "the operator's local write-off read is the same either way"
    );
    assert_eq!(bare.watermark(), attesting.watermark());
    assert!(bare.is_abandoned(GONE) && attesting.is_abandoned(GONE));

    assert_eq!(
        bare.attested(GONE),
        None,
        "a bare abandonment supplies none"
    );
    assert_eq!(attesting.attested(GONE), Some(1));
    assert_eq!(attesting.attested(1), None, "and speaks for nobody else");
    assert_eq!(attesting.attested(404), None);
}

/// The agreed coordinate is latched: a second attestation at another bound
/// is refused rather than absorbed.
///
/// Peers may already have sealed over it, and a seal does not roll back.
#[test]
fn re_attesting_at_another_bound_is_refused() {
    let mut stability = settled(&ROSTER, &base());
    let _ = stability
        .abandon_attested(&attested(&[(1, 4), (2, 4)]))
        .expect("a family remains");
    assert!(
        stability
            .abandon_attested(&attested(&[(1, 4), (2, 4)]))
            .is_ok(),
        "the same bound absorbs"
    );
    assert_eq!(
        stability.abandon_attested(&attested(&[(1, 4), (2, 9)])),
        Err(AbandonRefusal::Reattested {
            station: GONE,
            held: 4,
            offered: 9,
        })
    );
    assert_eq!(stability.attested(GONE), Some(4));
}

/// The attested door inherits `abandon`'s guards unchanged, and orders its
/// own refusals so each answers the question the caller asked.
///
/// Off the roster first, because nothing else is meaningful about a
/// non-member --- an attestation naming any family at all still cannot
/// install the belief that a stranger was a member. Then the latch. Then the
/// family, which is a fact about the attestation rather than about the
/// station. And `LastSurvivor` still fires underneath all three.
#[test]
fn the_attested_door_still_guards_the_family() {
    let mut stability = Stability::new(SURVIVORS);
    assert_eq!(
        stability.abandon_attested(&attested(&[(1, 1)])),
        Err(AbandonRefusal::UnknownStation { station: GONE }),
        "off the roster, and the narrow family it names does not change that"
    );
    let mut pair = settled(&SURVIVORS, &base());
    let _ = pair.abandon(2).expect("a family remains");
    let mut round = Departure::opened(1, 2, [2]).expect("a surviving family");
    let _ = round
        .propose(&prefix(1, 3), &DotSet::new())
        .expect("gap-free");
    let last = round.try_seal().expect("complete").clone();
    assert_eq!(
        pair.abandon_attested(&last),
        Err(AbandonRefusal::LastSurvivor { station: 1 }),
        "an attestation does not license emptying the family"
    );
    assert_eq!(pair.attested(1), None, "and records nothing when refused");
    assert!(!pair.is_abandoned(1), "nor narrows anything");
}

/// A later departure that would strand an earlier attested bound is refused.
///
/// An attestation promises that some survivor still holds everything below
/// its bound --- that is what lets the survivors complete each other, and
/// what a sealed record naming that coordinate depends on. Narrow past the
/// only member that can serve it and the promise is broken with no way back:
/// the meet at the departed coordinate can never reach the bound.
///
/// Servability has two arms because they answer at different times. A
/// *proposal* is evidence of holding at round time, which no later report can
/// supply; a *report* is evidence of holding now, which is how the condition
/// clears once the repair lane has carried the traffic to somebody else.
#[test]
fn a_departure_that_would_strand_an_attested_bound_is_refused() {
    // Only station 2 proposed through the bound, so only station 2 is known
    // to hold what the attestation writes the fleet's coverage down to.
    let mut stability = settled(&ROSTER, &base());
    let _ = stability
        .abandon_attested(&attested(&[(1, 1), (2, 5)]))
        .expect("a family remains");
    assert_eq!(stability.attested(GONE), Some(5));

    assert_eq!(
        stability.abandon(2),
        Err(AbandonRefusal::StrandsAttestation {
            station: 2,
            stranded: GONE,
        }),
        "station 2 is the only member that can still serve the agreed bound"
    );
    assert!(!stability.is_abandoned(2), "and the tracker is unchanged");

    // It clears the moment another survivor reports that far: the report arm
    // is how the repair lane unblocks the proposal arm.
    stability
        .report_cut(
            1,
            &Cut::from_witnessed(vector(&[(1, 1), (2, 1), (GONE, 5)])),
        )
        .expect("on roster");
    let _ = stability
        .abandon(2)
        .expect("station 1 can serve what station 2 was holding alone");

    // And installing is guarded from the other end, by the join rather than
    // by servability: a bound that rested on a member since departed is not
    // one the remaining family agreed, so it is refused as it arrives.
    let mut late = settled(&ROSTER, &base());
    let _ = late.abandon(2).expect("a family remains");
    assert_eq!(
        late.abandon_attested(&attested(&[(1, 1), (2, 9)])),
        Err(AbandonRefusal::FamilyMismatch { station: GONE }),
        "station 2 proposed the 9 and station 2 has already left, so the \
         family that remains would have agreed 1"
    );
    assert!(late.attested(GONE).is_none() && !late.is_abandoned(GONE));

    // And re-declaring an existing departure absorbs, ahead of every
    // condition about what a removal would cost: this one is not a removal.
    let mut again = settled(&ROSTER, &base());
    let _ = again
        .abandon_attested(&attested(&[(1, 1), (2, 5)]))
        .expect("a family remains");
    let _ = again
        .abandon_attested(&attested(&[(1, 1), (2, 5)]))
        .expect("idempotent");
    let _ = again.abandon(GONE).expect("and so is the bare door");
}

/// Two departures agreed concurrently can each be the only server of the
/// other's bound, and the second is **refused, not fatal**.
///
/// The schedule is honest: each round seals over its own family, and whichever
/// attestation the tracker takes first removes the sole server of the other's.
/// The refusal is the design working --- taking both would leave a coordinate
/// no survivor can ever reach --- and it clears the moment a remaining member
/// reports through the stranded bound.
#[test]
fn concurrent_departures_that_strand_each_other_refuse_rather_than_wedge() {
    let mut round_for_two = Departure::opened(2, 1, [1, GONE]).expect("a surviving family");
    let _ = round_for_two
        .propose(&DotSet::new(), &DotSet::new())
        .expect("holding nothing of 2");
    round_for_two
        .fold(&Vouched::trust(GONE, 5))
        .expect("the other survivor holds it to 5");
    let two = round_for_two.try_seal().expect("complete").clone();

    let three = attested(&[(1, 0), (2, 5)]);

    let mut stability = settled(&ROSTER, &base());
    let _ = stability
        .abandon_attested(&three)
        .expect("the first one applies");
    assert_eq!(
        stability.abandon_attested(&two),
        Err(AbandonRefusal::FamilyMismatch { station: 2 }),
        "and the second rested on a member that has now left, so the family \
         that remains never agreed its bound"
    );
    assert!(!stability.is_abandoned(2), "recoverably: nothing moved");

    // The cure is the repair lane and then a fresh round. Station 1 must
    // first come to hold what the pair were holding alone --- otherwise
    // letting station 2 go would strand station 3's bound as well --- and
    // then it can propose on its own authority over the family that is
    // actually there.
    stability
        .report_cut(
            1,
            &Cut::from_witnessed(vector(&[(1, 1), (2, 1), (GONE, 5)])),
        )
        .expect("on roster");
    let mut again = Departure::opened(2, 1, [1]).expect("a surviving family");
    let _ = again
        .propose(&prefix(2, 5), &DotSet::new())
        .expect("gap-free after repair");
    let two_again = again.try_seal().expect("complete").clone();
    let _ = stability
        .abandon_attested(&two_again)
        .expect("a round over the family that remains binds");
    assert_eq!(stability.attested(2), Some(5));
}

/// **The agreement law.** Two rounds for the same station over different
/// families must not leave two replicas holding different bounds.
///
/// Coverage alone admits a round that ranged *wider*, and a wider round can
/// agree a higher bound because a member since departed proposed it. Two
/// honest replicas holding different such rounds would then substitute
/// different coordinates into their seal records over the same surviving
/// family: one seals, the other waits forever for a watermark its own fence
/// permanently refuses. So the bound must be the join over the family that
/// *remains*, which every replica computes the same way.
#[test]
fn a_wider_rounds_higher_bound_is_not_the_remaining_familys_agreement() {
    const WIDE: [u32; 4] = [1, 2, 3, 4];
    let narrow = attested(&[(1, 1), (2, 1)]);

    let mut wider = Departure::opened(GONE, 1, [1, 2, 4]).expect("a surviving family");
    let _ = wider
        .propose(&prefix(GONE, 1), &DotSet::new())
        .expect("gap-free");
    wider.fold(&Vouched::trust(2, 1)).expect("a proposal");
    wider
        .fold(&Vouched::trust(4, 5))
        .expect("station 4 holds more of it");
    let wider = wider.try_seal().expect("complete").clone();
    assert_eq!(wider.bound(), 5, "the wider round agrees the higher bound");
    assert_eq!(narrow.bound(), 1);

    // Station 4 leaves first at both replicas, so both now meet over {1,2}.
    let mut holding_narrow = Stability::new(WIDE);
    let mut holding_wider = Stability::new(WIDE);
    for tracker in [&mut holding_narrow, &mut holding_wider] {
        for station in WIDE {
            tracker.report_cut(station, &base()).expect("on roster");
        }
        let mut leaving = Departure::opened(4, 1, [1, 2, GONE]).expect("a surviving family");
        let _ = leaving
            .propose(&DotSet::new(), &DotSet::new())
            .expect("holding nothing of station 4");
        for station in [2, GONE] {
            leaving
                .fold(&Vouched::trust(station, 0))
                .expect("a proposal");
        }
        let leaving = leaving.try_seal().expect("complete").clone();
        let _ = tracker
            .abandon_attested(&leaving)
            .expect("nothing of station 4's is written off");
    }

    let _ = holding_narrow
        .abandon_attested(&narrow)
        .expect("the round over exactly this family binds");
    assert_eq!(
        holding_wider.abandon_attested(&wider),
        Err(AbandonRefusal::FamilyMismatch { station: GONE }),
        "and the wider round does not, because its bound rested on station 4"
    );
    let _ = holding_wider
        .abandon_attested(&narrow)
        .expect("its only route to this departure is the round both families agree");
    assert_eq!(
        holding_narrow.attested(GONE),
        holding_wider.attested(GONE),
        "so both replicas retain the same bound over the same family"
    );
}

// -- The rounds the attestation completes ------------------------------------

/// Drives the window to wherever it gets, with `family` confirming and
/// adopting and `stability` supplying the license.
fn drive(stability: &mut Stability, family: &[u32]) -> (Epochs, Option<SealedEpoch>) {
    let (mut epochs, epoch) = confirmed(stability, family);
    adopt_all(&mut epochs, epoch, stability, family);
    let sealed = epochs.try_seal(stability).cloned();
    (epochs, sealed)
}

/// Opens the window and runs the confirmation round over `family`.
fn confirmed(stability: &mut Stability, family: &[u32]) -> (Epochs, crate::metis::EpochAddress) {
    let mut epochs = Epochs::new(ROSTER, horizon());
    let declaration = epochs
        .declare(DECLARATION, rank(), stability, &Cut::bottom())
        .expect("the settled watermark licenses the declaration");
    let epoch = declaration.address();
    for &station in family {
        stability
            .report_cut(station, &delivered())
            .expect("on roster");
        epochs
            .confirm(epoch, &Vouched::trust(station, delivered()))
            .expect("a confirmation for the delivered declaration");
    }
    (epochs, epoch)
}

/// Runs the adoption round over `family`, where the confirmation round
/// licenses it.
fn adopt_all(
    epochs: &mut Epochs,
    epoch: crate::metis::EpochAddress,
    stability: &Stability,
    family: &[u32],
) {
    if epochs.adopt(1, 2, stability).is_err() {
        return;
    }
    for &station in family {
        if station == 1 {
            continue;
        }
        epochs
            .adopt_report(
                epoch,
                &Vouched::trust(station, delivered().as_vector().get(station)),
            )
            .expect("an adoption report for a delivered candidate");
    }
}

/// **The payoff.** An attested departure completes the very rounds the bare
/// one deliberately leaves frozen.
///
/// The control is `membership::abandoning_cures_the_watermark_and_leaves_the_rounds_frozen`,
/// which pins the same fixture stalling at `Unconfirmed` with the watermark
/// already cured. The single difference here is that the departure carries
/// an agreed bound, so the rounds have a coordinate to read instead of a
/// report that is never coming.
#[test]
fn an_attested_departure_completes_the_rounds_a_bare_one_freezes() {
    let mut stability = settled(&ROSTER, &base());
    let _ = stability
        .abandon_attested(&attested(&[(1, 1), (2, 1)]))
        .expect("a family remains");
    let (epochs, sealed) = drive(&mut stability, &SURVIVORS);
    let sealed = sealed.expect("the rounds complete over the surviving family");
    assert_eq!(sealed.declaration().declaration(), DECLARATION);
    assert_eq!(
        epochs.sealed().count(),
        1,
        "and the lineage advanced exactly once"
    );
}

/// The sealed join carries the *agreed bound* at the departed coordinate,
/// which is what makes the record identical at every survivor.
#[test]
fn the_sealed_join_carries_the_agreed_bound_at_the_departed_coordinate() {
    let mut stability = settled(&ROSTER, &base());
    let _ = stability
        .abandon_attested(&attested(&[(1, 1), (2, 1)]))
        .expect("a family remains");
    let (_, sealed) = drive(&mut stability, &SURVIVORS);
    let sealed = sealed.expect("the rounds complete");
    assert_eq!(
        sealed.sealed_join().get(GONE),
        1,
        "the bound the round agreed, not the departed member's own claim"
    );
    for station in SURVIVORS {
        assert_eq!(
            sealed.sealed_join().get(station),
            delivered().as_vector().get(station),
            "and every survivor's own coordinate is still its own report"
        );
    }
}

/// The agreed bound is a bar the watermark must still clear: the survivors
/// have to *hold* what they wrote off up to.
///
/// This is the half that keeps the substitution honest at the consignment
/// door. A seal whose join names the departed coordinate at `b` promises
/// every survivor's shadow covers the departed member's traffic through `b`;
/// the round guarantees some survivor holds it, and the repair lane carries
/// it to the rest. Until it has, the seal does not fire --- a recoverable
/// stall, in the one place where the alternative is a record no peer can
/// consign.
#[test]
fn the_attested_bound_waits_for_the_survivors_to_hold_it() {
    let mut stability = settled(&ROSTER, &base());
    let _ = stability
        .abandon_attested(&attested(&[(1, 5), (2, 3)]))
        .expect("a family remains");
    let (mut epochs, epoch) = confirmed(&mut stability, &SURVIVORS);
    adopt_all(&mut epochs, epoch, &stability, &SURVIVORS);
    assert!(
        epochs.try_seal(&stability).is_none(),
        "the survivors have delivered the departed member only to 1, and the \
         agreed bound is 5"
    );

    // The repair lane completes them, and the same monotone check fires.
    let repaired = Cut::from_witnessed(vector(&[(1, 2), (2, 1), (GONE, 5)]));
    for station in SURVIVORS {
        stability.report_cut(station, &repaired).expect("on roster");
    }
    adopt_all(&mut epochs, epoch, &stability, &SURVIVORS);
    let sealed = epochs
        .try_seal(&stability)
        .expect("the watermark has covered the agreed bound");
    assert_eq!(
        sealed.sealed_join().get(GONE),
        5,
        "and the record carries the agreed bound, which no survivor's own \
         report could have supplied"
    );
}

/// **Failure of recovery** (R-35's fourth pre-registered falsifier), and the
/// confluence constraint the departure note fixed.
///
/// Whether a replica folded the departing member's last report before or
/// after its operator evicted it is a per-replica race no round arbitrates.
/// So the substitution is unconditional: an attested member's own testimony
/// is discarded even when it is sitting in the slot. Two survivors that saw
/// different halves of that race seal byte-identical records.
#[test]
fn the_substitution_discards_the_departed_members_own_testimony() {
    let heard = {
        let mut stability = settled(&ROSTER, &base());
        let mut epochs = Epochs::new(ROSTER, horizon());
        let declaration = epochs
            .declare(DECLARATION, rank(), &stability, &Cut::bottom())
            .expect("licensed");
        let epoch = declaration.address();
        // The departing member's last words reach this replica first: a full
        // confirmation and an adoption report far above what the survivors
        // will agree.
        let ahead = Cut::from_witnessed(vector(&[(1, 2), (2, 1), (GONE, 9)]));
        epochs
            .confirm(epoch, &Vouched::trust(GONE, ahead))
            .expect("a confirmation for the delivered declaration");
        epochs
            .adopt_report(epoch, &Vouched::trust(GONE, 9))
            .expect("an adoption report");
        let _ = stability
            .abandon_attested(&attested(&[(1, 1), (2, 1)]))
            .expect("a family remains");
        for &station in &SURVIVORS {
            stability
                .report_cut(station, &delivered())
                .expect("on roster");
            epochs
                .confirm(epoch, &Vouched::trust(station, delivered()))
                .expect("a confirmation");
        }
        let _ = epochs.adopt(1, 2, &stability).expect("confirmed");
        epochs
            .adopt_report(epoch, &Vouched::trust(2, 1))
            .expect("an adoption report");
        epochs
            .try_seal(&stability)
            .cloned()
            .expect("the seal fires")
    };
    let unheard = {
        let mut stability = settled(&ROSTER, &base());
        let _ = stability
            .abandon_attested(&attested(&[(1, 1), (2, 1)]))
            .expect("a family remains");
        drive(&mut stability, &SURVIVORS).1.expect("the seal fires")
    };
    assert_eq!(
        heard, unheard,
        "the record is a function of the family and the attestation alone"
    );
}

/// **The recovering-evictee race** (R-35's first pre-registered falsifier,
/// the charter's own): eviction testimony that arrives after a peer has
/// already sealed cannot move the record.
///
/// The fixture is the dangerous half of that schedule. One replica completes
/// the whole honest round and seals; another, short of the departing
/// member's adoption report, evicts it instead. The two must agree, and they
/// do --- not by luck but because the seal *forces* the bound. Sealing
/// requires the watermark to cover the join, so every survivor had already
/// delivered the departing member's traffic up to its adoption counter,
/// which puts every survivor's gap-free floor there; and an adopted member
/// mints nothing further old-addressed, which keeps every floor from going
/// past it. The join of those floors has exactly one value available to it.
#[test]
fn eviction_testimony_after_a_peer_has_sealed_cannot_move_the_record() {
    let honest = {
        let mut stability = settled(&ROSTER, &base());
        drive(&mut stability, &ROSTER).1.expect("the honest seal")
    };
    let evicting = {
        let mut stability = settled(&ROSTER, &base());
        // What the honest seal forced: the departing member's own coordinate
        // in the sealed join is exactly what every survivor had delivered,
        // so it is exactly what every survivor proposes.
        let forced = honest.sealed_join().get(GONE);
        let _ = stability
            .abandon_attested(&attested(&[(1, forced), (2, forced)]))
            .expect("a family remains");
        drive(&mut stability, &SURVIVORS)
            .1
            .expect("the evicting seal")
    };
    assert_eq!(
        honest, evicting,
        "a peer that sealed honestly and one that evicted agree on the record"
    );
}

/// An attestation for a member that is *not* silent changes nothing about
/// the other members: the round's reach is one coordinate.
#[test]
fn attesting_one_departure_leaves_every_other_slot_asking() {
    let mut stability = settled(&ROSTER, &base());
    let _ = stability
        .abandon_attested(&attested(&[(1, 1), (2, 1)]))
        .expect("a family remains");
    let mut epochs = Epochs::new(ROSTER, horizon());
    let declaration = epochs
        .declare(DECLARATION, rank(), &stability, &Cut::bottom())
        .expect("licensed");
    let epoch = declaration.address();
    stability.report_cut(1, &delivered()).expect("on roster");
    epochs
        .confirm(epoch, &Vouched::trust(1, delivered()))
        .expect("a confirmation");
    assert_eq!(
        epochs.adopt(1, 2, &stability),
        Err(EpochRefusal::Unconfirmed),
        "survivor 2 is still owed, and no attestation speaks for it"
    );
    assert!(epochs.try_seal(&stability).is_none());
}

/// PRD 0028 falsifier 8, pinned as it fell: riding the record *relocates*
/// the S340 re-key rather than retiring it. An admission-bearing seal
/// widens the roster (R-89), so the tracker rebuilt at that boundary
/// ranges over a family the old attestation never named --- an
/// attestation carried across over its *old* family refuses the widened
/// tracker's family check, and the cure is exactly the archive branch's:
/// `Departed::refounded` re-keys over the family the new plane opens
/// with. Coverage and possession part in the re-key: holders carry the
/// base prefix, a joiner's entry is zero until its own report proves
/// possession (the companion pin below). The fleet-grade kill is
/// `fleet::scenarios::arrival::an_attested_fleet_admits_and_the_carry_crosses_the_widening_seal`.
#[test]
fn a_refounded_attestation_binds_the_family_the_new_plane_opens_with() {
    let departed = attested(&[(1, 2), (2, 2)]);

    // The widened tracker: the roster after an admission-bearing seal.
    // An empty base is the nothing-survived case, so the carry is bottom.
    let mut widened = Stability::new([1, 2, GONE, 4]);
    let narrow = departed.refounded([1, 2], &DotSet::new());
    assert_eq!(
        widened.abandon_attested(&narrow),
        Err(AbandonRefusal::FamilyMismatch { station: GONE }),
        "an attestation re-keyed over the old family refuses the widened tracker"
    );

    // The re-key over the new plane's family binds --- the roster passes
    // directly, the departing seat filtered out --- at the compacted
    // prefix the consigned base carries for the departed station. The
    // coordinate is derived inside the door from the base itself: a raw
    // counter is not expressible, so an under- or over-stated prefix
    // requires a forged base, the trusted-local input the caller opens
    // the plane with.
    let wide = departed.refounded([1, 2, GONE, 4], &prefix(GONE, 1));
    assert!(wide.family().eq([1u32, 2, 4]));
    assert_eq!(
        wide.bound(),
        1,
        "the door reads the base's gap-free prefix, nothing else"
    );
    let mut ready = Stability::new([1, 2, GONE, 4]);
    ready
        .report_cut(1, &Cut::from_witnessed(vector(&[(GONE, 1)])))
        .expect("on roster");
    let _ = ready
        .abandon_attested(&wide)
        .expect("the re-keyed attestation binds the widened family");
    assert_eq!(
        ready.attested(GONE),
        Some(1),
        "the carried coordinate is the base prefix, not bottom"
    );

    // A ragged base cannot overstate: the door reads the gap-free
    // prefix, so a dot above a hole contributes nothing.
    let ragged = departed.refounded([1, 2, GONE, 4], &holding(GONE, &[1, 3]));
    assert_eq!(ragged.bound(), 1, "the prefix stops at the first hole");

    // The stationary case is unchanged: the same re-key over the same
    // family is the S339 carry, at the same base prefix.
    let mut stationary = Stability::new(ROSTER);
    stationary
        .report_cut(1, &Cut::from_witnessed(vector(&[(GONE, 1)])))
        .expect("on roster");
    let _ = stationary
        .abandon_attested(&departed.refounded(ROSTER, &prefix(GONE, 1)))
        .expect("the stationary carry still binds");
    assert_eq!(stationary.attested(GONE), Some(1));
}

/// The re-key's coverage/possession split, pinned as the review found it
/// (S343, the third round's P1): a widened carry gives an admitted joiner
/// a *family-coverage* entry, never *possession evidence*. A joiner's
/// admission proves nothing about what it holds --- it may not have
/// bootstrapped yet (PRD 0027 R5) --- so a synthetic proposal at the
/// prefix would let the incumbents abandon every actual holder while the
/// offline joiner's fabricated entry satisfies the serving check, and the
/// attested data would be gone for good. The joiner's entry is zero:
/// `binds` still counts the key, the join is unmoved, and holder status
/// is earned through the joiner's own report.
#[test]
fn a_widened_carry_claims_no_possession_for_the_joiner() {
    let departed = attested(&[(1, 2), (2, 2)]);
    let wide = departed.refounded([1, 2, GONE, 4], &prefix(GONE, 2));
    assert_eq!(wide.bound(), 2);
    assert_eq!(
        wide.proposal_of(4),
        Some(0),
        "the joiner's entry is coverage, not a claim of holding"
    );

    let mut tracker = Stability::new([1, 2, GONE, 4]);
    for holder in [1, 2] {
        tracker
            .report_cut(holder, &Cut::from_witnessed(vector(&[(GONE, 2)])))
            .expect("on roster");
    }
    let _ = tracker
        .abandon_attested(&wide)
        .expect("the re-keyed attestation binds the widened family");

    // The actual holders leave one by one. The second refusal is the pin:
    // the offline joiner's synthetic entry must not serve the bound.
    let _ = tracker
        .abandon(1)
        .expect("station 2 still serves the bound");
    assert_eq!(
        tracker.abandon(2),
        Err(AbandonRefusal::StrandsAttestation {
            station: 2,
            stranded: GONE,
        }),
        "abandoning the last real holder strands the attestation; the joiner's \
         coverage entry is not possession"
    );

    // The joiner earns holder status the honest way: its own report
    // reaches the bound (it bootstrapped and holds the base), and the
    // abandon clears.
    tracker
        .report_cut(4, &Cut::from_witnessed(vector(&[(GONE, 2)])))
        .expect("on roster");
    let _ = tracker
        .abandon(2)
        .expect("the joiner's report now serves the bound");
}