minerva 0.2.0

//! One full-stack fleet replica, composed entirely from shipped parts.
//!
//! The replica is the module note's consumer recipe made executable. Its
//! document is the text and movement causal pairs; its clock is a
//! [`TickCounter`]-backed [`Clock`]; delivery bookkeeping is a [`DotSet`]
//! have-set whose floor is the honest [`Stability`] report; the epoch
//! lifecycle is [`Epochs`]; the boundary is the stratum rebuilt from the
//! generation base plus the logged deltas the winning cut covers, the
//! [`EpochShadow`] for the window, the [`EpochIdeal`] for native holdback,
//! and the deterministic seal annex. Everything a replica says to the
//! fleet leaves through its outbox as a [`Note`]; everything it hears
//! arrives through [`Replica::handle`].

extern crate alloc;

use alloc::collections::{BTreeMap, BTreeSet};
use alloc::vec::Vec;

use crate::kairos::{Clock, Kairos, TickCounter};
use crate::metis::{
    Abandoned, Admission, ArrivalRefusal, Consigned, Cut, Departed, Departure, DepartureRefusal,
    Dot, DotSet, Dotted, EpochAddress, EpochBootstrapError, EpochConsignmentError, EpochIdeal,
    EpochProjection, EpochRefusal, EpochShadow, Epochs, Fenced, Metatheses, Retirement, Rhapsody,
    SealRecord, SealedEpoch, Stability, VersionVector, Vouched,
};

use super::{Note, OldDelta};

mod data;
mod lifecycle;
mod persistence;
mod receive;

pub type Text = Dotted<Rhapsody>;
pub type Moves = Dotted<Metatheses>;
/// One native-log entry: the event dots, the delta, and the original
/// stamp where one was spoken (`None` for entries learned as a sealed
/// peer's next-generation old traffic).
type NativeEntry = (Vec<Dot>, OldDelta, Option<Kairos>);

/// The modeled durable store: exactly what the consumer has fenced, and
/// nothing else. A checkpoint written once per seal (the generation, its
/// deterministic base pair, the epoch record with its retained lineage,
/// the dot allocator's basis, and the harness's seal agreement record,
/// all of which the seal computes in one place anyway), plus the fenced
/// notes since, in handled order: the replica's own mints and everything
/// it has heard. Nothing else is durable; everything else is rebuilt by
/// [`Replica::rehydrate`] replaying these notes through the ordinary
/// inbound door, or re-served by peers (the repair lane).
///
/// The fields are open to the fleet's tests deliberately: a falsifier
/// models a consumer that violated the stable-before-emit duty by
/// dropping from the journal a note the pre-crash replica had already
/// spoken for (that is what a lying fence looks like from the restart
/// side), and the boundary exhibits show what each such lie costs.
#[derive(Clone)]
pub struct Journal {
    /// The checkpointed generation.
    pub(super) generation: u64,
    /// The generation's deterministic base pair.
    pub(super) base_text: Text,
    pub(super) base_moves: Moves,
    /// The epoch record at the checkpoint: the retained sealed lineage
    /// (the duplicate recognizers), window closed. The lineage is part of
    /// the compaction output, so it rides the checkpoint: it is not
    /// reconstructible from the base pair and the log alone, and losing
    /// it would forfeit the replay-recognition contract.
    pub(super) epochs: Epochs,
    /// The old-plane dot allocator's checkpointed basis.
    pub(super) counter: u64,
    /// The clock's checkpointed watermark: a stamp minted at the
    /// checkpoint moment, so it dominates every stamp this replica spoke
    /// before it (the clock is monotone and minted them all). Needed
    /// because the re-foundation re-bases ranks: the consigned plane's
    /// ranks are re-minted small, so the sealed window's spoken stamps
    /// survive in no plane a restart could walk.
    pub(super) clock: Kairos,
    /// The harness's cross-fleet agreement record (every performed seal's
    /// address and final projection). Written only at the seal, so it
    /// rides the checkpoint for free; a consumer has no analogue.
    pub(super) seals: Vec<(EpochAddress, EpochProjection)>,
    /// The departures in force at the checkpoint: the stations abandoned
    /// bare, and the stations whose departure round sealed. The bounds are
    /// deliberately absent --- a checkpoint is written at a seal, every
    /// attestation crosses that boundary at the departed station's
    /// compacted prefix in the consigned base (`Departed::refounded`), and
    /// that prefix re-derives from the restored base itself, so the
    /// durable fact is the *departure*. Unlike the bare set, this half is
    /// not merely caller configuration: a restart that forgot it would
    /// admit the departed member's traffic above a coordinate its peers
    /// have already sealed over.
    pub(super) departed: BTreeSet<u32>,
    /// In departure order, for the reason the live field records.
    pub(super) attested: Vec<u32>,
    /// Fenced notes since the checkpoint, in handled order: own mints
    /// (journaled at emission, the mint duty) and received traffic
    /// (flushed from the volatile buffer no later than the next
    /// emission, the stable-before-emit duty).
    pub(super) notes: Vec<Note>,
}

