minerva 0.2.0

//! The exposure table: where the crash-only guarantees stop.
//!
//! # What this module is, and is not
//!
//! It is **not** a claim that Minerva tolerates Byzantine faults. The
//! charter's model is crash-only honest peers and this module does not move
//! it. Every other arm of the fleet runs an honest fabric, and every law
//! stated elsewhere is still proven under the model the charter states.
//!
//! It is a *boundary map*. Alma's `mod byzantium` is about to stand its
//! walls on this foundation, and a foundation whose failure surface is
//! unmeasured is sandstone however solid it happens to be. This module
//! measures it: one voice says two things, and the table below records which
//! laws survive, which fail closed, and which diverge.
//!
//! # The adversary
//!
//! The weakest one that reaches anything. A seated [`Voice`](super::fabric::Voice)
//! may only reshape traffic its own station genuinely emitted; it cannot
//! forge another station's identity, invent traffic from silence, or reach
//! inside a peer. A finding here is therefore a finding under every stronger
//! adversary, and a law that survives here has survived *only* equivocation,
//! which is the only claim this table is allowed to make.
//!
//! # The table
//!
//! The epoch protocol has exactly **three** peer-report channels, and the
//! sharp result is that only one of them is a safety surface. Each row is
//! measured, not argued; the argument beside it is why the measurement came
//! out that way.
//!
//! *The stability report* (`Stability::report_cut`, the channel the
//! watermark is a meet over) is a **liveness** lever only. Understating
//! freezes the fleet. Overstating reaches the victim's *latch* --- a liar
//! that is genuinely behind can make its peers fix a winner on stability
//! that does not exist --- and dies there, because sealing needs an adoption
//! report from every roster member and a member adopts only under its **own**
//! watermark. A liar cannot profit from a lie it must itself believe
//! ([`a_forked_report_cannot_force_a_seal_the_liar_has_not_licensed`]).
//!
//! *The confirmation cut* (`Epochs::confirm`, the channel the winner latch
//! reads) cannot fork the winner, which would be the worst outcome
//! available: a split in the seal record's `declaration` rather than its
//! join. Understating at a declaration's own coordinate is **refused at the
//! door** ([`EpochRefusal::ConfirmationDoesNotCover`]), so a declaration's
//! minter always confirms with a cut covering its own dot and every
//! declaration dot reaches every member's confirmation join
//! ([`a_forked_confirmation_cannot_hide_its_own_declaration`]). At the
//! coordinates that check does not reach, overstating freezes and
//! understating changes nothing
//! ([`a_forked_confirmation_either_freezes_the_fleet_or_changes_nothing`]).
//!
//! *The adoption counter* (`Epochs::adopt_report`) is the one that reaches
//! the record, and it is worth seeing why: it is the only one of the three
//! that is a **bare self-report with no second source**. A confirmation is
//! re-proven against the declaration it names; a stability report is
//! re-proven against the liar's own latch; an adoption counter is a station's
//! unchecked claim about its own output. The direction of the lie decides
//! the rest:
//!
//! * *Overstating* to one peer is **fail-closed**. The victim never seals:
//!   `try_seal`'s `join <= watermark` gate refuses, because the inflated
//!   join outruns the watermark that the liar's own honest stability reports
//!   hold down. The window wedges open exactly as it does for a silent
//!   member, and repair cures it. The lie defeats itself.
//! * *Understating* to one peer **diverges the record, and is caught one
//!   door later**. A smaller join trivially passes the watermark gate, so
//!   the victim seals a `SealedEpoch` whose `sealed_join` differs from its
//!   peers'. The two honest members then answer `recognize` differently for
//!   the same dot, one granting the duplicate verdict and the other refusing
//!   `AddressMiss`. The document plane is still protected: the next door,
//!   `EpochShadow::consign`, refuses the shrunken join on its
//!   no-less-and-no-more coverage check, so an honest consumer halts rather
//!   than folding a divergent base.
//!
//! The halt is *asymmetric*, and that is part of the result rather than a
//! harness artifact. `Epochs::try_seal` has already advanced the lineage by
//! the time the consignment door refuses, and a seal does not roll back, so
//! the wedged replica's epoch ledger stands one generation ahead of its
//! document plane and stays there. The window itself survives, because the
//! door hands the shadow back instead of consuming it on refusal:
//! recoverable at the shadow, not at the seal. And the wedge is **durable**:
//! restart replays the fenced report and re-derives the same divergent seal,
//! while the bootstrap door cannot start because a wedged replica has no
//! sealed checkpoint to offer. No shipped door recovers it
//! ([`the_wedge_is_durable_and_no_shipped_door_recovers_it`]); recovery
//! would need a door that adopts a peer's agreed plane while keeping one's
//! own identity, and no such door exists here.
//!
//! # What this means for the crossing
//!
//! Two readings, and the second is the actionable one.
//!
//! *The document plane is defended in depth.* Both directions fail closed
//! before any divergent state is folded. No BFT retrofit of the commit path
//! is implied by this table: the seal's own gates and the consignment door
//! catch what one voice can do.
//!
//! *The seal record was not, and now is.* `Epochs::try_seal` hands back a
//! divergent `SealedEpoch` **before** the consignment door refuses, and
//! `recognize` answers from it immediately. That record is exactly what
//! Alma's B5 certificate binds over, so a certificate assembled between the
//! two doors could certify a record its own peers do not hold and stay
//! internally valid while doing it: the quorum argument quietly weakening
//! from "`Q` members signed *the* digest" to "*a* digest".
//!
//! S335 closed that window by typing the duty. The full reasoning is at
//! [`Consigned`](crate::metis::Consigned); the short form is that
//! [`EpochShadow::consign`](crate::metis::EpochShadow::consign) is the only
//! mint of the certifiable grade, so a record its own document plane refuses
//! can no longer reach a certificate through this crate's types
//! ([`a_divergent_seal_record_never_reaches_the_certifiable_grade`]).
//!
//! # What the typed doors do and do not close
//!
//! S333 gave both peer-report doors a [`Vouched`](crate::metis::Vouched)
//! grade, so a claim now arrives bound to the station that made it and a
//! caller's verification has an unavoidable place to sit. That closes
//! *misattribution* and it is worth having. **It does not close what this
//! module measures, and these tests still pass because of that.**
//!
//! The reason is structural rather than incidental. A Byzantine station may
//! validly vouch two different claims to two peers. Each recipient's vouch
//! is then genuine, both fold, and the records still diverge. The local
//! machine sees one voice saying one thing and has no basis to suspect
//! otherwise: **detecting a contradiction requires comparing what different
//! members were told**, which is cross-member evidence Minerva structurally
//! cannot hold. That is precisely why Alma's B7 exists, and no amount of
//! typing at a report door substitutes for it.
//!
//! S335's [`Consigned`](crate::metis::Consigned) does not close it either,
//! and claims less than it might appear to. It does not detect the lie, does
//! not recover the wedged member, and is not a quorum. It removes exactly
//! one failure --- a locally-refused record reaching a certificate --- and
//! it can, because the consignment door is the only one in the protocol that
//! reads a *second, independent* source: the sealed join comes from peer
//! **reports**, the coverage check from the deltas actually **delivered**.
//!
//! That raises the sufficiency question, and it is answered:
//! [`withholding_the_covering_delta_freezes_instead_of_hiding_the_lie`]
//! shows the obvious escalation --- silence the second source too ---
//! degenerates, because the delta channel is also what the watermark is made
//! of. An understated adoption report therefore either meets a victim
//! holding the covering delta and is refused at the consignment door, or
//! meets one that does not and freezes the fleet. There is no third case
//! under one voice.
//!
//! # The lie family
//!
//! [`Voice`](super::fabric::Voice) has five implementors, and they compose
//! through [`Then`](super::fabric::Then) because B8's adversary is named as
//! simultaneous rather than sequential: `ForkedAdoption` (equivocation on
//! the adoption counter), `ForkedCut` (equivocation on either cut-carrying
//! channel, which is what let the two quiet rows above be measured at all),
//! `Withholding` (per-peer, per-kind silence, which no severance models
//! because a severance is symmetric and curable), `Replaying` (the first
//! stateful lie, and the reason `Voice` is a trait), and their composition.
//! Reordering is deliberately absent: the fabric's scheduler already reaches
//! every interleaving, so a reordering voice would duplicate transport the
//! harness owns rather than add adversary power.
//! Two exhibits, because the lies are not peers.
//! [`withholding_from_one_peer_freezes_the_whole_fleet`] shows that
//! withholding *dominates*: one station declining to report to one peer
//! freezes the roster-wide meet and nothing anywhere seals, so composed with
//! anything it wins first and masks the rest. The composition that reaches
//! the seal is therefore replay plus equivocation
//! ([`a_composed_adversary_still_cannot_fold_divergent_state`]), which holds
//! the line that matters: members either agree or refuse, and none folds a
//! base its peers reject. Liveness is expressly not claimed in either.
//!
//! # A note to whoever hardens this further
//!
//! The rows are of two kinds and they age differently. The *positive* rows
//! (the report and confirmation channels, and the sufficiency bound) are
//! laws: if one of them ever fails, the protocol lost a defense. The
//! *characterization* rows on the adoption channel pin the divergence as
//! something that happens, not as something guaranteed, and
//! [`an_understated_adoption_diverges_the_seal_record`] asserts the split it
//! describes. If a future change ever makes a local door able to refuse an
//! equivocated report, this module is **expected to fail**, and that failure
//! is the hardening landing rather than a regression. Rewrite the table
//! then; do not restore the assertions.

