minerva 0.2.0

extern crate alloc;

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

use crate::kairos::Kairos;

use super::super::super::metathesis::{Metatheses, Metathesis};
use super::super::super::{Composer, Dotted};
use super::super::ancestry::{
    AncestryQueryError, AncestryRelation, Coordinate, CoordinateError, QueryError,
};
use super::super::children::ChildPlane;
use super::super::identity::IdentityPlane;
use super::super::placement::Dot;
use super::super::{Anchor, Locus, PlacementEffect, Rhapsody};
#[cfg(any(test, feature = "timing"))]
use super::cycle::{CycleCheck, EnteredCycleGuard};
use super::cycle::{CycleOutcome, WalkBudget, move_forms_cycle};
#[cfg(test)]
use super::cycle::{CycleTerminal, move_forms_a_cycle};
#[cfg(feature = "timing")]
use super::{MovementBatchMoveProfile, ProfiledMovement};
use super::{Play, Recension, Verdict, as_identity, synchronize_coordinate, woven_target};
use crate::metis::dot::RawDot;

/// A forward movement batch refused before it could mutate the record.
///
/// The batch fast path is deliberately narrower than the total
/// [`Recension::collate`] contract. It accepts only placed-to-placed moves
/// whose ranks strictly extend the replay. Those are the live editor shape:
/// each accepted move changes the batch's effective topology immediately,
/// while the positional thread is materialized once when the batch ends.
#[derive(Clone, Copy, Debug, PartialEq, Eq, thiserror::Error)]
pub enum MovementBatchError {
    /// The composer cannot mint another fresh testimony dot at this station.
    #[error("the movement composer station exhausted its dot space")]
    StationExhausted {
        /// The exhausted station.
        station: u32,
    },
    /// Some woven identity is not placed in the current effective topology.
    #[error("the movement batch requires a fully placed effective topology")]
    UnplacedReading,
    /// A fully placed reading could not admit its exact ancestry coordinate.
    #[error("the exact ancestry coordinate is unavailable")]
    AncestryUnavailable(#[source] CoordinateError),
    /// Tentative suffix replay did not restore a fully placed result.
    #[error("moving {target:?} leaves {first_unplaced:?} unplaced")]
    UnplacedResult {
        /// The requested movement target.
        target: RawDot,
        /// The first effective identity left without a place.
        first_unplaced: RawDot,
    },
    /// The target has no effective locus in this recension.
    #[error("the movement target is not woven")]
    UnknownTarget {
        /// The requested target.
        target: RawDot,
    },
    /// The target is pending behind an unborn anchor.
    #[error("the movement target is not placed")]
    UnplacedTarget {
        /// The requested target.
        target: RawDot,
    },
    /// The destination anchor has no effective locus in this recension.
    #[error("the movement anchor is not woven")]
    UnknownAnchor {
        /// The requested anchor dot.
        anchor: RawDot,
    },
    /// The destination anchor is pending behind an unborn anchor.
    #[error("the movement anchor is not placed")]
    UnplacedAnchor {
        /// The requested anchor dot.
        anchor: RawDot,
    },
    /// The move does not extend the replay's rank order.
    #[error("the movement rank does not strictly extend the replay")]
    NonmonotonicRank,
}

trait MovementObserver {
    #[cfg(any(test, feature = "timing"))]
    const OBSERVES_CYCLE_DETAILS: bool = false;

    type Mark: Copy;
    type Output;

    fn mark(&self) -> Self::Mark;
    fn play_searched(
        &mut self,
        since: Self::Mark,
        until: Self::Mark,
        visits: usize,
        suffix_start: usize,
        raw_suffix_len: usize,
    );
    fn suffix_unwound(
        &mut self,
        since: Self::Mark,
        until: Self::Mark,
        applied: usize,
        refused: usize,
    );
    fn suffix_edited(
        &mut self,
        since: Self::Mark,
        until: Self::Mark,
        superseded: usize,
        replay_len: usize,
    );
    #[cfg(any(test, feature = "timing"))]
    fn cycle_checked(&mut self, since: Self::Mark, until: Self::Mark, check: CycleCheck);
    fn locus_replaced(&mut self);
    fn suffix_replayed(&mut self, since: Self::Mark, until: Self::Mark);
    fn topology_placed(&mut self, since: Self::Mark);
    fn record_composed(&mut self, since: Self::Mark);
    fn finish(self) -> Self::Output;
}

struct UnobservedMovement;

type MovementReceipt = (Dot, Dotted<Metatheses>);

impl MovementObserver for UnobservedMovement {
    type Mark = ();
    type Output = ();