/// The data-plane half of a fleet bootstrap, exported only at a seal.
///
/// The authenticated lineage remains separate: this value carries the
/// local compaction output and allocator state, while
/// [`crate::metis::LineageProofRecord`] carries the peer-evidenced
/// recognizer horizon.
/// Keeping the two inputs distinct prevents a data checkpoint from
/// smuggling lifecycle authority through the joiner door.
///
/// The *attested* departures ride the checkpoint, exactly as they ride
/// the [`Journal`]: an attestation is an agreed protocol object the
/// seat's world carries, and a joiner without it holds a meet the
/// evicted member pins forever and window rounds no substitution covers
/// (`scenarios::arrival::an_attested_fleet_admits_and_the_carry_crosses_the_widening_seal`).
/// A *bare* abandonment deliberately does not ride: it is one
/// operator's own policy read (R-84, departure is configuration), so
/// the joiner's operator makes its own.
pub(super) struct JoinerCheckpoint {
    station: u32,
    generation: u64,
    base_text: Text,
    base_moves: Moves,
    counter: u64,
    clock: Kairos,
    seals: Vec<(EpochAddress, EpochProjection)>,
    attested: Vec<u32>,
}

/// A fleet joiner refuses mismatched checkpoint and lineage authorities.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(super) enum JoinerBootstrapError {
    Epoch(EpochBootstrapError),
    StationOutsideRoster { station: u32 },
    GenerationMismatch { checkpoint: u64, lineage: u64 },
}

impl From<EpochBootstrapError> for JoinerBootstrapError {
    fn from(error: EpochBootstrapError) -> Self {
        Self::Epoch(error)
    }
}

/// The pre-dispatch snapshot effect-based fencing compares against: which
/// tracker a protocol note addresses (compared whole after dispatch, the
/// exact no-effect test), or a data note's pre-computed novelty.
enum Probe {
    Data(bool),
    Stability(Stability),
    Retirement(Retirement),
    // Boxed: `Epochs` grew past the variant-size lint with the
    // machine-held arrival round (S342), and the probe is a cold
    // pre-dispatch snapshot.
    Epochs(alloc::boxed::Box<Epochs>),
    Departures(BTreeMap<u32, Departure>),
}

/// Why a replica declined to declare: the dry-run's verdict, surfaced so a
/// scenario can pin the self-supporting-cut duty instead of opening a
/// window no replica could adopt.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Decline {
    /// No watermark cut is witnessed yet (the R1 license is absent).
    Unwitnessed,
    /// The witnessed cut cannot rebuild a foldable stratum: the named dot
    /// is the first above-cut carriage or unplaced identity the rebuild
    /// surfaced. Declaring now would open a window whose stratum every
    /// replica refuses identically; the cure is to keep gossiping.
    NotSelfSupporting {
        dot: Dot,
    },
    Refused(EpochRefusal),
}

/// The open-window state between adoption and seal.
struct Transition {
    adopted: crate::metis::Adopted,
    shadow: EpochShadow,
    /// The editing view: the pristine refounded core plus every native
    /// delta folded so far. The next generation's base comes from the
    /// shadow's seal consignment, never from this schedule-dependent view.
    native_text: Text,
    native_moves: Moves,
    native_counter: u64,
    /// Native data-note identities represented in the native log.
    /// Kept apart from causal context because a retract can cover a weave
    /// without carrying its ordering locus.
    native_recorded: DotSet,
    /// Every native delta this window saw (own mints and received), the
    /// next generation's opening log, one entry per unique event, each
    /// with its original stamp where one was spoken (`None` for entries
    /// learned as a sealed peer's next-generation old traffic, which
    /// never carried one). Retaining the stamp is what lets the repair
    /// lane re-serve without minting: a freshly minted repair stamp
    /// would be spoken-but-unfenced, exactly the class the recipe
    /// forbids.
    native_log: Vec<NativeEntry>,
}

