minerva 0.2.0

//! The membership schedule tape: the recorded-membership machinery under
//! a byte-driven adversary (R-90's named future work, taken at S346).
//!
//! The epoch schedule tape drives the lifecycle over a fixed roster; this
//! sibling drives the roster itself. Two ops join the alphabet: an
//! operator's admission (`open_arrival` at one member --- deliberately not
//! at all members, so partial rounds, parked endorsements, and racing
//! declarations are the common case, not the exception) and an operator's
//! eviction (sever, then every survivor abandons and opens the departure
//! round). A joiner installs through the shipped carriage the moment any
//! member's retained record admits it: re-stamped checkpoint, lineage
//! proof, `bootstrap`, repair lane --- under whatever severance and
//! traffic the tape has in flight.
//!
//! A sibling, not an amendment. The base tape's bytes are frozen
//! reproducers (its pinned catches and its libFuzzer corpus must keep
//! their meaning), so a new op alphabet gets a new interpreter --- the
//! same discipline that gives a wire frame a sibling version instead of
//! an in-place evolution.
//!
//! Eviction is permanent by the recorded model: an attested member is
//! gone (the exhibits' posture --- an attested-but-alive member that
//! seals past its eviction is outside the crash-only model), so a heal
//! op restores every link *except* the evicted members', and the closing
//! laws range over the survivors and installed joiners.
//!
//! After the tape: heal (minus the evicted), drain, install any joiner
//! whose admission sealed during the drain, drain again, and assert the
//! fleet-wide laws --- convergence, quiescence, and the membership laws
//! (every survivor's roster carries every installed joiner; every
//! survivor holds the attestation of every completed eviction).

extern crate alloc;

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

#[cfg(test)]
use proptest::prelude::*;

use super::fabric::Fabric;
use super::replica::Replica;
use super::tape::{Cursor, Delivery, assert_quiescent};
use super::{act, assert_converged, crash, fleet_of, lineage_of, resync};

/// The founding roster, the base tape's own.
const ROSTER: [u32; 3] = [1, 2, 3];

/// The op cap: admissions and evictions are heavier than lifecycle ops,
/// so the membership tape runs shorter programs than the base tape's 48.
const MAX_OPS: usize = 40;

/// The admittable pool. Two candidates bound the fleet at five members
/// and give the mutator a second admission to race against the first.
const CANDIDATES: [u32; 2] = [4, 5];

/// The retained-lineage depth every member runs.
fn horizon() -> NonZeroUsize {
    NonZeroUsize::new(2).expect("positive")
}

/// The live members: every replica in the fleet that has not been
/// evicted. Ops select their station from this set, so a tape byte
/// always names a member that can lawfully act.
fn live(fleet: &BTreeMap<u32, Replica>, evicted: &BTreeSet<u32>) -> Vec<u32> {
    fleet
        .keys()
        .copied()
        .filter(|station| !evicted.contains(station))
        .collect()
}

/// The current roster of `replica`, as the caller's configured slice.
fn roster_of(replica: &Replica) -> Vec<u32> {
    replica.epochs().roster().collect()
}

/// The oldest retained generation's roster: the ground `bootstrap` takes
/// (PRD 0028 R4). The current roster minus every admission a retained
/// seal carries --- admissions are disjoint (a station admits once), so
/// the subtraction is exact.
fn oldest_ground(replica: &Replica) -> Vec<u32> {
    let epochs = replica.epochs();
    let admitted: BTreeSet<u32> = epochs
        .sealed()
        .filter_map(|sealed| sealed.admission())
        .flat_map(crate::metis::Admission::joiners)
        .collect();
    epochs
        .roster()
        .filter(|station| !admitted.contains(station))
        .collect()
}

