minerva 0.2.0

//! The epoch lifecycle: declaration, confirmation, adoption, seal
//! (PRD 0024 stage two).
//!
//! [`Epochs`] records one replica's re-foundation rounds and owns exactly one
//! question: when a re-foundation is *licensed*, and when it is *settled*.
//! The fold itself is [`Rhapsody::refound`](crate::metis::Rhapsody::refound);
//! old-addressed judgment is [`EpochShadow`]; cadence, transport, and
//! membership stay caller-side.
//!
//! # The rounds
//!
//! [`declare`](Epochs::declare) mints a [`Declaration`] only over a witnessed
//! [`Stability::watermark_cut`] that has *risen* to the caller-declared epoch
//! basis, so the sealed stratum is complete and identical everywhere before a
//! window opens. Witnessed is not risen: a fresh tracker witnesses the bottom
//! cut, and declaring on that alone opens a window no replica can adopt
//! ([`EpochRefusal::Unwitnessed`], [`EpochRefusal::Unrisen`]).
//!
//! A delivered declaration is *provisional*. Members keep minting
//! old-addressed and each returns a confirmation report: its delivered cut at
//! delivery. Any concurrent declaration was minted before its own minter's
//! confirmation cut, so once the watermark covers the join of those cuts the
//! candidate set is complete everywhere and every member latches the same
//! winner by the public `(rank, dot)` rule. Losers stay visible as superseded
//! candidates rather than being dropped.
//!
//! Members then [`adopt`](Epochs::adopt) and report their own-station cut.
//! The seal fires when the watermark covers the join of those reports: no
//! old-addressed delta can lawfully remain in flight, the window closes, and
//! the sealed join retires into the lineage as the duplicate recognizer.
//!
//! # The lineage horizon
//!
//! The lineage retains the last `k` sealed recognizers, `k` caller-declared.
//! Within it, [`recognize`](Epochs::recognize) answers an old-addressed dot
//! with the duplicate verdict or [`EpochRefusal::AddressMiss`]; below it the
//! machine cannot tell duplicate from violation and refuses to guess
//! ([`EpochRefusal::BeyondHorizon`]). A transport replaying frames older than
//! `k` sealed epochs is outside the contract the caller declared.
//!
//! # What the machine trusts
//!
//! Crash-only honest peers. Reports are cut claims it cannot verify
//! (PRD 0011 R8). The roster belongs to the current generation. It is
//! initialized at construction and widened only by an admission-bearing seal (PRD 0028
//! R1). It must be the family that the caller's [`Stability`] tracker
//! declares, since a mismatched family misreads the watermark: after a
//! widening seal the caller rebuilds its tracker from the record, exactly
//! as the recipe rebuilds at every seal. A silent member freezes both
//! rounds, which stalls without corrupting.
//!
//! [`Vouched`] binds a peer's claim to the station that made it. That closes
//! *misattribution* and not equivocation: a Byzantine station can validly
//! vouch different claims to different peers, and detecting the contradiction
//! needs cross-member evidence this crate structurally cannot hold.
//!
//! # Two duties the types do not carry
//!
//! *A seal record is certifiable only once its consignment is accepted.*
//! [`try_seal`](Epochs::try_seal) returning a [`SealedEpoch`] means this
//! replica's rounds completed, not that its peers hold the same record. Under
//! equivocated reports those differ, and
//! [`EpochShadow::consign`](crate::metis::EpochShadow::consign) is where the
//! difference is caught and the sole mint of [`Consigned`], which a caller
//! binds its certificate over. Acceptance is a *local* fact: no quorum, no
//! signature, no evidence a peer accepted the same record.
//!
//! *Each entry door re-proves a different subset of the ledger invariants*,
//! because each arrives with a different warrant. The audit table is
//! `docs/metis-epoch-entry-doors.adoc`; the Byzantine failure surface is
//! mapped in `tests/fleet/byzantine.rs`.

extern crate alloc;

mod arrival;
mod bootstrap;
mod consigned;
mod gate;
mod preparation;
mod shadow;
mod snapshot;
mod vouched;
mod wire;

pub use arrival::{Admission, ArrivalRefusal, ArrivalRound, Endorsement};
pub use bootstrap::EpochBootstrapError;
pub use consigned::Consigned;
pub use gate::{EpochAddress, EpochAdoptionMismatch, EpochGate, InvalidEpochAddress};
pub use preparation::{EpochPreparation, EpochPreparationFold, EpochPreparationMiss};
#[cfg(feature = "instrumentation")]
pub use shadow::EpochShadowProfile;
pub use shadow::{
    EpochConsignment, EpochConsignmentError, EpochProjection, EpochShadow, EpochStratum,
    EpochStratumError, EpochTransition, SweptIdentity,
};
pub use snapshot::{
    EpochLedgerDecodeBudget, EpochLedgerDecodeError, EpochLedgerRehydrateError, EpochLedgerSnapshot,
};
pub use vouched::Vouched;
pub use wire::{
    AdoptionReportDecodeError, AdoptionReportRecord, ConfirmationDecodeError, ConfirmationRecord,
    DeclarationDecodeError, EndorsementDecodeBudget, EndorsementDecodeError,
    LineageProofDecodeBudget, LineageProofDecodeError, LineageProofEntry, LineageProofRecord,
    SealDecodeBudget, SealDecodeError, SealRecord, WireCutError,
};

use alloc::collections::{BTreeMap, BTreeSet, VecDeque};
use alloc::vec::Vec;
use core::num::{NonZeroU64, NonZeroUsize};

use crate::kairos::Kairos;

use super::dot::{Dot, RawDot};
use super::stability::{Stability, UnknownStation};
use super::{Cut, VersionVector};