use crate::metis::dot::RawDot;
extern crate alloc;

use alloc::boxed::Box;
use alloc::vec::Vec;
use core::num::NonZeroUsize;

use crate::kairos::Kairos;
use crate::metis::tests::support::dot as d;
use crate::metis::{
    Cut, Dot, EpochAddress, EpochRefusal, Epochs, SealedEpoch, Stability, VersionVector, Vouched,
};

use super::fabric::{
    Equivocator, Fabric, ForkedAdoption, ForkedCut, Note, Replaying, Then, Withholding,
};
use super::replica::Replica;
use super::{act, fleet_of};

const ROSTER: [u32; 3] = [1, 2, 3];
/// The declaring station's dot, one above the base cut's ceiling for it.
/// Kept in its raw spelling because [`EpochAddress::try_from_parts`] reads
/// a raw pair; [`declaration_dot`] is the identity form (ruling R-91).
const DECLARATION: (u32, u64) = (1, 2);
/// The liar, and the adoption counter it truthfully holds.
const LIAR: u32 = 3;
/// The roster minus the liar.
const SURVIVING: [u32; 2] = [1, 2];
const LIAR_TRUTH: u64 = 1;

/// The declaration's identity spelling.
fn declaration_dot() -> Dot {
    d(DECLARATION.0, DECLARATION.1)
}

/// The liar's own dot, at the counter it truthfully holds.
fn liar_dot(counter: u64) -> Dot {
    d(LIAR, counter)
}

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

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

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

fn address() -> EpochAddress {
    EpochAddress::try_from_parts(1, DECLARATION).expect("a well-formed address")
}

/// Drives one honest machine through a whole round, folding `liar_says` as
/// the liar's adoption counter, and returns the machine with whatever it
/// sealed.
///
/// Every input but that one counter is identical across calls, so any
/// difference in outcome is attributable to the lie alone.
fn round(own: u32, own_counter: u64, liar_says: u64) -> (Epochs, Option<SealedEpoch>) {
    let mut stability = Stability::new(ROSTER);
    for &station in &ROSTER {
        stability.report_cut(station, &base()).expect("on roster");
    }
    let mut epochs = Epochs::new(ROSTER, NonZeroUsize::new(2).expect("positive"));

    let declaration = epochs
        .declare(
            declaration_dot(),
            Kairos::new(2, 0, 1, 0u16),
            &stability,
            &Cut::bottom(),
        )
        .expect("the settled watermark licenses the declaration");
    let epoch = declaration.address();

    for &station in &ROSTER {
        stability
            .report_cut(station, &delivered())
            .expect("on roster");
        epochs
            .confirm(epoch, &Vouched::trust(station, delivered()))
            .expect("a confirmation for the delivered declaration");
    }
    let _ = epochs
        .adopt(own, own_counter, &stability)
        .expect("the confirmation round is complete under the watermark");

    for &station in &ROSTER {
        if station == own {
            continue;
        }
        let counter = if station == LIAR {
            liar_says
        } else {
            delivered().as_vector().get(station)
        };
        epochs
            .adopt_report(epoch, &Vouched::trust(station, counter))
            .expect("an adoption report for a delivered candidate");
    }

    let sealed = epochs.try_seal(&stability).cloned();
    (epochs, sealed)
}

/// The control: one voice, one truth, one record.
#[test]
fn an_honest_round_seals_one_record() {
    let (first, sealed_first) = round(1, 2, LIAR_TRUTH);
    let (second, sealed_second) = round(2, 1, LIAR_TRUTH);

    assert_eq!(
        sealed_first, sealed_second,
        "honest members seal the same record"
    );
    let sealed = sealed_first.expect("the round seals");
    assert_eq!(
        sealed.sealed_join().get(LIAR),
        LIAR_TRUTH,
        "the join carries the liar's honest counter"
    );
    for machine in [&first, &second] {
        assert_eq!(
            machine.recognize(address(), liar_dot(LIAR_TRUTH)),
            Ok(()),
            "both grant the duplicate verdict for a covered dot"
        );
    }
}

/// Overstating the adoption counter to one peer **fails closed**: the victim
/// never seals.
///
/// The inflated join must clear `try_seal`'s `join <= watermark` gate, and
/// the watermark is held down by the liar's own *honest* stability reports,
/// which it did not inflate. So the lie defeats itself: the window wedges
/// open exactly as a silent member wedges it, and repair cures it. This is
/// the direction that needs no defending.
#[test]
fn an_overstated_adoption_wedges_the_window() {
    let (_, honest) = round(1, 2, LIAR_TRUTH);
    let (victim, overstated) = round(2, 1, LIAR_TRUTH + 4);

    assert!(honest.is_some(), "the undeceived member seals");
    assert!(
        overstated.is_none(),
        "an overstated join cannot clear the watermark, so nothing seals"
    );
    assert_eq!(
        victim.recognize(address(), liar_dot(LIAR_TRUTH)),
        Err(EpochRefusal::WindowOpen { open: address() }),
        "the victim's window is still open: wedged, not diverged"
    );
}

/// Understating the adoption counter to one peer **diverges the seal
/// record**, and nothing at the seal catches it.
///
/// A smaller join trivially satisfies `join <= watermark`, so the victim
/// seals a record its peers do not hold. The two honest members then answer
/// the duplicate contract differently for the same dot. This is the crack,
/// and it is in the artifact Alma's certificate binds over rather than in
/// the document plane.
#[test]
fn an_understated_adoption_diverges_the_seal_record() {
    let (honest, sealed_honest) = round(1, 2, LIAR_TRUTH);
    let (victim, sealed_victim) = round(2, 1, LIAR_TRUTH - 1);

    let sealed_honest = sealed_honest.expect("the undeceived member seals");
    let sealed_victim = sealed_victim.expect("the understated join seals too: nothing refuses it");
    assert_ne!(
        sealed_honest, sealed_victim,
        "one voice saying two things splits the sealed record"
    );
    assert_eq!(sealed_honest.sealed_join().get(LIAR), LIAR_TRUTH);
    assert_eq!(
        sealed_victim.sealed_join().get(LIAR),
        LIAR_TRUTH - 1,
        "the victim's join carries the counter it was told"
    );
    assert_eq!(
        sealed_honest.declaration(),
        sealed_victim.declaration(),
        "the winner still agrees: the split is in the join, not the choice"
    );

    // The duplicate contract (R6) splits with it: the same old-addressed dot
    // absorbs at one honest member and is refused as a violation at the
    // other. This is the R-82 recognizer hole reached by equivocation rather
    // than by a trimmed proof, and it does not heal by sealing forward.
    assert_eq!(honest.recognize(address(), liar_dot(LIAR_TRUTH)), Ok(()));
    assert_eq!(
        victim.recognize(address(), liar_dot(LIAR_TRUTH)),
        Err(EpochRefusal::AddressMiss {
            epoch: address(),
            dot: RawDot::new(LIAR, LIAR_TRUTH),
        }),
        "the victim refuses what its peer grants"
    );
}

/// The backstop: an understated adoption reaches the document plane and is
/// **refused at the consignment door**, so an honest consumer halts instead
/// of folding a divergent base.
///
/// End-to-end through the whole shipped stack, with the lie seated in the
/// transport and the replicas composed entirely of shipped parts. The
/// refusal is the consignment's no-less-and-no-more coverage check catching
/// the shrunken join, and it is why the table's second row reads "caught one
/// door later" rather than "corrupts".
///
/// The halt is the point, and so is its lateness: the seal record had
/// already diverged and `recognize` had already split before this door said
/// no.
#[test]
fn an_understated_adoption_halts_the_consumer_at_the_consignment_door() {
    let mut fleet = fleet_of(&ROSTER, NonZeroUsize::new(2).expect("positive"));
    for replica in fleet.values_mut() {
        replica.tolerate_refusals();
    }
    let mut fabric = Fabric::new(0xB17E_0002, &ROSTER, 0);
    fabric.corrupt(
        Equivocator::within(1).speaking(LIAR, Box::new(ForkedAdoption { to: 2, delta: -1 })),
    );

    for (offset, &station) in ROSTER.iter().enumerate() {
        let _ = act(&mut fabric, &mut fleet, station, |replica, out| {
            replica.insert_visible(offset, out)
        });
    }
    fabric.drain(&mut fleet);
    let _ = act(&mut fabric, &mut fleet, 1, |replica, out| {
        replica.try_declare(out)
    })
    .expect("station 1 declares over the settled watermark");
    fabric.drain(&mut fleet);

    let halted: Vec<u32> = fleet
        .values()
        .filter(|replica| replica.halted().is_some())
        .map(super::replica::Replica::id)
        .collect();
    assert!(
        !halted.is_empty(),
        "the consignment door must refuse a divergent sealed join, not fold it"
    );
    assert!(
        halted.contains(&2),
        "the deceived member is the one that halts, got {halted:?}"
    );
}