/// One fleet member running the whole shipped stack.
pub struct Replica {
    id: u32,
    roster: Vec<u32>,
    clock: Clock<TickCounter>,
    epochs: Epochs,
    generation: u64,
    stability: Stability,
    /// This station's one dot allocator for the current old plane: text
    /// weaves, delete markers, movement testimonies, and declarations all
    /// draw from it, so the plane the witness ranges over is total.
    counter: u64,
    text: Text,
    moves: Moves,
    /// The generation's deterministic base: identical at every replica by
    /// construction (empty at birth, refound plus seal annex afterwards).
    base_text: Text,
    base_moves: Moves,
    /// This generation's dotted deltas (own and received), any order.
    log: Vec<(Vec<Dot>, OldDelta)>,
    /// Causal coverage of the current old plane. A delta's context can
    /// advance this beyond the data notes received directly.
    have: DotSet,
    /// Data-note identities represented in the generation base, log, or
    /// open shadow. Unlike `have`, this is a possession witness: causal
    /// context alone never suppresses a note that carries grow-only state.
    recorded: DotSet,
    reported: Option<VersionVector>,
    /// The generation-scoped retirement round: per member, the applied
    /// removals it has acknowledged, gathered under the cut gate (an
    /// acknowledgement counts only once its carried floor is delivered
    /// here). Reset at every seal: the re-foundation sweeps retired
    /// identity wholesale and the next plane is a new dot space, so
    /// retirement evidence never crosses a boundary.
    retirement: Retirement,
    /// The last applied-removal set this replica acknowledged, so the
    /// acknowledgement round publishes only on change. Starts empty: a
    /// member with nothing applied and the empty acknowledgement are the
    /// same bottom slot at every tracker.
    acked: DotSet,
    confirmed: BTreeSet<EpochAddress>,
    /// Declaration dots delivered this generation: window annex members.
    declaration_dots: Vec<Dot>,
    gate: EpochIdeal<(Vec<Dot>, OldDelta)>,
    /// Event dots held in the pre-adoption gate: the bounded-record
    /// dedup for native traffic that arrives before this replica adopts
    /// (the gate itself would hold duplicates). Reset at the seal.
    gated: DotSet,
    transition: Option<Transition>,
    /// Notes that outran their record (a report before its declaration, a
    /// faster peer's next-generation traffic); redelivery cures them.
    parked: Vec<Note>,
    /// Every seal this replica performed: the winning address and the
    /// final projection, the cross-fleet agreement surface.
    pub seals: Vec<(EpochAddress, EpochProjection)>,
    /// The modeled durable store (the rehydration recipe's fenced image).
    journal: Journal,
    /// Received notes not yet fenced: delivered, live, and lost at a
    /// crash. Flushed into the journal before anything leaves through the
    /// outbox, so no emission ever claims state the journal does not
    /// hold; a crash between emissions loses only what the repair lane
    /// re-serves.
    volatile: Vec<Note>,
    /// Armed only by the `byzantine` exposure module: turns the
    /// consignment door's refusal from a harness panic into an observable
    /// halt. Every other arm leaves it false, so the honest laws keep
    /// their strict assertion.
    halt_on_refusal: bool,
    /// The refusal that halted this replica, if one did.
    halted: Option<EpochConsignmentError>,
    /// Every seal record whose consignment this replica *accepted*: the
    /// certifiable grade (S335), and the only records a checkpoint
    /// certificate may lawfully bind over.
    ///
    /// Deliberately separate from `epochs().sealed()`. A wedged replica's
    /// ledger runs one generation ahead of this list, and that gap is the
    /// certification duty made observable: what `try_seal` retired, minus
    /// what the document plane accepted.
    certified: Vec<Consigned>,
    /// Members this replica's operator has abandoned without a departure
    /// round: the family narrows for the watermark, and the epoch rounds
    /// keep asking. Held beside the tracker because the recipe rebuilds the
    /// tracker at every seal, and the family a meet ranges over is
    /// caller-declared configuration that must survive that rebuild
    /// (`docs/metis-membership-departure.adoc`).
    departed: BTreeSet<u32>,
    /// Departure rounds this replica is running or has run, by departing
    /// station. A sealed one is this replica's fence, consulted on every
    /// inbound dot, and outlives the round for exactly that reason.
    departures: BTreeMap<u32, Departure>,
    /// The attestations sealed rounds produced, *in departure order*,
    /// re-applied at every seal (at the consigned base's compacted prefix,
    /// per `Departed::refounded`) exactly as `departed` is. The order is load-bearing: each attestation
    /// names the family that was surviving when it was agreed, so replaying
    /// a later departure before an earlier one presents a family the tracker
    /// has not narrowed to yet.
    attested: Vec<Departed>,
    /// Every dot the fence refused *after* its round sealed: the resurgence
    /// verdicts, in arrival order. A provisional refusal is a hold and is
    /// deliberately not recorded here.
    resurgences: Vec<Fenced>,
}

impl Replica {
    pub const fn id(&self) -> u32 {
        self.id
    }

    /// Arms the observable halt (see [`Replica::halted`]). The exposure
    /// module is the only caller.
    pub const fn tolerate_refusals(&mut self) {
        self.halt_on_refusal = true;
    }