/// The replicated re-foundation record: "this epoch re-founds at cut `C`".
///
/// Its address pairs the lineage generation with its own minting dot, so two
/// concurrent declarations and a tuple reused after horizon eviction can
/// never be confused (requirement R5); its rank is the declaration's
/// [`Kairos`] stamp, the first component of the fixed winner rule. The carried [`Cut`] is the sealed stratum's
/// boundary, witnessed by construction: the only door that mints one is
/// [`Epochs::declare`], which reads the license off the caller's
/// [`Stability`] tracker and refuses a watermark that has not risen to the
/// caller-declared epoch basis.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Declaration {
    /// The durable address: the lineage generation and the minting dot,
    /// each carrying its own law on the type (ruling R-91).
    address: EpochAddress,
    rank: Kairos,
    cut: Cut,
    /// The membership boundary this declaration carries, or `None`: the
    /// admission's one carrier (PRD 0028 R2, ruling R-89). Minted only
    /// from a complete [`ArrivalRound`], validated at
    /// [`Epochs::deliver`] on agreed inputs, and copied into the
    /// [`SealedEpoch`] the window seals. A version-1 wire frame never
    /// carries one; the version-2 sibling spelling does.
    admission: Option<Admission>,
    /// The record commitment the tagging round agreed (PRD 0028 R7),
    /// present exactly when `admission` is: the winner carries the
    /// concord's whole value, so an adopter's assent compares the
    /// commitment too and a forked predecessor under one address cannot
    /// feed a recorded admission.
    admission_commitment: Option<[u8; 32]>,
}

impl Declaration {
    /// The declaration's causal identity within its parent epoch.
    #[must_use]
    pub const fn dot(&self) -> Dot {
        self.address.declaration()
    }

    /// The durable epoch address: lineage generation plus declaration dot.
    #[must_use]
    pub const fn address(&self) -> EpochAddress {
        self.address
    }

    /// The declaration's stamp: the winner rule's leading component.
    #[must_use]
    pub const fn rank(&self) -> Kairos {
        self.rank
    }

    /// The witnessed cut the epoch re-founds at: the sealed stratum's
    /// boundary, complete and identical at every member by the license.
    #[must_use]
    pub const fn cut(&self) -> &Cut {
        &self.cut
    }

    /// The membership boundary this declaration carries, or `None`.
    #[must_use]
    pub const fn admission(&self) -> Option<&Admission> {
        self.admission.as_ref()
    }

    /// The tagging round's record commitment, present exactly when an
    /// admission is (PRD 0028 R7).
    #[must_use]
    pub const fn admission_commitment(&self) -> Option<&[u8; 32]> {
        self.admission_commitment.as_ref()
    }

    /// The fixed public winner key, `(rank, dot)`: the same shape the move
    /// replay orders by, so concurrent declarations resolve without
    /// negotiation.
    const fn key(&self) -> (Kairos, Dot) {
        (self.rank, self.address.declaration())
    }

    /// Whether this declaration's witnessed past contains `dot`.
    fn covers(&self, dot: Dot) -> bool {
        self.cut.as_vector().get(dot.station()) >= dot.counter()
    }
}

/// Opaque proof that this replica completed confirmation and adopted the
/// fixed winner.
///
/// Only [`Epochs::adopt`] constructs this witness. Holding a declaration is
/// insufficient: it may still be provisional or may lose the fixed race.
#[derive(Debug, PartialEq, Eq)]
pub struct Adopted {
    declaration: Declaration,
}

impl Adopted {
    /// The fixed winning declaration.
    #[must_use]
    pub const fn declaration(&self) -> &Declaration {
        &self.declaration
    }

    /// The winning declaration's causal identity within its parent epoch.
    #[must_use]
    pub const fn dot(&self) -> Dot {
        self.declaration.dot()
    }

    /// The durable winning epoch address.
    #[must_use]
    pub const fn address(&self) -> EpochAddress {
        self.declaration.address()
    }

    /// The winning declaration's fixed-rule rank.
    #[must_use]
    pub const fn rank(&self) -> Kairos {
        self.declaration.rank()
    }

    /// The witnessed stratum boundary.
    #[must_use]
    pub const fn cut(&self) -> &Cut {
        self.declaration.cut()
    }
}

/// A sealed epoch's retained record: the winning address, every candidate
/// address the window closed over, and the duplicate recognizer
/// (requirement R6).
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct SealedEpoch {
    /// The durable winning epoch address.
    declaration: EpochAddress,
    /// Every candidate address the window closed over, the winner
    /// included. Losing declarations and their reports remain lawful
    /// duplicates.
    candidates: BTreeSet<EpochAddress>,
    /// Every declaration dot the window accepted, displaced provisional
    /// records included: the generation's protocol-plane event ledger,
    /// candidate-count-bound.
    protocol: BTreeSet<Dot>,
    /// The sealed join: per member, its own-station cut at adoption. An
    /// old-addressed dot at or below it is an already-delivered duplicate;
    /// one above it is a violation.
    sealed_join: VersionVector,
    /// The membership boundary this record carries, or `None`: exactly
    /// what the winning declaration carried (PRD 0028 R1/R2). The joiners
    /// are roster members for every generation after this record and for
    /// none before it, at every replica, without any replica consulting
    /// another.
    admission: Option<Admission>,
}

impl SealedEpoch {
    /// The durable winning epoch address.
    #[must_use]
    pub const fn declaration(&self) -> EpochAddress {
        self.declaration
    }

    /// The sealed join of the adoption reports, the duplicate recognizer.
    #[must_use]
    pub const fn sealed_join(&self) -> &VersionVector {
        &self.sealed_join
    }

    /// Whether `declaration` was a candidate of this seal's window, the
    /// winner included.
    #[must_use]
    pub fn contains(&self, declaration: EpochAddress) -> bool {
        self.candidates.contains(&declaration)
    }

    /// Every candidate address this seal closed over, the winner included,
    /// ascending: the seal's first
    /// retained set, mirrored from the wire twin's `candidates` read so
    /// a certificate layer binds over both retained sets from the live
    /// entry without cloning a record (the S287 under-coverage rule
    /// names them part of the covered tuple).
    pub fn candidates(&self) -> impl Iterator<Item = EpochAddress> + '_ {
        self.candidates.iter().copied()
    }

    /// The membership boundary this record carries, or `None`.
    #[must_use]
    pub const fn admission(&self) -> Option<&Admission> {
        self.admission.as_ref()
    }

    /// The declaration dots this seal accounts for: the winner's, every
    /// retired candidate's, and every displaced provisional record's, the
    /// generation's protocol-plane events outside the two document pairs. The
    /// consignment door derives its sealed-join coverage from this record.
    /// Nothing caller-claimed can therefore mask a missing delivery: the
    /// witnessed seam stays witnessed end to end. A declaration that was
    /// delivered and then displaced by its own predecessor still counts. Its
    /// dot raised floors everywhere, and losing it would wedge the consignment
    /// on `Incomplete` with no cure.
    pub fn declaration_dots(&self) -> impl Iterator<Item = Dot> + '_ {
        self.protocol.iter().copied()
    }
}