/// The fault set refuses to outgrow the budget it declares.
///
/// A harness that silently over-corrupted would produce findings outside the
/// model it cites, which is worse than no finding: every claim in the table
/// above is scoped to one voice, and this is what keeps that scope honest.
#[test]
#[should_panic(expected = "may not exceed its declared budget")]
fn the_fault_set_refuses_to_outgrow_its_budget() {
    let _ = Equivocator::within(1)
        .speaking(2, Box::new(ForkedAdoption { to: 1, delta: -1 }))
        .speaking(3, Box::new(ForkedAdoption { to: 1, delta: -1 }));
}

/// Builds the wedge: station 3 understates its adoption counter to station
/// 2 only, and the fleet runs to quiescence.
fn wedged_fleet() -> (
    alloc::collections::BTreeMap<u32, super::replica::Replica>,
    Fabric,
) {
    let mut fleet = fleet_of(&ROSTER, NonZeroUsize::new(2).expect("positive"));
    for replica in fleet.values_mut() {
        replica.tolerate_refusals();
    }
    let mut fabric = Fabric::new(0xB17E_0002, &ROSTER, 0);
    fabric.corrupt(
        Equivocator::within(1).speaking(LIAR, Box::new(ForkedAdoption { to: 2, delta: -1 })),
    );
    for (offset, &station) in ROSTER.iter().enumerate() {
        let _ = act(&mut fabric, &mut fleet, station, |replica, out| {
            replica.insert_visible(offset, out)
        });
    }
    fabric.drain(&mut fleet);
    let _ = act(&mut fabric, &mut fleet, 1, |replica, out| {
        replica.try_declare(out)
    })
    .expect("station 1 declares over the settled watermark");
    fabric.drain(&mut fleet);
    (fleet, fabric)
}

/// The wedge is **durable**, and no shipped door recovers from it.
///
/// This is the recovery question the exposure table left open, answered. The
/// asymmetry is exactly one generation: the wedged replica's epoch ledger
/// sealed while its document plane did not follow. Two candidate recoveries
/// exist in the crate, and neither works:
///
/// * *Crash and restart* does not cure it. The equivocated adoption report
///   was fenced into the journal before the seal (the durability duty
///   requires exactly that), so the replay re-derives the identical
///   divergent seal and refuses again at the same door. Restart is the cure
///   for *lost* state, and this state is not lost, it is wrong.
/// * *Re-join through the bootstrap door* cannot start. `Replica::bootstrap`
///   composes a peer's verified lineage proof with the joiner's **own**
///   sealed checkpoint, and a wedged replica has no such checkpoint to
///   offer: `joiner_checkpoint` requires the document plane to stand
///   exactly at the seal, which is precisely the invariant the wedge broke.
///
/// So a station wedged this way stays wedged. That is fail-closed and it
/// never corrupts the fleet's document plane, but it is a *permanent*
/// liveness loss for that member, and recovering it would need a door that
/// does not exist here: adopting a peer's agreed plane while keeping one's
/// own identity and counter. Naming that door is a design question and this
/// test does not answer it; it pins that the question is real.
#[test]
fn the_wedge_is_durable_and_no_shipped_door_recovers_it() {
    let (mut fleet, mut fabric) = wedged_fleet();

    let wedged = fleet.get(&2).expect("roster member");
    assert!(wedged.halted().is_some(), "station 2 is wedged");
    assert_eq!(
        (wedged.generation(), wedged.epochs().generation()),
        (1, 2),
        "the ledger stands exactly one generation ahead of the plane"
    );
    assert!(
        wedged.joiner_checkpoint().is_none(),
        "a wedged replica cannot offer the sealed checkpoint the bootstrap door needs"
    );
    for &station in &[1u32, 3] {
        assert!(
            fleet[&station].halted().is_none(),
            "station {station} was told the truth and crossed the seal"
        );
    }

    // Restart from the durable journal: the fenced report replays and the
    // same seal is re-derived, so the wedge returns rather than clearing.
    let journal = fleet[&2].journal().clone();
    let _ = fleet.insert(
        2,
        super::replica::Replica::rehydrate_tolerating_refusals(
            2,
            &ROSTER,
            NonZeroUsize::new(2).expect("positive"),
            &journal,
        ),
    );
    super::resync(&mut fabric, &fleet, 2);
    fabric.drain(&mut fleet);

    let restarted = fleet.get(&2).expect("roster member");
    assert!(
        restarted.halted().is_some(),
        "restart re-derives the divergent seal: the wedge is durable, not volatile"
    );
    assert_ne!(
        fleet[&1].epochs(),
        fleet[&2].epochs(),
        "the ledgers stay split across the restart"
    );
}

/// The divergent seal record **never becomes certifiable**, and that is the
/// crack closed rather than merely mapped.
///
/// The exposure table's second row is that an understated adoption counter
/// diverges the seal record and is caught one door later, at the
/// consignment. Until S335 "one door later" was a *duty stated in prose*: a
/// consumer assembling Alma's B5 checkpoint certificate read
/// `Epochs::try_seal`'s record, and nothing in the types stopped it reading
/// the one its peers reject.
///
/// [`Consigned`](crate::metis::Consigned) closes that window. The wedged
/// replica's ledger holds the divergent record --- `try_seal` retired it and
/// a seal does not roll back --- but its certifiable list does not, because
/// the only mint is the door that refused. So the two planes disagree by
/// exactly one generation, and every certifiable record the fleet holds is
/// one its peers hold too.
///
/// The wedge is still a wedge: the type does not recover the member, and
/// [`the_wedge_is_durable_and_no_shipped_door_recovers_it`] says so. What it
/// removes is the *silent* failure beside it, where a halted member's
/// divergent record reaches a certificate anyway and the quorum argument
/// weakens from "`Q` members signed *the* digest" to "*a* digest".
#[test]
fn a_divergent_seal_record_never_reaches_the_certifiable_grade() {
    let (fleet, _) = wedged_fleet();

    let wedged = fleet.get(&2).expect("roster member");
    assert!(wedged.halted().is_some(), "station 2 is wedged");

    // Non-vacuity: the divergent record really is retired in the ledger.
    let retired = wedged
        .epochs()
        .sealed()
        .next()
        .expect("the wedged member retired the divergent record");
    assert_eq!(
        retired.sealed_join().get(LIAR),
        LIAR_TRUTH - 1,
        "the retired record carries the counter the liar forked"
    );
    // The law: it is not certifiable, because the door that refused it is
    // the only door that grants the grade.
    assert_eq!(
        wedged.certified().count(),
        0,
        "the divergent record reached the certifiable grade"
    );

    // And the members that crossed hold a certifiable record, identical to
    // one another and *not* the divergent one.
    for &station in &[1u32, 3] {
        let certified: Vec<&SealedEpoch> = fleet[&station].certified().collect();
        assert_eq!(
            certified.len(),
            1,
            "station {station} accepted exactly one consignment"
        );
        assert_eq!(
            certified[0].sealed_join().get(LIAR),
            LIAR_TRUTH,
            "station {station} certified a record carrying the honest counter"
        );
    }
    assert_eq!(
        fleet[&1].certified().collect::<Vec<_>>(),
        fleet[&3].certified().collect::<Vec<_>>(),
        "the certifiable grade must agree across every member that holds it"
    );
}