/// Installs every candidate some live member's retained record admits
/// and the fleet does not yet run, through the shipped carriage: the
/// fabric learns the station, an incumbent at its seal re-stamps a
/// checkpoint, the joiner bootstraps from the very proof that admits it,
/// and the repair lane serves it into the live traffic. Returns whether
/// anything installed (the closing drain loops until this goes quiet).
///
/// `joiner_checkpoint_for` returning `None` is lawful (the incumbent has
/// state the re-stamp cannot yet speak for); the install retries on a
/// later sweep. A `bootstrap` refusal from an honest incumbent's own
/// checkpoint and proof is *not* lawful --- that is the finding this
/// tape exists to catch, so it panics with the tape as reproducer.
fn install_admitted(
    fabric: &mut Fabric,
    fleet: &mut BTreeMap<u32, Replica>,
    evicted: &BTreeSet<u32>,
) -> bool {
    let mut installed = false;
    for candidate in CANDIDATES {
        if fleet.contains_key(&candidate) {
            continue;
        }
        let sponsor = live(fleet, evicted).into_iter().find(|station| {
            fleet[station]
                .epochs()
                .sealed()
                .filter_map(|sealed| sealed.admission())
                .any(|admission| admission.admits(candidate))
        });
        let Some(sponsor) = sponsor else {
            continue;
        };
        let Some(checkpoint) = fleet[&sponsor].joiner_checkpoint_for(candidate) else {
            continue;
        };
        let ground = oldest_ground(&fleet[&sponsor]);
        let lineage = lineage_of(&fleet[&sponsor]);
        let joiner = Replica::bootstrap(checkpoint, &ground, horizon(), &lineage)
            .expect("an admitted joiner bootstraps from the proof that admits it");
        fabric.admit(candidate);
        let _ = fleet.insert(candidate, joiner);
        resync(fabric, fleet, candidate);
        installed = true;
    }
    installed
}