/// The epoch lifecycle's typed refusals: PRD 0024's refusal vocabulary,
/// plus the adoption door's guard. Every variant names its evidence; none
/// is ever a guess or a positional re-anchor.
///
/// `#[non_exhaustive]` so a future refusal is additive.
#[non_exhaustive]
#[derive(Clone, Copy, Debug, PartialEq, Eq, thiserror::Error)]
pub enum EpochRefusal {
    /// `EpochUnwitnessed` (R1): the declaration license is
    /// [`Stability::watermark_cut`] and nothing else, and the tracker could
    /// not produce a witnessed cut (a bare report withdrew the witness, or
    /// nothing is stable yet).
    #[error("epoch unwitnessed: no watermark cut licenses a declaration")]
    Unwitnessed,
    /// `EpochUnrisen` (R1's risen half, ruling R-40): the witnessed
    /// watermark has not risen to the caller-declared epoch basis, so the
    /// declaration would re-found at a stratum that does not cover the
    /// generation's base. Witnessed is necessary, not sufficient: a fresh
    /// tracker witnesses the bottom cut, and a legal frontier is not a
    /// complete one. The evidence is the first basis entry the cut leaves
    /// uncovered.
    #[error("epoch unrisen: witnessed {witnessed} for station {station} is below basis {required}")]
    Unrisen {
        /// The first basis station the witnessed cut leaves uncovered.
        station: u32,
        /// The declared basis floor for `station`.
        required: u64,
        /// The witnessed watermark's count for `station`.
        witnessed: u64,
    },
    /// `EpochWindowOpen` (R5): one window is open at a time. Minting a new
    /// declaration causally after a delivered one and before the seal, or
    /// delivering a declaration the completed confirmation round proved
    /// cannot lawfully exist, refuses with the open window's address.
    #[error("epoch window open: declaration {open:?} is not yet sealed")]
    WindowOpen {
        /// The open window's provisional (or fixed) declaration dot.
        open: EpochAddress,
    },
    /// The adoption guard (R5's "no replica mints natively before the
    /// confirmation watermark fixes the winner"): the confirmation round
    /// has not completed under the delivery watermark, so there is no
    /// irrevocable winner to adopt.
    #[error("epoch unconfirmed: the confirmation watermark has not fixed a winner")]
    Unconfirmed,
    /// A declaration dot is already covered by its own witnessed cut.
    /// Declaration identity must be fresh in the old epoch's causal space.
    #[error("epoch declaration {dot} is not fresh above ceiling {ceiling}")]
    StaleDeclaration {
        /// The already-covered declaration identity.
        dot: Dot,
        /// The witnessed high water for `dot`'s station.
        ceiling: u64,
    },
    /// A declaration was minted by a station outside the current roster.
    #[error("epoch declaration station {station} is outside the roster")]
    UnknownDeclarationStation {
        /// The unrecognized declaration minter.
        station: u32,
    },
    /// A confirmation report does not prove receipt of the declaration it
    /// names.
    #[error("epoch confirmation from {station} covers {covered}, below declaration {epoch:?}")]
    ConfirmationDoesNotCover {
        /// The declaration address being confirmed.
        epoch: EpochAddress,
        /// The reporting roster member.
        station: u32,
        /// The report's high water for the declaration minter.
        covered: u64,
    },
    /// `EpochAddressMiss` (R6): an old dot not covered by its epoch's
    /// sealed join, or a report or delta addressed to a declaration this
    /// replica has never seen. A roster violation or a replay from outside
    /// the lawful window: typed, evidence-carrying, never absorbed.
    #[error("epoch address miss: dot {dot} is not covered by epoch {epoch:?}")]
    AddressMiss {
        /// The epoch address the traffic named.
        epoch: EpochAddress,
        /// The offending coordinate. `(station, 0)` when only the station
        /// is in evidence: a roster refusal or a report door names its
        /// speaker, not an event, and the raw spelling carries that
        /// honestly ([`RawDot`], ruling R-91).
        dot: RawDot,
    },
    /// `EpochBeyondHorizon` (R6): the named epoch is below the caller's
    /// declared replay horizon (or was never sealed at all, which the
    /// machine can no longer distinguish), so duplicate cannot be told from
    /// violation and the machine refuses to guess.
    #[error("epoch beyond horizon: {epoch:?} is below the retained lineage")]
    BeyondHorizon {
        /// The epoch address the traffic named.
        epoch: EpochAddress,
    },
    /// A tagged declaration whose boundary's base is not this machine's
    /// newest sealed record (PRD 0028 R2). Prevents, on agreed inputs: by
    /// the time a declaration for this generation folds, every honest
    /// replica holds the same predecessor record, so this refusal is
    /// identical everywhere. Unreachable from an honest declarer, whose
    /// machine tags only a round it located against the same lineage.
    #[error("admission base discord: offered {offered:?}, sealed {sealed:?}")]
    AdmissionBaseDiscord {
        /// The tagged boundary's base.
        offered: EpochAddress,
        /// This machine's newest sealed address, `None` before a first
        /// seal.
        sealed: Option<EpochAddress>,
    },
    /// A tagged declaration admitting a station already on the roster
    /// (PRD 0028 R2). Prevents, on agreed inputs; with R-84's permanent
    /// roster seats this is also the door that refuses a departed station
    /// re-entering under its own identity.
    #[error("admission already present: station {station} is on the roster")]
    AdmissionAlreadyPresent {
        /// The joining station already present.
        station: u32,
    },
    /// A tagged winner this machine has not endorsed (PRD 0028 R2, ruling
    /// R-89): the adoption is the assent, and it reads this machine's own
    /// arrival round, never the network. Prevents --- the winner's
    /// adoption round cannot complete anywhere without every speaking
    /// member's assent, so a fabricated boundary wedges before any seal
    /// exists instead of riding into an agreed record. Recoverable by
    /// [`Epochs::open_arrival`] over the same boundary.
    #[error(
        "unassented admission: winner {winner:?} carries a boundary this machine did not endorse"
    )]
    UnassentedAdmission {
        /// The tagged winner awaiting this machine's assent.
        winner: EpochAddress,
    },
}