/// A **composed** adversary: one station that replays a stale confirmation
/// *and* equivocates on its adoption counter, which is the simultaneity
/// Alma's B8 rung names.
///
/// The point is not a new failure mode but that the family composes and that
/// composition does not escape the fail-closed envelope: members either
/// agree or refuse, and none proceeds on a base its peers do not hold.
///
/// Note what is asserted *positively*. A first cut of this test checked
/// agreement only across the members that crossed, and under its schedule
/// nobody crossed at all, so the headline law held vacuously and the test
/// measured nothing (a review round caught it). The lesson is pinned here as
/// assertions rather than prose: this schedule must produce a real crossing
/// and a real halt, or it is not exercising the composition.
#[test]
fn a_composed_adversary_still_cannot_fold_divergent_state() {
    fn is_confirm(note: &Note) -> bool {
        matches!(note, Note::Confirm { .. })
    }

    let mut fleet = fleet_of(&ROSTER, NonZeroUsize::new(2).expect("positive"));
    for replica in fleet.values_mut() {
        replica.tolerate_refusals();
    }
    let mut fabric = Fabric::new(0xB17E_0003, &ROSTER, 0);
    fabric.corrupt(Equivocator::within(1).speaking(
        LIAR,
        Box::new(Then(
            Replaying::of(is_confirm),
            ForkedAdoption { to: 2, delta: -1 },
        )),
    ));

    for (offset, &station) in ROSTER.iter().enumerate() {
        let _ = act(&mut fabric, &mut fleet, station, |replica, out| {
            replica.insert_visible(offset, out)
        });
    }
    fabric.drain(&mut fleet);
    let _ = act(&mut fabric, &mut fleet, 1, |replica, out| {
        replica.try_declare(out)
    })
    .expect("station 1 declares over the settled watermark");
    fabric.drain(&mut fleet);

    let crossed: Vec<u32> = fleet
        .values()
        .filter(|replica| replica.halted().is_none() && replica.generation() > 1)
        .map(super::replica::Replica::id)
        .collect();
    let halted: Vec<u32> = fleet
        .values()
        .filter(|replica| replica.halted().is_some())
        .map(super::replica::Replica::id)
        .collect();

    // Non-vacuity first: the schedule must actually exercise both outcomes.
    assert_eq!(
        crossed,
        [1, 3],
        "the truthfully-told members must really cross, or the law below is vacuous"
    );
    assert_eq!(
        halted,
        [2],
        "the deceived member must really halt, or the composition reached nothing"
    );

    // Then the law: every member that crossed folded the same plane.
    for &station in &crossed {
        assert_eq!(
            fleet[&station].text().store().to_bytes(),
            fleet[&crossed[0]].text().store().to_bytes(),
            "station {station} folded a document plane its fellow crossers do not hold"
        );
        assert_eq!(
            fleet[&station].epochs(),
            fleet[&crossed[0]].epochs(),
            "station {station} crossed on a ledger its fellow crossers do not hold"
        );
    }
}

/// Selective withholding by one station is a **fleet-wide liveness attack**,
/// and it dominates every other lie.
///
/// Station 3 simply declines to send its stability reports to station 1,
/// telling everyone else the truth. Because the watermark is a roster-wide
/// meet, station 1's can never rise, so station 1 never confirms, so no
/// member's confirmation round completes and *nothing anywhere seals*. This
/// is the silent-member freeze reached by a lie rather than a failure, and
/// it is why the composed adversary above deliberately omits withholding:
/// composed with anything, withholding wins first and masks the rest.
///
/// Fail-closed and total: no member crosses, no member halts on a divergent
/// record, and nothing is folded. Liveness is the price, exactly as the
/// charter's R8 honesty says it is.
#[test]
fn withholding_from_one_peer_freezes_the_whole_fleet() {
    fn is_report(note: &Note) -> bool {
        matches!(note, Note::Report { .. })
    }

    let mut fleet = fleet_of(&ROSTER, NonZeroUsize::new(2).expect("positive"));
    for replica in fleet.values_mut() {
        replica.tolerate_refusals();
    }
    let mut fabric = Fabric::new(0xB17E_0004, &ROSTER, 0);
    fabric.corrupt(Equivocator::within(1).speaking(
        LIAR,
        Box::new(Withholding {
            from: 1,
            kind: is_report,
        }),
    ));

    for (offset, &station) in ROSTER.iter().enumerate() {
        let _ = act(&mut fabric, &mut fleet, station, |replica, out| {
            replica.insert_visible(offset, out)
        });
    }
    fabric.drain(&mut fleet);
    let _ = act(&mut fabric, &mut fleet, 1, |replica, out| {
        replica.try_declare(out)
    })
    .expect("station 1 declares over the watermark it had before the withholding bit");
    fabric.drain(&mut fleet);

    for &station in &ROSTER {
        let replica = &fleet[&station];
        assert_eq!(
            replica.generation(),
            1,
            "station {station} sealed under a withheld watermark"
        );
        assert!(
            replica.halted().is_none(),
            "station {station} halted: the freeze must be a stall, never a refusal"
        );
        assert!(
            replica.epochs().sealed().next().is_none(),
            "station {station} retired a seal the frozen round cannot license"
        );
    }
}

/// Abandoning the withholder releases the meet, and that cures *this*
/// freeze completely (S336).
///
/// R-83 left eviction as the sole liveness path under a withholder.
/// `Stability::abandon` narrows the family the meet ranges over, and here
/// the meet was the only thing short: the withholder equivocates on the
/// report channel alone, so both epoch rounds already held its testimony.
/// Releasing the meet therefore carries the fleet all the way through ---
/// seal, consignment, certifiable grade, identical records --- from a state
/// that was stalled forever a moment earlier.
///
/// The rounds themselves deliberately do **not** narrow, and that is the
/// other half of the design rather than an omission.
///
/// A `sealed_join` is built from the family's *self*-reports. Each member's
/// adoption report is its own-station counter, so an honest member's own
/// traffic enters the join *because that member is there to report it*.
/// Narrow the rounds and the join loses the departed member's coordinate ---
/// but the survivors' shadows still hold deltas it minted above the declared
/// cut. `EpochShadow::consign` then refuses `BeyondSeal`, one door *after*
/// `Epochs::try_seal` has advanced the lineage, and a seal does not roll
/// back: the recoverable stall becomes the durable wedge measured at
/// [`the_wedge_is_durable_and_no_shipped_door_recovers_it`]. Measured, not
/// argued: that is exactly what the first cut of this session shipped.
///
/// Here the meet was the *only* thing short --- the withholder equivocates
/// on the report channel alone, so both epoch rounds already had its
/// testimony --- and releasing the meet cures the freeze end to end: the
/// survivors seal, consign, and reach the certifiable grade on the identical
/// record. That is the whole of the liveness R-83 said membership had to
/// buy, for this fault.
///
/// It is not the whole of membership. A member silent on *every* channel
/// leaves the rounds themselves short, and there the tracker's cure stops:
/// `membership::abandoning_cures_the_watermark_and_leaves_the_rounds_frozen`
/// pins that the stall stays, recoverably, rather than being traded for the
/// wedge. Closing it needs an abandonment bound that is **agreed** across
/// survivors and **enforced** against later traffic --- R-35's eviction
/// attestation, reached from the other end, and the charter's first
/// requirement (`docs/metis-membership-departure.adoc`).
#[test]
fn abandoning_the_withholder_releases_the_meet_and_cures_this_freeze() {
    fn is_report(note: &Note) -> bool {
        matches!(note, Note::Report { .. })
    }

    let mut fleet = fleet_of(&ROSTER, NonZeroUsize::new(2).expect("positive"));
    for replica in fleet.values_mut() {
        replica.tolerate_refusals();
    }
    let mut fabric = Fabric::new(0xB17E_0005, &ROSTER, 0);
    fabric.corrupt(Equivocator::within(1).speaking(
        LIAR,
        Box::new(Withholding {
            from: 1,
            kind: is_report,
        }),
    ));

    for (offset, &station) in ROSTER.iter().enumerate() {
        let _ = act(&mut fabric, &mut fleet, station, |replica, out| {
            replica.insert_visible(offset, out)
        });
    }
    fabric.drain(&mut fleet);
    let _ = act(&mut fabric, &mut fleet, 1, |replica, out| {
        replica.try_declare(out)
    })
    .expect("station 1 declares over the watermark it had before the withholding bit");
    fabric.drain(&mut fleet);

    let pinned = fleet[&1].watermark();
    for &station in &SURVIVING {
        let abandoned = act(&mut fabric, &mut fleet, station, |replica, out| {
            replica.abandon(LIAR, out)
        });
        assert_eq!(abandoned.station(), LIAR);
        assert_eq!(
            abandoned.counter(),
            1,
            "the bound is what the survivors vouched for of the withholder's minting"
        );
    }
    fabric.drain(&mut fleet);

    assert!(
        fleet[&1].watermark() > pinned,
        "the meet is released: every watermark consumer moves again"
    );
    let survivors: Vec<&Replica> = SURVIVING.iter().map(|station| &fleet[station]).collect();
    for replica in &survivors {
        assert_eq!(
            replica.generation(),
            2,
            "and the rounds, which were never short, complete and seal"
        );
        assert!(
            replica.halted().is_none(),
            "with no replica wedged at the consignment door"
        );
    }
    // The read's honest character, in one line: it answers differently
    // depending on whom you ask. Station 2 still receives the withholder's
    // reports and sees it resurgent --- correctly, since it is alive.
    // Station 1, the one the lie targeted, sees nothing at all.
    assert_eq!(
        fleet[&2].resurgent().collect::<Vec<_>>(),
        [LIAR],
        "the survivor still being reported to observes the departed member speaking"
    );
    assert!(
        fleet[&1].resurgent().next().is_none(),
        "the survivor the lie targeted observes nothing, on the same fleet at the same moment"
    );
    let first: Vec<&SealedEpoch> = survivors[0].certified().collect();
    assert!(!first.is_empty(), "a record reached the certifiable grade");
    for replica in &survivors[1..] {
        assert_eq!(
            replica.certified().collect::<Vec<_>>(),
            first,
            "and every survivor accepted the identical record"
        );
    }
    assert_eq!(
        first[0].sealed_join().get(LIAR),
        1,
        "carrying the withholder's own coordinate, because it reported on this channel"
    );
}

