minerva 0.2.0

extern crate alloc;

use alloc::collections::BTreeSet;
use alloc::vec::Vec;

use super::super::super::metathesis::{Metatheses, Metathesis};
use super::super::ancestry::CoordinateError;
#[cfg(test)]
use super::super::ancestry::Relation;
use super::super::placement::Dot;
use super::super::possession::OccupancyPlane;
use super::super::{Anchor, Locus, Rhapsody};
use super::cycle::{CycleOutcome, WalkBudget, move_forms_cycle};
use super::{Play, PlayKey, Recension, Verdict, as_identity, synchronize_coordinate, woven_target};
use crate::metis::dot::RawDot;

impl Recension {
    fn cycle_outcome(&mut self, target: RawDot, anchor: Anchor) -> CycleOutcome {
        let performed = &self.performed;
        move_forms_cycle(
            self.ancestry.as_mut().ok(),
            target,
            anchor,
            &self.anchored,
            WalkBudget::for_skeleton(performed.skeleton_len()),
            |link| {
                as_identity(link)
                    .and_then(|link| performed.skeleton.get(&link))
                    .map(|locus| locus.anchor.dot())
            },
        )
    }

    /// Collates this recension against the pair's current state: after the
    /// call every read of `self` equals a fresh
    /// [`text.recension(moves)`](Rhapsody::recension).
    ///
    /// Exact agreement with that eager replay is the one law; this is its
    /// cost shell. The view diffs possession marks against the pair and
    /// replays only the changed regions, with suffix re-placement capped at
    /// the eager replay's own cost class so no arrival order makes that work
    /// dearer than not maintaining the view. The cap covers suffix replay and
    /// not the whole call.
    ///
    /// Total for *any* pair, not only a causal successor of the last one: a
    /// shrunken skeleton falls back to one whole replay. The skeleton diff
    /// reads presence alone, which extends exactly the immutable-locus trust
    /// [`causal_merge`](crate::metis::DotStore::causal_merge) already
    /// extends; a document reusing another's dot space under different loci
    /// is outside this contract exactly as it is outside the merge's. Debug
    /// builds assert the basis.
    ///
    /// The cost model, the amplification cap, and what the presence-only
    /// diff does and does not trust are
    /// `docs/metis-incremental-recension.adoc`.
    ///
    /// # Panics
    ///
    /// Never in practice: every `expect` reads state the possession marks or
    /// the play log just established, so a panic here signals a mechanism
    /// bug rather than caller error.
    pub fn collate(&mut self, text: &Rhapsody, moves: &Metatheses) {
        // A shrunken skeleton (condense, or an unrelated pair) cannot be
        // unwound from the log: the whole replay is the lawful fallback,
        // and its cadence is the caller's retirement round, never a hot
        // path.
        if self.woven_mark.difference(&text.woven).next().is_some() {
            self.rebuild(text, moves);
            return;
        }
        self.debug_assert_text_basis(text);
        let births: Vec<Dot> = text.woven.difference(&self.woven_mark).collect();
        let born_now: BTreeSet<Dot> = births.iter().copied().collect();

        // Visibility flips for dots already collated (a fresh birth folds
        // in below carrying its current visibility). Movement never reads
        // visibility, so flips commute with everything here. One paged
        // occupancy walk carries both directions (S205): the changed
        // pages fall out of an `O(pages)` word-compare merge, and the
        // flipped dots out of each page's XOR, so the scan costs the
        // pair's actual difference plus the page walk, never a probe per
        // retained exception.
        let mut mark_batch: Vec<((u32, u64), u64)> = Vec::new();
        let mut performed_batch: Vec<((u32, u64), u64)> = Vec::new();
        for (key, held, live) in self.visible_mark.changed_pages(&text.visible_pages) {
            mark_batch.push((key, live));
            // The performed plane's row is the live mask minus the bits
            // this call's births carry (they fold with their visibility
            // below, through the per-dot choke point).
            let mut performed_mask = live;
            for dot in OccupancyPlane::dots_of(key, held & !live) {
                if !born_now.contains(&dot) {
                    let _ = self.performed.visible.remove(dot);
                    let _ = self.performed.thread.set_visible(dot, false);
                }
            }
            for dot in OccupancyPlane::dots_of(key, live & !held) {
                if born_now.contains(&dot) {
                    performed_mask &= !(1u64 << (dot.counter() % 64));
                } else {
                    let _ = self.performed.visible.insert(dot);
                    let _ = self.performed.thread.set_visible(dot, true);
                }
            }
            if performed_mask != held {
                performed_batch.push((key, performed_mask));
            }
        }
        // One batched settle per plane (the gate review's quadratic
        // finding: per-dot and per-page row removal from the sorted
        // vectors made a bulk retirement `O(pages^2)`; the batch is one
        // point write per surviving row or one linear merge).
        self.performed.visible_pages.apply_pages(&performed_batch);

        // Phase one: find the earliest affected play key WITHOUT touching
        // the performed store. The unwind below must run on exactly the
        // state the plays were earned against: a displaced locus is only
        // guaranteed acyclic in the state that once applied it, so a
        // birth folded before the unwind could make a historical restore
        // land inside its own reading region (the gate review's P1: an
        // older move anchored at an unborn dot, displaced by a newer
        // move, then the dot born beneath the target).
        let mut earliest: Option<PlayKey> = None;
        let note = |key: PlayKey, earliest: &mut Option<PlayKey>| {
            *earliest = Some(earliest.map_or(key, |held| held.min(key)));
        };
        for &dot in &births {
            // The unborn-anchor guard is keyed by anchor COORDINATES (a
            // payload seat), so an arriving identity is read there in raw
            // spelling.
            if let Some(entry) = self.dangling.remove(&RawDot::from(dot)) {
                // A folded locus anchors to this arrival: the verdicts
                // from the recorded key (or the whole log) re-earn.
                let from = entry.or_else(|| self.plays.first().map(|play| play.key));
                if let Some(key) = from {
                    note(key, &mut earliest);
                }
            }
        }
        if !births.is_empty() {
            for &witness in &self.pending {
                let testimony = self.folded[&witness];
                if as_identity(testimony.target).is_some_and(|target| born_now.contains(&target)) {
                    note((testimony.to.rank, witness), &mut earliest);
                }
            }
        }

        // The record's diff: a sorted merge walk comparing presence and
        // content (a replaced testimony diffs as departed plus arrived).
        // Departures fold first, so a replacement's old entry leaves
        // before its new one lands. Participation is decided against the
        // pair's own skeleton (the final truth), since the performed
        // store has not folded this call's births yet.
        let (arrived, departed) = self.diff_record(moves);
        // The classification the note below already needs, kept as its
        // by-product: a departure found in the pending vector earned no
        // verdict, so it changes no move order and only has to leave the
        // vector. Its POSITION is recorded here and the vector is
        // compacted once, after this loop, rather than spliced per
        // witness (see below). Positions rise with the witnesses, since
        // both sequences are ascending.
        let mut parked: Vec<usize> = Vec::new();
        for &witness in &departed {
            let testimony = self
                .folded
                .remove(&witness)
                .expect("a departed testimony was folded");
            match self.pending.binary_search(&witness) {
                Ok(at) => parked.push(at),
                Err(_) => note((testimony.to.rank, witness), &mut earliest),
            }
        }
        // The parked departures leave in ONE compaction over the tail
        // from the first of them: `O(pending - first + departures)`, and
        // nothing at all when no departure was parked. Splicing each one
        // out with `Vec::remove` made a mass retraction `O(M * pending)`,
        // and it is the one path the suffix cap below cannot cover
        // (these removals set no earliest key, so they never consult the
        // replay budget) --- against a fresh eager replay, which is
        // constant time when the text and the record are both empty,
        // that was exactly the S200 shape: maintaining the view dearer
        // than not having it. The probe above stays a probe per
        // departure, so a collation retracting only unparked testimonies
        // reads the pending vector and never walks it.
        if let Some(&first) = parked.first() {
            let mut leaving = parked.iter().skip(1).peekable();
            let mut write = first;
            for read in first + 1..self.pending.len() {
                if leaving.peek() == Some(&&read) {
                    let _ = leaving.next();
                } else {
                    self.pending[write] = self.pending[read];
                    write += 1;
                }
            }
            #[cfg(test)]
            {
                self.compacted += self.pending.len() - first;
            }
            self.pending.truncate(write);
        }
        for &(witness, testimony) in &arrived {
            let _ = self.folded.insert(witness, testimony);
            if as_identity(testimony.target).is_some_and(|target| text.skeleton.contains(target)) {
                note((testimony.to.rank, witness), &mut earliest);
            } else {
                let at = self.pending.partition_point(|&held| held < witness);
                self.pending.insert(at, witness);
            }
        }

        // Update the possession marks before any replay: the fallback
        // rebuilds them from the pair itself, and the phases below read
        // only the pair, the performed store, and the log.
        for &dot in &births {
            let _ = self.woven_mark.insert(dot);
        }
        // The mark adopts exactly the rows the walk found changed (the
        // steady state writes nothing and copies nothing; a bulk change
        // is one linear merge).
        self.visible_mark.apply_pages(&mark_batch);

        let Some(from) = earliest else {
            // No move-order changes: the births fold incrementally and
            // every collated verdict stands.
            self.fold_births(&births, text);
            return;
        };
        self.replay_suffix_from(from, &births, &arrived, &departed, text, moves);
    }