    /// The operator's departure decision. Narrows the family, records it so
    /// the seal-time tracker rebuild re-applies it, and polls: a departure
    /// enables monotone steps exactly as an arriving note does.
    pub fn abandon(&mut self, station: u32, out: &mut Vec<Note>) -> Abandoned {
        let _ = self.departed.insert(station);
        let abandoned = self.stability.abandon(station).expect("a family remains");
        self.poll(out);
        abandoned
    }

    /// Opens (or re-enters) the departure round for `departing`, proposes
    /// this replica's own gap-free prefix of its dots, and fences there.
    ///
    /// The operator's half. The proposal leaves through the outbox for the
    /// other survivors to fold; the round seals here once theirs arrive.
    ///
    /// # Errors
    ///
    /// Whatever [`Departure::propose`] refuses, unchanged: a ragged holding
    /// or a broken fence is a fact about this replica, not about the fleet.
    pub fn open_departure(
        &mut self,
        departing: u32,
        out: &mut Vec<Note>,
    ) -> Result<u64, DepartureRefusal> {
        let (id, generation) = (self.id, self.generation);
        let held = self.have.clone();
        // The successor-plane witness: everything this replica possesses in
        // the plane the open epoch founds. A departing station's coordinate
        // there is bottom, so a dot of its already possessed is one no bound
        // agreed here could cover.
        //
        // Both halves are possession. `native_recorded` is what a transition
        // has folded; `gated` is what the pre-adoption gate is *holding* for
        // a window this replica has not adopted, which `drain_gate` releases
        // straight into the successor plane. Reading only the first would let
        // a gated dot survive a departure that refounds beneath it.
        let mut successor = self.gated.clone();
        if let Some(transition) = &self.transition {
            successor = successor.merge(&transition.native_recorded);
        }
        let prefix = self
            .enter_departure(departing)?
            .propose(&held, &successor)?;
        // The mint duty, in its sharpest form. A proposal is testimony under
        // this replica's own identity *and* a promise about its future
        // intake, so it must be durable before any peer can price a bound on
        // it: a restart that forgot the fence would admit traffic above a
        // bound its peers have already sealed over
        // (`Departure::reinstate` is the replay door).
        self.emit_mint(
            out,
            Note::Depart {
                generation,
                departing,
                by: id,
                prefix,
            },
        );
        self.settle_departures();
        self.poll(out);
        Ok(prefix)
    }

    /// The round for `departing`, opened over the family that survives it if
    /// this replica has not opened one yet.
    fn enter_departure(&mut self, departing: u32) -> Result<&mut Departure, DepartureRefusal> {
        if !self.departures.contains_key(&departing) {
            let survivors: Vec<u32> = self
                .roster
                .iter()
                .copied()
                .filter(|&station| {
                    station != departing
                        && !self.departed.contains(&station)
                        && self.attested(station).is_none()
                })
                .collect();
            let _ = self
                .departures
                .insert(departing, Departure::opened(departing, self.id, survivors)?);
        }
        Ok(self
            .departures
            .get_mut(&departing)
            .expect("just inserted or already present"))
    }

    /// Seals whatever departure rounds are complete and applies their
    /// attestations to the tracker. The caller polls: an attestation enables
    /// monotone steps exactly as an arriving report does.
    pub(super) fn settle_departures(&mut self) {
        let ready: Vec<Departed> = self
            .departures
            .values_mut()
            .filter_map(|round| round.try_seal().cloned())
            .collect();
        for departed in ready {
            if self.attested(departed.station()).is_some() {
                continue;
            }
            // A refusal here is a *recoverable* outcome, not a recipe error.
            // Two departures agreed concurrently can each be the sole server
            // of the other's bound, so whichever is applied first refuses the
            // second until some survivor's report reaches that coordinate.
            // The sealed round stays where it is and this runs again on every
            // poll, so the retry needs no separate lane.
            if self.stability.abandon_attested(&departed).is_ok() {
                self.attested.push(departed);
            }
        }
    }