/// Interprets a byte tape over a fleet whose membership moves.
///
/// The first eight bytes seed the fabric; the rest are `(opcode,
/// argument)` pairs capped at [`MAX_OPS`], addresses reading from the
/// tape's second end exactly as the base tape's addressed arm. `opcode %
/// 10` selects the op; the argument selects operands over the *live*
/// member set:
///
/// * `0`--`7`: the base tape's alphabet --- insert, delete, move,
///   declare, step, sever, heal, crash --- over the live members, with
///   two membership-aware differences: heal restores every link except
///   an evicted member's, and a crash rehydrates against the crashing
///   replica's own current roster (after a widening seal the configured
///   roster is the record's, the recipe's own rebuild rule).
/// * `8` Admit: the selected member's operator opens the arrival for the
///   next uninstalled candidate and endorses it. One member, not all:
///   further opens come from further `8`s, so incomplete rounds, parked
///   peer endorsements, and `NoRound` schedules are the default. Every
///   refusal is a lawful outcome and absorbed.
/// * `9` Evict: every survivor opens the departure round, then the selected
///   victim is cut off for good. Before the cut, the fabric heals and
///   drains: the departure door's `Ragged` refusal makes that repair a
///   caller duty, not an adversarial schedule. The proposals themselves
///   remain in flight, so their attestation still races later tape
///   operations. Guarded to keep at least two survivors, so departure
///   rounds stay multi-party and convergence stays a claim about a fleet.
///
/// After every op, any admission a retained record carries installs its
/// joiner through the shipped carriage ([`install_admitted`]).
#[allow(clippy::too_many_lines)]
fn interpret_with(tape: &[u8], delivery: Delivery) {
    let mut seed = [0u8; 8];
    for (slot, byte) in seed.iter_mut().zip(tape.iter()) {
        *slot = *byte;
    }
    let mut cursor = Cursor::new(tape);

    let mut fleet = fleet_of(&ROSTER, horizon());
    let mut fabric = Fabric::new(u64::from_le_bytes(seed), &ROSTER, 7);
    let mut evicted: BTreeSet<u32> = BTreeSet::new();

    for _ in 0..MAX_OPS {
        let Some((opcode, argument)) = cursor.op() else {
            break;
        };
        let members = live(&fleet, &evicted);
        let station = members[usize::from(argument) % members.len()];
        let at = (usize::from(argument) / 3) % 8;
        let to = (usize::from(argument) / 24) % 8;
        match opcode % 10 {
            0 => {
                let _ = act(&mut fabric, &mut fleet, station, |replica, out| {
                    replica.insert_visible(at, out)
                });
            }
            1 => {
                let _ = act(&mut fabric, &mut fleet, station, |replica, out| {
                    replica.delete_visible(at, out)
                });
            }
            2 => {
                // The recipe's deferral duty, as the base tape honors it:
                // an adopted replica edits natively instead of minting a
                // movement testimony.
                if fleet[&station].adopted() {
                    let _ = act(&mut fabric, &mut fleet, station, |replica, out| {
                        replica.delete_visible(at, out)
                    });
                } else {
                    let _ = act(&mut fabric, &mut fleet, station, |replica, out| {
                        replica.move_visible(at, Some(to), out)
                    });
                }
            }
            3 => {
                let _ = act(&mut fabric, &mut fleet, station, |replica, out| {
                    replica.try_declare(out)
                });
            }
            4 => {
                let steps = usize::from(argument) % 32 + 1;
                for _ in 0..steps {
                    let address = match delivery {
                        Delivery::Seeded => None,
                        Delivery::Addressed => cursor.address(),
                    };
                    let _ = fabric.step_addressed(&mut fleet, address);
                }
            }
            5 => fabric.sever(&[station]),
            6 => {
                fabric.heal();
                let gone: Vec<u32> = evicted.iter().copied().collect();
                fabric.sever(&gone);
            }
            7 => {
                let roster = roster_of(&fleet[&station]);
                crash(&mut fleet, &roster, horizon(), station);
                resync(&mut fabric, &fleet, station);
            }
            8 => {
                if let Some(candidate) = CANDIDATES
                    .iter()
                    .copied()
                    .find(|candidate| !fleet.contains_key(candidate))
                {
                    // Acceptance and refusal are both lawful: Unfounded,
                    // Premature, AlreadyPresent, a competing round --- the
                    // machine's vocabulary is the shield, exactly as the
                    // base tape treats a mid-traffic declaration.
                    let _ = act(&mut fabric, &mut fleet, station, |replica, out| {
                        replica.open_arrival(&[candidate], out)
                    });
                }
            }
            _ => {
                if members.len() > 2 {
                    // `Ragged` is a refusal before mutation and explicitly
                    // assigns repair to the caller. Satisfy that duty before
                    // silencing the victim; the Depart notes emitted below
                    // are left scheduled, preserving the attestation race.
                    fabric.heal();
                    let gone: Vec<u32> = evicted.iter().copied().collect();
                    fabric.sever(&gone);
                    fabric.drain(&mut fleet);
                    while install_admitted(&mut fabric, &mut fleet, &evicted) {
                        fabric.drain(&mut fleet);
                    }

                    let current = live(&fleet, &evicted);
                    // Do not overlap two incomplete evictions: a departure
                    // round's family is the survivors known when it opens.
                    // Also wait until every recorded joiner is installed;
                    // otherwise the round would name a roster member with no
                    // replica to speak its proposal.
                    let prior_complete = evicted.iter().all(|&prior| {
                        current
                            .iter()
                            .all(|member| fleet[member].attested(prior).is_some())
                    });
                    let expected: BTreeSet<u32> = current.iter().copied().collect();
                    let roster_ready = current.iter().all(|member| {
                        roster_of(&fleet[member])
                            .into_iter()
                            .filter(|station| !evicted.contains(station))
                            .collect::<BTreeSet<_>>()
                            == expected
                    });
                    let transition_open = current.iter().any(|member| fleet[member].adopted());
                    let survivors: Vec<u32> = current
                        .into_iter()
                        .filter(|&member| member != station)
                        .collect();
                    if survivors.len() >= 2 && prior_complete && roster_ready && !transition_open {
                        for &survivor in &survivors {
                            let _ = act(&mut fabric, &mut fleet, survivor, |replica, out| {
                                replica.open_departure(station, out)
                            })
                            .expect("the repaired survivor proposes a gap-free departure prefix");
                        }
                        let _ = evicted.insert(station);
                        fabric.sever(&[station]);
                    }
                }
            }
        }
        let _ = install_admitted(&mut fabric, &mut fleet, &evicted);
    }

    // Settle: reconnect the survivors (never the evicted) and drive the
    // tape's pending membership decisions to their records, as a real
    // fleet's operators would. The arrival round is operator-local state
    // (each member's round carries its own voice), so convergence is a
    // claim about a fleet whose rounds have been consumed by a seal or
    // superseded --- the exhibits' own posture. Each settle pass:
    //
    // 1. drain;
    // 2. retry every survivor's departure round for every eviction not
    //    yet attested everywhere (a ragged holding heals as traffic
    //    drains, and the retry needs no separate lane);
    // 3. for the first candidate that is admitted-but-uninstalled or
    //    named by any member's open round: every operator opens (the
    //    decision reaches every operator eventually --- a completed round
    //    IS every operator's word), every member writes once, and each
    //    member offers a declaration --- a complete round tags, a frozen
    //    round rides plain and dies by supersession, and the re-open on
    //    the next pass completes over the surviving family;
    // 4. install whatever the record now admits.
    //
    // The pass count is bounded: two candidates, each worth at most a
    // freeze-supersede-readmit cycle, plus eviction retries. A fleet
    // still holding an open round, an uninstalled admission, or a
    // half-attested eviction after the bound is wedged, and the tape is
    // the reproducer.
    fabric.heal();
    let gone: Vec<u32> = evicted.iter().copied().collect();
    fabric.sever(&gone);
    for _ in 0..8 {
        fabric.drain(&mut fleet);
        // The consumer's anti-entropy duty, run among the survivors: an
        // evicted member's unflushed sends are lost with it, so a
        // survivor that held its traffic re-serves the others up to the
        // agreed bound. Without this lane two survivors lawfully diverge
        // on the departed station's data forever --- the tape's first
        // catch, and the recipe's own reason the duty exists.
        let wedged = live(&fleet, &evicted)
            .iter()
            .any(|member| fleet[member].epochs().candidates().count() > 0);
        if !evicted.is_empty() || wedged {
            for member in live(&fleet, &evicted) {
                resync(&mut fabric, &fleet, member);
            }
            fabric.drain(&mut fleet);
        }
        let members = live(&fleet, &evicted);
        for &station in &evicted {
            let unattested = members
                .iter()
                .any(|member| fleet[member].attested(station).is_none());
            if unattested {
                for &survivor in &members {
                    let _ = act(&mut fabric, &mut fleet, survivor, |replica, out| {
                        replica.open_departure(station, out)
                    });
                }
                fabric.drain(&mut fleet);
            }
        }
        let pending = CANDIDATES.iter().copied().find(|&candidate| {
            !fleet.contains_key(&candidate)
                && live(&fleet, &evicted).iter().any(|member| {
                    let held = fleet[member]
                        .epochs()
                        .arrival()
                        .is_some_and(|round| round.admission().admits(candidate));
                    let admitted = fleet[member]
                        .epochs()
                        .sealed()
                        .filter_map(|sealed| sealed.admission())
                        .any(|admission| admission.admits(candidate));
                    held || admitted
                })
        });
        let _ = install_admitted(&mut fabric, &mut fleet, &evicted);
        let Some(candidate) = pending else {
            let members = live(&fleet, &evicted);
            let open_rounds = members
                .iter()
                .any(|member| fleet[member].epochs().arrival().is_some());
            let open_windows = members
                .iter()
                .any(|member| fleet[member].epochs().candidates().count() > 0);
            let unattested = evicted.iter().any(|&gone| {
                members
                    .iter()
                    .any(|member| fleet[member].attested(gone).is_none())
            });
            if open_rounds {
                // A round over a boundary whose joiner is already
                // installed (or a stale base): a plain boundary crossing
                // supersedes it.
                settle_boundary(&mut fabric, &mut fleet, &evicted);
            }
            if !open_rounds && !open_windows && !unattested {
                break;
            }
            // An open window or a pending attestation needs cadence, not
            // new work: the next pass's repair lane re-states risen
            // floors and round positions, and the drain polls the
            // machines forward. A state that survives every pass reaches
            // the closing asserts as the wedge it is.
            continue;
        };
        for &member in &live(&fleet, &evicted) {
            let _ = act(&mut fabric, &mut fleet, member, |replica, out| {
                replica.open_arrival(&[candidate], out)
            });
        }
        settle_boundary(&mut fabric, &mut fleet, &evicted);
        let _ = install_admitted(&mut fabric, &mut fleet, &evicted);
    }
    fabric.drain(&mut fleet);
    while install_admitted(&mut fabric, &mut fleet, &evicted) {
        fabric.drain(&mut fleet);
    }

    // The evicted are gone by the recorded model; the laws range over
    // the members that remain.
    for station in &evicted {
        let _ = fleet.remove(station);
    }
    let survivors: Vec<u32> = fleet.keys().copied().collect();
    for &station in &survivors {
        assert!(
            fleet[&station].epochs().arrival().is_none(),
            "station {station}: an arrival round survived the settle --- a wedge",
        );
    }
    assert_converged(&fleet);
    assert_quiescent(&fleet);

    // The membership laws. Every installed joiner is on every survivor's
    // roster (the seal that admitted it widened every member
    // identically), and every eviction is attested at every survivor
    // (the departure round sealed fleet-wide; the settle retried it) ---
    // half-applied membership is the divergence this tape exists to
    // catch.
    for &station in &survivors {
        let roster: BTreeSet<u32> = fleet[&station].epochs().roster().collect();
        for candidate in CANDIDATES {
            if fleet.contains_key(&candidate) {
                assert!(
                    roster.contains(&candidate),
                    "station {station}: installed joiner {candidate} is missing from the roster"
                );
            }
        }
        for &gone in &evicted {
            assert!(
                fleet[&station].attested(gone).is_some(),
                "station {station}: eviction of {gone} never attested --- a wedge",
            );
        }
    }
}