    fn mark(&self) -> Self::Mark {}
    fn play_searched(&mut self, (): Self::Mark, (): Self::Mark, _: usize, _: usize, _: usize) {}
    fn suffix_unwound(&mut self, (): Self::Mark, (): Self::Mark, _: usize, _: usize) {}
    fn suffix_edited(&mut self, (): Self::Mark, (): Self::Mark, _: usize, _: usize) {}
    #[cfg(any(test, feature = "timing"))]
    fn cycle_checked(&mut self, (): Self::Mark, (): Self::Mark, _: CycleCheck) {}
    fn locus_replaced(&mut self) {}
    fn suffix_replayed(&mut self, (): Self::Mark, (): Self::Mark) {}
    fn topology_placed(&mut self, (): Self::Mark) {}
    fn record_composed(&mut self, (): Self::Mark) {}
    fn finish(self) -> Self::Output {}
}

#[cfg(feature = "timing")]
#[derive(Default)]
struct MovementTimer {
    profile: MovementBatchMoveProfile,
}

#[cfg(feature = "timing")]
impl MovementObserver for MovementTimer {
    const OBSERVES_CYCLE_DETAILS: bool = true;

    type Mark = std::time::Instant;
    type Output = MovementBatchMoveProfile;

    fn mark(&self) -> Self::Mark {
        std::time::Instant::now()
    }

    fn play_searched(
        &mut self,
        since: Self::Mark,
        until: Self::Mark,
        visits: usize,
        suffix_start: usize,
        raw_suffix_len: usize,
    ) {
        self.profile.topology.phases.play_search = until.duration_since(since);
        self.profile.topology.work.play_search_visits = visits;
        self.profile.topology.work.suffix_start = suffix_start;
        self.profile.topology.work.raw_suffix_len = raw_suffix_len;
    }

    fn suffix_unwound(
        &mut self,
        since: Self::Mark,
        until: Self::Mark,
        applied: usize,
        refused: usize,
    ) {
        self.profile.topology.phases.suffix_unwind = until.duration_since(since);
        self.profile.topology.work.applied_plays_unwound = applied;
        self.profile.topology.work.refused_plays_skipped_unwind = refused;
    }

    fn suffix_edited(
        &mut self,
        since: Self::Mark,
        until: Self::Mark,
        superseded: usize,
        replay_len: usize,
    ) {
        self.profile.topology.phases.suffix_edit = until.duration_since(since);
        self.profile.topology.work.superseded_plays_removed = superseded;
        self.profile.topology.work.replay_len = replay_len;
    }

    fn cycle_checked(&mut self, since: Self::Mark, until: Self::Mark, check: CycleCheck) {
        let validation = &mut self.profile.topology.cycle_validation;
        validation.total += until.duration_since(since);
        check.add_to(&mut validation.work);
    }

    fn locus_replaced(&mut self) {
        self.profile.topology.work.locus_replacements += 1;
    }

    fn suffix_replayed(&mut self, since: Self::Mark, until: Self::Mark) {
        self.profile.topology.phases.suffix_replay = until.duration_since(since);
    }

    fn topology_placed(&mut self, since: Self::Mark) {
        self.profile.topology.total = since.elapsed();
        #[allow(deprecated)]
        {
            self.profile.topology_placement = self.profile.topology.total;
        }
    }

    fn record_composed(&mut self, since: Self::Mark) {
        self.profile.record_composition = since.elapsed();
    }