    /// Re-applies the departures a checkpoint recorded, at the restored
    /// base's compacted prefix: a sealed round whose bound crossed the
    /// boundary is exactly a fence at that prefix over the surviving
    /// family, and re-deriving it needs no round because the prefix is
    /// what the consigned base itself holds of the departed member ---
    /// testimony every survivor can give from the plane it restored
    /// (`Departed::refounded`; bottom when nothing of the member's
    /// survived).
    ///
    /// Coverage and possession part here exactly as they do in the live
    /// carry, and the provenance is *derived, never persisted*: a member
    /// has proven possession of the restored base iff it participated in
    /// the seal that consigned it, and the newest sealed record's own
    /// admission names precisely the members that have not --- its
    /// joiners, whose first seal is still ahead of them. Every older
    /// joiner co-sealed each later window (N-of-N over the widened
    /// roster), so the one unproven set is the newest record's, which the
    /// checkpoint already carries: recorded membership answering its own
    /// restore question. A coverage-only member's word folds at zero (the
    /// join is unmoved; the family check still counts the key) and it
    /// earns holder status through its own report. The restoring replica
    /// itself holds its base by construction, whichever door it came
    /// through, so its own word is the prefix.
    fn restore_departures(&mut self, departed: &BTreeSet<u32>, attested: &[u32]) {
        for &station in departed {
            let _ = self.departed.insert(station);
            let _ = self.stability.abandon(station).expect("a family remains");
        }
        let coverage_only: BTreeSet<u32> = self
            .epochs
            .newest_sealed()
            .and_then(SealedEpoch::admission)
            .map(|admission| admission.joiners().collect())
            .unwrap_or_default();
        for &station in attested {
            let held = self.have.floor().get(station);
            let mut round = self
                .enter_departure(station)
                .expect("a checkpointed departure left a surviving family")
                .clone();
            round
                .reinstate(held)
                .expect("the base prefix cannot lower a fresh fence");
            let family: Vec<u32> = round.proposals().map(|(station, _)| station).collect();
            for surviving in family {
                if surviving != self.id {
                    let word = if coverage_only.contains(&surviving) {
                        0
                    } else {
                        held
                    };
                    round
                        .fold(&Vouched::trust(surviving, word))
                        .expect("a survivor of the checkpointed family");
                }
            }
            let departed = round
                .try_seal()
                .expect("every slot holds the base prefix")
                .clone();
            let _ = self
                .stability
                .abandon_attested(&departed)
                .expect("a family remains");
            self.attested.push(departed);
            let _ = self.departures.insert(station, round);
        }
    }

    /// The reconstructed attestation's word for `member`: the provenance
    /// read the restore regression pins (a coverage-only joiner's word is
    /// zero, a holder's is the base prefix).
    pub fn attested_proposal(&self, departed: u32, member: u32) -> Option<u64> {
        self.attested
            .iter()
            .find(|attestation| attestation.station() == departed)
            .and_then(|attestation| attestation.proposal_of(member))
    }

    /// The operator's admission decision: opens the arrival round for
    /// `joiners` over this machine's newest sealed base, endorses it in
    /// the same act, and broadcasts the endorsement (PRD 0028 R2).
    ///
    /// The commitment beside the boundary is this replica's own digest of
    /// the base's exact seal frame ([`record_commitment`]), so the round's
    /// concord compares records, not addresses: two members holding forked
    /// records under one winner refuse each other's word as
    /// `CommitmentDiscord` instead of agreeing past the fork (R7).
    ///
    /// The endorsement is journaled before it travels (the mint duty in
    /// its R3 form: mint durably, then act), and a restart re-opens the
    /// round from the journaled note through the ordinary replay door.
    ///
    /// # Errors
    ///
    /// Whatever [`Epochs::open_arrival`] refuses, unchanged: the refusal
    /// is a fact about this replica's own lineage and family, surfaced so
    /// a scenario can pin it.
    pub fn open_arrival(
        &mut self,
        joiners: &[u32],
        out: &mut Vec<Note>,
    ) -> Result<(), ArrivalRefusal> {
        let base = self
            .epochs
            .newest_sealed()
            .ok_or(ArrivalRefusal::Unfounded)?;
        let commitment = record_commitment(base);
        let admission = Admission::new(base.declaration(), joiners.iter().copied())?;
        // The machine door is idempotent for the identical boundary, and
        // the journal must be too (the bounded-record rule): only the
        // open that creates the round journals its mint; a retry
        // re-emits the already-durable word without appending it again.
        let fresh = self.epochs.arrival().is_none();
        let endorsement =
            self.epochs
                .open_arrival(admission, commitment, self.id, &self.stability)?;
        let note = Note::Endorse {
            by: self.id,
            admission: endorsement.admission().clone(),
            commitment,
        };
        if fresh {
            self.emit_mint(out, note);
        } else {
            self.emit(out, note);
        }
        // The full drive, not a bare poll: a peer endorsement that
        // arrived before this operator's decision parked as `NoRound`,
        // and this very act is what unparks it.
        self.drive(out);
        Ok(())
    }

    /// The first dot in `dots` this replica's fences refuse, if any.
    ///
    /// Sealed refusals are recorded as resurgence verdicts on the way past;
    /// a provisional one is a hold, and redelivery after the seal admits it.
    fn fenced(&mut self, dots: &[Dot]) -> Option<Fenced> {
        let refusal = self.old_plane_refusal(dots);
        self.refuse(refusal)
    }

