minerva 0.2.0

//! The positional reads' named cases (arc 11 phase three, S200): offsets
//! forward and backward, tombstone slots, the unplaced refusal, the
//! Before-side region-start arm, and offsets through condense.

extern crate alloc;

use alloc::vec::Vec;

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

use super::super::d;
use super::{Seq, clock, woven};
use crate::metis::dot::RawDot;

/// A locus with a hand-picked rank (full control over sibling order, the
/// wire suites' idiom).
fn locus_at(anchor: Anchor, physical: u64, station: u32) -> Locus {
    Locus {
        anchor,
        rank: Kairos::new(physical, 0, station, 0u16),
    }
}

#[test]
fn test_order_at_reads_the_document_by_offset() {
    let clk = clock(1);
    let mut store = Rhapsody::new();
    let mut prev: Option<Dot> = None;
    for index in 1..=12u64 {
        let anchor = prev.map_or(Anchor::Origin, |dot| Anchor::After(dot.into()));
        assert!(store.weave(
            d(1, index),
            Locus {
                anchor,
                rank: clk.now(0u16),
            }
        ));
        prev = Some(d(1, index));
    }
    let order = store.order();
    assert_eq!(store.order_len(), order.len());
    for (offset, &dot) in order.iter().enumerate() {
        assert_eq!(store.order_at(offset), Some(dot));
        assert_eq!(store.offset_of(dot), Some(offset));
    }
    assert_eq!(store.order_at(order.len()), None, "out of range refuses");
}

#[test]
fn test_offset_of_a_tombstone_bounds_its_slot() {
    let clk = clock(1);
    let mut seq = Seq::new();
    let mut prev: Option<Dot> = None;
    for _ in 0..5 {
        let dot = seq.next_dot(1);
        let anchor = prev.map_or(Anchor::Origin, |dot| Anchor::After(dot.into()));
        let mut store = Rhapsody::new();
        assert!(store.weave(
            dot,
            Locus {
                anchor,
                rank: clk.now(0u16),
            }
        ));
        seq = seq.merge(&woven(dot, store.locus(dot).unwrap()));
        prev = Some(dot);
    }
    // Delete the middle element: its slot persists as an order tombstone.
    let mut ctx = DotSet::new();
    let _ = ctx.insert(d(1, 3));
    seq = seq.merge(&crate::metis::Dotted::from_context(ctx));
    let store = seq.store();

    assert_eq!(store.order(), [d(1, 1), d(1, 2), d(1, 4), d(1, 5)]);
    assert_eq!(store.order_len(), 4);
    // The tombstone's offset is the count of visible elements before its
    // slot: it bounds the offset its slot holds open, and the successor
    // element reads at the same offset.
    assert_eq!(store.offset_of(d(1, 3)), Some(2));
    assert_eq!(store.offset_of(d(1, 4)), Some(2));
    assert_eq!(store.order_at(2), Some(d(1, 4)), "selection skips the slot");
    // The reverse resume at the tombstone yields the reversed prefix
    // before its slot, exactly as the forward resume yields the suffix.
    let rev: Vec<Dot> = store.order_walk_rev_before(d(1, 3)).unwrap().collect();
    assert_eq!(rev, [d(1, 2), d(1, 1)]);
}

#[test]
fn test_reverse_walk_reads_the_document_backward() {
    let clk = clock(1);
    let mut store = Rhapsody::new();
    let mut prev: Option<Dot> = None;
    for index in 1..=8u64 {
        let anchor = prev.map_or(Anchor::Origin, |dot| Anchor::After(dot.into()));
        assert!(store.weave(
            d(1, index),
            Locus {
                anchor,
                rank: clk.now(0u16),
            }
        ));
        prev = Some(d(1, index));
    }
    let mut expected = store.order();
    expected.reverse();
    let rev: Vec<Dot> = store.order_walk_rev().collect();
    assert_eq!(rev, expected);
    // The scroll-up window: the last three, newest first, exact-sized.
    let walk = store.order_walk_rev();
    assert_eq!(walk.len(), 8);
    let tail: Vec<Dot> = walk.take(3).collect();
    assert_eq!(tail, [d(1, 8), d(1, 7), d(1, 6)]);
    // The resume before a visible element mirrors the forward suffix.
    let before: Vec<Dot> = store.order_walk_rev_before(d(1, 3)).unwrap().collect();
    assert_eq!(before, [d(1, 2), d(1, 1)]);
}

