minerva 0.2.0

//! The deterministic fabric: a seed-replayable adversarial transport.
//!
//! Every in-flight note sits in one pool; a seeded generator picks which
//! deliverable note lands next, so any interleaving of text, movement,
//! report, declaration, confirmation, and adoption traffic is reachable
//! from a seed and every failure replays from its seed alone. Duplication
//! re-enqueues an already-delivered note once; a severed station neither
//! receives nor is heard from until healed (messages wait, nothing is
//! lost: the fabric models delay and disorder, and the protocol's own
//! watermarks are what turn "not yet delivered" into "not yet sealed").
//!
//! The chooser has two doors. [`Fabric::step`] draws the envelope from the
//! seed, which is what a scaffolded suite wants (one seed replays one whole
//! interleaving). [`Fabric::step_addressed`] takes the choice from outside,
//! so a byte-holding driver addresses the transport itself; ruling R-71's
//! counterweight named that door as the prerequisite for any wedge class
//! needing envelope-level addressing, and the schedule tape's addressed
//! arm (S331) is its caller.

extern crate alloc;

use alloc::boxed::Box;
use alloc::collections::BTreeMap;
use alloc::collections::BTreeSet;
use alloc::vec::Vec;

use crate::kairos::Kairos;
use crate::metis::{
    Admission, Cut, Declaration, Dot, DotSet, Dotted, EpochAddress, Metatheses, Rhapsody,
};

use super::Replica;

/// A splitmix-style generator: tiny, seedable, and good enough to scatter
/// schedules; determinism is the requirement, statistical quality is not.
pub struct Schedule(u64);

impl Schedule {
    pub const fn new(seed: u64) -> Self {
        Self(seed ^ 0x9E37_79B9_7F4A_7C15)
    }

    pub fn next(&mut self) -> u64 {
        self.0 = self.0.wrapping_add(0x9E37_79B9_7F4A_7C15);
        let mut z = self.0;
        z = (z ^ (z >> 30)).wrapping_mul(0xBF58_476D_1CE4_E5B9);
        z = (z ^ (z >> 27)).wrapping_mul(0x94D0_49BB_1331_11EB);
        z ^ (z >> 31)
    }

    /// A position below `bound` (`bound > 0`).
    pub fn below(&mut self, bound: usize) -> usize {
        let bound = u64::try_from(bound).expect("test-scale bound");
        usize::try_from(self.next() % bound).expect("below a usize bound")
    }

    /// True roughly once in `out_of` draws.
    pub fn one_in(&mut self, out_of: u64) -> bool {
        self.next().is_multiple_of(out_of)
    }
}

/// One old-plane delta: the payload shape both the old-addressed and the
/// native lanes carry (a native delta is simply next-plane traffic). The
/// text pair rides boxed so the note enum stays note-sized.
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum OldDelta {
    Text(Box<Dotted<Rhapsody>>),
    Moves(Dotted<Metatheses>),
}

