minerva 0.2.0

//! The boundary-schedule note's pinned laws (S282, extended S285): the
//! lag lemma made executable, the declaration-oath shape (the fourth
//! soundness shape for a continuously maintained cut-state) carried to
//! its proof, and the oath meet's policy axis (chooser scoping, the
//! note's section 6) pinned beside it: the full-roster meet is the
//! largest chooser-agnostic eager bound, the declarer's own held join
//! is the kernel every scoped bound inherits, and a scoped meet's
//! out-of-scope miss is the base seam's detected rebuild, never a
//! wrong stratum.
//!
//! The note (`docs/metis-boundary-schedule.adoc`) owns the derivations;
//! this module owns the executables. Eighteen directed pins drive the
//! *shipped* doors (the [`EpochPreparation`] base seam is the exact
//! overshoot detector, so every counterexample lands as its
//! `UnpinnedBase` refusal) and the modeled fleet, three generated laws
//! sweep the epoch lifecycle, the maximality family, and the relay
//! under-claim lemma (section 7), and each of
//! the thirteen defenses' mutants dies deterministically under a
//! fixed schedule. The laws are conditional
//! exactly where the note says: the generation-scoped meet of
//! per-member advertised watermarks (join-folded per emitter, read
//! only when every roster member has advertised within the current
//! generation, reset at each seal exactly as `Stability` is, stale
//! generations excluded at the fold door because the reborn
//! coordinate space reuses counter numerals) bounds every
//! same-generation declaration wherever the own-station stamp fence
//! passes, and where delivery order breaks (nothing grants it; the
//! skewed delivery op breaks it at will, and the replay op models the
//! cross-session redelivery that outlives any one ordered session) the
//! fence detects before any fold is trusted, so an overshoot is never
//! silent.
//! No mechanism ships with this module; the modeled advertisement is a
//! plain [`VersionVector`] beside the shipped types it composes with.

extern crate alloc;

mod model;
mod shipped;

use alloc::vec::Vec;
use core::array;
use core::cmp::Ordering;
use core::num::NonZeroUsize;

use proptest::prelude::*;

use crate::kairos::Kairos;
use crate::metis::dot::RawDot;
use crate::metis::{
    Anchor, Cut, Dot, DotSet, Dotted, EpochPreparation, EpochPreparationMiss, Epochs, Locus,
    Metatheses, Metathesis, Rhapsody, Stability, VersionVector,
};

/// Builds an identity literal. Panics on the non-dot counter zero (R-91).
#[track_caller]
fn d(station: u32, counter: u64) -> Dot {
    Dot::from_parts(station, counter).expect("test literal names the non-dot counter zero")
}

type Text = Dotted<Rhapsody>;
type Moves = Dotted<Metatheses>;

fn rank(physical_ns: u64) -> Kairos {
    Kairos::new(physical_ns, 0, 1, 0u16)
}

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

fn text_delta(dot: Dot, anchor: Anchor, at: Kairos) -> Text {
    let mut text = Rhapsody::new();
    assert!(text.weave(dot, Locus { anchor, rank: at }));
    Dotted::from_store(text)
}

fn move_delta(testimony: Dot, target: (u32, u64), at: Kairos) -> Moves {
    Dotted::from_store(Metatheses::singleton(
        testimony,
        Metathesis {
            target: target.into(),
            to: Locus {
                anchor: Anchor::Origin,
                rank: at,
            },
        },
    ))
}