#[test]
fn test_positional_reads_refuse_the_unplaced() {
    let mut store = Rhapsody::new();
    // A dangling delta: its anchor has not arrived, so it is visible but
    // has no place in the current order.
    assert!(store.weave(
        d(2, 1),
        locus_at(
            Anchor::After(RawDot {
                station: 1,
                counter: 1
            }),
            500,
            2
        )
    ));
    assert_eq!(store.visible_len(), 1, "visibility counts the dangling");
    assert_eq!(store.order_len(), 0, "the order does not");
    assert_eq!(store.offset_of(d(2, 1)), None);
    assert!(store.order_walk_rev_before(d(2, 1)).is_none());
    assert_eq!(store.order_at(0), None);
    // The repair merge places it, and every read agrees again.
    assert!(store.weave(d(1, 1), locus_at(Anchor::Origin, 100, 1)));
    assert_eq!(store.order_len(), 2);
    assert_eq!(store.offset_of(d(2, 1)), Some(1));
    assert_eq!(store.order_at(1), Some(d(2, 1)));
    let rev: Vec<Dot> = store.order_walk_rev().collect();
    assert_eq!(rev, [d(2, 1), d(1, 1)]);
}

/// A repaired subtree is one contiguous reading region. Thread it in walk
/// order when the missing anchor arrives, including both anchor sides.
#[test]
fn test_a_repaired_region_threads_in_walk_order() {
    let mut store = Rhapsody::new();
    assert!(store.weave(d(1, 1), locus_at(Anchor::Origin, 100, 1)));

    // Deliver the region inside out. Every element remains visible but
    // unplaced until (1, 2) connects the region to the origin.
    assert!(store.weave(
        d(1, 4),
        locus_at(
            Anchor::Before(RawDot {
                station: 1,
                counter: 3
            }),
            400,
            1
        )
    ));
    assert!(store.weave(
        d(1, 5),
        locus_at(
            Anchor::After(RawDot {
                station: 1,
                counter: 3
            }),
            500,
            1
        )
    ));
    assert!(store.weave(
        d(1, 3),
        locus_at(
            Anchor::After(RawDot {
                station: 1,
                counter: 2
            }),
            300,
            1
        )
    ));
    store.check_order_thread();
    assert_eq!(store.order(), [d(1, 1)]);
    assert_eq!(store.offset_of(d(1, 3)), None);

    assert!(store.weave(
        d(1, 2),
        locus_at(
            Anchor::After(RawDot {
                station: 1,
                counter: 1
            }),
            200,
            1
        )
    ));
    store.check_order_thread();
    let expected = [d(1, 1), d(1, 2), d(1, 4), d(1, 3), d(1, 5)];
    assert_eq!(store.order(), expected);
    assert_eq!(store.order_len(), expected.len());
    for (offset, dot) in expected.into_iter().enumerate() {
        assert_eq!(store.order_at(offset), Some(dot));
        assert_eq!(store.offset_of(dot), Some(offset));
    }
    assert_eq!(
        store.order_walk_rev().collect::<Vec<_>>(),
        [d(1, 5), d(1, 3), d(1, 4), d(1, 2), d(1, 1),]
    );
}

#[test]
fn test_a_lower_rank_before_sibling_positions_by_region_start() {
    let mut store = Rhapsody::new();
    // d at the origin; b0 hangs Before(d) with the higher rank and carries
    // its own Before-child c, so b0's reading region starts at c; then b1
    // arrives Before(d) with a LOWER rank (stored after b0, reading before
    // b0's whole region: the non-head Before arm, whose position is the
    // region-start descent through c).
    assert!(store.weave(d(1, 1), locus_at(Anchor::Origin, 100, 1)));
    assert!(store.weave(
        d(1, 2),
        locus_at(
            Anchor::Before(RawDot {
                station: 1,
                counter: 1
            }),
            200,
            1
        )
    ));
    assert!(store.weave(
        d(1, 3),
        locus_at(
            Anchor::Before(RawDot {
                station: 1,
                counter: 2
            }),
            300,
            1
        )
    ));
    assert!(store.weave(
        d(2, 1),
        locus_at(
            Anchor::Before(RawDot {
                station: 1,
                counter: 1
            }),
            150,
            2
        )
    ));
    let order = store.order();
    assert_eq!(
        order,
        [d(2, 1), d(1, 3), d(1, 2), d(1, 1)],
        "the low-rank sibling reads before the high-rank region"
    );
    for (offset, &dot) in order.iter().enumerate() {
        assert_eq!(store.offset_of(dot), Some(offset));
        assert_eq!(store.order_at(offset), Some(dot));
    }
    store.check_order_thread();
}