/// Attesting the departure of a member silent on **every** channel carries
/// the fleet through the seal (S339).
///
/// This is the half [`abandoning_the_withholder_releases_the_meet_and_cures_this_freeze`]
/// named as left uncured, and it is the harder one: the withholder there
/// equivocated on the report channel alone, so both epoch rounds already
/// held its testimony and the meet was the only thing short. Here the member
/// is severed --- crash, partition, and total withholding reach the same
/// state --- so the rounds themselves are short, and narrowing the family
/// under them is what the departure note measured as a durable wedge.
///
/// What closes it is the departure *round*. Each survivor proposes the
/// gap-free prefix of the silent member's dots it holds and fences there;
/// the survivors' join is the agreed bound; and the epoch rounds substitute
/// that bound for the testimony that is not coming. Both directions matter
/// and both are asserted below: the fleet completes, and it completes on
/// *identical* records that every survivor accepts at the consignment door.
///
/// The bound is not a free parameter. It is what the survivors hold, and the
/// seal still waits for the watermark to cover it, so the record a departure
/// licenses is one the survivors can actually consign.
#[test]
fn attesting_the_silent_members_departure_carries_the_fleet_through_the_seal() {
    let mut fleet = fleet_of(&ROSTER, NonZeroUsize::new(2).expect("positive"));
    for replica in fleet.values_mut() {
        replica.tolerate_refusals();
    }
    let mut fabric = Fabric::new(0xB17E_0006, &ROSTER, 0);

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

    // The member goes silent on every channel at once, after its own
    // traffic reached the survivors.
    fabric.sever(&[LIAR]);
    let _ = act(&mut fabric, &mut fleet, 1, |replica, out| {
        replica.try_declare(out)
    })
    .expect("station 1 declares over the settled watermark");
    fabric.drain(&mut fleet);

    for &station in &SURVIVING {
        let replica = &fleet[&station];
        assert_eq!(
            replica.generation(),
            1,
            "station {station} sealed a window the silent member never confirmed"
        );
        assert!(replica.epochs().sealed().next().is_none());
    }

    // The operators evict. Each survivor proposes what it holds of the
    // silent member and fences there; the proposals cross the (still
    // severed) fabric between the survivors.
    for &station in &SURVIVING {
        let prefix = act(&mut fabric, &mut fleet, station, |replica, out| {
            replica.open_departure(LIAR, out)
        })
        .expect("a gap-free holding of the silent member's traffic");
        assert_eq!(
            prefix, LIAR_TRUTH,
            "station {station} proposes exactly the prefix it holds"
        );
    }
    fabric.drain(&mut fleet);

    let survivors: Vec<&Replica> = SURVIVING.iter().map(|station| &fleet[station]).collect();
    for replica in &survivors {
        assert_eq!(
            replica.attested(LIAR),
            Some(1),
            "the departure crossed the boundary at the compacted prefix the \
             new base carries for the evictee (its surviving element), never \
             at the old plane's number"
        );
        assert_eq!(
            replica.generation(),
            2,
            "and the rounds the silence froze complete and seal"
        );
        assert!(
            replica.halted().is_none(),
            "with no replica wedged at the consignment door: the bound the \
             record names is one the survivors hold"
        );
    }
    let first: Vec<&SealedEpoch> = survivors[0].certified().collect();
    assert!(!first.is_empty(), "a record reached the certifiable grade");
    for replica in &survivors[1..] {
        assert_eq!(
            replica.certified().collect::<Vec<_>>(),
            first,
            "and every survivor accepted the identical record"
        );
    }
    assert_eq!(
        first[0].sealed_join().get(LIAR),
        LIAR_TRUTH,
        "carrying the agreed bound, which no self-report of the silent \
         member's supplied"
    );
    for replica in &survivors {
        assert!(
            replica.resurgences().next().is_none(),
            "and nothing arrived above the fence, so no verdict is claimed"
        );
    }
}

/// A departed member that speaks again is refused **at the fence**, and that
/// refusal is the crate's only resurgence *verdict* (S339).
///
/// [`Stability::resurgent`] observes a report absorbed after local
/// abandonment, and R-84's counterweight is that the read proves nothing: an
/// in-flight report is a false positive, an abandoned member's traffic on
/// another channel leaves no mark, and two survivors of one fleet answer it
/// differently at the same moment. All three failings have the same cause
/// --- there is no fence, so an observation is not evidence about anything.
///
/// The departure round supplies one. Every survivor promised to admit
/// nothing of the departed member's above the agreed bound, and the promise
/// is what makes the refusal mean something: the dot named below reached a
/// replica that had already written it off, on a fleet whose seal record
/// says it does not exist.
///
/// What it does *not* prove is the member's liveness, and the fixture is
/// built so that distinction is visible: the traffic here is old-plane
/// traffic minted before the healing, arriving late. The verdict is about
/// the decision, not the station.
///
/// The counterfactual is measured rather than argued. Absorb this delta and
/// the shadow carries a dot above its sealed join, which is the refusal
/// [`the_wedge_is_durable_and_no_shipped_door_recovers_it`] measures as
/// unrecoverable.
#[test]
fn a_departed_member_that_speaks_again_is_refused_at_the_fence() {
    let mut fleet = fleet_of(&ROSTER, NonZeroUsize::new(2).expect("positive"));
    for replica in fleet.values_mut() {
        replica.tolerate_refusals();
    }
    let mut fabric = Fabric::new(0xB17E_0007, &ROSTER, 0);
    for (offset, &station) in ROSTER.iter().enumerate() {
        let _ = act(&mut fabric, &mut fleet, station, |replica, out| {
            replica.insert_visible(offset, out)
        });
    }
    fabric.drain(&mut fleet);
    fabric.sever(&[LIAR]);
    let _ = act(&mut fabric, &mut fleet, 1, |replica, out| {
        replica.try_declare(out)
    })
    .expect("station 1 declares over the settled watermark");
    fabric.drain(&mut fleet);
    for &station in &SURVIVING {
        let _ = act(&mut fabric, &mut fleet, station, |replica, out| {
            replica.open_departure(LIAR, out)
        })
        .expect("a gap-free holding");
    }
    fabric.drain(&mut fleet);
    let sealed: Vec<&SealedEpoch> = fleet[&1].certified().collect();
    let record = sealed[0].clone();
    assert_eq!(record.sealed_join().get(LIAR), LIAR_TRUTH);

    // The departed member, which never heard any of it, keeps editing its
    // own plane and is healed back into the fabric.
    let _ = act(&mut fabric, &mut fleet, LIAR, |replica, out| {
        replica.insert_visible(9, out)
    });
    fabric.heal();
    fabric.drain(&mut fleet);

    for &station in &SURVIVING {
        let replica = &fleet[&station];
        let verdicts: Vec<&crate::metis::Fenced> = replica.resurgences().collect();
        assert_eq!(
            verdicts.len(),
            1,
            "station {station} recorded exactly one resurgence verdict"
        );
        assert_eq!(verdicts[0].station, LIAR);
        assert!(
            verdicts[0].counter > record.sealed_join().get(LIAR),
            "and the dot it names is one the sealed record says does not exist"
        );
        assert!(
            verdicts[0].is_resurgence(),
            "a sealed fence refused it, so it is evidence rather than a prompt"
        );
        assert!(
            replica.halted().is_none(),
            "station {station} halted: the fence is what keeps this a refusal \
             at the door rather than a wedge one door later"
        );
        assert_eq!(
            replica.certified().collect::<Vec<_>>(),
            [&record],
            "and the record it accepted is untouched"
        );
    }
}

