minerva 0.2.0

//! The replay laws (PRD 0022): recension convergence across replicas and
//! fold orders, agreement with an independent replay oracle computed from
//! public reads alone, and the empty-record identity.

extern crate alloc;

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

use crate::kairos::Kairos;
use crate::metis::{Anchor, Dot, Locus, Metatheses, Rhapsody};
use proptest::prelude::*;

use super::{Moves, Text, clock, locus_at, move_to, type_after};

const REPLICAS: usize = 3;

#[derive(Clone, Debug)]
enum Op {
    Insert {
        replica: usize,
        caret_pick: usize,
    },
    Delete {
        replica: usize,
        victim_pick: usize,
    },
    Move {
        replica: usize,
        target_pick: usize,
        dest_pick: usize,
        before: bool,
    },
    Undo {
        replica: usize,
        testimony_pick: usize,
    },
    Gossip {
        from: usize,
        to: usize,
    },
}

fn arb_ops() -> impl Strategy<Value = Vec<Op>> {
    let op = prop_oneof![
        3 => (0..REPLICAS, 0usize..8)
            .prop_map(|(replica, caret_pick)| Op::Insert { replica, caret_pick }),
        1 => (0..REPLICAS, 0usize..8)
            .prop_map(|(replica, victim_pick)| Op::Delete { replica, victim_pick }),
        3 => (0..REPLICAS, 0usize..8, 0usize..8, any::<bool>()).prop_map(
            |(replica, target_pick, dest_pick, before)| Op::Move {
                replica,
                target_pick,
                dest_pick,
                before,
            }
        ),
        1 => (0..REPLICAS, 0usize..8)
            .prop_map(|(replica, testimony_pick)| Op::Undo { replica, testimony_pick }),
        3 => (0..REPLICAS, 0..REPLICAS).prop_map(|(from, to)| Op::Gossip { from, to }),
    ];
    prop::collection::vec(op, 0..32)
}

/// Ships everything `from` owes `to` across both pairs (text and moves).
fn gossip(fleet: &mut [(Text, Moves)], from: usize, to: usize) {
    let text_delta = fleet[from].0.owed_to(fleet[to].0.state().context());
    let move_delta = fleet[from].1.owed_to(fleet[to].1.state().context());
    let sender = u32::try_from(from + 1).unwrap();
    let _ = fleet[to].0.absorb(sender, &text_delta);
    let _ = fleet[to].1.absorb(sender, &move_delta);
}