/// One boundary crossing during the settle: every live member writes
/// once, the traffic drains, each member offers a declaration (a Decline
/// is lawful; a member holding a complete round tags), and the traffic
/// drains again --- `seal_once`'s shape, tolerant of refusals.
fn settle_boundary(
    fabric: &mut Fabric,
    fleet: &mut BTreeMap<u32, Replica>,
    evicted: &BTreeSet<u32>,
) {
    let members = live(fleet, evicted);
    for &member in &members {
        let _ = act(fabric, fleet, member, |replica, out| {
            replica.insert_visible(0, out)
        });
    }
    fabric.drain(fleet);
    for &member in &members {
        let _ = act(fabric, fleet, member, Replica::try_declare);
    }
    fabric.drain(fleet);
}

/// The libFuzzer leg's entry point: the addressed arm, for the same
/// reason the base tape's leg runs it --- the mutator's edge is steering
/// which envelope moves, and membership schedules (a parked endorsement
/// racing its round's completion, an attestation racing a tag) are
/// envelope-order schedules.
pub(in crate::metis) fn run_membership_tape(tape: &[u8]) {
    interpret_with(tape, Delivery::Addressed);
}

#[cfg(test)]
#[test]
fn a_directed_admission_and_eviction_survive_the_tape() {
    // A hand-laid program: settle, admit 4 (all three operators), step,
    // evict 3, step, then let the closing drain finish whatever the ops
    // left open. Smoke, not schedule: the adversarial schedules are the
    // property's and the fuzzer's job.
    let mut tape = alloc::vec![7u8, 3, 9, 1, 4, 200];
    tape.extend_from_slice(&[0, 0, 0, 1, 0, 2, 4, 255, 3, 0, 4, 255, 4, 255]);
    tape.extend_from_slice(&[8, 0, 8, 1, 8, 2, 4, 255, 3, 0, 4, 255, 4, 255]);
    tape.extend_from_slice(&[9, 2, 4, 255, 3, 0, 4, 255, 4, 255]);
    interpret_with(&tape, Delivery::Seeded);
}

#[cfg(test)]
proptest! {
    #![proptest_config(ProptestConfig::with_cases(48))]

    /// The membership law under arbitrary bytes: any tape of edits,
    /// declarations, partial admissions, evictions, severance, and
    /// crashes heals, drains, installs what the record admits, and
    /// converges over the surviving membership.
    #[test]
    fn any_membership_tape_heals_drains_and_converges(
        tape in prop::collection::vec(any::<u8>(), 0..96),
    ) {
        interpret_with(&tape, Delivery::Seeded);
    }

    /// The same law under the addressed arm: the in-tree gate for the
    /// door the libFuzzer leg drives, so the fuzz target is never the
    /// only thing exercising it.
    #[test]
    fn any_addressed_membership_tape_heals_drains_and_converges(
        tape in prop::collection::vec(any::<u8>(), 0..96),
    ) {
        run_membership_tape(&tape);
    }
}