/// The S200 gate review's coverage finding, pinned deterministically: a
/// deletion arriving through the IN-PLACE fold (`merge_from`, the hot
/// editor receive path) into a live document must move the positional
/// aggregates through the settle's expelled-flip arm, the arm a pure-merge
/// rebuild never runs and a from-empty fold can never reach. A wrong or
/// absent flip here was invisible to every prior suite (the thread is
/// excluded from equality and `order()` does not read it).
#[test]
fn test_a_deletion_through_the_in_place_fold_moves_the_offsets() {
    let clk = clock(1);
    let mut origin = Seq::new();
    let mut prev: Option<Dot> = None;
    for _ in 0..5 {
        let dot = origin.next_dot(1);
        let anchor = prev.map_or(Anchor::Origin, |dot| Anchor::After(dot.into()));
        let mut store = Rhapsody::new();
        assert!(store.weave(
            dot,
            Locus {
                anchor,
                rank: clk.now(0u16),
            }
        ));
        origin = origin.merge(&woven(dot, store.locus(dot).unwrap()));
        prev = Some(dot);
    }
    // The replica holds the full document, thread settled and live.
    let mut replica = origin.clone();
    assert_eq!(replica.store().order_len(), 5);
    assert_eq!(replica.store().offset_of(d(1, 4)), Some(3));

    // The origin deletes the middle element; the replica receives it
    // through the in-place fold, whose expelled residual carries (1, 3).
    let mut ctx = DotSet::new();
    let _ = ctx.insert(d(1, 3));
    origin = origin.merge(&crate::metis::Dotted::from_context(ctx.clone()));
    replica.merge_from(&crate::metis::Dotted::from_context(ctx));
    replica.store().check_order_thread();

    assert_eq!(replica.store().order(), origin.store().order());
    assert_eq!(replica.store().order_len(), 4);
    assert_eq!(replica.store().order_at(2), Some(d(1, 4)));
    assert_eq!(
        replica.store().offset_of(d(1, 3)),
        Some(2),
        "the tombstone bounds its slot"
    );
    assert_eq!(replica.store().offset_of(d(1, 4)), Some(2));
    let rev: Vec<Dot> = replica.store().order_walk_rev().collect();
    assert_eq!(rev, [d(1, 5), d(1, 4), d(1, 2), d(1, 1)]);
}

/// The S200 gate review's finding, pinned at the store and through decode:
/// an element at the dot ceiling (`u64::MAX`) whose walk successor probes
/// fragment extension must refuse by checked arithmetic, on the
/// incremental weave path and on the decode rebuild (whose "never panics"
/// contract reaches the thread's bulk build).
#[test]
fn test_a_ceiling_dot_weaves_and_decodes_without_overflow() {
    let mut store = Rhapsody::new();
    assert!(store.weave(d(1, u64::MAX), locus_at(Anchor::Origin, 100, 1)));
    assert!(store.weave(
        d(1, 1),
        locus_at(
            Anchor::After(RawDot {
                station: 1,
                counter: u64::MAX
            }),
            200,
            1
        )
    ));
    store.check_order_thread();
    assert_eq!(store.order(), [d(1, u64::MAX), d(1, 1)]);
    assert_eq!(store.offset_of(d(1, u64::MAX)), Some(0));
    assert_eq!(store.offset_of(d(1, 1)), Some(1));

    let decoded = Rhapsody::from_bytes(&store.to_bytes()).expect("the ceiling value decodes");
    decoded.check_order_thread();
    assert_eq!(decoded, store);
    assert_eq!(decoded.order_at(0), Some(d(1, u64::MAX)));
    assert_eq!(decoded.order_at(1), Some(d(1, 1)));
}