/// The S285 maximality exhibit (the RQ3 brief's two-participant
/// counterexample, driven through the shipped doors): the full-roster
/// oath meet is the largest chooser-agnostic eager bound. Sufficiency
/// is the S282 law (every fence-passing declaration is bounded by
/// it); maximality at family grade is the generated law below
/// (`prop_any_bound_strictly_above_the_roster_meet_is_undershot`,
/// the universal property: a bound strictly above the meet exceeds
/// some member's oath, and that member's lawful declaration at its
/// oath undershoots the bound). This pin is that law's smallest
/// exhibit: the bound is one member's whole oath, the member whose
/// oath excludes it declares, and the shipped seam refuses the
/// advanced base with the exact first offender. The refusal is the
/// price tag, not a soundness hole: safety against a declarer whose
/// oath never bounded the base was never the bound's claim, and the
/// base comparison detects the miss on data already in hand.
#[test]
fn the_full_roster_meet_is_the_largest_chooser_agnostic_eager_bound() {
    // Two members over two stations. Member 1's oath (its held join)
    // sits high, member 2's sits low; the full meet is member 2's.
    let oath_one = vector(&[(1, 4), (2, 4)]);
    let oath_two = vector(&[(1, 1), (2, 1)]);
    let full_meet = oath_one.meet(&oath_two);
    assert_eq!(full_meet, oath_two);

    // Member 2 lawfully declares at (1, 2), (2, 2): above its own
    // oath (the declare door's only obligation), below member 1's.
    let mut stability = Stability::new([1, 2]);
    let declared = Cut::from_witnessed(vector(&[(1, 2), (2, 2)]));
    stability.report_cut(1, &declared).unwrap();
    stability.report_cut(2, &declared).unwrap();
    let mut epochs = Epochs::new([1, 2], NonZeroUsize::new(1).unwrap());
    let declaration = epochs
        .declare(d(2, 3), rank(3), &stability, &Cut::bottom())
        .unwrap();
    assert!(&oath_two <= declaration.cut().as_vector());

    // The meet scoped to member 1 alone sits above that cut: a
    // trailing base advanced to it is unusable for this boundary, and
    // the seam names the first offender.
    assert!(declaration.cut().as_vector().happens_before(&oath_one));
    let trailing = Text::from_context(Cut::from_witnessed(oath_one).to_have_set());
    assert_eq!(
        EpochPreparation::begin(&declaration, trailing, Moves::new()).unwrap_err(),
        EpochPreparationMiss::UnpinnedBase {
            dot: RawDot::new(1, 4)
        }
    );

    // The full meet serves the same declaration: chooser-agnostic,
    // and no strictly larger vector is, because a strict rise above
    // the meet rises above some member's oath, which that member may
    // lawfully undershoot exactly as member 2 just did.
    let trailing = Text::from_context(Cut::from_witnessed(full_meet).to_have_set());
    assert!(EpochPreparation::begin(&declaration, trailing, Moves::new()).is_ok());
}

/// A generated oath: an advertised watermark over the three shared
/// stations, empty included (a member that has advertised only the
/// bottom still gates the read).
fn arb_oath() -> impl Strategy<Value = VersionVector> {
    prop::collection::btree_map(1u32..=3, 1u64..=6, 0..=3).prop_map(|counts| {
        let mut oath = VersionVector::new();
        for (station, count) in counts {
            oath.observe(station, count);
        }
        oath
    })
}