    /// Folds fresh births into the performed store through its own
    /// incremental insertion, exactly as the in-place merge folds novel
    /// loci, recording the walk-guard anchors and the unborn-anchor
    /// references as each locus lands. Births are canonical (never
    /// refused), so the fold is unconditional; the caller decides WHEN it
    /// runs (after any unwind, so historical restores never meet a birth
    /// the plays were not earned against).
    fn fold_births(&mut self, births: &[Dot], text: &Rhapsody) {
        for &dot in births {
            let locus = text
                .skeleton
                .get(&dot)
                .expect("a birth enumerated from the skeleton diff is woven");
            let repaired = self
                .performed
                .record_locus_transition(dot, locus)
                .expect("a birth is novel to the performed store");
            self.synchronize_birth(dot, locus, &repaired);
            if text.visible.contains(dot) {
                self.performed.note_visibility(dot, true);
                let _ = self.performed.thread.set_visible(dot, true);
            }
            if let Some(anchor_dot) = locus.anchor.dot() {
                let _ = self.anchored.insert(anchor_dot);
                // A birth anchored to a dot unborn even now (checked
                // against the pair, so a same-batch anchor never
                // records): the whole log is sensitive to that arrival.
                if !as_identity(anchor_dot).is_some_and(|dot| text.skeleton.contains(dot)) {
                    let _ = self.dangling.insert(anchor_dot, None);
                }
            }
        }
    }