/// A low-rank sibling under an early anchor resolves the predecessor's
/// document-deep region end through the maintained endpoint forest. The
/// read stays exact without walking the anchor spine or rebuilding.
#[test]
fn test_a_deep_seam_insert_uses_the_region_endpoint() {
    let mut store = Rhapsody::new();
    let mut physical = 100u64;
    let mut prev: Option<Dot> = None;
    for index in 1..=200u64 {
        let anchor = prev.map_or(Anchor::Origin, |dot| Anchor::After(dot.into()));
        assert!(store.weave(d(1, index), locus_at(anchor, physical, 1)));
        physical += 10;
        prev = Some(d(1, index));
    }
    // Ranked below (1, 2): stored after it in the After((1, 1)) bucket, so
    // its position is (1, 2)'s region end, behind a 198-edge anchor spine.
    assert!(store.weave(
        d(9, 1),
        locus_at(
            Anchor::After(RawDot {
                station: 1,
                counter: 1
            }),
            105,
            9
        )
    ));
    store.check_order_thread();
    assert_eq!(store.order_len(), 201);
    assert_eq!(
        store.offset_of(d(9, 1)),
        Some(200),
        "the low-rank sibling reads last"
    );
    assert_eq!(store.order_at(200), Some(d(9, 1)));
    assert_eq!(store.order_at(0), Some(d(1, 1)));
    let order = store.order();
    assert_eq!(order[200], d(9, 1));
    assert_eq!(&order[..3], [d(1, 1), d(1, 2), d(1, 3)]);
}

/// Split peer deltas crossing distinct deep gaps remain on the dynamic
/// endpoint path. Each insertion targets a different predecessor region.
#[test]
fn test_split_deltas_cross_independent_deep_seams() {
    const SEAMS: usize = 4;
    const DEPTH: usize = 80;
    let mut state = Rhapsody::new();
    let mut deltas = Vec::new();
    let mut inserted = Vec::new();
    let mut next = 1u64;
    for seam in 0..SEAMS {
        let anchor = d(1, next);
        next += 1;
        assert!(state.weave(anchor, locus_at(Anchor::Origin, seam as u64, 1)));
        let predecessor = d(1, next);
        next += 1;
        assert!(state.weave(predecessor, locus_at(Anchor::After(anchor.into()), 300, 1)));
        let mut tail = predecessor;
        for _ in 1..DEPTH {
            let dot = d(1, next);
            next += 1;
            assert!(state.weave(dot, locus_at(Anchor::After(tail.into()), next, 1)));
            tail = dot;
        }
        let successor = d(1, next);
        next += 1;
        assert!(state.weave(successor, locus_at(Anchor::After(anchor.into()), 100, 1)));

        let dot = d(9, seam as u64 + 1);
        let mut delta = Rhapsody::new();
        assert!(delta.weave(dot, locus_at(Anchor::After(anchor.into()), 200, 9)));
        deltas.push(crate::metis::Dotted::from_store(delta));
        inserted.push(dot);
    }

    let mut replica = crate::metis::Dotted::from_store(state);
    for delta in &deltas {
        replica.merge_from(delta);
        replica.store().check_order_thread();
    }
    assert_eq!(replica.store().order_len(), SEAMS * (DEPTH + 3));
    for dot in inserted {
        assert!(replica.store().offset_of(dot).is_some());
    }
}