proptest! {
    #![proptest_config(ProptestConfig::with_cases(256))]

    /// The maximality law at family grade (the S285 review's first
    /// finding, absorbed): over any oath family and any bound
    /// strictly above its meet, some member's oath excludes the bound
    /// (the meet's universal property, asserted executable), that
    /// member's lawful declaration at exactly its oath is refused
    /// against a base advanced to the bound (the shipped seam, the
    /// detected rebuild), and the same declaration is served from the
    /// full meet (sufficiency, the chooser-agnostic direction). The
    /// directed two-member pin above is this law's smallest exhibit.
    #[test]
    fn prop_any_bound_strictly_above_the_roster_meet_is_undershot(
        oaths in prop::collection::vec(arb_oath(), 2..=4),
        bump_station in 1u32..=3,
        bump in 1u64..=4,
    ) {
        let meet = oaths
            .iter()
            .skip(1)
            .fold(oaths[0].clone(), |folded, oath| folded.meet(oath));
        let mut bound = meet.clone();
        bound.observe(bump_station, meet.get(bump_station) + bump);

        // The universal property: a strict rise above the meet cannot
        // sit below every oath, or it would sit below the meet.
        let excluded = oaths.iter().position(|oath| {
            !matches!(
                bound.partial_cmp(oath),
                Some(Ordering::Less | Ordering::Equal)
            )
        });
        let Some(member) = excluded else {
            return Err(TestCaseError::fail(
                "a bound strictly above the meet must exceed some member's oath",
            ));
        };

        // The excluded member declares at exactly its oath: its views
        // never rose past what it advertised, nothing adversarial.
        let station = u32::try_from(member + 1).expect("small rosters");
        let cut = Cut::from_witnessed(oaths[member].clone());
        let mut stability = Stability::new([1, 2, 3, 4]);
        for reporter in [1, 2, 3, 4] {
            stability.report_cut(reporter, &cut).unwrap();
        }
        let fresh = oaths[member].get(station) + 1;
        let mut epochs = Epochs::new([1, 2, 3, 4], NonZeroUsize::new(1).unwrap());
        let declaration = epochs
            // `fresh` is one above a held count, so it is always a dot.
            .declare(d(station, fresh), rank(1), &stability, &Cut::bottom())
            .unwrap();
        prop_assert_eq!(declaration.cut().as_vector(), &oaths[member]);

        // Maximality: the base advanced to the bound is refused.
        let trailing = Text::from_context(Cut::from_witnessed(bound).to_have_set());
        let refused = EpochPreparation::begin(&declaration, trailing, Moves::new());
        let refused = matches!(refused, Err(EpochPreparationMiss::UnpinnedBase { .. }));
        prop_assert!(refused, "the bound-advanced base is refused at the seam");

        // Sufficiency: the base advanced to the full meet is served.
        let trailing = Text::from_context(Cut::from_witnessed(meet).to_have_set());
        let served = EpochPreparation::begin(&declaration, trailing, Moves::new());
        prop_assert!(served.is_ok());
    }
}

/// The one transported-claim door, shared by the relay pins: the PRD
/// 0025 section 4 discipline made a helper. A cut travels as its
/// canonical gap-free have-set frame; the receiver decodes, *proves*
/// gap-freedom (the decoded set equals its own floor's witnessed
/// embedding), and re-enters the witnessed world through the honest
/// [`Cut::floor_of`] door. This is the path real transported floor
/// reports take (the confirmation frame's exact posture), so the
/// relay pins exercise the genuine trust boundary rather than the
/// in-crate constructor: the witness standing is *preserved*, and the
/// bare `Stability::report` door (which would withdraw the witnessed
/// read for good, by its R8 charter) is deliberately not the relay
/// path.
fn transported(cut: &Cut) -> Cut {
    let bytes = DotSet::from_witnessed(cut).to_bytes();
    let decoded = DotSet::from_bytes(&bytes).expect("a canonical frame decodes");
    let floor = Cut::floor_of(&decoded);
    assert_eq!(
        DotSet::from_witnessed(&floor),
        decoded,
        "the receiver proves gap-freedom before the cut-shaped role"
    );
    floor
}