/// Everything the fleet's stations say to one another. The envelope names
/// its event dots explicitly (the have-set counts events, never context
/// spillover; see the module note) and its plane: `Old` traffic belongs to
/// `generation`'s plane, `Native` traffic to the plane the named epoch
/// founds, which becomes `generation + 1`'s old plane at the seal.
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum Note {
    Old {
        generation: u64,
        dots: Vec<Dot>,
        delta: OldDelta,
    },
    Native {
        epoch: EpochAddress,
        stamp: Kairos,
        dots: Vec<Dot>,
        delta: OldDelta,
    },
    Report {
        generation: u64,
        station: u32,
        cut: Cut,
    },
    Declare {
        declaration: Declaration,
    },
    Confirm {
        generation: u64,
        epoch: EpochAddress,
        station: u32,
        cut: Cut,
    },
    Adoption {
        generation: u64,
        epoch: EpochAddress,
        station: u32,
        counter: u64,
    },
    /// A retirement acknowledgement: `station` testifies it has *applied*
    /// the removal of every dot in `applied`, and `at` is its delivered
    /// floor at acknowledgement time. A receiver counts the acknowledgement
    /// only once its own floor covers `at` (the gathering gate the replica
    /// note records): every write the acker minted while a removed element
    /// was still visible to it precedes the acknowledgement, so it sits at
    /// or below `at`, and delivering `at` first is what proves no in-flight
    /// write can still anchor through a dot the meet would license for
    /// excision.
    Retire {
        generation: u64,
        station: u32,
        applied: DotSet,
        at: Cut,
    },
    /// A departure proposal: `by` testifies that it holds the gap-free
    /// prefix `prefix` of `departing`'s dots in `generation`'s plane, and
    /// promises to admit nothing of `departing`'s above it. The survivors'
    /// join of these is the agreed abandonment bound, which is what the
    /// epoch rounds substitute for the testimony `departing` will never
    /// give (`docs/metis-membership-departure.adoc`).
    Depart {
        generation: u64,
        departing: u32,
        by: u32,
        prefix: u64,
    },
    /// An arrival endorsement: `by`'s authenticated word that the boundary
    /// `admission` may seal, beside the record commitment its round agreed
    /// (PRD 0028 R2/R7). The transported shape of
    /// [`Endorsement`](crate::metis::Endorsement), rebuilt at the receiver
    /// through `Endorsement::from_parts` and bound to its speaker with
    /// `Vouched::trust` --- the same posture every report note takes. The
    /// boundary names its own plane (the admission's base), so the note
    /// carries no generation field.
    Endorse {
        by: u32,
        admission: Admission,
        commitment: [u8; 32],
    },
}

struct Envelope {
    src: u32,
    dst: u32,
    /// A duplicated copy is not duplicated again, so the pool drains.
    replayable: bool,
    note: Note,
}

/// What one station actually *says* to one peer.
///
/// The lying lives in the transport, never in the [`Replica`]: replicas stay
/// composed entirely of shipped parts (the studio discipline), and the fabric
/// asks each nominated station's voice what to put on the wire for each
/// destination. A voice may only reshape traffic its station genuinely
/// emitted; it cannot forge another station's identity, invent traffic from
/// silence, or reach inside a peer.
///
/// A trait rather than a function pointer because the interesting lies are
/// *stateful*. Equivocation is memoryless, but replay must remember a stale
/// note and reordering must buffer one, and those are named in the same
/// adversary family Alma's B8 rung will have to face ("withholds, replays,
/// reorders, and equivocates simultaneously"). `&mut self` is what lets the
/// family grow here rather than be rebuilt later.
pub trait Voice {
    /// What `dst` actually hears in place of `note`.
    fn speak(&mut self, dst: u32, note: &Note) -> Utterance;
}

/// What a [`Voice`] does to one outgoing note.
///
/// Three outcomes rather than two, because suppression is a distinct lie
/// from substitution and a family that cannot express it would quietly bound
/// every finding to talkative adversaries.
pub enum Utterance {
    /// Say the honest thing.
    Truth,
    /// Say this instead.
    Instead(Note),
    /// Say nothing to this peer. The note is simply never queued, which is
    /// selective withholding rather than a severance: other peers still hear
    /// it, so the fleet cannot read the silence as a partition.
    Silence,
}

/// Withholds a whole class of traffic from one peer while telling everyone
/// else the truth.
///
/// Distinct from `Fabric::sever`, which is symmetric, total, and curable by
/// healing. This is one station choosing, per peer and per note kind, not to
/// speak, which no honest failure produces.
pub struct Withholding {
    pub from: u32,
    pub kind: fn(&Note) -> bool,
}

impl Voice for Withholding {
    fn speak(&mut self, dst: u32, note: &Note) -> Utterance {
        if dst == self.from && (self.kind)(note) {
            Utterance::Silence
        } else {
            Utterance::Truth
        }
    }
}

