minerva 0.2.0

//! The order thread arena (crdt-vision arc 11 phase three): the maintained,
//! monoid-annotated positional structure over the placed walk order.
//!
//! The reads this carries have no home in the walk machinery: the walk
//! answers "what comes after this element" in `O(window)`, but an editor's
//! native coordinate is the *offset*, and before this structure every
//! offset-shaped question (the element at offset `k`, the offset of an
//! element, the visible length, anything read backward) cost one whole
//! `order()` pass, `O(document)` per call (the S181/S183 `order_cold`
//! number). The thread is the crate's answer in the shape the S189 scoping
//! fixed: a counted tree whose nodes cache the two subtree aggregates
//! `(slots, visible)`, the dynamic-programming trade taken once
//! (incremental aggregate maintenance on the `&mut` write path, `O(log)`
//! per mutation) so every positional read is a descent instead of a scan.
//!
//! Shape: an arena-backed tree (`Vec<Node>` plus a free list, `u32` ids,
//! parent pointers; the identity plane's allocation idiom) whose leaves
//! hold *fragments*, maximal walk-contiguous dot runs of one station
//! (capped at 64 slots, the fiber page width) with a per-slot visibility
//! bitmask. Honest chained traffic factors the whole document into one
//! fragment per 64-slot segment of each maximal run (the S192 coalescing
//! rate, the same factoring the wire codec and the run-coalesced plane
//! already speak; the probe's 64-char chunks tile the cap exactly, so
//! there fragments equal runs). Dynamic region endpoints are the second
//! coordinate, run-coalesced since arc 11 phase five: the ends side
//! contracts successor-shaped edges into segments (`SegmentPlane`) and
//! the starts side allocates lazily (`SparseForest`), so the honest
//! document's region state is `O(run segments)` beside the fragments
//! while insertion-position lookup stays amortized logarithmic across
//! independent deep seams (S200's guarantee, kept by construction: the
//! forest still answers every stored-edge path, and the arithmetic
//! segments cost two range probes).
//!
//! This is a *derived coordinate* under the shell discipline: the walk
//! stays the contract form (the order semantics live in `traversal.rs` and
//! nothing here decides an order), the thread re-realizes the same
//! sequence for positional cost, and the one law owed is agreement,
//! pinned property-by-property against `order()` and its oracle. It is
//! excluded from equality, hashing, and the wire, and it is never
//! interior mutable (the C8 constraint): slot aggregates and region
//! endpoints are maintained together on the `&mut` write path, including
//! repaired regions and retirement.
//!
//! Deliberate simplifications, recorded once: removal never rebalances
//! (excision only shrinks; an underfull path persists until the next
//! rebuild restores balance), and a node count is bounded by the identity
//! plane's own live-entry ceiling: leaves never outnumber fragments,
//! fragments never outnumber slots, and internal levels shrink at least
//! eightfold per level up on construction (deletion can leave one-child
//! internals, and the bulk build's last chunk per level can be one child,
//! but never grows the count), so live nodes stay within a small factor
//! of fragments and ids fit `u32` while slots stay under the plane's
//! `2^31 - 2`; the allocation `expect` carries this argument.

extern crate alloc;

use alloc::collections::BTreeMap;
use alloc::vec::Vec;
use core::num::NonZeroU64;

use super::super::placement::Dot;
use super::NIL;
use super::region::{RegionBoundary, RegionEnds};

/// Fanout cap for internal nodes.
const BRANCH: usize = 8;
/// Fragments per leaf.
const LEAF_FRAGS: usize = 8;
/// Slots per fragment: the visibility mask's width, matching the identity
/// plane's page width (honest chunks tile it exactly).
const FRAG_CAP: usize = 64;
/// One maximal walk-contiguous dot run of one station: slots for dots
/// `base .. base + len`, in walk order (a chain run reads in dot order),
/// with per-slot visibility in the mask's low `len` bits (bit `k` is dot
/// `base + k`).
#[derive(Clone, Copy, Debug)]
struct Fragment {
    /// The run's station.
    station: u32,
    /// The run's first dot counter (one-based, so a fragment's base is an
    /// identity by construction: ruling R-91).
    base: NonZeroU64,
    /// Slots held (`1..=FRAG_CAP`).
    len: u8,
    /// Per-slot visibility; bits at and above `len` are zero.
    visible: u64,
}

