minerva 0.2.0

//! The run-grade write's agreement laws (R-19): `weave_run` IS the per-dot
//! weave, and the in-place fold's run arm IS the pure merge, on every read.

extern crate alloc;

use alloc::vec::Vec;
use core::num::{NonZeroU32, NonZeroU64};

use crate::kairos::{Clock, TickCounter};
use crate::metis::{Anchor, Dot, Locus};
use proptest::prelude::*;

use super::super::d;
use super::{Seq, arb_ops_with_paste, run};
use crate::metis::dot::RawDot;

proptest! {
    /// The bulk weave is the per-dot weave, exactly: against any
    /// model-reached document, a fresh station's `weave_run` and a twin
    /// weaving the same reservation's members one at a time (each interior
    /// anchored `After` its predecessor, the rank read straight off the
    /// reservation) produce one value and agree on every positional,
    /// visibility, and reachability read, with the thread and occupancy
    /// invariants held. Anchors sample placed dots on BOTH sides (`After`
    /// and `Before` heads; the interiors chain after the head either
    /// way), the origin, and a dangling dot (the parked arm).
    #[test]
    fn prop_weave_run_agrees_with_the_per_dot_weave(
        ops in arb_ops_with_paste(),
        len in 1u32..200,
        anchor_pick in 0usize..11,
    ) {
        let (live, _) = run(&ops);
        let state = live.iter().fold(Seq::new(), |acc, r| acc.merge(r));
        let station = 9u32;
        let clock = Clock::with_default_config(TickCounter::new(), station).unwrap();
        let reservation = clock.now_run(NonZeroU32::new(len).unwrap(), 0u16);
        let first = state.next_dot(station);
        let visible = state.store().order();
        let anchor = if anchor_pick == 9 {
            // A dangling anchor: the run parks whole until repair.
            Anchor::After(RawDot { station: 7, counter: 1_000_000 })
        } else if visible.is_empty() || anchor_pick == 8 {
            Anchor::Origin
        } else if anchor_pick == 10 {
            // A Before-anchored head (the gate review's coverage note):
            // the run hangs before its anchor, interiors still chain
            // after the head.
            Anchor::Before(visible[len as usize % visible.len()].into())
        } else {
            Anchor::After(visible[anchor_pick % visible.len()].into())
        };

        let members: Vec<_> = reservation.iter().collect();
        let mut bulk = state.store().clone();
        prop_assert!(bulk.weave_run(first, anchor, reservation));

        let mut pointwise = state.store().clone();
        let mut prev = first;
        for k in 0..len {
            let dot = d(station, first.counter() + u64::from(k));
            let locus = Locus {
                anchor: if k == 0 { anchor } else { Anchor::After(prev.into()) },
                rank: members[k as usize],
            };
            prop_assert!(pointwise.weave(dot, locus));
            prev = dot;
        }

        prop_assert_eq!(&bulk, &pointwise);
        bulk.check_order_thread();
        prop_assert_eq!(bulk.order(), pointwise.order());
        prop_assert_eq!(bulk.order_len(), pointwise.order_len());
        for offset in 0..bulk.order_len() {
            prop_assert_eq!(bulk.order_at(offset), pointwise.order_at(offset));
        }
        for k in 0..len {
            let dot = d(station, first.counter() + u64::from(k));
            prop_assert_eq!(bulk.offset_of(dot), pointwise.offset_of(dot));
            prop_assert_eq!(bulk.is_reachable(dot), pointwise.is_reachable(dot));
            prop_assert!(bulk.is_visible(dot));
            prop_assert_eq!(bulk.locus(dot), pointwise.locus(dot));
        }
    }

    /// The run delta round-trips at run grade: a paste composed as one
    /// covered run delta absorbs into any model-reached peer through the
    /// in-place fold exactly as through the pure merge (the fold's run arm
    /// beside its per-dot arm), and a stale re-delivery is idempotent.
    #[test]
    fn prop_a_run_delta_absorbs_like_the_pure_merge(
        ops in arb_ops_with_paste(),
        len in 1u32..200,
    ) {
        let (live, _) = run(&ops);
        let state = live.iter().fold(Seq::new(), |acc, r| acc.merge(r));
        let station = 9u32;
        let clock = Clock::with_default_config(TickCounter::new(), station).unwrap();
        let reservation = clock.now_run(NonZeroU32::new(len).unwrap(), 0u16);
        let first = state.next_dot(station);
        let anchor = state
            .store()
            .order()
            .last()
            .map_or(Anchor::Origin, |&tail| Anchor::After(tail.into()));
        let mut fragment = crate::metis::Rhapsody::new();
        prop_assert!(fragment.weave_run(first, anchor, reservation));
        let delta = Seq::from_store(fragment);

        let pure = state.merge(&delta);
        let mut folded = state;
        folded.merge_from(&delta);
        prop_assert_eq!(&folded, &pure);
        prop_assert_eq!(folded.store().order(), pure.store().order());
        folded.store().check_order_thread();

        // Idempotence: the same run delta lands twice without effect.
        folded.merge_from(&delta);
        prop_assert_eq!(&folded, &pure);
    }
}