/// Replays a stale note: remembers the first note of a chosen kind and keeps
/// re-speaking it in place of every later one.
///
/// The first genuinely *stateful* lie, and the reason [`Voice`] is a trait
/// rather than a function pointer. A station whose reports freeze at an old
/// value while it keeps emitting is saying something it once truthfully
/// said, which no signature check can refuse.
pub struct Replaying {
    pub kind: fn(&Note) -> bool,
    pub held: Option<Note>,
}

impl Replaying {
    #[must_use]
    pub const fn of(kind: fn(&Note) -> bool) -> Self {
        Self { kind, held: None }
    }
}

impl Voice for Replaying {
    fn speak(&mut self, _dst: u32, note: &Note) -> Utterance {
        if !(self.kind)(note) {
            return Utterance::Truth;
        }
        match &self.held {
            None => {
                self.held = Some(note.clone());
                Utterance::Truth
            }
            Some(stale) => Utterance::Instead(stale.clone()),
        }
    }
}

/// Composes two voices: `first` speaks, and `second` reshapes whatever came
/// out of it.
///
/// B8's adversary "withholds, replays, reorders, and equivocates
/// simultaneously", so the family has to compose or it models one lie at a
/// time and calls it an adversary. `Silence` short-circuits: nothing said
/// cannot be reshaped.
pub struct Then<A, B>(pub A, pub B);

impl<A: Voice, B: Voice> Voice for Then<A, B> {
    fn speak(&mut self, dst: u32, note: &Note) -> Utterance {
        match self.0.speak(dst, note) {
            Utterance::Silence => Utterance::Silence,
            Utterance::Truth => self.1.speak(dst, note),
            Utterance::Instead(spoken) => match self.1.speak(dst, &spoken) {
                Utterance::Truth => Utterance::Instead(spoken),
                other => other,
            },
        }
    }
}

// Reordering is deliberately NOT a `Voice`. The fabric's scheduler already
// reaches every interleaving from a seed, and the addressed tape chooses
// them directly, so a reordering voice would duplicate transport the harness
// owns rather than add adversary power. Withholding and replay are voices
// because no honest schedule produces them.

/// Forks the adoption counter: `to` hears the counter shifted by `delta`,
/// every other peer hears the truth.
///
/// The narrowest lie that reaches the seal record, and the one the exposure
/// table is built on. `delta` is signed because the two directions have
/// *opposite* outcomes, which is the finding: overstating wedges the window
/// (fail-closed), understating diverges the record (fail-open at the record,
/// caught one door later).
pub struct ForkedAdoption {
    pub to: u32,
    pub delta: i64,
}

impl Voice for ForkedAdoption {
    fn speak(&mut self, dst: u32, note: &Note) -> Utterance {
        match note {
            Note::Adoption {
                generation,
                epoch,
                station,
                counter,
            } if dst == self.to => Utterance::Instead(Note::Adoption {
                generation: *generation,
                epoch: *epoch,
                station: *station,
                counter: counter.saturating_add_signed(self.delta),
            }),
            _ => Utterance::Truth,
        }
    }
}

/// Forks a *witnessed cut* on one of the two cut-carrying protocol
/// channels: `to` hears this station's reported or confirmed cut with
/// coordinate `at` shifted by `delta`, and every other peer hears the
/// truth.
///
/// The epoch protocol has three peer-report channels, and until S335 only
/// the third was ever forked. This voice reaches the other two, selected by
/// `kind` in the same idiom [`Withholding`] uses: `Note::Report` is the
/// stability channel the watermark is a meet over, and `Note::Confirm` is
/// the channel the winner latch reads. Shifting a foreign coordinate is the
/// sharp lie in both directions: it claims to have delivered more (or less)
/// of some third station's output than the liar really holds, which is
/// exactly the claim neither channel can check.
pub struct ForkedCut {
    pub to: u32,
    pub kind: fn(&Note) -> bool,
    pub at: u32,
    pub delta: i64,
}