    fn synchronize_birth(&mut self, dot: Dot, locus: Locus, repaired: &[Dot]) {
        let first_unplaced = self.performed.unplaced.first().copied();
        let synchronized = match &mut self.ancestry {
            Ok(coordinate) => (|| {
                let prior_first = coordinate.first_unplaced();
                let placed = !repaired.is_empty();
                coordinate.add(
                    dot,
                    locus.anchor.dot(),
                    placed,
                    if placed { prior_first } else { first_unplaced },
                )?;
                if placed {
                    for &member in &repaired[1..] {
                        let parent = self
                            .performed
                            .skeleton
                            .get(&member)
                            .ok_or(CoordinateError::InvariantViolation)?
                            .anchor
                            .dot();
                        coordinate.repair(member, parent)?;
                    }
                    coordinate.publish_state(first_unplaced);
                }
                Ok(())
            })(),
            Err(_) => return,
        };
        if let Err(error) = synchronized {
            self.ancestry = Err(error);
        }
    }

    fn re_place_performed(&mut self, target: RawDot, locus: Locus) -> usize {
        let transition = self.performed.re_place(woven_target(target), locus);
        let first_unplaced = self.performed.unplaced.first().copied();
        if let Ok(coordinate) = &mut self.ancestry
            && let Err(error) = synchronize_coordinate(
                coordinate,
                &self.performed.skeleton,
                first_unplaced,
                transition.effect,
            )
        {
            self.ancestry = Err(error);
        }
        transition.work
    }