/// Runs a tape over the parallel pairs (text beside moves, per replica),
/// then fully exchanges both, so every replica converges.
fn run(ops: &[Op]) -> Vec<(Text, Moves)> {
    let mut fleet: Vec<(Text, Moves)> = (0..REPLICAS)
        .map(|i| {
            let station = u32::try_from(i + 1).unwrap();
            (Text::new(station), Moves::new(station))
        })
        .collect();
    let clocks: Vec<_> = (0..REPLICAS)
        .map(|i| clock(u32::try_from(i + 1).unwrap()))
        .collect();

    for op in ops {
        match *op {
            Op::Insert {
                replica,
                caret_pick,
            } => {
                let visible = fleet[replica].0.state().store().order();
                let caret = if visible.is_empty() {
                    None
                } else {
                    Some(visible[caret_pick % visible.len()].into())
                };
                let _ = type_after(&mut fleet[replica].0, &clocks[replica], caret);
            }
            Op::Delete {
                replica,
                victim_pick,
            } => {
                let visible = fleet[replica].0.state().store().order();
                if visible.is_empty() {
                    continue;
                }
                let victim = visible[victim_pick % visible.len()];
                let mut gone = crate::metis::DotSet::new();
                let _ = gone.insert(victim);
                let _ = fleet[replica].0.retract(gone);
            }
            Op::Move {
                replica,
                target_pick,
                dest_pick,
                before,
            } => {
                // Targets and destinations from the replica's own woven
                // set (tombstones includable: moving a tombstone is
                // lawful); `dest` may equal `target`, so self-anchoring
                // moves (a degenerate cycle) are generated and must refuse.
                let woven: Vec<Dot> = fleet[replica].0.state().store().woven().dots().collect();
                if woven.len() < 2 {
                    continue;
                }
                let target = woven[target_pick % woven.len()];
                let dest = woven[dest_pick % woven.len()];
                let anchor = if before {
                    Anchor::Before(dest.into())
                } else {
                    Anchor::After(dest.into())
                };
                let to = locus_at(&clocks[replica], anchor);
                let _ = move_to(&mut fleet[replica].1, target.into(), to);
            }
            Op::Undo {
                replica,
                testimony_pick,
            } => {
                let held: Vec<Dot> = fleet[replica]
                    .1
                    .state()
                    .store()
                    .iter()
                    .map(|(dot, _)| dot)
                    .collect();
                if held.is_empty() {
                    continue;
                }
                let victim = held[testimony_pick % held.len()];
                let mut undo = crate::metis::DotSet::new();
                let _ = undo.insert(victim);
                let _ = fleet[replica].1.retract(undo);
            }
            Op::Gossip { from, to } => {
                if from != to {
                    gossip(&mut fleet, from, to);
                }
            }
        }
    }
    // Full exchange, twice around, so everyone holds everything.
    for _ in 0..2 {
        for from in 0..REPLICAS {
            for to in 0..REPLICAS {
                if from != to {
                    gossip(&mut fleet, from, to);
                }
            }
        }
    }
    fleet
}

/// `(order, refused, pending)`, the three dot-lists a replay produces.
type Replay = (Vec<Dot>, Vec<Dot>, Vec<Dot>);

/// The raw-to-identity funnel an anchor read crosses: a coordinate naming
/// no dot ends the chain, exactly as an anchor no element carries did.
fn anchor_identity(anchor: Anchor) -> Option<Dot> {
    anchor.dot().and_then(|at| Dot::try_from(at).ok())
}

/// The independent replay oracle, computed from PUBLIC reads alone: births
/// via `woven()` + `locus()` applied first (canonical, never refused), then
/// moves via the record's own iteration in (rank, metathesis-dot) order,
/// each applied with a naive graph cycle check (only moves are refutable),
/// then walked by the naive in-order reference (buckets re-sorted per
/// anchor, Before reversed / self / After).
fn reference_recension(text: &Rhapsody, moves: &Metatheses) -> Replay {
    struct Play {
        key: (Kairos, Dot),
        target: Dot,
        locus: Locus,
        witness: Dot,
    }
    enum Frame {
        Visit(Dot),
        Emit(Dot),
    }
    // Births first: canonical placements, never refused.
    let mut effective: BTreeMap<Dot, Locus> = BTreeMap::new();
    let born: Vec<Dot> = text.woven().dots().collect();
    let born_set: BTreeSet<Dot> = born.iter().copied().collect();
    for &dot in &born {
        let locus = text.locus(dot).unwrap();
        let _ = effective.insert(dot, locus);
    }
    // Moves: the only refutable testimonies, in (rank, dot) order. A
    // target coordinate naming no born dot parks, the non-dot counter
    // included (it is born nowhere), so the funnel keeps the old verdict.
    let mut plays: Vec<Play> = Vec::new();
    let mut pending: Vec<Dot> = Vec::new();
    for (dot, metathesis) in moves {
        let target = Dot::try_from(metathesis.target)
            .ok()
            .filter(|target| born_set.contains(target));
        if let Some(target) = target {
            plays.push(Play {
                key: (metathesis.to.rank, dot),
                target,
                locus: metathesis.to,
                witness: dot,
            });
        } else {
            pending.push(dot);
        }
    }
    plays.sort_by_key(|play| play.key);

    let budget = born.len() + plays.len() + 1;
    let mut refused: Vec<Dot> = Vec::new();
    for play in &plays {
        // Naive cycle check: walk the whole chain with a step budget.
        let mut cursor = anchor_identity(play.locus.anchor);
        let mut steps = 0usize;
        let mut cycles = false;
        while let Some(at) = cursor {
            if at == play.target {
                cycles = true;
                break;
            }
            steps += 1;
            if steps > budget {
                break; // a cycle not through the target
            }
            cursor = effective
                .get(&at)
                .and_then(|locus| anchor_identity(locus.anchor));
        }
        if cycles {
            refused.push(play.witness);
        } else {
            let _ = effective.insert(play.target, play.locus);
        }
    }

    // The naive in-order walk over the effective loci.
    let bucket = |anchor: Anchor| -> Vec<Dot> {
        let mut kids: Vec<Dot> = effective
            .iter()
            .filter(|(_, locus)| locus.anchor == anchor)
            .map(|(&dot, _)| dot)
            .collect();
        kids.sort_by(|x, y| {
            let (rx, ry) = (effective[x].rank, effective[y].rank);
            ry.cmp(&rx).then_with(|| x.cmp(y))
        });
        kids
    };
    let mut order = Vec::new();
    let mut stack: Vec<Frame> = bucket(Anchor::Origin)
        .into_iter()
        .rev()
        .map(Frame::Visit)
        .collect();
    while let Some(frame) = stack.pop() {
        match frame {
            Frame::Emit(dot) => {
                if text.is_visible(dot) {
                    order.push(dot);
                }
            }
            Frame::Visit(dot) => {
                for kid in bucket(Anchor::After(dot.into())).into_iter().rev() {
                    stack.push(Frame::Visit(kid));
                }
                stack.push(Frame::Emit(dot));
                for kid in bucket(Anchor::Before(dot.into())) {
                    stack.push(Frame::Visit(kid));
                }
            }
        }
    }
    (order, refused, pending)
}