/// A crash between the promise and the seal does not lower the fence
/// (S339).
///
/// A proposal is a promise about this replica's *future* intake, and peers
/// price the agreed bound on it the moment they hear it. So the fence has to
/// be durable before the proposal is emitted, and re-derived at its recorded
/// value rather than from whatever the rebuilt state happens to hold: a
/// restart that re-derived a lower number would admit exactly the traffic its
/// peers have already sealed over.
///
/// This crashes the proposer immediately after it promises, which is the
/// worst moment available --- the promise is on the wire, the round is not
/// sealed, and every receipt since the last emission is gone.
#[test]
fn a_crash_between_the_promise_and_the_seal_does_not_lower_the_fence() {
    let horizon = NonZeroUsize::new(2).expect("positive");
    let mut fleet = fleet_of(&ROSTER, horizon);
    let mut fabric = Fabric::new(0xB17E_0008, &ROSTER, 0);
    for (offset, &station) in ROSTER.iter().enumerate() {
        let _ = act(&mut fabric, &mut fleet, station, |replica, out| {
            replica.insert_visible(offset, out)
        });
    }
    fabric.drain(&mut fleet);
    fabric.sever(&[LIAR]);

    let promised = act(&mut fabric, &mut fleet, 1, |replica, out| {
        replica.open_departure(LIAR, out)
    })
    .expect("a gap-free holding");
    assert_eq!(promised, LIAR_TRUTH);

    super::crash(&mut fleet, &ROSTER, horizon, 1);
    let restarted = &fleet[&1];
    assert_eq!(
        restarted.departure_fence(LIAR),
        Some(promised),
        "the promise came back at the value it was emitted at"
    );
    assert!(
        restarted.attested(LIAR).is_none(),
        "and the round is still open: a restart re-enters, it does not seal"
    );

    // The other survivor completes the round against the restarted replica,
    // and the fleet goes through on the same bound.
    let _ = act(&mut fabric, &mut fleet, 2, |replica, out| {
        replica.open_departure(LIAR, out)
    })
    .expect("a gap-free holding");
    let _ = act(&mut fabric, &mut fleet, 1, |replica, out| {
        replica.try_declare(out)
    })
    .expect("station 1 declares over the settled watermark");
    fabric.drain(&mut fleet);
    for &station in &SURVIVING {
        assert_eq!(
            fleet[&station].attested(LIAR),
            Some(1),
            "station {station} sealed and carried the departure across, at \
             the compacted prefix the new base holds of the evictee"
        );
        assert!(fleet[&station].certified().next().is_some());
    }
}

/// Two departures survive a restart, in the order they were agreed (S339).
///
/// Each attestation names the family that was surviving when it was agreed,
/// so a replica rebuilding from its checkpoint must re-apply them in that
/// order: the second names a family the tracker has not narrowed to until the
/// first has been applied. The checkpoint therefore carries the departures as
/// a sequence rather than a set, and this is the fixture that tells the
/// difference --- a `BTreeSet` would replay station 2 before station 3 and
/// present a family that never existed.
#[test]
fn two_departures_survive_a_restart_in_the_order_they_were_agreed() {
    let horizon = NonZeroUsize::new(2).expect("positive");
    let mut fleet = fleet_of(&ROSTER, horizon);
    let mut fabric = Fabric::new(0xB17E_000A, &ROSTER, 0);
    for (offset, &station) in ROSTER.iter().enumerate() {
        let _ = act(&mut fabric, &mut fleet, station, |replica, out| {
            replica.insert_visible(offset, out)
        });
    }
    fabric.drain(&mut fleet);

    // Station 3 leaves first, agreed by 1 and 2; then station 2 leaves,
    // agreed by 1 alone.
    for &station in &SURVIVING {
        let _ = act(&mut fabric, &mut fleet, station, |replica, out| {
            replica.open_departure(LIAR, out)
        })
        .expect("a gap-free holding");
    }
    fabric.drain(&mut fleet);
    let _ = act(&mut fabric, &mut fleet, 1, |replica, out| {
        replica.open_departure(2, out)
    })
    .expect("a gap-free holding");
    fabric.drain(&mut fleet);
    assert_eq!(fleet[&1].attested(LIAR), Some(1));
    assert_eq!(fleet[&1].attested(2), Some(1));

    super::crash(&mut fleet, &ROSTER, horizon, 1);
    let restarted = &fleet[&1];
    assert_eq!(
        restarted.attested(LIAR),
        Some(1),
        "the earlier departure came back"
    );
    assert_eq!(
        restarted.attested(2),
        Some(1),
        "and so did the one agreed over the family the first one left"
    );
    assert_eq!(restarted.departure_fence(LIAR), Some(1));
    assert_eq!(restarted.departure_fence(2), Some(1));
}

/// The successor plane's fence: a departing member's native traffic never
/// rides the seal into the next generation (S339).
///
/// The old-plane fence bounds what a departing member minted *before* the
/// eviction. A member that has already adopted can also mint in the plane the
/// epoch founds, and that plane's attestation is bottom
/// (`Departed::refounded`), so its coordinate in the next generation's record
/// is bottom too. Traffic that slipped into the transition would therefore
/// sit above the coordinate the next round's record names --- the same
/// `BeyondSeal` shape the old-plane fence exists to prevent, one generation
/// later.
#[test]
fn a_departing_members_native_traffic_does_not_ride_the_seal() {
    let mut fleet = fleet_of(&ROSTER, NonZeroUsize::new(2).expect("positive"));
    for replica in fleet.values_mut() {
        replica.tolerate_refusals();
    }
    let mut fabric = Fabric::new(0xB17E_0009, &ROSTER, 0);
    for (offset, &station) in ROSTER.iter().enumerate() {
        let _ = act(&mut fabric, &mut fleet, station, |replica, out| {
            replica.insert_visible(offset, out)
        });
    }
    fabric.drain(&mut fleet);

    // The survivors evict, and the member is never told: it keeps taking
    // part in the protocol, which is what makes this the hard case.
    for &station in &SURVIVING {
        let _ = act(&mut fabric, &mut fleet, station, |replica, out| {
            replica.open_departure(LIAR, out)
        })
        .expect("a gap-free holding");
    }
    fabric.drain(&mut fleet);
    let _ = act(&mut fabric, &mut fleet, 1, |replica, out| {
        replica.try_declare(out)
    })
    .expect("station 1 declares over the settled watermark");
    fabric.drain(&mut fleet);
    for &station in &SURVIVING {
        assert_eq!(
            fleet[&station].generation(),
            2,
            "station {station} should have sealed over the substituted bound"
        );
    }
    assert!(
        fleet[&LIAR].adopted() || fleet[&LIAR].generation() == 2,
        "the evicted member adopted the epoch it was never told it left"
    );

    // And now it writes in the plane the epoch founded.
    let native = act(&mut fabric, &mut fleet, LIAR, |replica, out| {
        replica.insert_visible(0, out)
    });
    fabric.drain(&mut fleet);

    for &station in &SURVIVING {
        let replica = &fleet[&station];
        assert!(
            !replica.effective_order().contains(&native),
            "station {station} folded the evicted member's successor-plane \
             dot {native:?}"
        );
        assert!(
            replica.halted().is_none(),
            "station {station} halted: the refusal must be at the fence, not \
             at the consignment door"
        );
        assert!(
            replica.resurgences().any(|verdict| verdict.station == LIAR),
            "station {station} refused it silently rather than as a verdict"
        );
    }
}

/// Successor-plane traffic minted **before** the departure round seals is
/// refused too, and the refusal is what keeps the boundary safe (S339).
///
/// The dangerous schedule, and it is an honest one. The target adopts, mints
/// in the plane the epoch founds, and only then does its round complete. If
/// the successor fence waited for the seal, a survivor would judge that dot
/// against the *old*-plane promise, admit it, and carry it across a boundary
/// where the attestation is bottom --- putting a held dot above the
/// coordinate the next generation's record names, which is `BeyondSeal` one
/// generation later.
///
/// So the successor fence is bottom from the moment the round opens, and the
/// phase decides only what a refusal means: a hold here, a verdict after the
/// seal. What the exhibit measures is that nothing of the departing station's
/// crosses, and that no survivor is wedged by it.
#[test]
fn successor_traffic_minted_before_the_round_seals_is_still_refused() {
    let mut fleet = fleet_of(&ROSTER, NonZeroUsize::new(2).expect("positive"));
    for replica in fleet.values_mut() {
        replica.tolerate_refusals();
    }
    let mut fabric = Fabric::new(0xB17E_000B, &ROSTER, 0);
    for (offset, &station) in ROSTER.iter().enumerate() {
        let _ = act(&mut fabric, &mut fleet, station, |replica, out| {
            replica.insert_visible(offset, out)
        });
    }
    fabric.drain(&mut fleet);
    let _ = act(&mut fabric, &mut fleet, 1, |replica, out| {
        replica.try_declare(out)
    })
    .expect("station 1 declares over the settled watermark");
    // Everything except the target's own adoption report: every replica
    // adopts and opens its transition, and no survivor can seal until the
    // departure supplies the coordinate that report would have.
    let mut without_the_targets_adoption =
        |_: u32, note: &Note| !matches!(note, Note::Adoption { station: LIAR, .. });
    fabric.drain_selected(&mut fleet, &mut without_the_targets_adoption);
    for &station in &SURVIVING {
        assert!(
            fleet[&station].adopted(),
            "station {station} must be holding the window open"
        );
    }
    assert_eq!(
        fleet[&LIAR].generation(),
        2,
        "and the target, which folds its own report, has already crossed: \
         everything it mints now belongs to the successor plane"
    );

    // One survivor opens the round; the target mints natively *before* it
    // can seal; only then does the second survivor propose.
    let _ = act(&mut fabric, &mut fleet, 1, |replica, out| {
        replica.open_departure(LIAR, out)
    })
    .expect("a gap-free holding");
    fabric.drain_selected(&mut fleet, &mut without_the_targets_adoption);
    let native = act(&mut fabric, &mut fleet, LIAR, |replica, out| {
        replica.insert_visible(0, out)
    });
    fabric.drain_selected(&mut fleet, &mut without_the_targets_adoption);
    assert!(
        fleet[&1].attested(LIAR).is_none(),
        "the round is still open when that dot arrives, which is the point"
    );
    assert!(
        !fleet[&1].effective_order().contains(&native),
        "and it is held out anyway: the successor fence does not wait for the \
         seal"
    );

    let _ = act(&mut fabric, &mut fleet, 2, |replica, out| {
        replica.open_departure(LIAR, out)
    })
    .expect("a gap-free holding");
    fabric.drain_selected(&mut fleet, &mut without_the_targets_adoption);
    for &station in &SURVIVING {
        let replica = &fleet[&station];
        assert!(
            !replica.effective_order().contains(&native),
            "station {station} carried the departing member's successor-plane \
             dot {native:?} across the boundary"
        );
        assert!(
            replica.halted().is_none(),
            "station {station} wedged at the consignment door, which is exactly \
             what the fence exists to prevent"
        );
        assert_eq!(
            replica.attested(LIAR),
            Some(1),
            "and the departure crossed the boundary at the base's compacted \
             prefix for the evictee"
        );
        assert_eq!(
            replica.generation(),
            2,
            "the attestation supplied the report the target never sent"
        );
    }
}

