minerva 0.2.0

//! The oracle's own falsifiers and its scoping, executable.
//!
//! An oracle that cannot refuse is theater, so the first two exhibits drive
//! the checker to its refusal on the two anomaly shapes that define it: the
//! pair flip (two states ordering one pair both ways) and the three-state
//! rotation, which every state-local check passes and only the
//! execution-wide view refuses (the weak/strong separation). The third
//! exhibit earns a verdict from a real two-replica execution through the
//! shipped store, restating the specification's clause directly: every
//! observed state is a subsequence of the one witness order. The fourth pins
//! the scoping boundary: block interleaving satisfies strong-list, so the
//! `FugueMax` corner (PRD 0018 R6) is a residual this oracle deliberately
//! does not judge.

extern crate alloc;

use alloc::vec::Vec;

use crate::kairos::Kairos;
use crate::metis::{Anchor, Composer, DotSet, Locus, Rhapsody};

use super::super::support::dot;
use super::{Dot, OrderLedger};

/// The specification clause restated: `state` read as a subsequence of
/// `witness` (its elements in witness order are exactly the state).
fn embeds(witness: &[Dot], state: &[Dot]) -> bool {
    let members: alloc::collections::BTreeSet<Dot> = state.iter().copied().collect();
    let projected: Vec<Dot> = witness
        .iter()
        .copied()
        .filter(|dot| members.contains(dot))
        .collect();
    projected == state
}

#[test]
fn test_the_pair_flip_is_refused_with_both_observations_as_evidence() {
    let a = dot(1, 1);
    let b = dot(2, 1);
    let mut ledger = OrderLedger::new();
    let first = ledger.observe(&[a, b]);
    let second = ledger.observe(&[b, a]);
    let refusal = ledger.verdict().expect_err("a flipped pair has no order");
    assert_eq!(refusal.cycle, [a, b]);
    assert_eq!(refusal.witnesses, [first, second]);
}

#[test]
fn test_the_rotation_is_refused_where_every_pairwise_view_passes() {
    // The weak/strong separation: [a b], [b c], [c a] are pairwise
    // consistent (any two admit a total order), and each state alone is
    // trivially fine, so no per-state or per-pair check can refuse. The
    // execution-wide constraint graph is the 3-cycle.
    let a = dot(1, 1);
    let b = dot(2, 1);
    let c = dot(3, 1);
    let states: [&[Dot]; 3] = [&[a, b], &[b, c], &[c, a]];

    for skip in 0..states.len() {
        let mut pairwise = OrderLedger::new();
        for (index, state) in states.iter().enumerate() {
            if index != skip {
                let _ = pairwise.observe(state);
            }
        }
        assert!(
            pairwise.verdict().is_ok(),
            "any two of the three states admit an order",
        );
    }

    let mut ledger = OrderLedger::new();
    for state in states {
        let _ = ledger.observe(state);
    }
    let refusal = ledger.verdict().expect_err("the rotation has no order");
    assert_eq!(refusal.cycle, [a, b, c]);
    assert_eq!(refusal.witnesses, [0, 1, 2]);
}