impl ForkedCut {
    fn shift(&self, cut: &Cut) -> Cut {
        let mut shifted = crate::metis::VersionVector::new();
        for (station, counter) in cut.as_vector() {
            let counter = if station == self.at {
                counter.saturating_add_signed(self.delta)
            } else {
                counter
            };
            if counter > 0 {
                shifted.observe(station, counter);
            }
        }
        if self.delta > 0 {
            shifted.observe(
                self.at,
                cut.as_vector()
                    .get(self.at)
                    .saturating_add_signed(self.delta),
            );
        }
        Cut::from_witnessed(shifted)
    }
}

impl Voice for ForkedCut {
    fn speak(&mut self, dst: u32, note: &Note) -> Utterance {
        if dst != self.to || !(self.kind)(note) {
            return Utterance::Truth;
        }
        match note {
            Note::Report {
                generation,
                station,
                cut,
            } => Utterance::Instead(Note::Report {
                generation: *generation,
                station: *station,
                cut: self.shift(cut),
            }),
            Note::Confirm {
                generation,
                epoch,
                station,
                cut,
            } => Utterance::Instead(Note::Confirm {
                generation: *generation,
                epoch: *epoch,
                station: *station,
                cut: self.shift(cut),
            }),
            _ => Utterance::Truth,
        }
    }
}

/// The Byzantine set: which stations lie, how, and under what budget.
///
/// The budget is mechanism, not decoration. Alma's threshold rule derives
/// `Q` from a configured `f`, and every claim the exposure table makes is
/// scoped to a fault set no larger than the one it declares, so the harness
/// refuses to seat more voices than the budget allows rather than letting a
/// test quietly exceed the model it cites.
pub struct Equivocator {
    voices: BTreeMap<u32, Box<dyn Voice>>,
    budget: usize,
}

impl Equivocator {
    /// A fault set of at most `budget` stations.
    pub const fn within(budget: usize) -> Self {
        Self {
            voices: BTreeMap::new(),
            budget,
        }
    }

    /// Seats `voice` at `station`.
    ///
    /// # Panics
    ///
    /// When the fault set would exceed its declared budget: a harness that
    /// silently over-corrupted would produce findings outside the model it
    /// claims, which is worse than no finding.
    #[must_use]
    pub fn speaking(mut self, station: u32, voice: Box<dyn Voice>) -> Self {
        assert!(
            self.voices.len() < self.budget,
            "the fault set may not exceed its declared budget of {}",
            self.budget
        );
        // Counting map size alone would let a second seat at the same
        // station replace the first without growing the set, so an
        // authoring slip could corrupt one station while the budget still
        // read as spent elsewhere.
        assert!(
            self.voices.insert(station, voice).is_none(),
            "station {station} is already seated"
        );
        self
    }

    fn speak(&mut self, src: u32, dst: u32, note: &Note) -> Utterance {
        self.voices
            .get_mut(&src)
            .map_or(Utterance::Truth, |voice| voice.speak(dst, note))
    }
}

/// The fleet's transport and scheduler.
pub struct Fabric {
    schedule: Schedule,
    roster: Vec<u32>,
    pool: Vec<Envelope>,
    severed: BTreeSet<u32>,
    /// Zero disables duplication; `n` re-enqueues one delivered note in `n`.
    duplicate_one_in: u64,
    /// `None` for every honest arm, which is every arm but `byzantine`.
    equivocator: Option<Equivocator>,
}

impl Fabric {
    pub fn new(seed: u64, roster: &[u32], duplicate_one_in: u64) -> Self {
        Self {
            schedule: Schedule::new(seed),
            roster: roster.to_vec(),
            pool: Vec::new(),
            severed: BTreeSet::new(),
            duplicate_one_in,
            equivocator: None,
        }
    }

    /// Seats a Byzantine fault set on this fabric.
    ///
    /// Only the `byzantine` exposure module calls this. Every other arm
    /// leaves the fabric honest, so the crash-only model the charter states
    /// is what every other law is still proven under.
    pub fn corrupt(&mut self, equivocator: Equivocator) {
        self.equivocator = Some(equivocator);
    }