/// One watermark round's per-member report slots: `None` until the member's
/// first report, folded by join thereafter (duplicated, reordered, or stale
/// reports absorb, the [`Stability`] posture). Completion means every slot
/// reported; the round's join is only read then.
#[derive(Clone, Debug, PartialEq, Eq)]
struct Round {
    reports: BTreeMap<u32, Option<VersionVector>>,
}

impl Round {
    fn over(roster: &BTreeSet<u32>) -> Self {
        Self {
            reports: roster.iter().map(|&station| (station, None)).collect(),
        }
    }

    fn fold(&mut self, station: u32, report: &VersionVector) -> Result<(), UnknownStation> {
        let slot = self
            .reports
            .get_mut(&station)
            .ok_or(UnknownStation { station })?;
        *slot = Some(
            slot.take()
                .map_or_else(|| report.clone(), |held| held.merge(report)),
        );
        Ok(())
    }

    /// The join of every member's report, or `None` while any is silent. A
    /// late or repeated report can only raise the bar the watermark must
    /// clear, never lower it.
    ///
    /// Ranges over the whole roster. A member the caller merely abandoned
    /// ([`Stability::abandon`]) is still asked for its report, and its
    /// silence still freezes the round: narrowing the family here would seal
    /// a record with no coordinate for it while the survivors' shadows still
    /// hold its deltas, so the consignment door would refuse one door after
    /// the lineage advanced --- a recoverable stall traded for a durable
    /// wedge.
    ///
    /// An *attested* departure ([`Stability::abandon_attested`]) supplies
    /// that coordinate instead, from the bound its round agreed, and the
    /// substitution is unconditional: an attested member's own report is
    /// discarded even when it is present. That is what makes the round a
    /// function of the family and the attestation alone, never of the
    /// per-replica race about whether a departing member's last report
    /// arrived before its operator evicted it
    /// (`docs/metis-membership-departure.adoc`).
    fn join_when_complete(&self, stability: &Stability) -> Option<VersionVector> {
        self.reports
            .iter()
            .try_fold(VersionVector::new(), |mut join, (&station, slot)| {
                if let Some(bound) = stability.attested(station) {
                    join.observe(station, bound);
                    return Some(join);
                }
                Some(join.merge(slot.as_ref()?))
            })
    }
}

/// The open window: the candidate set, the two report rounds, and the
/// winner latch.
#[derive(Clone, Debug, PartialEq, Eq)]
struct Window {
    candidates: BTreeMap<EpochAddress, Declaration>,
    /// Every declaration dot this window accepted, displaced ones included:
    /// the generation's protocol-plane event ledger. A displaced dot still
    /// raised floors, so omitting it here makes the consignment's
    /// sealed-join coverage unsatisfiable.
    protocol: BTreeSet<Dot>,
    confirmations: Round,
    /// Adoption reports are candidate-addressed. Only the fixed winner's
    /// round can seal; loser-addressed reports remain quarantined there.
    adoptions: BTreeMap<EpochAddress, Round>,
    /// The latched winner, `Some` from the first local observation of
    /// confirmation completeness under the watermark. Never cleared, never
    /// changed: R5 irrevocability, held locally.
    fixed: Option<EpochAddress>,
    /// Whether this replica has adopted (reported its own adoption).
    adopted: bool,
}

/// One replica's epoch lifecycle record: the declaration door, the two
/// report rounds over the caller's [`Stability`] witness, the seal, and the
/// retained lineage.
///
/// Mechanism, not policy. The machine refuses what the witness cannot license
/// and settles what the rounds prove; *when* to declare, how declarations
/// travel, and who is on the roster are the caller's. The roster is the
/// current generation's. Construction initializes it. Only an
/// admission-bearing seal widens it (PRD 0028 R1, the one door that moves it).
/// It must be the family the caller's tracker declares: after a widening seal,
/// rebuild the tracker from the record just sealed.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Epochs {
    roster: BTreeSet<u32>,
    horizon: NonZeroUsize,
    /// The next window's lineage generation, advanced only at seal. The
    /// lineage starts at one and only rises, so the one-based law rides
    /// the type and no address door re-proves it (ruling R-91).
    generation: NonZeroU64,
    window: Option<Window>,
    /// Sealed epochs, oldest first, at most `horizon` retained.
    lineage: VecDeque<SealedEpoch>,
    /// The open arrival round, or `None` (PRD 0028 R2/R3). Opened before
    /// a boundary's window, read by [`declare`](Self::declare) for the
    /// tagging license and by [`adopt`](Self::adopt) for assent, and
    /// ended by every seal: consumed when the record carries its
    /// boundary, superseded otherwise.
    arrival: Option<ArrivalRound>,
}

impl Epochs {
    /// Creates the record over the founding roster, retaining the last
    /// `horizon` sealed recognizers. Duplicate roster ids collapse. The
    /// roster then widens only at an admission-bearing seal (PRD 0028 R1).
    ///
    /// The bound is mechanism; its value is the caller's transport fact.
    #[must_use]
    pub fn new(roster: impl IntoIterator<Item = u32>, horizon: NonZeroUsize) -> Self {
        Self {
            roster: roster.into_iter().collect(),
            horizon,
            generation: NonZeroU64::MIN,
            window: None,
            lineage: VecDeque::new(),
            arrival: None,
        }
    }