/// The relay under-claim lemma (S286, the note's section 7; the RQ4
/// brief's certified silence claimed by pin): the tracker's two
/// standing postures, join-per-reporter folding and reports as claims
/// that may lawfully under-state, compose into exact hierarchical
/// aggregation as pure caller transport policy. A relay that meets a
/// subtree's claims and reports that meet *as each covered member's
/// claim* emits sound under-claims (a meet sits below each member's
/// floor), so safety holds at every prefix of any interleaving; and
/// the watermark over a complete wave equals the direct all-to-all
/// watermark exactly, because the meet over members of subtree meets
/// is the global meet (associativity and idempotence). Later direct
/// traffic refines entries upward without moving the read. One
/// vector per tree edge per wave instead of one per member per
/// member; freshness costs one forwarding step per level, a
/// structural bound, never a temporal one.
///
/// The whole path is public and *witness-preserving* (the S286
/// review's finding, absorbed): every hop rides the gap-free
/// embedding and re-enters through `Cut::floor_of`, so
/// `watermark_cut` keeps its standing and the epoch license
/// (`Epochs::declare`) remains available over a relayed tracker,
/// which this pin drives to a successful declaration. What the
/// lemma costs is trust *concentration*, not trust *grade*: a relay
/// reports for every member it covers, so a false relay carries the
/// R8 over-claim exposure of its whole subtree; choosing relays is
/// choosing whom you let claim for whom (caller policy, the
/// trust-grades note's territory).
#[test]
fn a_relay_of_subset_meets_serves_the_exact_watermark() {
    // Four members' claims, deliberately crossing so no single claim
    // is the meet and every column has a distinct minimum owner.
    let claims: [(u32, VersionVector); 4] = [
        (1, vector(&[(1, 9), (2, 3), (3, 5), (4, 7)])),
        (2, vector(&[(1, 4), (2, 8), (3, 6), (4, 2)])),
        (3, vector(&[(1, 6), (2, 5), (3, 9), (4, 4)])),
        (4, vector(&[(1, 8), (2, 2), (3, 7), (4, 9)])),
    ];
    // The direct tracker receives each member's own floor through the
    // same transported door the relays use.
    let mut direct = Stability::new([1, 2, 3, 4]);
    for (member, claim) in &claims {
        direct
            .report_cut(*member, &transported(&Cut::from_witnessed(claim.clone())))
            .unwrap();
    }
    let exact = direct.watermark();
    assert_eq!(exact, vector(&[(1, 4), (2, 2), (3, 5), (4, 2)]));
    assert!(direct.watermark_cut().is_some());

    // Two relays receive their subtrees' transported floors, meet
    // them (a meet of cuts is a cut, the closure theorems), and
    // forward one transported frame per subtree; the receiver reports
    // the decoded meet as each covered member's claim.
    let left = transported(&Cut::from_witnessed(claims[0].1.clone()))
        .meet(&transported(&Cut::from_witnessed(claims[1].1.clone())));
    let right = transported(&Cut::from_witnessed(claims[2].1.clone()))
        .meet(&transported(&Cut::from_witnessed(claims[3].1.clone())));
    let mut relayed = Stability::new([1, 2, 3, 4]);
    for member in [1u32, 2] {
        assert!(relayed.watermark() <= exact, "safe at every prefix");
        relayed.report_cut(member, &transported(&left)).unwrap();
    }
    for member in [3u32, 4] {
        assert!(relayed.watermark() <= exact, "safe at every prefix");
        relayed.report_cut(member, &transported(&right)).unwrap();
    }
    assert_eq!(
        relayed.watermark(),
        exact,
        "a complete relay wave reads the exact global meet"
    );

    // The witness standing survived the whole relayed path, so the
    // epoch license stays available: a declaration mints over the
    // relayed tracker exactly as over the direct one.
    assert!(relayed.watermark_cut().is_some());
    let mut epochs = Epochs::new([1, 2, 3, 4], NonZeroUsize::new(1).unwrap());
    assert!(
        epochs
            .declare(d(1, 10), rank(3), &relayed, &Cut::bottom())
            .is_ok(),
        "the relayed tracker licenses a declaration"
    );

    // Direct traffic arriving after (or interleaved with) the wave
    // refines entries upward and the read does not move: compression
    // costs attribution granularity, never exactness.
    for (member, claim) in &claims {
        relayed
            .report_cut(*member, &transported(&Cut::from_witnessed(claim.clone())))
            .unwrap();
        assert_eq!(relayed.watermark(), exact);
    }

    // The contrast that keeps the bare door honest: feeding the same
    // relayed vectors through `Stability::report` computes the same
    // watermark but withdraws the witnessed read for good (a bare
    // vector is a claim without its gap-freedom proof), which is why
    // the embedding path above is the transported discipline.
    let mut bare = Stability::new([1, 2, 3, 4]);
    for member in [1u32, 2] {
        bare.report(member, left.as_vector()).unwrap();
    }
    for member in [3u32, 4] {
        bare.report(member, right.as_vector()).unwrap();
    }
    assert_eq!(bare.watermark(), exact);
    assert!(
        bare.watermark_cut().is_none(),
        "the bare door costs the witness"
    );
}