proptest! {
    /// The replay law: after any history of inserts, deletes, moves,
    /// undos, and gossip, every replica's recension agrees with every
    /// other's AND with the independent oracle: same effective order,
    /// same refusals (in replay order), same pending set. And the empty
    /// record is the identity at every replica.
    #[test]
    fn prop_recension_converges_and_matches_the_reference(ops in arb_ops()) {
        let fleet = run(&ops);
        let (reference_text, reference_moves) = (
            fleet[0].0.state().store(),
            fleet[0].1.state().store(),
        );
        let (oracle_order, oracle_refused, oracle_pending) =
            reference_recension(reference_text, reference_moves);
        for (text, moves) in &fleet {
            prop_assert_eq!(text.state().store(), reference_text);
            prop_assert_eq!(moves.state().store(), reference_moves);
            let recension = text.state().store().recension(moves.state().store());
            prop_assert_eq!(recension.order(), oracle_order.clone());
            prop_assert_eq!(recension.refused(), &oracle_refused[..]);
            prop_assert_eq!(recension.pending(), &oracle_pending[..]);
            // The empty record is the identity.
            let identity = text.state().store().recension(&Metatheses::new());
            prop_assert_eq!(identity.order(), text.state().store().order());
            prop_assert!(identity.refused().is_empty());
            prop_assert!(identity.pending().is_empty());
        }
    }

    /// Move-free histories refuse nothing: the refusal surface exists
    /// only where concurrent movement can genuinely conflict.
    #[test]
    fn prop_no_moves_means_no_refusals(ops in arb_ops()) {
        let move_free: Vec<Op> = ops
            .into_iter()
            .filter(|op| !matches!(op, Op::Move { .. } | Op::Undo { .. }))
            .collect();
        let fleet = run(&move_free);
        for (text, moves) in &fleet {
            prop_assert!(moves.state().store().is_empty());
            let recension = text.state().store().recension(moves.state().store());
            prop_assert!(recension.refused().is_empty());
            prop_assert_eq!(recension.order(), text.state().store().order());
        }
    }
}