    pub fn schedule(&mut self) -> &mut Schedule {
        &mut self.schedule
    }

    /// Queues `note` from `src` to every other roster member, reshaped per
    /// destination when `src` is a seated voice.
    pub fn broadcast(&mut self, src: u32, note: &Note) {
        for &dst in &self.roster {
            if dst != src {
                let spoken = match self
                    .equivocator
                    .as_mut()
                    .map_or(Utterance::Truth, |seated| seated.speak(src, dst, note))
                {
                    Utterance::Truth => note.clone(),
                    Utterance::Instead(spoken) => spoken,
                    // Withheld from this peer only: no envelope is queued, so
                    // the fleet sees silence it cannot distinguish from delay.
                    Utterance::Silence => continue,
                };
                self.pool.push(Envelope {
                    src,
                    dst,
                    replayable: true,
                    note: spoken,
                });
            }
        }
    }

    /// Queues everything a replica emitted.
    pub fn post(&mut self, src: u32, outbox: Vec<Note>) {
        for note in outbox {
            self.broadcast(src, &note);
        }
    }

    /// Queues one targeted note: the repair lane's transport. Resync
    /// traffic rides the same scheduled pool as everything else, so a
    /// repair interleaves adversarially with live traffic like any other
    /// delivery.
    pub fn send(&mut self, src: u32, dst: u32, note: &Note) {
        // The repair lane speaks with the same voice as the broadcast lane.
        // A Byzantine station has no reason to turn honest when re-serving
        // its record to a restarted peer, and leaving this door truthful
        // would silently bound every future finding to fault-free repair.
        let spoken = match self
            .equivocator
            .as_mut()
            .map_or(Utterance::Truth, |seated| seated.speak(src, dst, note))
        {
            Utterance::Truth => note.clone(),
            Utterance::Instead(spoken) => spoken,
            Utterance::Silence => return,
        };
        self.pool.push(Envelope {
            src,
            dst,
            replayable: true,
            note: spoken,
        });
    }

    /// Adds an admitted station to the transport's roster, so broadcasts
    /// reach it from now on. The fabric's half of a membership admission:
    /// the *record's* half is the sealed admission (PRD 0028 R1), and the
    /// driver calls this once the joiner's replica is installed.
    pub fn admit(&mut self, station: u32) {
        assert!(
            !self.roster.contains(&station),
            "station {station} is already on the fabric's roster"
        );
        self.roster.push(station);
    }

    /// Cuts the named stations off in both directions; their traffic waits.
    pub fn sever(&mut self, stations: &[u32]) {
        self.severed.extend(stations.iter().copied());
    }

    /// Restores full connectivity.
    pub fn heal(&mut self) {
        self.severed.clear();
    }

    pub fn in_flight(&self) -> usize {
        self.pool.len()
    }

    /// Delivers one schedulable note to its replica and posts whatever that
    /// replica said in response. Returns `false` when nothing can move
    /// (empty pool, or everything waits behind a severance).
    ///
    /// The envelope is drawn from the seeded schedule, which is what every
    /// scaffolded suite and every pinned seed wants: the seed replays the
    /// whole interleaving.
    pub fn step(&mut self, fleet: &mut BTreeMap<u32, Replica>) -> bool {
        self.step_addressed(fleet, None)
    }