/// A generated relay-claim seed: per-member claims over the shared
/// station alphabet, absence included.
fn arb_relay_claims() -> impl Strategy<Value = [VersionVector; 4]> {
    prop::collection::vec(prop::collection::btree_map(1u32..=4, 1u64..=20, 0..=4), 4).prop_map(
        |claims| {
            let mut out: [VersionVector; 4] = array::from_fn(|_| VersionVector::new());
            for (slot, counts) in out.iter_mut().zip(claims) {
                for (station, count) in counts {
                    slot.observe(station, count);
                }
            }
            out
        },
    )
}

proptest! {
    #![proptest_config(ProptestConfig::with_cases(256))]

    /// The relay under-claim lemma at family grade: over generated
    /// claims, an arbitrary two-relay partition of the roster, and an
    /// arbitrary interleaving of relayed and direct reports, every
    /// report riding the transported gap-free-embedding door, the
    /// watermark is a sound under-approximation of the global meet at
    /// every prefix, equals it exactly once every member's report
    /// (relayed or direct) has arrived, and the witness standing
    /// survives the whole interleaving.
    #[test]
    fn prop_relayed_subset_meets_are_sound_at_every_prefix_and_exact_when_complete(
        claims in arb_relay_claims(),
        partition in prop::collection::vec(any::<bool>(), 4),
        keys in prop::collection::vec(0u32..1_000_000, 8),
    ) {
        let exact = claims
            .iter()
            .skip(1)
            .fold(claims[0].clone(), |meet, claim| meet.meet(claim));

        // Each member's relayed value is its group's meet of
        // transported cuts (a group may be a singleton or the whole
        // roster; both are lawful trees).
        let group_meet = |flag: bool| {
            claims
                .iter()
                .zip(&partition)
                .filter(|(_, in_group)| **in_group == flag)
                .map(|(claim, _)| transported(&Cut::from_witnessed(claim.clone())))
                .fold(None::<Cut>, |meet, cut| {
                    Some(meet.map_or_else(|| cut.clone(), |held| held.meet(&cut)))
                })
        };
        let meets = [group_meet(false), group_meet(true)];

        // Eight report events (a relayed and a direct report per
        // member), shuffled by the generated keys.
        let mut events: Vec<(u32, usize, bool)> = (0..8)
            .map(|index| (keys[index], index % 4, index >= 4))
            .collect();
        events.sort_unstable();

        let mut tracker = Stability::new([1, 2, 3, 4]);
        let mut covered = [false; 4];
        let mut complete_at: Option<usize> = None;
        for (step, &(_, member, direct)) in events.iter().enumerate() {
            let station = u32::try_from(member + 1).expect("small roster");
            let report = if direct {
                transported(&Cut::from_witnessed(claims[member].clone()))
            } else {
                meets[usize::from(partition[member])]
                    .clone()
                    .expect("the member's own group is never empty")
            };
            tracker.report_cut(station, &transported(&report)).unwrap();
            covered[member] = true;
            if complete_at.is_none() && covered.iter().all(|&seen| seen) {
                complete_at = Some(step);
            }
            // Safety at every prefix: never past the global meet.
            let watermark = tracker.watermark();
            prop_assert!(watermark <= exact, "prefix watermark overshoots the meet");
            // Exactness from the first moment every member is covered
            // by at least its group meet, and thereafter.
            if complete_at.is_some() {
                prop_assert_eq!(watermark, exact.clone());
            }
        }
        // The witness standing survived the whole transported
        // interleaving.
        prop_assert!(tracker.watermark_cut().is_some() || events.is_empty());
    }
}