    /// Mints a declaration over the tracker's witnessed watermark and opens
    /// the window with it as the provisional candidate.
    ///
    /// `dot` is minted off the causal context exactly as any write's dot is;
    /// `rank` is the winner rule's leading component. `basis` is the
    /// generation's base frontier --- bottom for a first epoch born empty,
    /// the seal annex's gap-free floor thereafter --- and the watermark must
    /// have risen to it or the declaration is vacuous. The returned
    /// declaration is the caller's to replicate; this replica has trivially
    /// delivered its own.
    ///
    /// # Errors
    ///
    /// [`EpochRefusal::WindowOpen`] while a delivered declaration is
    /// unsealed; [`EpochRefusal::Unwitnessed`] when the tracker has no
    /// witnessed cut; [`EpochRefusal::Unrisen`] when that cut does not cover
    /// `basis`; [`EpochRefusal::UnknownDeclarationStation`] for an
    /// off-roster minter; [`EpochRefusal::StaleDeclaration`] for a dot the
    /// cut already covers.
    pub fn declare(
        &mut self,
        dot: Dot,
        rank: Kairos,
        stability: &Stability,
        basis: &Cut,
    ) -> Result<Declaration, EpochRefusal> {
        if !self.roster.contains(&dot.station()) {
            return Err(EpochRefusal::UnknownDeclarationStation {
                station: dot.station(),
            });
        }
        if let Some(window) = &self.window {
            return Err(EpochRefusal::WindowOpen {
                open: window
                    .fixed
                    .or_else(|| window.candidates.keys().next().copied())
                    .unwrap_or_else(|| EpochAddress::new(self.generation, dot)),
            });
        }
        let cut = stability.watermark_cut().ok_or(EpochRefusal::Unwitnessed)?;
        if let Some(refusal) = Self::unrisen(basis, &cut) {
            return Err(refusal);
        }
        let ceiling = cut.as_vector().get(dot.station());
        let address = EpochAddress::new(self.generation, dot);
        if dot.counter() <= ceiling {
            return Err(EpochRefusal::StaleDeclaration { dot, ceiling });
        }
        // The tagging license (PRD 0028 R2): the boundary rides exactly
        // when this machine holds its round complete and no member of the
        // round's family is attested-departed --- a precondition evaluated
        // on own state, never a duty a drive loop can break by working
        // normally. An open round's base is this machine's newest seal by
        // invariant (every seal ends the round), so no base or roster
        // re-check exists to disagree with `deliver`'s. The attestation
        // half is the freeze, enforced at both doors: a departed
        // identity's word cannot arrive (`fold_endorsement`), and a word
        // it left behind cannot be spent (`declare`) --- the arrival
        // re-agrees over a later base with the surviving family.
        let tag = self
            .arrival
            .as_ref()
            .filter(|round| {
                round.complete()
                    && round
                        .family()
                        .all(|station| stability.attested(station).is_none())
            })
            .map(|round| (round.admission().clone(), *round.commitment()));
        let (admission, admission_commitment) = match tag {
            Some((admission, commitment)) => (Some(admission), Some(commitment)),
            None => (None, None),
        };
        let declaration = Declaration {
            address,
            rank,
            cut,
            admission,
            admission_commitment,
        };
        self.window = Some(Window {
            candidates: BTreeMap::from([(address, declaration.clone())]),
            protocol: BTreeSet::from([dot]),
            confirmations: Round::over(&self.roster),
            adoptions: BTreeMap::from([(address, Round::over(&self.roster))]),
            fixed: None,
            adopted: false,
        });
        Ok(declaration)
    }

    /// Folds a peer's declaration: a new candidate while the round is open,
    /// an absorbed duplicate once it is known or sealed.
    ///
    /// Opens the window if none is open. A redelivered declaration of an
    /// epoch still inside the horizon absorbs as a duplicate; older than
    /// that is outside the contract the caller declared. `basis` is the
    /// same deterministic value an honest declarer holds, so the risen
    /// refusal is identical at every member.
    ///
    /// # Errors
    ///
    /// [`EpochRefusal::WindowOpen`] for an unknown candidate arriving after
    /// this replica saw the confirmation round complete: that round proves
    /// no lawful concurrent declaration remains undelivered, so the arrival
    /// is a violation rather than something to absorb.
    /// [`EpochRefusal::Unrisen`] for a candidate whose cut does not cover
    /// `basis`, which would open a window no replica can adopt.
    /// [`EpochRefusal::UnknownDeclarationStation`] for an off-roster
    /// minter. [`EpochRefusal::StaleDeclaration`] for a dot the delivered
    /// cut already covers. [`EpochRefusal::AddressMiss`] and
    /// [`EpochRefusal::BeyondHorizon`] locating the address against the
    /// retained lineage. [`EpochRefusal::AdmissionBaseDiscord`] and
    /// [`EpochRefusal::AdmissionAlreadyPresent`] validating a carried
    /// membership boundary against the agreed inputs.
    pub fn deliver(
        &mut self,
        declaration: Declaration,
        stability: &Stability,
        basis: &Cut,
    ) -> Result<(), EpochRefusal> {
        let dot = declaration.dot();
        if !self.roster.contains(&dot.station()) {
            return Err(EpochRefusal::UnknownDeclarationStation {
                station: dot.station(),
            });
        }
        let address = declaration.address();
        if self.lineage.iter().any(|sealed| sealed.contains(address)) {
            return Ok(());
        }
        // The named epoch is this address by the guard: a retained seal at
        // the same generation whose join already covers this very dot.
        if self.lineage.iter().any(|sealed| {
            sealed.declaration.generation() == address.generation()
                && sealed.sealed_join.get(dot.station()) >= dot.counter()
        }) {
            return Err(EpochRefusal::AddressMiss {
                epoch: address,
                dot: dot.into(),
            });
        }
        if address.generation() < self.generation.get() {
            return Err(EpochRefusal::BeyondHorizon { epoch: address });
        }
        if address.generation() > self.generation.get() {
            return Err(EpochRefusal::AddressMiss {
                epoch: address,
                dot: dot.into(),
            });
        }
        if let Some(refusal) = Self::unrisen(basis, &declaration.cut) {
            return Err(refusal);
        }
        let ceiling = declaration.cut.as_vector().get(dot.station());
        if dot.counter() <= ceiling {
            return Err(EpochRefusal::StaleDeclaration { dot, ceiling });
        }
        // A carried boundary is validated on agreed inputs only --- the
        // predecessor record's address and the roster, both derived from
        // the lineage --- so every honest replica refuses an invalid
        // tagged candidate identically, at this door, before it can enter
        // any candidate set (PRD 0028 R2). Honest machines cannot mint
        // one: `declare` tags from a round whose base and joiners passed
        // the same checks, so these refusals price a liar, not a race.
        if let Some(admission) = declaration.admission() {
            let sealed = self.lineage.back().map(SealedEpoch::declaration);
            if sealed != Some(admission.base()) {
                return Err(EpochRefusal::AdmissionBaseDiscord {
                    offered: admission.base(),
                    sealed,
                });
            }
            if let Some(station) = admission
                .joiners()
                .find(|station| self.roster.contains(station))
            {
                return Err(EpochRefusal::AdmissionAlreadyPresent { station });
            }
        }
        let declaration_address = declaration.address();
        match &mut self.window {
            None => {
                self.window = Some(Window {
                    candidates: BTreeMap::from([(declaration_address, declaration)]),
                    protocol: BTreeSet::from([dot]),
                    confirmations: Round::over(&self.roster),
                    adoptions: BTreeMap::from([(declaration_address, Round::over(&self.roster))]),
                    fixed: None,
                    adopted: false,
                });
                Ok(())
            }
            Some(window) => {
                if window.candidates.contains_key(&declaration.address()) {
                    return Ok(());
                }
                // Receipt recording is verdict-independent: a causally
                // later declaration is refused here in one arrival order
                // and displaced in the other, but its dot reached this
                // replica either way, and the seal's ledger must read the
                // same whichever order the fabric chose.
                let _ = window.protocol.insert(dot);
                if let Some(&open) = window
                    .candidates
                    .keys()
                    .find(|&&candidate| declaration.covers(candidate.declaration()))
                {
                    return Err(EpochRefusal::WindowOpen { open });
                }
                Self::latch_winner(window, stability);
                if let Some(open) = window.fixed {
                    return Err(EpochRefusal::WindowOpen { open });
                }
                let later: Vec<EpochAddress> = window
                    .candidates
                    .iter()
                    .filter_map(|(&address, held)| held.covers(dot).then_some(address))
                    .collect();
                for displaced in later {
                    let _ = window.candidates.remove(&displaced);
                    let _ = window.adoptions.remove(&displaced);
                }
                let address = declaration.address();
                let _ = window.candidates.insert(address, declaration);
                let _ = window.adoptions.insert(address, Round::over(&self.roster));
                Ok(())
            }
        }
    }