    fn finish(self) -> Self::Output {
        self.profile
    }
}

/// A scoped batch of strictly forward movement testimony.
///
/// The batch couples the movement [`Composer`] to the maintained reading so
/// record and decision state cannot advance independently. Each successful
/// [`move_to`](Self::move_to) updates a topology-only reading and its exact
/// ancestry coordinate immediately;
/// subsequent placement reads therefore see exactly the state the preceding
/// testimony chose. The positional thread is materialized once from the final
/// effective identity plane before [`Recension::with_movement_batch`] returns;
/// `Drop` repeats the same idempotent finalizer only during unwinding. A live batch
/// exposes only the placement reads that the topology can answer exactly; full
/// order and offset reads become available again only after the batch borrow
/// ends.
///
/// This is not a general collation transaction. Delayed, externally
/// retracting, pending, or out-of-rank testimony belongs on
/// [`Recension::collate`], whose undo-redo replay is total for those shapes. A
/// strictly forward cycle-former is admissible record state and remains a
/// surfaced replay refusal here.
pub struct MovementBatch<'a> {
    recension: &'a mut Recension,
    text: &'a Rhapsody,
    moves: &'a mut Composer<Metatheses>,
    placement: BatchPlacement,
    changed: bool,
}

/// The effective placement coordinates maintained inside a movement batch.
/// No positional thread exists here by construction.
struct BatchPlacement {
    skeleton: IdentityPlane,
    children: ChildPlane,
    unplaced: BTreeSet<Dot>,
    ancestry: Option<Coordinate>,
    anchored: BTreeSet<RawDot>,
    highest_rank: Option<Kairos>,
    plays: Vec<Play>,
}

/// A movement proven admissible against the current batch topology.
#[derive(Clone, Copy)]
struct ValidatedMovement {
    witness: Dot,
    testimony: Metathesis,
}

impl BatchPlacement {
    fn from_recension(recension: &Recension, ancestry: Coordinate) -> Self {
        Self {
            skeleton: recension.performed.skeleton.clone(),
            children: recension.performed.children.clone(),
            unplaced: recension.performed.unplaced.clone(),
            ancestry: Some(ancestry),
            anchored: recension.anchored.clone(),
            highest_rank: recension
                .folded
                .values()
                .map(|testimony| testimony.to.rank)
                .max(),
            plays: recension.plays.clone(),
        }
    }

    fn locus(&self, target: Dot) -> Option<Locus> {
        self.skeleton.get(&target)
    }