impl Fragment {
    /// A one-slot fragment.
    const fn single(dot: Dot, visible: bool) -> Self {
        Self {
            station: dot.station(),
            base: dot.counter_nonzero(),
            len: 1,
            visible: if visible { 1 } else { 0 },
        }
    }

    /// Slots held.
    const fn slots(&self) -> usize {
        self.len as usize
    }

    /// Visible slots held.
    const fn visible_count(&self) -> usize {
        self.visible.count_ones() as usize
    }

    /// The base dot, the fragment index's key.
    const fn base_dot(&self) -> Dot {
        Dot::new(self.station, self.base)
    }
}

/// A contiguous slot range lifted out of the thread by
/// [`OrderThread::extract_range`], carried as whole fragments and
/// re-entered whole by [`OrderThread::insert_fragments`]: the region
/// splice's carrier (PRD 0022 R7's fragment-grade re-placement, S204).
/// Opaque past the fragment count on purpose: the slots inside are in
/// flight between two thread positions and answer no read.
#[derive(Debug)]
pub(in crate::metis::rhapsody) struct Extraction {
    frags: Vec<Fragment>,
}

impl Extraction {
    /// Fragments carried: the splice's honest work unit.
    pub(in crate::metis::rhapsody) const fn fragments(&self) -> usize {
        self.frags.len()
    }
}

/// A tree node: an internal fanout row or a leaf of fragments, each caching
/// its subtree's two aggregates.
#[derive(Clone, Debug)]
struct Node {
    /// The parent id (`NIL` at the root).
    parent: u32,
    /// Subtree slot total.
    slots: usize,
    /// Subtree visible total.
    visible: usize,
    /// The node's payload.
    kind: NodeKind,
}

#[derive(Clone, Debug)]
enum NodeKind {
    /// Child ids, in walk order.
    Internal(Vec<u32>),
    /// Fragments, in walk order.
    Leaf(Vec<Fragment>),
}

/// The maintained order thread. Reads are `&self`; every mutation is
/// `&mut self` and leaves the aggregates exact.
#[derive(Clone, Debug)]
pub(in crate::metis::rhapsody) struct OrderThread {
    /// The node arena; freed ids are reused.
    nodes: Vec<Node>,
    /// Retired node ids.
    free: Vec<u32>,
    /// The root id, `NIL` when empty.
    root: u32,
    /// Per fragment-base dot, the leaf holding it: the dot-to-position
    /// door. `O(fragments)` entries, run-coalesced like everything else.
    index: BTreeMap<Dot, u32>,
    /// The placed anchor tree's dynamic region boundaries. These keep
    /// insertion-position lookup logarithmic across independent deep seams.
    regions: RegionEnds,
}

// Derived `Default` would zero `root`, which is a live id, not the empty
// sentinel; the empty thread is `new()`.
impl Default for OrderThread {
    fn default() -> Self {
        Self::new()
    }
}

impl OrderThread {
    /// The empty thread.
    pub(in crate::metis::rhapsody) const fn new() -> Self {
        Self {
            nodes: Vec::new(),
            free: Vec::new(),
            root: NIL,
            index: BTreeMap::new(),
            regions: RegionEnds::new(),
        }
    }

    /// Registers one newly placed dot in both endpoint forests.
    pub(in crate::metis::rhapsody) fn insert_region_dot(&mut self, dot: Dot) -> bool {
        self.regions.insert(dot)
    }

    /// Registers a freshly placed chain run `head ..= tail` as one region
    /// segment (the bulk-ingest form of per-dot registration plus the
    /// successor merges, in one write): every member's region end resolves
    /// to `tail` and its start to itself until edges say otherwise, which
    /// is exactly the chain's reading shape.
    pub(in crate::metis::rhapsody) fn insert_region_run(&mut self, head: Dot, tail: Dot) -> bool {
        self.regions.insert_run(head, tail)
    }

    /// Rebuilds both region planes in one pass: the whole-document
    /// construction path, where the incremental insert-then-merge
    /// protocol would churn the segment sets once per placed dot. The
    /// segments, breaks, and stored edges are derived by a single scan
    /// over the sorted placed dots, and the sets build from sorted data;
    /// edges naming an unplaced endpoint are skipped exactly as the
    /// incremental path refuses them.
    pub(in crate::metis::rhapsody) fn rebuild_regions(
        &mut self,
        placed: impl Iterator<Item = Dot>,
        ends_edges: &BTreeMap<Dot, Dot>,
        starts_edges: &BTreeMap<Dot, Dot>,
    ) {
        self.regions.rebuild(placed, ends_edges, starts_edges);
    }