    /// Folds a member's confirmation report: its delivered cut when the
    /// window's declaration reached it, addressed by the candidate it
    /// confirmed.
    ///
    /// The station is read out of the [`Vouched`] evidence rather than
    /// passed beside it. Repeated reports fold by join, and reports for a
    /// sealed epoch or a latched winner absorb.
    ///
    /// # Errors
    ///
    /// [`EpochRefusal::AddressMiss`] for a report addressed to a
    /// declaration this replica has not delivered --- the report outran its
    /// record, and redelivery cures it --- or from an off-roster station.
    /// [`EpochRefusal::ConfirmationDoesNotCover`] when the reported cut
    /// omits the declaration it claims to confirm, which is what stops a
    /// minter hiding its own declaration.
    pub fn confirm(
        &mut self,
        epoch: EpochAddress,
        report: &Vouched<Cut>,
    ) -> Result<(), EpochRefusal> {
        let station = report.by();
        let delivered = report.claim();
        if self.lineage.iter().any(|sealed| sealed.contains(epoch)) {
            return Ok(());
        }
        let Some(window) = &mut self.window else {
            return Err(EpochRefusal::AddressMiss {
                epoch,
                dot: RawDot::new(station, 0),
            });
        };
        if !window.candidates.contains_key(&epoch) {
            return Err(EpochRefusal::AddressMiss {
                epoch,
                dot: RawDot::new(station, 0),
            });
        }
        if window.fixed.is_some() {
            return Ok(());
        }
        let declaration = epoch.declaration();
        let covered = delivered.as_vector().get(declaration.station());
        if covered < declaration.counter() {
            return Err(EpochRefusal::ConfirmationDoesNotCover {
                epoch,
                station,
                covered,
            });
        }
        window
            .confirmations
            .fold(station, delivered.as_vector())
            .map_err(|UnknownStation { station }| EpochRefusal::AddressMiss {
                epoch,
                dot: RawDot::new(station, 0),
            })
    }

    /// Adopts the fixed winner: the local half of the adoption round.
    ///
    /// Succeeds exactly when the confirmation round is complete under the
    /// witnessed watermark. Latches the winner, irrevocably from here on,
    /// and folds this replica's own report; the caller replicates the same
    /// report to its peers. Idempotent.
    ///
    /// `own_counter` is this station's high water at adoption: a measure,
    /// not an identity, and zero is lawful (a member that has minted
    /// nothing reports `0`). The pair is deliberately not a [`Dot`]
    /// (ruling R-91).
    ///
    /// # Errors
    ///
    /// [`EpochRefusal::Unconfirmed`] while there is no window or the
    /// watermark has not covered the round's join, so no irrevocable winner
    /// exists and nothing may mint natively yet.
    /// [`EpochRefusal::UnassentedAdmission`] for a tagged winner whose
    /// membership boundary this machine has not endorsed.
    /// [`EpochRefusal::AddressMiss`] for an adopter off the round's roster.
    pub fn adopt(
        &mut self,
        station: u32,
        own_counter: u64,
        stability: &Stability,
    ) -> Result<Adopted, EpochRefusal> {
        let Some(window) = &mut self.window else {
            return Err(EpochRefusal::Unconfirmed);
        };
        Self::latch_winner(window, stability);
        let Some(winner) = window.fixed else {
            return Err(EpochRefusal::Unconfirmed);
        };
        // Assent rides the adoption (PRD 0028 R2, ruling R-89): a tagged
        // winner is adopted only over this machine's own endorsement of
        // the same boundary, held as round state. An open round always
        // carries its opener's word, so "round with this boundary" is the
        // assent. Honest fleets never stall here --- a tag requires the
        // complete round, which is every family member's endorsement ---
        // so this refusal prices a fabricated tag: the window wedges
        // before any seal exists, loudly, at every honest member, instead
        // of an agreed wrong record after retention. Recoverable by
        // [`open_arrival`](Self::open_arrival) over the same boundary.
        if let Some(candidate) = window.candidates.get(&winner)
            && let Some(required) = candidate.admission()
        {
            let assented = self.arrival.as_ref().is_some_and(|round| {
                round.own() == station
                    && round.admission() == required
                    && Some(round.commitment()) == candidate.admission_commitment()
            });
            if !assented {
                return Err(EpochRefusal::UnassentedAdmission { winner });
            }
        }
        let mut own = VersionVector::new();
        own.observe(station, own_counter);
        let Some(round) = window.adoptions.get_mut(&winner) else {
            return Err(EpochRefusal::Unconfirmed);
        };
        round
            .fold(station, &own)
            .map_err(|UnknownStation { station }| EpochRefusal::AddressMiss {
                epoch: winner,
                dot: RawDot::new(station, 0),
            })?;
        window.adopted = true;
        Ok(Adopted {
            declaration: window.candidates[&winner].clone(),
        })
    }