    fn children_of(&self, anchor: Anchor) -> impl Iterator<Item = Dot> + '_ {
        self.children.iter(&self.skeleton, anchor)
    }

    fn anchor_for_visual_insert(&self, after: Option<RawDot>) -> Anchor {
        let base = after.map_or(Anchor::Origin, Anchor::After);
        let Some(head) = self
            .children
            .bucket(&self.skeleton, base)
            .map(|bucket| bucket.first())
        else {
            return base;
        };
        let mut successor = head;
        while let Some(next) = self
            .children
            .bucket(&self.skeleton, Anchor::Before(successor.into()))
            .map(|bucket| bucket.last())
        {
            successor = next;
        }
        Anchor::Before(successor.into())
    }

    fn collect_region(&self, root: Dot) -> Vec<Dot> {
        let mut region = Vec::new();
        let mut stack = alloc::vec![root];
        while let Some(dot) = stack.pop() {
            region.push(dot);
            for anchor in [Anchor::After(dot.into()), Anchor::Before(dot.into())] {
                for child in self.children.iter(&self.skeleton, anchor) {
                    if !self.unplaced.contains(&child) {
                        stack.push(child);
                    }
                }
            }
        }
        region
    }

    fn repair_placement_from(&mut self, root: Dot) -> Vec<Dot> {
        let mut repaired = alloc::vec![root];
        let mut worklist = alloc::vec![root];
        while let Some(parent) = worklist.pop() {
            for anchor in [Anchor::After(parent.into()), Anchor::Before(parent.into())] {
                for child in self.children.iter(&self.skeleton, anchor) {
                    if self.unplaced.remove(&child) {
                        repaired.push(child);
                        worklist.push(child);
                    }
                }
            }
        }
        repaired
    }

    #[cfg(any(test, feature = "timing"))]
    fn entered_cycle_guard(&self, target: RawDot, anchor: Anchor) -> Option<EnteredCycleGuard> {
        if anchor.dot() == Some(target) {
            return Some(EnteredCycleGuard::SelfAnchor);
        }
        if !self.anchored.contains(&target) {
            return None;
        }
        let has_live_child = [Anchor::Before(target), Anchor::After(target)]
            .into_iter()
            .any(|anchor| self.children.bucket(&self.skeleton, anchor).is_some());
        if has_live_child {
            Some(EnteredCycleGuard::LiveChild)
        } else {
            Some(EnteredCycleGuard::StaleMonotone)
        }
    }

    fn cycle_outcome(&mut self, target: RawDot, anchor: Anchor) -> CycleOutcome {
        move_forms_cycle(
            self.ancestry.as_mut(),
            target,
            anchor,
            &self.anchored,
            WalkBudget::for_skeleton(self.skeleton.len()),
            |link| {
                as_identity(link)
                    .and_then(|link| self.skeleton.get(&link))
                    .map(|locus| locus.anchor.dot())
            },
        )
    }

    fn replace_locus(&mut self, raw_target: RawDot, to: Locus) -> Locus {
        let target = woven_target(raw_target);
        let old = self
            .skeleton
            .get(&target)
            .expect("a replayed target remains woven");
        if old == to {
            return old;
        }
        let was_placed = !self.unplaced.contains(&target);
        let placed_new = match to.anchor.dot().map(Dot::try_from) {
            None => true,
            Some(Ok(anchor)) => {
                anchor != target
                    && self.skeleton.contains(anchor)
                    && !self.unplaced.contains(&anchor)
            }
            // A non-dot destination coordinate names no woven element
            // (ruling R-91).
            Some(Err(_)) => false,
        };
        let parked = (was_placed && !placed_new).then(|| self.collect_region(target));
        let old = self
            .skeleton
            .remove(target)
            .expect("a replayed target remains woven");
        let _ = self.children.remove(&self.skeleton, old.anchor, target);
        let inserted = self.skeleton.insert(target, to);
        debug_assert!(inserted, "a removed target re-inserts under the same cap");
        self.children.insert(&self.skeleton, to.anchor, target);
        let effect = match (was_placed, placed_new) {
            (true, true) => PlacementEffect::Reparented {
                dot: target,
                parent: to.anchor.dot(),
            },
            (true, false) => {
                let parked = parked.expect("the placed region was collected");
                self.unplaced.extend(parked.iter().copied());
                PlacementEffect::Parked(parked)
            }
            (false, true) => {
                let _ = self.unplaced.remove(&target);
                PlacementEffect::Repaired(self.repair_placement_from(target))
            }
            (false, false) => PlacementEffect::Unchanged,
        };
        let first_unplaced = self.unplaced.first().copied();
        let mut ancestry = self
            .ancestry
            .take()
            .expect("the batch returns its coordinate only during finalization");
        let synchronized =
            synchronize_coordinate(&mut ancestry, &self.skeleton, first_unplaced, effect);
        self.ancestry = Some(ancestry);
        synchronized.expect("an admitted batch transition preserves its exact coordinate");
        old
    }

    fn validate(
        &self,
        witness: Dot,
        testimony: Metathesis,
    ) -> Result<ValidatedMovement, MovementBatchError> {
        if self
            .highest_rank
            .is_some_and(|highest| testimony.to.rank <= highest)
        {
            return Err(MovementBatchError::NonmonotonicRank);
        }
        // The target and the destination anchor are payload coordinates:
        // a non-dot one is woven nowhere, so it takes the unknown arm
        // exactly as an unwoven coordinate does (ruling R-91).
        let Some(target) = as_identity(testimony.target).filter(|&d| self.skeleton.contains(d))
        else {
            return Err(MovementBatchError::UnknownTarget {
                target: testimony.target,
            });
        };
        if self.unplaced.contains(&target) {
            return Err(MovementBatchError::UnplacedTarget {
                target: testimony.target,
            });
        }
        if let Some(raw_anchor) = testimony.to.anchor.dot() {
            let Some(anchor) = as_identity(raw_anchor).filter(|&d| self.skeleton.contains(d))
            else {
                return Err(MovementBatchError::UnknownAnchor { anchor: raw_anchor });
            };
            if self.unplaced.contains(&anchor) {
                return Err(MovementBatchError::UnplacedAnchor { anchor: raw_anchor });
            }
        }
        Ok(ValidatedMovement { witness, testimony })
    }

    fn apply<O: MovementObserver>(
        &mut self,
        movement: ValidatedMovement,
        observer: &mut O,
    ) -> Result<(), MovementBatchError> {
        let ValidatedMovement { witness, testimony } = movement;
        let old_highest_rank = self.highest_rank;
        let search_started = observer.mark();
        let mut search_visits = 0usize;
        let from = self
            .plays
            .iter()
            .position(|play| {
                search_visits += 1;
                play.target == testimony.target
            })
            .unwrap_or(self.plays.len());
        let raw_suffix_len = self.plays.len() - from;
        let search_finished = observer.mark();
        observer.play_searched(
            search_started,
            search_finished,
            search_visits,
            from,
            raw_suffix_len,
        );
        let unwind_started = observer.mark();
        let mut applied_unwound = 0usize;
        let mut refused_skipped = 0usize;
        for position in (from..self.plays.len()).rev() {
            if let Verdict::Applied { displaced } = self.plays[position].verdict {
                applied_unwound += 1;
                let _ = self.replace_locus(self.plays[position].target, displaced);
            } else {
                refused_skipped += 1;
            }
        }
        let unwind_finished = observer.mark();
        observer.suffix_unwound(
            unwind_started,
            unwind_finished,
            applied_unwound,
            refused_skipped,
        );
        let edit_started = observer.mark();
        let old_suffix = self.plays.split_off(from);
        let mut suffix: Vec<Play> = old_suffix
            .iter()
            .filter(|play| play.target != testimony.target)
            .cloned()
            .collect();
        let superseded = raw_suffix_len - suffix.len();
        suffix.push(Play {
            key: (testimony.to.rank, witness),
            target: testimony.target,
            locus: testimony.to,
            verdict: Verdict::Refused,
        });
        let replay_len = suffix.len();
        let edit_finished = observer.mark();
        observer.suffix_edited(edit_started, edit_finished, superseded, replay_len);
        let mut inserted_anchors = Vec::new();
        self.replay_suffix(suffix, &mut inserted_anchors, observer);
        if let Some(first_unplaced) = self.unplaced.first().copied() {
            self.rollback_suffix(from, &old_suffix, inserted_anchors);
            self.highest_rank = old_highest_rank;
            debug_assert!(
                self.unplaced.is_empty(),
                "rollback restores admitted readiness"
            );
            return Err(MovementBatchError::UnplacedResult {
                target: testimony.target,
                first_unplaced: first_unplaced.into(),
            });
        }
        self.highest_rank = Some(testimony.to.rank);
        Ok(())
    }

    fn replay_suffix<O: MovementObserver>(
        &mut self,
        suffix: Vec<Play>,
        inserted_anchors: &mut Vec<RawDot>,
        observer: &mut O,
    ) {
        let replay_started = observer.mark();
        for mut play in suffix {
            #[cfg(any(test, feature = "timing"))]
            let validation_started = observer.mark();
            let outcome = self.cycle_outcome(play.target, play.locus.anchor);
            #[cfg(any(test, feature = "timing"))]
            let validation_finished = observer.mark();
            #[cfg(any(test, feature = "timing"))]
            if O::OBSERVES_CYCLE_DETAILS {
                let check = match (
                    self.entered_cycle_guard(play.target, play.locus.anchor),
                    outcome,
                ) {
                    (_, CycleOutcome::Skipped) => CycleCheck::Skipped,
                    (Some(guard), CycleOutcome::Coordinate(terminal)) => {
                        CycleCheck::Coordinate { guard, terminal }
                    }
                    (Some(guard), CycleOutcome::Walk(walk)) => CycleCheck::Walk { guard, walk },
                    _ => unreachable!("the exact guard classification agrees with the cycle walk"),
                };
                observer.cycle_checked(validation_started, validation_finished, check);
            }
            if outcome.must_refuse() {
                play.verdict = Verdict::Refused;
            } else {
                let displaced = self.replace_locus(play.target, play.locus);
                if let Some(anchor) = play.locus.anchor.dot()
                    && self.anchored.insert(anchor)
                {
                    inserted_anchors.push(anchor);
                }
                observer.locus_replaced();
                play.verdict = Verdict::Applied { displaced };
            }
            self.plays.push(play);
        }
        let replay_finished = observer.mark();
        observer.suffix_replayed(replay_started, replay_finished);
    }

    fn rollback_suffix(&mut self, from: usize, old_suffix: &[Play], inserted_anchors: Vec<RawDot>) {
        for position in (from..self.plays.len()).rev() {
            if let Verdict::Applied { displaced } = self.plays[position].verdict {
                let _ = self.replace_locus(self.plays[position].target, displaced);
            }
        }
        self.plays.truncate(from);
        for play in old_suffix {
            if matches!(play.verdict, Verdict::Applied { .. }) {
                let _ = self.replace_locus(play.target, play.locus);
            }
            self.plays.push(play.clone());
        }
        for anchor in inserted_anchors {
            let removed = self.anchored.remove(&anchor);
            debug_assert!(removed, "only newly inserted anchors roll back");
        }
    }
}