    /// The `&self` half, so the inbound door can ask before it decides
    /// whether a note taught this replica anything worth journaling.
    fn old_plane_refusal(&self, dots: &[Dot]) -> Option<Fenced> {
        dots.iter().find_map(|&dot| {
            self.departures
                .get(&dot.station())
                .and_then(|round| round.admits(dot).err())
        })
    }

    /// The same, for the *successor* plane, where a departing member's
    /// coordinate is bottom from the moment the round opens
    /// (`Departure::admits_successor`).
    fn fenced_native(&mut self, dots: &[Dot]) -> Option<Fenced> {
        let refusal = self.native_plane_refusal(dots);
        self.refuse(refusal)
    }

    fn native_plane_refusal(&self, dots: &[Dot]) -> Option<Fenced> {
        // The old `dot.max(1)` clamp is gone: a one-based counter is the
        // identity type's own law now (ruling R-91), so the successor
        // plane's bottom coordinate is unreachable from a dot.
        dots.iter().find_map(|&dot| {
            self.departures
                .get(&dot.station())?
                .admits_successor(dot)
                .err()
        })
    }

    /// Whether a data note carries a dot this replica's fences refuse, in
    /// the plane that note addresses. Pure: the recording half runs at
    /// dispatch, once, for the note that is actually being judged.
    fn note_refused(&self, note: &Note) -> bool {
        match note {
            Note::Old {
                dots, generation, ..
            } if *generation <= self.generation => self.old_plane_refusal(dots).is_some(),
            Note::Old { dots, .. } | Note::Native { dots, .. } => {
                self.native_plane_refusal(dots).is_some()
            }
            Note::Declare { declaration } => self.old_plane_refusal(&[declaration.dot()]).is_some(),
            _ => false,
        }
    }

    /// Records a sealed refusal as a resurgence verdict on the way past.
    fn refuse(&mut self, refusal: Option<Fenced>) -> Option<Fenced> {
        let refusal = refusal?;
        if refusal.is_resurgence() && !self.resurgences.contains(&refusal) {
            self.resurgences.push(refusal);
        }
        Some(refusal)
    }

    /// The dots this replica's sealed fences refused: the resurgence
    /// verdicts. See [`Fenced::is_resurgence`] for exactly what one proves.
    pub fn resurgences(&self) -> impl Iterator<Item = &Fenced> {
        self.resurgences.iter()
    }

    /// The promise in force for a departing member, if this replica has made
    /// one: the durable half of a departure, surfaced so a restart can be
    /// asked whether it kept it.
    pub fn departure_fence(&self, station: u32) -> Option<u64> {
        self.departures.get(&station).and_then(Departure::fence)
    }

    /// The agreed bound this replica holds for a departed member, if its
    /// round sealed.
    pub fn attested(&self, station: u32) -> Option<u64> {
        self.attested
            .iter()
            .find(|departed| departed.station() == station)
            .map(Departed::bound)
    }