/// Successor-plane traffic folded **before** anyone moves to evict blocks the
/// eviction rather than wedging the boundary (S339).
///
/// The fence stops what arrives after a round opens. This is the ordering it
/// cannot reach: the member adopts, mints in the plane the epoch founds, and
/// a survivor folds that dot while nobody is evicting anything. Refounding to
/// bottom underneath a held dot is exactly the `BeyondSeal` wedge, so the
/// round refuses to start (`SuccessorHeld`).
///
/// The refusal is recoverable and the exhibit shows the recovery: seal the
/// window, and that plane becomes the next generation's old plane, where a
/// round proposes a prefix that covers the dot.
#[test]
fn successor_traffic_folded_before_the_eviction_blocks_it_rather_than_wedging() {
    let mut fleet = fleet_of(&ROSTER, NonZeroUsize::new(2).expect("positive"));
    for replica in fleet.values_mut() {
        replica.tolerate_refusals();
    }
    let mut fabric = Fabric::new(0xB17E_000C, &ROSTER, 0);
    for (offset, &station) in ROSTER.iter().enumerate() {
        let _ = act(&mut fabric, &mut fleet, station, |replica, out| {
            replica.insert_visible(offset, out)
        });
    }
    fabric.drain(&mut fleet);
    let _ = act(&mut fabric, &mut fleet, 1, |replica, out| {
        replica.try_declare(out)
    })
    .expect("station 1 declares over the settled watermark");
    let mut without_the_targets_adoption =
        |_: u32, note: &Note| !matches!(note, Note::Adoption { station: LIAR, .. });
    fabric.drain_selected(&mut fleet, &mut without_the_targets_adoption);

    // Nobody is evicting anything yet, so the survivors fold it.
    let native = act(&mut fabric, &mut fleet, LIAR, |replica, out| {
        replica.insert_visible(0, out)
    });
    fabric.drain_selected(&mut fleet, &mut without_the_targets_adoption);
    assert!(
        fleet[&1].effective_order().contains(&native),
        "the survivor folded it, lawfully: there was no departure to fence it"
    );

    // Only now does the operator move, and the round refuses to start.
    let refusal = act(&mut fabric, &mut fleet, 1, |replica, out| {
        replica.open_departure(LIAR, out)
    })
    .expect_err("a survivor already holding successor traffic cannot propose");
    assert!(
        matches!(
            refusal,
            crate::metis::DepartureRefusal::SuccessorHeld { station: LIAR, .. }
        ),
        "and it says why: {refusal:?}"
    );
    assert!(
        fleet[&1].departure_fence(LIAR).is_none(),
        "nothing was fenced, so nothing has to be unfenced"
    );

    // The cure: seal the window, and evict in the plane that dot now lives
    // in. The withheld adoption report is what the seal was waiting on.
    fabric.drain(&mut fleet);
    for &station in &SURVIVING {
        assert_eq!(fleet[&station].generation(), 2);
        let prefix = act(&mut fabric, &mut fleet, station, |replica, out| {
            replica.open_departure(LIAR, out)
        })
        .expect("the plane that held the dot is now the old one");
        assert!(
            prefix >= 1,
            "station {station} proposes a prefix that covers what it holds"
        );
    }
    fabric.drain(&mut fleet);
    for &station in &SURVIVING {
        let replica = &fleet[&station];
        assert!(
            replica.attested(LIAR).is_some(),
            "station {station} completed the eviction one boundary later"
        );
        assert!(
            replica.halted().is_none(),
            "and no replica was wedged getting there"
        );
    }
}

/// A forked confirmation cut **cannot hide the declaration it names**.
///
/// The narrowest useful lie on the confirmation channel would be to drop a
/// concurrent declaration's dot from the cut, so a victim's confirmation
/// round completes over a *partial* candidate set and latches a winner its
/// peers do not choose. That would split the seal record in its
/// `declaration` field rather than merely its join, which is strictly worse
/// than anything the adoption channel can do.
///
/// It is refused, by a check that shipped long before any Byzantine work
/// here: [`Epochs::confirm`] re-proves the report against the declaration it
/// addresses, so a cut below that declaration's own dot is
/// [`EpochRefusal::ConfirmationDoesNotCover`]. A declaration's minter must
/// therefore always confirm with a cut covering its own dot, and since the
/// minter is the one member guaranteed to hold its own declaration, *every*
/// declaration dot enters the confirmation round's join at every member, at
/// every fault budget. The lie has nowhere to stand.
///
/// This is the Byzantine reading of the latch-completeness finding
/// (`fleet::sweep`): what was discovered as a negative result about the
/// decision space is the reason the winner cannot fork.
#[test]
fn a_forked_confirmation_cannot_hide_its_own_declaration() {
    let mut stability = Stability::new(ROSTER);
    for &station in &ROSTER {
        stability.report_cut(station, &base()).expect("on roster");
    }
    let mut epochs = Epochs::new(ROSTER, NonZeroUsize::new(2).expect("positive"));
    let declaration = epochs
        .declare(
            declaration_dot(),
            Kairos::new(2, 0, 1, 0u16),
            &stability,
            &Cut::bottom(),
        )
        .expect("the settled watermark licenses the declaration");
    let epoch = declaration.address();

    // The forked cut: everything the honest confirmation carried, minus the
    // declaration's own dot.
    let understated = Cut::from_witnessed(vector(&[(1, DECLARATION.1 - 1), (2, 1), (3, 1)]));
    assert_eq!(
        epochs.confirm(epoch, &Vouched::trust(LIAR, understated)),
        Err(EpochRefusal::ConfirmationDoesNotCover {
            epoch,
            station: LIAR,
            covered: DECLARATION.1 - 1,
        }),
        "a confirmation below the declaration it names must be refused, or the \
         candidate set stops being complete at the latch"
    );
    assert_eq!(
        epochs.confirm(epoch, &Vouched::trust(LIAR, delivered())),
        Ok(()),
        "the honest cut confirms"
    );
}