impl MovementBatch<'_> {
    fn finish(&mut self) {
        if self.changed {
            let skeleton = core::mem::replace(&mut self.placement.skeleton, IdentityPlane::new());
            self.recension.performed = Rhapsody::from_plane(skeleton, self.text.visible.clone());
            self.recension.plays = core::mem::take(&mut self.placement.plays);
            self.recension.anchored = core::mem::take(&mut self.placement.anchored);
            self.recension.refused = self
                .recension
                .plays
                .iter()
                .filter_map(|play| match play.verdict {
                    Verdict::Refused => Some(play.key.1),
                    Verdict::Applied { .. } => None,
                })
                .collect();
            self.recension.folded = self
                .moves
                .state()
                .store()
                .iter()
                .map(|(dot, testimony)| (dot, *testimony))
                .collect();
            self.changed = false;
        }
        if let Some(ancestry) = self.placement.ancestry.take() {
            self.recension.ancestry = Ok(ancestry);
        }
    }

    /// The effective locus after every move accepted by this batch.
    #[must_use]
    pub fn locus(&self, dot: Dot) -> Option<Locus> {
        self.placement.locus(dot)
    }

    /// Effective children in stored sibling order after every accepted move.
    pub fn children_of(&self, anchor: Anchor) -> impl Iterator<Item = Dot> + '_ {
        self.placement.children_of(anchor)
    }

    /// The visual-insert anchor in the batch's current effective topology.
    #[must_use]
    pub fn anchor_for_visual_insert(&self, after: Option<RawDot>) -> Anchor {
        self.placement.anchor_for_visual_insert(after)
    }

    /// Returns the current batch reading's inclusive ancestry relation.
    ///
    /// # Errors
    ///
    /// Returns a typed error for unknown operands or bounded structural
    /// exhaustion. Successful movements always leave the batch ready for the
    /// next query.
    pub fn relation(
        &mut self,
        descendant: Dot,
        ancestor: Dot,
    ) -> Result<AncestryRelation, AncestryQueryError> {
        self.placement
            .ancestry
            .as_mut()
            .ok_or(AncestryQueryError::InvariantViolation)?
            .relation(descendant, ancestor)
            .map_err(QueryError::into_public)
    }

    /// Records and applies one strictly forward placed-to-placed move.
    ///
    /// A typed validation error is inert: the movement record and batch reading
    /// remain unchanged. Success supersedes every testimony this composer
    /// currently holds for `target`, returns the minted dot and causal delta,
    /// and makes the resulting replay verdict visible to the next placement
    /// read. A cycle-forming testimony is still a successful record write; its
    /// refusal is the replay's deterministic decision, just as in ordinary
    /// collation.
    ///
    /// # Errors
    ///
    /// Refuses before mutation unless the target and destination are placed
    /// in the batch reading and `to.rank` strictly extends every testimony
    /// already folded into it.
    pub fn move_to(
        &mut self,
        target: RawDot,
        to: Locus,
    ) -> Result<(Dot, Dotted<Metatheses>), MovementBatchError> {
        self.move_to_observed(target, to, UnobservedMovement)
            .map(|(receipt, ())| receipt)
    }

    /// Records, applies, and attributes one strictly forward move.
    ///
    /// This first-party measurement surface is enabled by the `timing`
    /// feature. It shares the implementation used by [`Self::move_to`]; normal
    /// builds contain no clock reads.
    ///
    /// # Errors
    ///
    /// Returns the same inert validation errors as [`Self::move_to`].
    #[cfg(feature = "timing")]
    pub fn move_to_profiled(
        &mut self,
        target: RawDot,
        to: Locus,
    ) -> Result<ProfiledMovement, MovementBatchError> {
        self.move_to_observed(target, to, MovementTimer::default())
            .map(|((witness, delta), profile)| ProfiledMovement {
                witness,
                delta,
                profile,
            })
    }

    fn move_to_observed<O: MovementObserver>(
        &mut self,
        target: RawDot,
        to: Locus,
        mut observer: O,
    ) -> Result<(MovementReceipt, O::Output), MovementBatchError> {
        let testimony = Metathesis { target, to };
        let witness = self.moves.state().next_dot(self.moves.station());
        if self.moves.state().context().contains(witness) {
            return Err(MovementBatchError::StationExhausted {
                station: witness.station(),
            });
        }
        let movement = self.placement.validate(witness, testimony)?;
        let phase_started = observer.mark();
        self.placement.apply(movement, &mut observer)?;
        observer.topology_placed(phase_started);
        let phase_started = observer.mark();
        let receipt = self.moves.compose_super(
            |assigned| Metatheses::singleton(assigned, testimony),
            |held| held.moves_of(target),
        );
        observer.record_composed(phase_started);
        debug_assert_eq!(receipt.0, witness, "the composer minted the previewed dot");
        self.changed = true;
        Ok((receipt, observer.finish()))
    }
}