    /// Retires one sterile placed dot from both endpoint forests.
    pub(in crate::metis::rhapsody) fn remove_region_dot(&mut self, dot: Dot) -> bool {
        self.regions.remove(dot)
    }

    /// First dot in `dot`'s region, in `O(log n)` amortized time.
    pub(in crate::metis::rhapsody) fn region_start(&mut self, dot: Dot) -> Option<Dot> {
        self.regions.endpoint(dot, RegionBoundary::Start)
    }

    /// Last dot in `dot`'s region, in `O(log n)` amortized time.
    pub(in crate::metis::rhapsody) fn region_end(&mut self, dot: Dot) -> Option<Dot> {
        self.regions.endpoint(dot, RegionBoundary::End)
    }

    /// Replaces the last `Before` child edge of `dot`.
    pub(in crate::metis::rhapsody) fn replace_region_start(
        &mut self,
        dot: Dot,
        child: Option<Dot>,
    ) -> bool {
        self.regions.replace(dot, child, RegionBoundary::Start)
    }

    /// Replaces the last `After` child edge of `dot`.
    pub(in crate::metis::rhapsody) fn replace_region_end(
        &mut self,
        dot: Dot,
        child: Option<Dot>,
    ) -> bool {
        self.regions.replace(dot, child, RegionBoundary::End)
    }

    /// Placed dots covered by the region planes (the coverage invariant's
    /// read: every placed slot is covered by exactly one segment).
    #[cfg(test)]
    pub(in crate::metis::rhapsody) fn region_nodes(&self) -> usize {
        self.regions.covered()
    }

    /// Region segments held: the endpoint plane's run-coalescing
    /// accounting read (arc 11 phase five); on honest chained traffic
    /// this is the run count, not the placed-dot count.
    #[cfg(test)]
    pub(in crate::metis::rhapsody) fn region_segments(&self) -> usize {
        self.regions.segments()
    }

    /// Link-cut nodes actually allocated across both region forests (the
    /// lazy-allocation accounting read).
    #[cfg(test)]
    pub(in crate::metis::rhapsody) fn region_forest_nodes(&self) -> usize {
        self.regions.forest_nodes()
    }

    /// Allocates a node, reusing a retired id first. The id space cannot
    /// exhaust below the identity plane's own live-entry ceiling: leaves
    /// never outnumber fragments, fragments never outnumber slots, and
    /// internal levels shrink per level up (a one-child internal can
    /// survive deletion or end a bulk-build level, but no path grows the
    /// node count past the construction bound), so ids stay under
    /// `2^32 - 1` while slots stay under the plane's `2^31 - 2`.
    fn alloc(&mut self, node: Node) -> u32 {
        if let Some(id) = self.free.pop() {
            self.nodes[id as usize] = node;
            id
        } else {
            let id = u32::try_from(self.nodes.len())
                .ok()
                .filter(|&id| id != NIL)
                .expect("the plane's live-entry ceiling bounds the thread's node count");
            self.nodes.push(node);
            id
        }
    }

    /// Recomputes one node's aggregates from its children or fragments.
    fn refresh(&mut self, id: u32) {
        let (slots, visible) = match &self.nodes[id as usize].kind {
            NodeKind::Internal(children) => children.iter().fold((0, 0), |(s, v), &child| {
                let node = &self.nodes[child as usize];
                (s + node.slots, v + node.visible)
            }),
            NodeKind::Leaf(frags) => frags.iter().fold((0, 0), |(s, v), frag| {
                (s + frag.slots(), v + frag.visible_count())
            }),
        };
        self.nodes[id as usize].slots = slots;
        self.nodes[id as usize].visible = visible;
    }

    /// Refreshes aggregates from `id` up to the root.
    fn refresh_to_root(&mut self, mut id: u32) {
        while id != NIL {
            self.refresh(id);
            id = self.nodes[id as usize].parent;
        }
    }