    /// The record's diff against the folded mark: a sorted merge walk
    /// comparing presence AND content. Within one causal system a
    /// testimony is immutable under its dot (axis law 3), but the value
    /// itself is freely constructible, and the testimonies are already in
    /// hand at each equal key, so the content check is free: a witness
    /// carrying a different testimony diffs as a departure plus an
    /// arrival, and the suffix replay replaces it (the gate review's
    /// finding: presence alone would keep the stale testimony silently).
    #[allow(clippy::type_complexity)]
    fn diff_record(&self, moves: &Metatheses) -> (Vec<(Dot, Metathesis)>, BTreeSet<Dot>) {
        let mut arrived: Vec<(Dot, Metathesis)> = Vec::new();
        let mut departed: BTreeSet<Dot> = BTreeSet::new();
        let mut held = self.folded.iter().peekable();
        let mut live = moves.iter().peekable();
        loop {
            match (held.peek(), live.peek().copied()) {
                (Some(&(&h, held_testimony)), Some((l, testimony))) => match h.cmp(&l) {
                    core::cmp::Ordering::Less => {
                        let _ = departed.insert(h);
                        let _ = held.next();
                    }
                    core::cmp::Ordering::Greater => {
                        arrived.push((l, *testimony));
                        let _ = live.next();
                    }
                    core::cmp::Ordering::Equal => {
                        if held_testimony != testimony {
                            let _ = departed.insert(h);
                            arrived.push((l, *testimony));
                        }
                        let _ = held.next();
                        let _ = live.next();
                    }
                },
                (Some(&(&h, _)), None) => {
                    let _ = departed.insert(h);
                    let _ = held.next();
                }
                (None, Some((l, testimony))) => {
                    arrived.push((l, *testimony));
                    let _ = live.next();
                }
                (None, None) => break,
            }
        }
        (arrived, departed)
    }