impl Drop for MovementBatch<'_> {
    fn drop(&mut self) {
        self.finish();
    }
}

impl Recension {
    /// Runs a scoped batch of strictly forward placed-to-placed moves.
    ///
    /// The method first collates any state already present in `moves`, then
    /// retains ownership of the batch while `use_batch` receives only a mutable
    /// borrow. After the callback returns, the method explicitly materializes
    /// the positional thread before releasing either record or recension back
    /// to the caller. Safe destructor suppression therefore cannot separate the
    /// two. [`MovementBatch`]'s `Drop` is only the panic/unwind backstop.
    ///
    /// # Errors
    ///
    /// Refuses when any woven identity is unplaced in either the canonical text
    /// or current effective reading, or when the exact coordinate is
    /// structurally unavailable. A tentative suffix that cannot return its
    /// privately parked history to a fully placed state is rolled back and
    /// returned as [`MovementBatchError::UnplacedResult`].
    pub fn with_movement_batch<'a, R>(
        &'a mut self,
        text: &'a Rhapsody,
        moves: &'a mut Composer<Metatheses>,
        use_batch: impl FnOnce(&mut MovementBatch<'a>) -> R,
    ) -> Result<R, MovementBatchError> {
        self.collate(text, moves.state().store());
        if !text.unplaced.is_empty() || !self.performed.unplaced.is_empty() {
            return Err(MovementBatchError::UnplacedReading);
        }
        let ancestry = match core::mem::replace(
            &mut self.ancestry,
            Err(CoordinateError::InvariantViolation),
        ) {
            Ok(ancestry) => ancestry,
            Err(error) => {
                self.ancestry = Err(error);
                return Err(MovementBatchError::AncestryUnavailable(error));
            }
        };
        let placement = BatchPlacement::from_recension(self, ancestry);
        let mut batch = MovementBatch {
            recension: self,
            text,
            moves,
            placement,
            changed: false,
        };
        let result = use_batch(&mut batch);
        batch.finish();
        Ok(result)
    }
}

#[cfg(test)]
mod tests;