    /// The leaf holding slot-space `position`, and the offset within it. A
    /// position on a boundary between two subtrees lands at the END of the
    /// left one (so `position == slot_len` lands past the last leaf's last
    /// fragment): landing left is what lets the tail-extend fast path fire
    /// across leaf boundaries, keeping a typed run one fragment however
    /// the leaves split around it.
    fn leaf_at_slot(&self, position: usize) -> (u32, usize) {
        let mut id = self.root;
        let mut remaining = position;
        loop {
            match &self.nodes[id as usize].kind {
                NodeKind::Internal(children) => {
                    let mut next = None;
                    for &child in children {
                        let slots = self.nodes[child as usize].slots;
                        if remaining <= slots {
                            next = Some(child);
                            break;
                        }
                        remaining -= slots;
                    }
                    id = next.expect("an in-range position always lands");
                }
                NodeKind::Leaf(_) => return (id, remaining),
            }
        }
    }

    /// Splits `leaf` if it exceeds its fragment cap, re-homing the moved
    /// fragments' index entries, and splits ancestors that overflow in
    /// turn (the root split grows the tree by one level).
    fn split_leaf_if_over(&mut self, leaf: u32) {
        let NodeKind::Leaf(frags) = &mut self.nodes[leaf as usize].kind else {
            unreachable!("only leaves split here");
        };
        if frags.len() <= LEAF_FRAGS {
            return;
        }
        let tail = frags.split_off(frags.len() / 2);
        let parent = self.nodes[leaf as usize].parent;
        let moved: Vec<Dot> = tail.iter().map(Fragment::base_dot).collect();
        let sibling = self.alloc(Node {
            parent,
            slots: 0,
            visible: 0,
            kind: NodeKind::Leaf(tail),
        });
        for dot in moved {
            let _ = self.index.insert(dot, sibling);
        }
        self.refresh(leaf);
        self.refresh(sibling);
        self.attach_after(leaf, sibling);
    }

    /// Attaches `sibling` immediately after `child` under its parent,
    /// growing a new root when `child` was the root, and splitting
    /// overflowing internals up the path.
    fn attach_after(&mut self, child: u32, sibling: u32) {
        let parent = self.nodes[child as usize].parent;
        if parent == NIL {
            let root = self.alloc(Node {
                parent: NIL,
                slots: 0,
                visible: 0,
                kind: NodeKind::Internal(alloc::vec![child, sibling]),
            });
            self.nodes[child as usize].parent = root;
            self.nodes[sibling as usize].parent = root;
            self.root = root;
            self.refresh(root);
            return;
        }
        self.nodes[sibling as usize].parent = parent;
        let NodeKind::Internal(children) = &mut self.nodes[parent as usize].kind else {
            unreachable!("a parent is internal");
        };
        let at = children
            .iter()
            .position(|&c| c == child)
            .expect("the child is under its parent");
        children.insert(at + 1, sibling);
        if children.len() > BRANCH {
            let tail = children.split_off(children.len() / 2);
            let split = self.alloc(Node {
                parent: NIL,
                slots: 0,
                visible: 0,
                kind: NodeKind::Internal(tail),
            });
            let NodeKind::Internal(tail_children) = &self.nodes[split as usize].kind else {
                unreachable!("just built internal");
            };
            let reparent: Vec<u32> = tail_children.clone();
            for moved in reparent {
                self.nodes[moved as usize].parent = split;
            }
            self.refresh(parent);
            self.refresh(split);
            self.attach_after(parent, split);
        } else {
            self.refresh(parent);
        }
    }

    /// Detaches an emptied node from its parent, recursing while parents
    /// empty; an emptied root leaves the thread empty.
    fn detach_empty(&mut self, id: u32) {
        let parent = self.nodes[id as usize].parent;
        self.free.push(id);
        if parent == NIL {
            self.root = NIL;
            return;
        }
        let NodeKind::Internal(children) = &mut self.nodes[parent as usize].kind else {
            unreachable!("a parent is internal");
        };
        let at = children
            .iter()
            .position(|&c| c == id)
            .expect("the child is under its parent");
        let _ = children.remove(at);
        if children.is_empty() {
            self.detach_empty(parent);
        } else {
            self.refresh_to_root(parent);
        }
    }