    /// [`Fabric::step`] with the choice of envelope optionally *addressed*
    /// from outside: `Some(byte)` indexes the deliverable set directly, so
    /// a driver holding bytes (the schedule tape, S331) chooses *which*
    /// note moves rather than only *when* a note moves. `None` restores
    /// the seeded draw exactly, so the addressing door costs the pinned
    /// seeds nothing.
    ///
    /// Duplication stays seeded under both: the tape addresses transport
    /// order, and re-enqueueing an already-delivered copy is a fabric
    /// fault, not an ordering choice.
    pub fn step_addressed(
        &mut self,
        fleet: &mut BTreeMap<u32, Replica>,
        address: Option<u8>,
    ) -> bool {
        let deliverable: Vec<usize> = self
            .pool
            .iter()
            .enumerate()
            .filter(|(_, envelope)| {
                !self.severed.contains(&envelope.src) && !self.severed.contains(&envelope.dst)
            })
            .map(|(index, _)| index)
            .collect();
        if deliverable.is_empty() {
            return false;
        }
        let index = address.map_or_else(
            || self.schedule.below(deliverable.len()),
            |byte| usize::from(byte) % deliverable.len(),
        );
        let chosen = deliverable[index];
        let envelope = self.pool.swap_remove(chosen);
        if envelope.replayable
            && self.duplicate_one_in > 0
            && self.schedule.one_in(self.duplicate_one_in)
        {
            self.pool.push(Envelope {
                src: envelope.src,
                dst: envelope.dst,
                replayable: false,
                note: envelope.note.clone(),
            });
        }
        let replica = fleet
            .get_mut(&envelope.dst)
            .expect("every envelope addresses a roster member");
        let mut outbox = Vec::new();
        replica.handle(&envelope.note, &mut outbox);
        self.post(envelope.dst, outbox);
        true
    }

    /// [`Fabric::step`] with the envelope chosen by a *predicate* over its
    /// destination and note rather than by seed or address: the third
    /// addressing mode, and the one a decision-point sweep needs.
    ///
    /// `step` and `step_addressed` both draw from the whole deliverable
    /// set, so neither can withhold traffic by kind. The latch-cut sweep
    /// must say "deliver this declaration to this station, and nothing else
    /// yet" in order to place a replica's latch at a chosen delivered set,
    /// which is the only order-sensitive variable the seal record has (the
    /// working note in `docs/miscellany/`). The first match in the pool's
    /// *current* order is taken, with no draw, so the door is deterministic;
    /// it is not enqueue-ordered, because `swap_remove` moves the tail
    /// element into each vacated slot and perturbs the order across
    /// successive calls. Nothing here depends on which matching envelope
    /// moves first: the caller narrows by predicate, and `drain_selected`
    /// delivers the whole matching set either way. Determinism is the
    /// claim; enqueue order is not.
    ///
    /// Returns `false` when nothing deliverable matches, which the sweep
    /// reads as "this phase is done", not as an error. The seeded and
    /// addressed doors are untouched, so no pinned seed moves.
    ///
    /// Duplication is off on this door, unlike the other two: a re-enqueued
    /// copy is a fabric fault, and the sweep's claim is over the decision
    /// space, not over transport faults. Redelivery already has its own
    /// pins.
    pub fn step_selected(
        &mut self,
        fleet: &mut BTreeMap<u32, Replica>,
        mut wanted: impl FnMut(u32, &Note) -> bool,
    ) -> bool {
        let Some(chosen) = self.pool.iter().position(|envelope| {
            !self.severed.contains(&envelope.src)
                && !self.severed.contains(&envelope.dst)
                && wanted(envelope.dst, &envelope.note)
        }) else {
            return false;
        };
        let envelope = self.pool.swap_remove(chosen);
        let replica = fleet
            .get_mut(&envelope.dst)
            .expect("every envelope addresses a roster member");
        let mut outbox = Vec::new();
        replica.handle(&envelope.note, &mut outbox);
        self.post(envelope.dst, outbox);
        true
    }

    /// Runs [`Fabric::step_selected`] until nothing deliverable matches.
    pub fn drain_selected(
        &mut self,
        fleet: &mut BTreeMap<u32, Replica>,
        mut wanted: impl FnMut(u32, &Note) -> bool,
    ) {
        while self.step_selected(fleet, &mut wanted) {}
    }

    /// Runs the fabric dry. With no severance the pool empties; under a
    /// severance this stops at the blocked residue.
    pub fn drain(&mut self, fleet: &mut BTreeMap<u32, Replica>) {
        while self.step(fleet) {}
    }
}