    /// Folds a peer's adoption report: its own-station cut at adoption,
    /// addressed by the candidate it adopted.
    ///
    /// The [`Vouched`] counter binds to the station that claimed it. This is
    /// the value the sealed join is built from, so every downstream digest
    /// and certificate rests on it transitively --- which is why this door
    /// carries the grade.
    ///
    /// A report may arrive before this replica has seen the confirmation
    /// round complete; it folds into that candidate's own round. Once the
    /// winner fixes, only its round can seal.
    ///
    /// # Errors
    ///
    /// [`EpochRefusal::AddressMiss`] for a report addressed to a
    /// declaration this replica has not delivered (redelivery cures), or
    /// from a station off the roster.
    pub fn adopt_report(
        &mut self,
        epoch: EpochAddress,
        report: &Vouched<u64>,
    ) -> Result<(), EpochRefusal> {
        let station = report.by();
        let own_counter = *report.claim();
        if self.lineage.iter().any(|sealed| sealed.contains(epoch)) {
            return Ok(());
        }
        let Some(window) = &mut self.window else {
            return Err(EpochRefusal::AddressMiss {
                epoch,
                dot: RawDot::new(station, own_counter),
            });
        };
        if !window.candidates.contains_key(&epoch) {
            return Err(EpochRefusal::AddressMiss {
                epoch,
                dot: RawDot::new(station, own_counter),
            });
        }
        let mut own = VersionVector::new();
        own.observe(station, own_counter);
        let Some(round) = window.adoptions.get_mut(&epoch) else {
            return Err(EpochRefusal::AddressMiss {
                epoch,
                dot: RawDot::new(station, own_counter),
            });
        };
        round
            .fold(station, &own)
            .map_err(|UnknownStation { station }| EpochRefusal::AddressMiss {
                epoch,
                dot: RawDot::new(station, own_counter),
            })
    }

    /// Fires the seal if the adoption round is complete under the witnessed
    /// watermark: the window closes, the candidate set and sealed join
    /// retire into the lineage as the epoch's duplicate recognizer, and the
    /// horizon truncates the oldest entry.
    ///
    /// `None` while the round is short or the watermark has not covered its
    /// join. The check is monotone, so call again as reports arrive.
    ///
    /// **A returned record is not a certifiable one.** It means this
    /// replica's rounds completed, not that its peers hold the same record;
    /// [`EpochShadow::consign`](crate::metis::EpochShadow::consign) is where
    /// that is caught, and is why it and not this door mints [`Consigned`].
    /// What the seal *does* establish is that no old-addressed delta can
    /// lawfully remain in flight, which is what licenses dropping the shadow.
    ///
    /// One refusal is terminal rather than monotone: a window at the
    /// `u64::MAX` generation ceiling never seals, since its successor would
    /// share its generation and break lineage consecutiveness. Unreachable
    /// by live sealing and refused at the bootstrap door, so this is defence
    /// in depth against the pairing; it runs before any mutation.
    pub fn try_seal(&mut self, stability: &Stability) -> Option<&SealedEpoch> {
        let window = self.window.as_mut()?;
        Self::latch_winner(window, stability);
        let winner = window.fixed?;
        if !window.adopted {
            return None;
        }
        let join = window
            .adoptions
            .get(&winner)?
            .join_when_complete(stability)?;
        let watermark = stability.watermark_cut()?;
        if !join
            .partial_cmp(watermark.as_vector())
            .is_some_and(core::cmp::Ordering::is_le)
        {
            return None;
        }
        let next = self.generation.checked_add(1)?;
        // The record carries what its winner declared, or nothing
        // (PRD 0028 R1/R2): no separate rule, no per-replica predicate.
        let admission = window
            .candidates
            .get(&winner)
            .and_then(Declaration::admission)
            .cloned();
        self.lineage.push_back(SealedEpoch {
            declaration: winner,
            // Retained raw: the set is complete, not merely
            // order-insensitive (`docs/metis-epoch-entry-doors.adoc`).
            candidates: window.candidates.keys().copied().collect(),
            // Canonicalized against the join, which is what separates a
            // receipt delivered everywhere from one that reached only some
            // members and would diverge the consignment contexts.
            protocol: window
                .protocol
                .iter()
                .copied()
                .filter(|dot| join.get(dot.station()) >= dot.counter())
                .collect(),
            sealed_join: join,
            admission: admission.clone(),
        });
        self.window = None;
        self.generation = next;
        // Widening is what sealing an admission-bearing record means
        // (PRD 0028 R1): the joiners are members from the next
        // generation, so the next window's rounds range over them. The
        // caller rebuilds its tracker from this record at the boundary,
        // exactly as the recipe already rebuilds at every seal. Every
        // seal also ends the arrival round --- consumed by this record,
        // or superseded with its base --- which is what makes a stale
        // endorsement structurally unspendable.
        if let Some(admission) = &admission {
            self.roster.extend(admission.joiners());
        }
        self.arrival = None;
        while self.lineage.len() > self.horizon.get() {
            let _ = self.lineage.pop_front();
        }
        self.lineage.back()
    }