    /// Unwinds the play log down to `from` (on the state the plays were
    /// earned against), folds this call's births, splices the pending
    /// activations, arrivals, and departures in, and replays forward,
    /// re-earning each verdict; abandons to one whole replay when the
    /// suffix work would exceed it (the eager-parity cap). Every affected
    /// key is at least `from` by construction (the caller noted each one),
    /// so the untouched prefix stays lawful. The phase order is
    /// load-bearing (the gate review's P1): a restored displaced locus is
    /// only guaranteed acyclic in the state that once applied it, so the
    /// births fold strictly AFTER the unwind, and every intermediate
    /// restore is a state the log actually passed through.
    fn replay_suffix_from(
        &mut self,
        from: PlayKey,
        births: &[Dot],
        arrived: &[(Dot, Metathesis)],
        departed: &BTreeSet<Dot>,
        text: &Rhapsody,
        moves: &Metatheses,
    ) {
        #[cfg(test)]
        {
            self.replays += 1;
        }
        // The cap: one whole replay costs the document plus the record,
        // so a suffix that works past that is dearer than starting over.
        let budget = self.performed.skeleton_len() + self.folded.len() + 1;
        let mut work = 0usize;
        let at = self.plays.partition_point(|play| play.key < from);
        for position in (at..self.plays.len()).rev() {
            if let Verdict::Applied { displaced } = self.plays[position].verdict {
                work += self.re_place_performed(self.plays[position].target, displaced);
                if work > budget {
                    self.rebuild(text, moves);
                    return;
                }
            }
        }
        // The births land on the unwound state: the same births-first
        // order the eager replay commits to, arrived at incrementally.
        self.fold_births(births, text);
        let mut suffix: Vec<Play> = self.plays.split_off(at);
        suffix.retain(|play| !departed.contains(&play.key.1));
        // Pending moves whose targets have now arrived join the order.
        let mut still_pending: Vec<Dot> = Vec::with_capacity(self.pending.len());
        for &witness in &self.pending {
            let testimony = self.folded[&witness];
            if as_identity(testimony.target)
                .is_some_and(|target| self.performed.skeleton.contains(target))
            {
                suffix.push(Play {
                    key: (testimony.to.rank, witness),
                    target: testimony.target,
                    locus: testimony.to,
                    verdict: Verdict::Refused,
                });
            } else {
                still_pending.push(witness);
            }
        }
        self.pending = still_pending;
        // Arrivals whose targets are born join the order; the pending ones
        // stayed in the pending list above.
        for &(witness, testimony) in arrived {
            if as_identity(testimony.target)
                .is_some_and(|target| self.performed.skeleton.contains(target))
            {
                suffix.push(Play {
                    key: (testimony.to.rank, witness),
                    target: testimony.target,
                    locus: testimony.to,
                    verdict: Verdict::Refused,
                });
            }
        }
        suffix.sort_unstable_by_key(|play| play.key);
        debug_assert!(
            suffix.first().is_none_or(|play| play.key >= from),
            "every spliced play sits at or past the unwind point"
        );
        for mut play in suffix {
            let cycles = self
                .cycle_outcome(play.target, play.locus.anchor)
                .must_refuse();
            if cycles {
                play.verdict = Verdict::Refused;
            } else {
                let displaced = self
                    .performed
                    .skeleton
                    .get(&woven_target(play.target))
                    .expect("a participating target is born");
                work += self.re_place_performed(play.target, play.locus);
                if work > budget {
                    self.rebuild(text, moves);
                    return;
                }
                play.verdict = Verdict::Applied { displaced };
                if let Some(anchor_dot) = play.locus.anchor.dot() {
                    let _ = self.anchored.insert(anchor_dot);
                    // An applied move anchored to an unborn dot: its
                    // arrival re-earns the verdicts from this play on.
                    if !as_identity(anchor_dot)
                        .is_some_and(|dot| self.performed.skeleton.contains(dot))
                    {
                        let earliest = self.dangling.entry(anchor_dot).or_insert(Some(play.key));
                        *earliest = (*earliest).min(Some(play.key));
                    }
                }
            }
            self.plays.push(play);
        }
        self.refused = self
            .plays
            .iter()
            .filter_map(|play| match play.verdict {
                Verdict::Refused => Some(play.key.1),
                Verdict::Applied { .. } => None,
            })
            .collect();
    }

    /// The debug-build belt on the text side of the collation's contract
    /// (the gate review's finding): every already-collated dot the text
    /// still carries must carry the locus it was collated with, unless
    /// movement testimony targets it (its effective locus lawfully
    /// differs from its birth locus). The release build extends trait
    /// law 3's trust exactly as `causal_merge` does; this assert turns a
    /// basis violation into a loud stop in every test and property run
    /// instead of a silently stale reading.
    #[cfg(debug_assertions)]
    fn debug_assert_text_basis(&self, text: &Rhapsody) {
        let targets: BTreeSet<RawDot> = self
            .folded
            .values()
            .map(|testimony| testimony.target)
            .collect();
        for (dot, locus) in text.skeleton.iter() {
            if targets.contains(&dot.into()) {
                continue;
            }
            if let Some(collated) = self.performed.skeleton.get(&dot) {
                debug_assert!(
                    collated == locus,
                    "a collated text keeps the honest basis: an already-collated dot \
                     re-appeared with a different locus (trait law 3; the eager read \
                     is the unconditioned form)"
                );
            }
        }
    }

    /// The release build extends trait law 3's trust exactly as
    /// `causal_merge` does; the belt compiles away.
    #[cfg(not(debug_assertions))]
    #[allow(clippy::unused_self)]
    const fn debug_assert_text_basis(&self, _text: &Rhapsody) {}