/// Forking a confirmation cut at an *unchecked* coordinate either **freezes
/// the whole fleet** or **changes nothing**. Neither direction reaches the
/// record.
///
/// The coordinates [`Epochs::confirm`] does not re-prove are the ones no
/// declaration names, and the grid below sweeps both directions at every one
/// of them under two concurrent declarations, which is the configuration a
/// winner split would need.
///
/// * *Overstating* inflates the confirmation round's join *at the victim*,
///   and the latch needs the roster-wide meet to cover that join. The meet
///   is built from honest stability reports the liar cannot raise, so the
///   victim never latches, so it never adopts, so no member's adoption round
///   completes and nothing seals anywhere: fail-closed and fleet-wide, by
///   the same unanimity that stops the forked report.
/// * *Understating* only lowers a bar the watermark had already cleared.
///   Candidacy comes from *delivery*, never from the join, so a smaller join
///   cannot remove a candidate; the fleet seals the same winner over the
///   same join it would have sealed honestly.
///
/// The negative half is the load-bearing one: it says the harmless direction
/// really is harmless rather than merely unexercised.
#[test]
fn a_forked_confirmation_either_freezes_the_fleet_or_changes_nothing() {
    fn is_confirmation(note: &Note) -> bool {
        matches!(note, Note::Confirm { .. })
    }

    for (at, delta, freezes) in [
        (2u32, -1i64, false),
        (2, 1, true),
        (1, 1, true),
        (3, 1, true),
    ] {
        let mut fleet = fleet_of(&ROSTER, NonZeroUsize::new(2).expect("positive"));
        for replica in fleet.values_mut() {
            replica.tolerate_refusals();
        }
        let mut fabric = Fabric::new(0x5EED_0C17, &ROSTER, 0);
        fabric.corrupt(Equivocator::within(1).speaking(
            LIAR,
            Box::new(ForkedCut {
                to: 2,
                kind: is_confirmation,
                at,
                delta,
            }),
        ));

        for (offset, &station) in ROSTER.iter().enumerate() {
            let _ = act(&mut fabric, &mut fleet, station, |replica, out| {
                replica.insert_visible(offset, out)
            });
        }
        fabric.drain(&mut fleet);
        // Two concurrent declarations: a real winner contest to split.
        for &station in &[1u32, LIAR] {
            let _ = act(&mut fabric, &mut fleet, station, |replica, out| {
                replica.try_declare(out)
            })
            .expect("the settled watermark licenses the declaration");
        }
        fabric.drain(&mut fleet);

        for &station in &ROSTER {
            let replica = &fleet[&station];
            assert!(
                replica.halted().is_none(),
                "at {at} delta {delta}: station {station} halted; \
                 a confirmation fork must freeze or pass, never diverge"
            );
            if freezes {
                assert!(
                    station != 2 || replica.epochs().fixed().is_none(),
                    "at {at} delta {delta}: the deceived member latched over an inflated join"
                );
                assert_eq!(
                    replica.generation(),
                    1,
                    "at {at} delta {delta}: station {station} latched over an inflated join"
                );
                assert!(
                    replica.epochs().sealed().next().is_none(),
                    "at {at} delta {delta}: station {station} retired a seal \
                     the frozen round cannot license"
                );
            } else {
                assert_eq!(
                    replica.generation(),
                    2,
                    "at {at} delta {delta}: station {station} did not cross"
                );
            }
        }
        if !freezes {
            let witness = fleet[&1]
                .epochs()
                .sealed()
                .next()
                .expect("the seal")
                .clone();
            for &station in &ROSTER {
                let sealed = fleet[&station]
                    .epochs()
                    .sealed()
                    .next()
                    .expect("every member sealed");
                assert_eq!(
                    sealed, &witness,
                    "at {at} delta {delta}: station {station} sealed a record its peers do not hold"
                );
            }
        }
    }
}

/// A liar cannot profit from a lie **it must itself believe**: a forked
/// stability report reaches the victim's latch and dies at the adoption
/// round.
///
/// This is the sharpest form of the stability-channel attack, and the one
/// the accountability intake (S204--S206) named as the safety-critical
/// surface to harden first. Station 3 is genuinely *behind* (it never
/// receives station 2's insert) and tells **both** peers that it is not, so
/// their watermarks rise above the fleet's true meet. The lie works exactly
/// as far as it can: stations 1 and 2 latch a winner on stability that does
/// not exist.
///
/// And then it stops. Sealing needs an adoption report from *every* roster
/// member, and a member adopts only under its **own** watermark, which is a
/// meet over its own honest slot. So to make anyone seal prematurely the
/// liar must first convince itself, which it cannot do by speaking: only by
/// actually delivering. Nothing seals, nothing halts, and no member folds a
/// base its peers do not hold.
///
/// The corollary is the crate's Byzantine posture in one line. The
/// roster-wide unanimity that costs *all* the liveness under a withholding
/// member ([`withholding_from_one_peer_freezes_the_whole_fleet`]) is the
/// same mechanism that buys this safety. They are one trade, not two facts,
/// and a quorum-sized adoption round would sell the second to buy the first.
#[test]
fn a_forked_report_cannot_force_a_seal_the_liar_has_not_licensed() {
    fn is_report(note: &Note) -> bool {
        matches!(note, Note::Report { .. })
    }
    /// The dot station 3 never receives and claims to hold anyway.
    const UNHELD_STATION: u32 = 2;
    let unheld = d(UNHELD_STATION, 1);

    let mut fleet = fleet_of(&ROSTER, NonZeroUsize::new(2).expect("positive"));
    for replica in fleet.values_mut() {
        replica.tolerate_refusals();
    }
    let mut fabric = Fabric::new(0x5EED_0C17, &ROSTER, 0);
    fabric.corrupt(Equivocator::within(1).speaking(
        LIAR,
        Box::new(Then(
            ForkedCut {
                to: 1,
                kind: is_report,
                at: UNHELD_STATION,
                delta: 1,
            },
            ForkedCut {
                to: 2,
                kind: is_report,
                at: UNHELD_STATION,
                delta: 1,
            },
        )),
    ));

    let held = |dst: u32, note: &Note| matches!(note, Note::Old { dots, .. } if dst == LIAR && dots.contains(&unheld));
    for (offset, &station) in ROSTER.iter().enumerate() {
        let _ = act(&mut fabric, &mut fleet, station, |replica, out| {
            replica.insert_visible(offset, out)
        });
    }
    fabric.drain_selected(&mut fleet, |dst, note| !held(dst, note));
    let _ = act(&mut fabric, &mut fleet, 1, |replica, out| {
        replica.try_declare(out)
    })
    .expect("station 1 declares over the watermark the lie inflated");
    fabric.drain_selected(&mut fleet, |dst, note| !held(dst, note));

    assert_eq!(
        fabric.in_flight(),
        1,
        "exactly the withheld delta waits: the liar is genuinely behind"
    );
    // Non-vacuity: the lie must really reach the latch, or the law is empty.
    for &station in &[1u32, 2] {
        assert!(
            fleet[&station].epochs().fixed().is_some(),
            "station {station} did not latch: the forked report reached nothing"
        );
    }
    assert!(
        fleet[&LIAR].epochs().fixed().is_none(),
        "the liar latched under its own honest meet"
    );
    // The law: no seal anywhere, and no halt anywhere.
    for &station in &ROSTER {
        let replica = &fleet[&station];
        assert_eq!(
            replica.generation(),
            1,
            "station {station} sealed on stability no member licensed"
        );
        assert!(
            replica.epochs().sealed().next().is_none(),
            "station {station} retired a seal the incomplete adoption round cannot license"
        );
        assert!(
            replica.halted().is_none(),
            "station {station} halted: the block must be a stall, never a refusal"
        );
    }
}

/// Withholding the delta that would expose an understated adoption counter
/// **freezes the fleet instead of hiding the lie**.
///
/// This closes the sufficiency question the certification duty leaves open.
/// The consignment door catches the understated join because it checks the
/// seal record against an *independent* channel: the sealed join is built
/// from peer **reports**, and the coverage check reads the deltas this
/// replica actually **delivered**. So the obvious escalation is to silence
/// the second channel too --- understate the counter to one peer *and*
/// withhold from that same peer the delta the counter would have named.
///
/// It degenerates. The delta channel is what the watermark is made of: a
/// victim that never receives the delta reports a smaller cut, the
/// roster-wide meet falls with it, and the honest members' confirmation cuts
/// still name the withheld dot, so no member's latch clears. Nothing seals
/// anywhere.
///
/// The two-line result: an understated adoption report either reaches a
/// victim that holds the covering delta, and is **refused at the consignment
/// door**, or reaches one that does not, and **nothing seals at all**. The
/// cross-check cannot be evaded by silencing its second source, because that
/// source is also the protocol's clock.
#[test]
fn withholding_the_covering_delta_freezes_instead_of_hiding_the_lie() {
    fn is_old_plane(note: &Note) -> bool {
        matches!(note, Note::Old { .. })
    }

    let mut fleet = fleet_of(&ROSTER, NonZeroUsize::new(2).expect("positive"));
    for replica in fleet.values_mut() {
        replica.tolerate_refusals();
    }
    let mut fabric = Fabric::new(0xB17E_0002, &ROSTER, 0);
    fabric.corrupt(Equivocator::within(1).speaking(
        LIAR,
        Box::new(Then(
            Withholding {
                from: 2,
                kind: is_old_plane,
            },
            ForkedAdoption { to: 2, delta: -1 },
        )),
    ));

    for (offset, &station) in ROSTER.iter().enumerate() {
        let _ = act(&mut fabric, &mut fleet, station, |replica, out| {
            replica.insert_visible(offset, out)
        });
    }
    fabric.drain(&mut fleet);
    let _ = act(&mut fabric, &mut fleet, 1, |replica, out| {
        replica.try_declare(out)
    })
    .expect("station 1 declares over the settled watermark");
    fabric.drain(&mut fleet);

    for &station in &ROSTER {
        let replica = &fleet[&station];
        assert_eq!(
            replica.generation(),
            1,
            "station {station} sealed under a watermark the withheld delta holds down"
        );
        assert!(
            replica.epochs().sealed().next().is_none(),
            "station {station} retired a seal the frozen round cannot license"
        );
        assert!(
            replica.halted().is_none(),
            "station {station} halted: the composed lie must stall, never diverge"
        );
    }
}