    /// Abandoned members whose slot absorbed a report afterwards, surfaced
    /// through the recipe. See [`Stability::resurgent`] for how narrow the
    /// claim is.
    pub fn resurgent(&self) -> impl Iterator<Item = u32> + '_ {
        self.stability.resurgent()
    }

    /// This replica's stability watermark: the read a departure releases.
    pub fn watermark(&self) -> VersionVector {
        self.stability.watermark()
    }

    /// The consignment refusal that stopped this replica, if any.
    pub const fn halted(&self) -> Option<&EpochConsignmentError> {
        self.halted.as_ref()
    }

    /// Re-establishes the certifiable grade for a lineage installed by a
    /// restore door rather than earned at a consignment. Both callers are
    /// named at their call sites; see [`Consigned::vouch`].
    fn vouch_restored_lineage(&mut self) {
        self.certified = self
            .epochs
            .sealed()
            .cloned()
            .map(Consigned::vouch)
            .collect();
    }

    /// The seal records this replica accepted at the consignment door,
    /// oldest first: the certifiable grade.
    pub fn certified(&self) -> impl Iterator<Item = &SealedEpoch> {
        self.certified.iter().map(Consigned::record)
    }

    pub const fn generation(&self) -> u64 {
        self.generation
    }

    pub const fn epochs(&self) -> &Epochs {
        &self.epochs
    }

    pub const fn text(&self) -> &Text {
        &self.text
    }

    pub const fn moves(&self) -> &Moves {
        &self.moves
    }

    pub fn adopted(&self) -> bool {
        self.transition.is_some()
    }

    /// The effective visible order of whatever plane this replica is
    /// editing right now.
    pub fn effective_order(&self) -> Vec<Dot> {
        self.transition.as_ref().map_or_else(
            || self.text.store().recension(self.moves.store()).order(),
            |transition| {
                transition
                    .native_text
                    .store()
                    .recension(transition.native_moves.store())
                    .order()
            },
        )
    }

    // ---- the durability boundary (the rehydration recipe) ------------------

    /// The modeled fsync: flushes the volatile buffer into the journal.
    fn fence(&mut self) {
        self.journal.notes.append(&mut self.volatile);
    }

    /// Parks a note that outran its record, once: an exact duplicate of
    /// an already-parked note re-teaches nothing, and counting it as
    /// effectful would compound the parked set, the volatile buffer, and
    /// the journal on every repair round against a lagging prerequisite
    /// (the bounded-record rule applied to the parked lane). A restate
    /// carrying genuinely new information (a higher cut, a wider
    /// acknowledgement) differs by value and parks beside the old copy;
    /// redelivery retires both through the retry loop's idempotent folds.
    fn park(&mut self, note: &Note) {
        if !self.parked.contains(note) {
            self.parked.push(note.clone());
        }
    }

    /// Fences, then emits: every outbound note passes through here or
    /// [`emit_mint`](Self::emit_mint), so nothing this replica says ever
    /// outruns its durable image (the stable-before-emit duty; Raft's
    /// persisted-before-respond rule leaderless). The note itself is not
    /// journaled: reports, acknowledgements, confirmations, and adoption
    /// reports are re-derived from the rebuilt state at rehydration, and
    /// re-deriving can only re-say them at an equal or higher basis.
    fn emit(&mut self, out: &mut Vec<Note>, note: Note) {
        self.fence();
        out.push(note);
    }

    /// Whether a data note carries an identity this replica does not possess
    /// in the addressed plane: the pre-dispatch half of effect-based
    /// fencing. Causal coverage is insufficient because a retract can cover
    /// a weave without carrying its ordering locus. Pure duplicates fold to
    /// nothing, so journaling them would only compound the record across
    /// redelivery and repair rounds.
    fn data_note_novel(&self, note: &Note) -> bool {
        match note {
            Note::Old {
                generation, dots, ..
            } => {
                if *generation < self.generation {
                    return false;
                }
                if *generation == self.generation + 1 {
                    return self.transition.as_ref().is_none_or(|transition| {
                        dots.iter()
                            .any(|&dot| !transition.native_recorded.contains(dot))
                    });
                }
                dots.iter().any(|&dot| !self.recorded.contains(dot))
            }
            Note::Native { epoch, dots, .. } => {
                if let Some(transition) = &self.transition
                    && transition.adopted.address() == *epoch
                {
                    return dots
                        .iter()
                        .any(|&dot| !transition.native_recorded.contains(dot));
                }
                if self
                    .epochs
                    .sealed()
                    .any(|sealed| sealed.declaration() == *epoch)
                {
                    return epoch.generation() + 1 == self.generation
                        && dots.iter().any(|&dot| !self.recorded.contains(dot));
                }
                // Pre-adoption: novel against the gate's held set.
                dots.iter().any(|&dot| !self.gated.contains(dot))
            }
            _ => unreachable!("only data notes are probed for novelty"),
        }
    }

    /// Fences, journals the mint, then emits: the mint duty. A dotted
    /// delta and a declaration are testimony under this replica's own
    /// identity, so they must be durable before any peer can hold them;
    /// a re-derived allocator that had forgotten an emitted dot would
    /// re-mint it for different content, and one identity would name two
    /// payloads forever (the boundary exhibit).
    fn emit_mint(&mut self, out: &mut Vec<Note>, note: Note) {
        self.fence();
        self.journal.notes.push(note.clone());
        out.push(note);
    }

    fn fold_current(&mut self, delta: &OldDelta) {
        match delta {
            OldDelta::Text(delta) => self.text.merge_from(delta),
            OldDelta::Moves(delta) => self.moves.merge_from(delta),
        }
    }

    // ---- the declaration door ---------------------------------------------

    /// The generation's base frontier, the declared epoch basis of the R1
    /// risen license: the floor the current plane was born with, bottom at
    /// birth and the seal annex's gap-free floor thereafter. A
    /// deterministic function of the sealed generation, so every honest
    /// member declares the same basis and the `EpochUnrisen` refusal is
    /// identical everywhere.
    fn base_frontier(&self) -> Cut {
        Cut::floor_of(&self.base_text.context().merge(self.base_moves.context()))
    }

    /// Declares a re-foundation at the currently witnessed watermark, after
    /// dry-running the stratum rebuild the way every adopter will run it.
    pub fn try_declare(&mut self, out: &mut Vec<Note>) -> Result<EpochAddress, Decline> {
        let Some(cut) = self.stability.watermark_cut() else {
            return Err(Decline::Unwitnessed);
        };
        if let Some(dot) = self.stratum_obstacle(&cut) {
            return Err(Decline::NotSelfSupporting { dot });
        }
        // The allocator's next counter: one past a count, so one-based by
        // construction and never the non-dot zero (ruling R-91).
        let dot = Dot::from_parts(self.id, self.counter + 1)
            .expect("the allocator's next counter is one past a count, hence nonzero");
        let rank = self.clock.now(0u16);
        let declaration = self
            .epochs
            .declare(dot, rank, &self.stability, &self.base_frontier())
            .map_err(Decline::Refused)?;
        self.counter += 1;
        let _ = self.have.insert(dot);
        self.declaration_dots.push(dot);
        let address = declaration.address();
        self.emit_mint(out, Note::Declare { declaration });
        self.poll(out);
        Ok(address)
    }

    /// The dry-run: the first dot at which the stratum rebuilt at `cut`
    /// would refuse, or `None` when the fold is licensed. Mirrors the
    /// `EpochStratum` checks without needing the adoption witness.
    fn stratum_obstacle(&self, cut: &Cut) -> Option<Dot> {
        let (text, moves) = self.rebuild_at(cut);
        let coverage = text.context().merge(moves.context());
        if let Some(dot) = cut.to_have_set().difference(&coverage).next() {
            return Some(dot);
        }
        let vector = cut.as_vector();
        if let Some(dot) = text
            .context()
            .dots()
            .chain(moves.context().dots())
            .find(|dot| dot.counter() > vector.get(dot.station()))
        {
            return Some(dot);
        }
        match text.store().refound(moves.store()) {
            Ok(_) => None,
            Err(unsealed) => Some(unsealed.dot),
        }
    }

    /// The stratum rebuild: the generation base plus exactly the logged
    /// deltas whose event dots the cut covers. Pair-merge idempotence makes
    /// this insensitive to how much of that traffic the live pairs already
    /// absorbed.
    fn rebuild_at(&self, cut: &Cut) -> (Text, Moves) {
        let vector = cut.as_vector();
        let mut text = self.base_text.clone();
        let mut moves = self.base_moves.clone();
        for (dots, delta) in &self.log {
            if dots
                .iter()
                .all(|dot| dot.counter() <= vector.get(dot.station()))
            {
                match delta {
                    OldDelta::Text(delta) => text.merge_from(delta),
                    OldDelta::Moves(delta) => moves.merge_from(delta),
                }
            }
        }
        (text, moves)
    }
}