    /// The whole-replay fallback: re-earn everything from the pair itself
    /// (always lawful, since the collation is a pure function of the
    /// pair's current state).
    fn rebuild(&mut self, text: &Rhapsody, moves: &Metatheses) {
        #[cfg(test)]
        let (replays, rebuilds, compacted) = (self.replays, self.rebuilds, self.compacted);
        *self = Self::replay_full(text, moves);
        #[cfg(test)]
        {
            self.replays = replays;
            self.rebuilds = rebuilds + 1;
            self.compacted = compacted;
        }
    }

    /// The collation's exhaustive agreement check against the eager
    /// contract form (test machinery): value equality, every positional
    /// read, the placement and visibility verdicts on every woven dot,
    /// and the maintained thread's own structural invariants.
    #[cfg(test)]
    pub(crate) fn check_collation(&self, eager: &Self) {
        assert_eq!(self, eager, "the collation equals the eager replay");
        assert_eq!(
            self.visible_mark, eager.visible_mark,
            "the possession mark adopted the pair's plane exactly (canonical \
             form included: the eager twin's mark IS the pair's plane)"
        );
        let order = self.order();
        assert_eq!(order, eager.order(), "one effective order");
        assert_eq!(
            self.order_len(),
            order.len(),
            "the length aggregate is exact"
        );
        for (offset, &dot) in order.iter().enumerate() {
            assert_eq!(self.order_at(offset), Some(dot), "selection agrees");
            assert_eq!(self.offset_of(dot), Some(offset), "offsets agree");
        }
        assert_eq!(self.order_at(order.len()), None, "out of range refuses");
        for (dot, _) in self.performed.skeleton.iter() {
            assert_eq!(
                self.is_reachable(dot),
                eager.is_reachable(dot),
                "placement verdicts agree"
            );
            assert_eq!(
                self.is_visible(dot),
                eager.is_visible(dot),
                "visibility verdicts agree"
            );
        }
        if self.performed.skeleton_len() <= 256 {
            let mut coordinate = self
                .ancestry
                .clone()
                .expect("the in-memory effective plane admits its coordinate");
            coordinate
                .validate(&self.performed.skeleton, &self.performed.unplaced)
                .expect("the coordinate matches every effective parent edge");
            if !self.performed.unplaced.is_empty() {
                self.performed.check_order_thread();
                return;
            }
            let dots: Vec<Dot> = self.performed.skeleton.iter().map(|(dot, _)| dot).collect();
            for &descendant in &dots {
                for &ancestor in &dots {
                    let expected = if descendant == ancestor {
                        Relation::Same
                    } else {
                        let mut at = descendant;
                        let mut found = false;
                        for _ in 0..self.performed.skeleton_len() {
                            let Some(parent) = self
                                .performed
                                .skeleton
                                .get(&at)
                                .and_then(|locus| locus.anchor.dot())
                                .and_then(as_identity)
                            else {
                                break;
                            };
                            if parent == ancestor {
                                found = true;
                                break;
                            }
                            at = parent;
                        }
                        if found {
                            Relation::Descendant
                        } else {
                            Relation::NotDescendant
                        }
                    };
                    assert_eq!(
                        coordinate.relation(descendant, ancestor),
                        Ok(expected),
                        "the exact coordinate agrees with the parent walk"
                    );
                }
            }
        }
        self.performed.check_order_thread();
    }

    /// Suffix replays run by collations on this view (test accounting).
    #[cfg(test)]
    pub(crate) const fn collation_replays(&self) -> usize {
        self.replays
    }

    /// Whole-replay fallbacks run by collations on this view (test
    /// accounting).
    #[cfg(test)]
    pub(crate) const fn collation_rebuilds(&self) -> usize {
        self.rebuilds
    }

    /// Pending-vector slots collations on this view have touched
    /// compacting parked departures out (test accounting): zero when no
    /// departure was parked, and the tail from the first parked one
    /// otherwise, never a walk per departure.
    #[cfg(test)]
    pub(crate) const fn collation_compacted(&self) -> usize {
        self.compacted
    }
}