    /// Packs coalesced fragments into leaves and builds the internal
    /// levels bottom-up over an empty thread: the bulk-construction tail
    /// shared by [`from_slots`](Self::from_slots) and the emptied-thread
    /// arm of [`insert_fragments`](Self::insert_fragments). `O(fragments)`.
    fn build_from_fragments(&mut self, frags: &[Fragment]) {
        debug_assert_eq!(self.root, NIL, "the bulk build starts empty");
        if frags.is_empty() {
            return;
        }
        let mut level: Vec<u32> = Vec::new();
        for chunk in frags.chunks(LEAF_FRAGS) {
            let taken = chunk.to_vec();
            let bases: Vec<Dot> = taken.iter().map(Fragment::base_dot).collect();
            let leaf = self.alloc(Node {
                parent: NIL,
                slots: 0,
                visible: 0,
                kind: NodeKind::Leaf(taken),
            });
            for dot in bases {
                let _ = self.index.insert(dot, leaf);
            }
            self.refresh(leaf);
            level.push(leaf);
        }
        while level.len() > 1 {
            let mut above: Vec<u32> = Vec::new();
            for group in level.chunks(BRANCH) {
                let node = self.alloc(Node {
                    parent: NIL,
                    slots: 0,
                    visible: 0,
                    kind: NodeKind::Internal(group.to_vec()),
                });
                for &child in group {
                    self.nodes[child as usize].parent = node;
                }
                self.refresh(node);
                above.push(node);
            }
            level = above;
        }
        self.root = level[0];
    }

    /// Builds a thread from the walk's slot sequence (dots with their
    /// visibility, in walk order): greedy fragment coalescing, leaves
    /// packed, levels built bottom-up. `O(n)`.
    pub(in crate::metis::rhapsody) fn from_slots(
        slots: impl IntoIterator<Item = (Dot, bool)>,
    ) -> Self {
        let mut frags: Vec<Fragment> = Vec::new();
        for (dot, visible) in slots {
            if let Some(last) = frags.last_mut()
                && last.station == dot.station()
                && last.base.checked_add(u64::from(last.len)) == Some(dot.counter_nonzero())
                && last.slots() < FRAG_CAP
            {
                if visible {
                    last.visible |= 1u64 << last.len;
                }
                last.len += 1;
            } else {
                frags.push(Fragment::single(dot, visible));
            }
        }
        let mut thread = Self::new();
        thread.build_from_fragments(&frags);
        thread
    }

    /// Exhaustive structural check for the test suites: parent links,
    /// aggregate exactness, fragment-index coverage, and mask hygiene.
    #[cfg(test)]
    pub(in crate::metis::rhapsody) fn check_invariants(&self) {
        self.regions.check_invariants();
        if self.root == NIL {
            assert!(self.index.is_empty(), "an empty thread indexes nothing");
            assert!(
                self.regions.is_empty(),
                "an empty thread has no region segments"
            );
            return;
        }
        assert_eq!(self.nodes[self.root as usize].parent, NIL);
        let mut seen_frags = 0usize;
        let mut stack = alloc::vec![self.root];
        while let Some(id) = stack.pop() {
            let node = &self.nodes[id as usize];
            match &node.kind {
                NodeKind::Internal(children) => {
                    assert!(!children.is_empty(), "internals hold children");
                    let (mut slots, mut visible) = (0, 0);
                    for &child in children {
                        assert_eq!(self.nodes[child as usize].parent, id, "parent links agree");
                        slots += self.nodes[child as usize].slots;
                        visible += self.nodes[child as usize].visible;
                        stack.push(child);
                    }
                    assert_eq!((node.slots, node.visible), (slots, visible));
                }
                NodeKind::Leaf(frags) => {
                    assert!(!frags.is_empty(), "leaves hold fragments");
                    assert!(
                        frags.len() <= LEAF_FRAGS,
                        "a bulk splice split every overfull leaf"
                    );
                    let (mut slots, mut visible) = (0, 0);
                    for frag in frags {
                        assert!(frag.len >= 1 && frag.slots() <= FRAG_CAP);
                        if frag.len < 64 {
                            assert_eq!(
                                frag.visible >> frag.len,
                                0,
                                "mask bits above len stay zero"
                            );
                        }
                        assert_eq!(self.index.get(&frag.base_dot()), Some(&id));
                        slots += frag.slots();
                        visible += frag.visible_count();
                        seen_frags += 1;
                    }
                    assert_eq!((node.slots, node.visible), (slots, visible));
                }
            }
        }
        assert_eq!(
            seen_frags,
            self.index.len(),
            "the index covers every fragment"
        );
    }
}

mod mutate;
mod reads;
mod splice;

#[cfg(test)]
use reads::nth_set_bit;

#[cfg(test)]
mod tests;