    /// Answers old-addressed traffic against the retained lineage: the
    /// duplicate contract (R6).
    ///
    /// `Ok(())` is the duplicate verdict: the dot is covered by its epoch's
    /// sealed join, so it was delivered everywhere before the seal and the
    /// delta absorbs as a no-op, before and after the seal identically.
    ///
    /// # Errors
    ///
    /// [`EpochRefusal::WindowOpen`] while the named epoch's window is still
    /// open (the delta is window traffic: judge it in the old epoch's
    /// shadow, stage three's territory, not here);
    /// [`EpochRefusal::AddressMiss`] for a dot above the sealed join (a
    /// roster violation or a replay from outside the lawful window); and
    /// [`EpochRefusal::BeyondHorizon`] for an epoch below the retained
    /// lineage, where duplicate can no longer be told from violation and
    /// the machine refuses to guess.
    pub fn recognize(&self, epoch: EpochAddress, dot: Dot) -> Result<(), EpochRefusal> {
        if let Some(window) = &self.window
            && window.candidates.contains_key(&epoch)
        {
            return Err(EpochRefusal::WindowOpen {
                open: window.fixed.unwrap_or(epoch),
            });
        }
        match self.lineage.iter().find(|sealed| sealed.contains(epoch)) {
            Some(sealed) if sealed.sealed_join.get(dot.station()) >= dot.counter() => Ok(()),
            Some(_) => Err(EpochRefusal::AddressMiss {
                epoch,
                dot: dot.into(),
            }),
            None => Err(EpochRefusal::BeyondHorizon { epoch }),
        }
    }

    /// The open window's candidates, ascending by dot: the provisional
    /// declaration and everything concurrent with it, empty when no window
    /// is open. Losers stay visible until the seal rather than being
    /// silently dropped.
    pub fn candidates(&self) -> impl Iterator<Item = &Declaration> {
        self.window
            .iter()
            .flat_map(|window| window.candidates.values())
    }

    /// The locally latched winner, `Some` once this replica has observed
    /// the confirmation round complete under the delivery watermark.
    /// Irrevocable from that observation on.
    #[must_use]
    pub fn fixed(&self) -> Option<&Declaration> {
        let window = self.window.as_ref()?;
        window.candidates.get(&window.fixed?)
    }

    /// Whether this replica has adopted the open window's winner.
    #[must_use]
    pub fn adopted(&self) -> bool {
        self.window.as_ref().is_some_and(|window| window.adopted)
    }

    /// The retained sealed epochs, oldest first, at most the declared
    /// horizon.
    pub fn sealed(&self) -> impl Iterator<Item = &SealedEpoch> {
        self.lineage.iter()
    }

    /// The next window's lineage generation, advanced only at seal. The
    /// live mirror of the snapshot's `generation`, without materializing a
    /// checkpoint to learn one number.
    #[must_use]
    pub const fn generation(&self) -> u64 {
        self.generation.get()
    }

    /// The current generation's roster, ascending --- the founding roster
    /// plus every admission a retained seal carried --- so a caller
    /// validating a proof or certificate against its configured family
    /// reads the machine directly.
    pub fn roster(&self) -> impl Iterator<Item = u32> + '_ {
        self.roster.iter().copied()
    }

    /// The retained-recognizer horizon, as declared at construction.
    #[must_use]
    pub const fn horizon(&self) -> NonZeroUsize {
        self.horizon
    }

    /// The newest retained sealed recognizer, when any generation has
    /// sealed within the horizon.
    ///
    /// Two reads ride on this one door. First, it is the in-band
    /// freshness floor (ruling R-69). The floor is stated in checkpoint
    /// terms, because an installable checkpoint names its *successor*
    /// generation while a seal names its completed one. A returning
    /// member refuses any checkpoint whose successor generation is at
    /// or below its own newest seal's, equivalently below the live
    /// [`Self::generation`] read. Without that floor, an equal-labeled
    /// pre-seal checkpoint rolls the member back across its durable
    /// seal. Second, it is the deterministic re-entry pivot after a
    /// caller crashes between its durable seal and its checkpoint mint
    /// (ruling R-76). Rehydration has already re-proven the retained
    /// lineage, and this read hands back the recognizer the interrupted
    /// mint consumed live: a read of re-proven state, never a second
    /// mint. `try_seal` remains the one live door and the affine
    /// adoption witness stays re-earned.
    #[must_use]
    pub fn newest_sealed(&self) -> Option<&SealedEpoch> {
        self.lineage.back()
    }

    /// The risen half of the R1 license (ruling R-40): the first basis
    /// entry `cut` leaves uncovered, as the typed refusal, or `None` when
    /// the cut covers the basis (coverage is inclusive: a cut exactly at
    /// the basis is risen). Canonical vector form means only genuinely
    /// required stations are walked.
    fn unrisen(basis: &Cut, cut: &Cut) -> Option<EpochRefusal> {
        basis.as_vector().iter().find_map(|(station, required)| {
            let witnessed = cut.as_vector().get(station);
            (witnessed < required).then_some(EpochRefusal::Unrisen {
                station,
                required,
                witnessed,
            })
        })
    }

    /// Latches the winner if the confirmation round is complete under the
    /// tracker's witnessed watermark: the maximum candidate by the fixed
    /// public `(rank, dot)` rule. Monotone: once latched, never recomputed,
    /// so a straggling (join-raising) confirmation report can delay a
    /// latch but never move one.
    fn latch_winner(window: &mut Window, stability: &Stability) {
        if window.fixed.is_some() {
            return;
        }
        let Some(join) = window.confirmations.join_when_complete(stability) else {
            return;
        };
        let Some(watermark) = stability.watermark_cut() else {
            return;
        };
        if !join
            .partial_cmp(watermark.as_vector())
            .is_some_and(core::cmp::Ordering::is_le)
        {
            return;
        }
        window.fixed = window
            .candidates
            .values()
            .max_by_key(|candidate| candidate.key())
            .map(Declaration::address);
    }
}