/// The S200 gate review's P1, pinned: a SELF-ANCHORED locus (which `weave`
/// accepts and the wire deliberately carries, the dangling-anchor posture)
/// must classify as unplaced, exactly as the structural rebuild classifies
/// it, on the incremental path and through decode-then-merge. Before the
/// fix the incremental placement read called it placed (the dot itself had
/// just entered the skeleton), which was a latent derived-coordinate
/// disagreement pre-S200 and a reachable panic in the thread's position
/// math after it.
#[test]
fn test_a_self_anchored_element_is_unplaced_on_every_path() {
    let mut store = Rhapsody::new();
    assert!(store.weave(
        d(1, 1),
        locus_at(
            Anchor::After(RawDot {
                station: 1,
                counter: 1
            }),
            100,
            1
        )
    ));
    assert!(store.weave(
        d(1, 2),
        locus_at(
            Anchor::Before(RawDot {
                station: 1,
                counter: 2
            }),
            200,
            1
        )
    ));
    store.check_order_thread();
    assert_eq!(store.order_len(), 0);
    assert_eq!(store.order(), []);
    assert!(!store.is_reachable(d(1, 1)));
    assert!(!store.is_reachable(d(1, 2)));
    assert_eq!(store.offset_of(d(1, 1)), None);
    assert_eq!(store.offset_of(d(1, 2)), None);
    assert_eq!(store.visible_len(), 2, "visibility still counts them");

    // The network shape: the value survives its frame, and a live replica
    // absorbs it through the in-place fold without panicking or placing.
    let decoded = Rhapsody::from_bytes(&store.to_bytes()).expect("the value decodes");
    assert_eq!(decoded, store);
    let clk = clock(2);
    let mut replica = Seq::new();
    let dot = replica.next_dot(2);
    let mut seeded = Rhapsody::new();
    assert!(seeded.weave(
        dot,
        Locus {
            anchor: Anchor::Origin,
            rank: clk.now(0u16),
        }
    ));
    replica = replica.merge(&woven(dot, seeded.locus(dot).unwrap()));
    replica.merge_from(&crate::metis::Dotted::from_store(decoded));
    replica.store().check_order_thread();
    assert_eq!(replica.store().order(), [dot]);
    assert_eq!(replica.store().order_len(), 1);
    assert_eq!(replica.store().offset_of(d(1, 1)), None);
}

#[test]
fn test_condense_keeps_every_surviving_offset() {
    let clk = clock(1);
    let mut seq = Seq::new();
    let mut prev: Option<Dot> = None;
    for _ in 0..6 {
        let dot = seq.next_dot(1);
        let anchor = prev.map_or(Anchor::Origin, |dot| Anchor::After(dot.into()));
        let mut store = Rhapsody::new();
        assert!(store.weave(
            dot,
            Locus {
                anchor,
                rank: clk.now(0u16),
            }
        ));
        seq = seq.merge(&woven(dot, store.locus(dot).unwrap()));
        prev = Some(dot);
    }
    // Retire the tail element (sterile once invisible), then condense.
    let mut ctx = DotSet::new();
    let _ = ctx.insert(d(1, 6));
    seq = seq.merge(&crate::metis::Dotted::from_context(ctx));
    let mut store = seq.store().clone();
    let mut retired = DotSet::new();
    let _ = retired.insert(d(1, 6));
    assert_eq!(store.condense(&Retired::trust(retired)), 1);
    store.check_order_thread();

    assert_eq!(store.order_len(), 5);
    assert_eq!(
        store.offset_of(d(1, 6)),
        None,
        "an excised slot has no offset"
    );
    for (offset, &dot) in store.order().iter().enumerate() {
        assert_eq!(store.offset_of(dot), Some(offset));
        assert_eq!(store.order_at(offset), Some(dot));
    }
}

#[test]
fn test_condense_relinks_the_region_endpoint() {
    let mut initial = Rhapsody::new();
    assert!(initial.weave(d(1, 1), locus_at(Anchor::Origin, 400, 1)));
    assert!(initial.weave(
        d(1, 2),
        locus_at(
            Anchor::After(RawDot {
                station: 1,
                counter: 1
            }),
            300,
            1
        )
    ));
    assert!(initial.weave(
        d(1, 3),
        locus_at(
            Anchor::After(RawDot {
                station: 1,
                counter: 1
            }),
            100,
            1
        )
    ));
    let mut seq = crate::metis::Dotted::from_store(initial);
    let mut removed = DotSet::new();
    let _ = removed.insert(d(1, 3));
    seq = seq.merge(&crate::metis::Dotted::from_context(removed));

    let mut retired = DotSet::new();
    let _ = retired.insert(d(1, 3));
    let mut store = seq.store().clone();
    assert_eq!(store.condense(&Retired::trust(retired)), 1);
    assert!(store.weave(
        d(2, 1),
        locus_at(
            Anchor::After(RawDot {
                station: 1,
                counter: 1
            }),
            200,
            2
        )
    ));
    store.check_order_thread();
    assert_eq!(store.order(), [d(1, 1), d(1, 2), d(2, 1)]);
    assert_eq!(store.offset_of(d(2, 1)), Some(2));
}