/// The harness's record commitment (PRD 0028 R7): a deterministic 32-byte
/// fold over the base's exact seal frame, minted by every member from its
/// own retained record. Content-sensitive --- any byte of the frame moves
/// every lane --- so forked records under one winner yield discordant
/// commitments and the round's concord sees what the address cannot.
/// Determinism is the claim; cryptographic strength stays the consumer's
/// (the caller-minted-digest posture of the lineage node).
fn record_commitment(sealed: &SealedEpoch) -> [u8; 32] {
    let frame = SealRecord::from_sealed(sealed).to_bytes();
    let mut commitment = [0u8; 32];
    for (lane, chunk) in (1u64..).zip(commitment.chunks_exact_mut(8)) {
        let mut acc = 0xcbf2_9ce4_8422_2325u64 ^ lane.wrapping_mul(0x9E37_79B9_7F4A_7C15);
        for &byte in &frame {
            acc ^= u64::from(byte);
            acc = acc.wrapping_mul(0x0000_0100_0000_01B3);
        }
        chunk.copy_from_slice(&acc.to_be_bytes());
    }
    commitment
}

/// Advances `clock` past every rank a pair's planes carry (loci and
/// movement testimonies): the rehydration recipe's clock re-derivation.
/// Volatile clock state is never persisted; the durable planes are its
/// recovery source, exactly as they are for everything else.
fn observe_pair_ranks(clock: &Clock<TickCounter>, text: &Text, moves: &Moves) {
    for dot in text.context().dots() {
        if let Some(locus) = text.store().locus(dot) {
            clock.observe(locus.rank);
        }
    }
    for (_, testimony) in moves.store() {
        clock.observe(testimony.to.rank);
    }
}

fn fold_native(transition: &mut Transition, dots: &[Dot], delta: &OldDelta, stamp: Option<Kairos>) {
    // The bounded-record rule: a redelivered native note still folds
    // (idempotent merge), but only a novel one appends to the native
    // log; without this gate, redelivery and repeated repair rounds
    // would compound the log and every journal built from it.
    let novel = dots
        .iter()
        .any(|&dot| !transition.native_recorded.contains(dot));
    for &dot in dots {
        let _ = transition.native_recorded.insert(dot);
    }
    match delta {
        OldDelta::Text(delta) => transition.native_text.merge_from(delta),
        OldDelta::Moves(delta) => transition.native_moves.merge_from(delta),
    }
    if novel {
        transition
            .native_log
            .push((dots.to_vec(), delta.clone(), stamp));
    }
}