#[test]
fn test_a_real_execution_earns_its_witness_order() {
    // A two-replica insert/delete execution through the shipped store:
    // concurrent weaves on both sides of a shared prefix, a retract, delta
    // gossip both ways. Every visible order along the way is observed; the
    // verdict must produce one witness order that every observation reads
    // as a subsequence of (the strong list clause, restated directly).
    let mut recorded: Vec<Vec<Dot>> = Vec::new();
    let mut ledger = OrderLedger::new();
    let mut observe = |ledger: &mut OrderLedger, state: Vec<Dot>| {
        let _ = ledger.observe(&state);
        recorded.push(state);
    };

    let mut one = Composer::<Rhapsody>::new(1);
    let mut two = Composer::<Rhapsody>::new(2);
    let mut tick = 0u64;
    let mut weave = |composer: &mut Composer<Rhapsody>, station: u32, anchor: Anchor| -> Dot {
        let rank = Kairos::new(tick, 0u16, station, 0u16);
        tick += 1;
        let (dot, _) = composer.compose(|assigned| {
            let mut rhapsody = Rhapsody::new();
            assert!(rhapsody.weave(assigned, Locus { anchor, rank }));
            (rhapsody, DotSet::new())
        });
        dot
    };

    // A shared prefix on replica one, gossiped whole.
    let a = weave(&mut one, 1, Anchor::Origin);
    let b = weave(&mut one, 1, Anchor::After(a.into()));
    observe(&mut ledger, one.state().store().order());
    let _ = two.absorb(1, &one.state().clone());
    observe(&mut ledger, two.state().store().order());

    // Concurrently: one appends after `b`, two prepends before `a` and
    // retracts `b`.
    let c = weave(&mut one, 1, Anchor::After(b.into()));
    observe(&mut ledger, one.state().store().order());
    let d = weave(&mut two, 2, Anchor::Before(a.into()));
    observe(&mut ledger, two.state().store().order());
    let mut superseded = DotSet::new();
    assert!(superseded.insert(b));
    let _ = two.retract(superseded);
    observe(&mut ledger, two.state().store().order());

    // Delta gossip both ways, observing each receiver.
    let owed_to_two = one.owed_to(&two.state().context().clone());
    let _ = two.absorb(1, &owed_to_two);
    observe(&mut ledger, two.state().store().order());
    let owed_to_one = two.owed_to(&one.state().context().clone());
    let _ = one.absorb(2, &owed_to_one);
    observe(&mut ledger, one.state().store().order());

    let witness = ledger.verdict().expect("the execution admits one order");
    assert_eq!(ledger.observations(), recorded.len());
    for state in &recorded {
        assert!(
            embeds(&witness, state),
            "an observed state must read as a subsequence of the witness: {state:?} in {witness:?}",
        );
    }
    // The converged reading is `d a c` (b retracted); it embeds too.
    assert_eq!(one.state().store().order(), two.state().store().order());
    assert_eq!(one.state().store().order(), [d, a, c]);
}

#[test]
fn test_a_skipped_pairwise_merge_can_hide_a_contradiction() {
    // The S288 review's second finding, pinned at the ledger level: a
    // convergence harness that walks one merge path (the first replica
    // absorbs the second, then the third meets their join) can read a
    // consistent stream even when a *skipped* pairwise join contradicts
    // it, because the full join heals the transient. Modeled with
    // sequences (the shipped merge is believed sound; the oracle exists
    // to catch it if it were not): replicas read [p q], [q], [p]; the
    // one-path convergence stream repeats [p q] and stays consistent,
    // while a faulty pairwise join of the second and third replicas that
    // answered [q p] is caught only if pairwise joins are observed. This
    // is why both convergence twins observe every pairwise merge from
    // the pre-convergence snapshot before the in-place loop runs.
    let p = dot(1, 1);
    let q = dot(2, 1);
    let path_only: [&[Dot]; 5] = [&[p, q], &[q], &[p], &[p, q], &[p, q]];

    let mut blind = OrderLedger::new();
    for state in path_only {
        let _ = blind.observe(state);
    }
    assert!(
        blind.verdict().is_ok(),
        "the one-path stream hides the faulty pairwise join",
    );

    let mut covered = OrderLedger::new();
    for state in path_only {
        let _ = covered.observe(state);
    }
    let faulty_pairwise = covered.observe(&[q, p]);
    let refusal = covered
        .verdict()
        .expect_err("the observed pairwise join is caught");
    assert_eq!(refusal.cycle, [p, q]);
    assert_eq!(refusal.witnesses, [0, faulty_pairwise]);
}

#[test]
fn test_block_interleaving_satisfies_the_specification() {
    // The scoping boundary, executable: two runs observed contiguous at
    // their home replicas and *interleaved* in the merged state still admit
    // one total order, so the strong list specification passes. Maximal
    // non-interleaving is a strictly stronger criterion the oracle
    // deliberately does not judge: the FugueMax corner stays PRD 0018 R6's
    // recorded residual, and the shipped same-direction contiguity cure is
    // pinned by the runs suites, not here.
    let f1 = dot(1, 1);
    let f2 = dot(1, 2);
    let k1 = dot(2, 1);
    let k2 = dot(2, 2);
    let mut ledger = OrderLedger::new();
    let _ = ledger.observe(&[f1, f2]);
    let _ = ledger.observe(&[k1, k2]);
    let _ = ledger.observe(&[f1, k1, f2, k2]);
    let witness = ledger
        .verdict()
        .expect("interleaved blocks still admit one order");
    assert!(embeds(&witness, &[f1, k1, f2, k2]));
}