/// The paste absorb hits the bulk arm structurally, not incidentally: a
/// contiguous fresh chain run absorbed by the in-place fold lands as whole
/// fragments (one per 64 slots plus the seam), the same factoring the
/// per-dot typing path coalesces into, so the claim "coalesces by
/// birthright" is pinned on the receive side too.
#[test]
fn test_a_run_delta_lands_as_whole_fragments() {
    let station = 2u32;
    let clock = Clock::with_default_config(TickCounter::new(), station).unwrap();
    let reservation = clock.now_run(NonZeroU32::new(300).unwrap(), 0u16);
    let mut fragment = crate::metis::Rhapsody::new();
    assert!(fragment.weave_run(d(station, 1), Anchor::Origin, reservation));
    let delta = Seq::from_store(fragment);

    let mut receiver = Seq::new();
    receiver.merge_from(&delta);
    assert_eq!(receiver.store().order().len(), 300);
    receiver.store().check_order_thread();
    assert_eq!(
        receiver.store().order_thread_fragments(),
        5,
        "three hundred slots tile five fragments"
    );
    assert_eq!(
        receiver.store().order_thread_region_segments(),
        1,
        "one chain run is one region segment"
    );
}

/// The fold's contested corner: a parked dot at the RECEIVER waits on a dot
/// inside an arriving run delta. The run arm must degrade to the per-dot
/// fold so the waiter's repair protocol runs, and the folded result equals
/// the pure merge exactly.
#[test]
fn test_an_arriving_run_repairs_a_parked_waiter_per_dot() {
    // The receiver holds a dangling element anchored to (1, 40), a dot the
    // coming run will carry.
    let mut receiver = Seq::new();
    let waiter_rank = Clock::with_default_config(TickCounter::new(), 3)
        .unwrap()
        .now(0u16);
    let mut waiter = crate::metis::Rhapsody::new();
    assert!(waiter.weave(
        d(3, 1),
        Locus {
            anchor: Anchor::After(RawDot {
                station: 1,
                counter: 40
            }),
            rank: waiter_rank,
        }
    ));
    receiver.merge_from(&Seq::from_store(waiter));
    assert!(!receiver.store().is_reachable(d(3, 1)));

    // The run delta: dots (1, 1..=80) as one chain.
    let clock = Clock::with_default_config(TickCounter::new(), 1).unwrap();
    let run = clock.now_run(NonZeroU32::new(80).unwrap(), 0u16);
    let mut fragment = crate::metis::Rhapsody::new();
    assert!(fragment.weave_run(d(1, 1), Anchor::Origin, run));
    let delta = Seq::from_store(fragment);

    let pure = receiver.merge(&delta);
    let mut folded = receiver;
    folded.merge_from(&delta);
    assert_eq!(&folded, &pure);
    folded.store().check_order_thread();
    assert!(
        folded.store().is_reachable(d(3, 1)),
        "the waiter was repaired"
    );
    assert_eq!(folded.store().order(), pure.store().order());
    assert_eq!(folded.store().order_len(), 81);
}

/// The visibility settle is run-grade on both sides: a bulk retract of a
/// contiguous span (the receive of one superseding delta) and a bulk
/// re-admission both settle the thread by fragment spans, agreeing with
/// the pure merge exactly.
#[test]
fn test_a_bulk_retract_settles_by_runs() {
    let station = 1u32;
    let clock = Clock::with_default_config(TickCounter::new(), station).unwrap();
    let reservation = clock.now_run(NonZeroU32::new(256).unwrap(), 0u16);
    let mut store = crate::metis::Rhapsody::new();
    assert!(store.weave_run(d(station, 1), Anchor::Origin, reservation));
    let state = Seq::from_store(store);

    // Retract an UNALIGNED interior span: dots 70..=190, one context-only
    // delta. Both boundaries sit mid-fragment on purpose: an aligned span
    // starts at a fragment base (offset zero), where a settle mask that
    // forgot its offset shift is invisible (this pin's own first survivor,
    // the mutation discipline working as intended).
    let mut superseded = crate::metis::DotSet::new();
    superseded.insert_run(
        station,
        NonZeroU64::new(70).expect("a run starts at a dot"),
        121,
    );
    let retract = Seq::from_context(superseded);

    let pure = state.merge(&retract);
    let mut folded = state;
    folded.merge_from(&retract);
    assert_eq!(&folded, &pure);
    folded.store().check_order_thread();
    assert_eq!(folded.store().order_len(), 135);
    let visible: Vec<Dot> = folded.store().order();
    assert_eq!(visible.first(), Some(&d(station, 1)));
    assert_eq!(
        visible.get(69),
        Some(&d(station, 191)),
        "the span reopens past the retract"
    );
    assert_eq!(visible.last(), Some(&d(station, 256)));
    // The thread's positional view agrees with the walk at EVERY offset:
    // clearing the wrong bits of a fragment can preserve the visible
    // count while moving every position (this pin's second survivor), so
    // the settle's masks are pinned position-exact, not count-exact.
    for (offset, &dot) in visible.iter().enumerate() {
        assert_eq!(folded.store().order_at(offset), Some(dot));
        assert_eq!(folded.store().offset_of(dot), Some(offset));
    }
}