minerva 0.2.0

//! Placement transitions and derived-coordinate maintenance.

extern crate alloc;

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

use super::children::ChildPlane;
use super::identity::IdentityPlane;
use super::placement::{Dot, RawDot};
use super::thread::OrderThread;
use super::wire;
use super::{Anchor, Locus, Rhapsody};

/// Groups an ascending dot list into its maximal contiguous per-station
/// runs, `(station, first, len)`: the bulk-settle read (R-19). A span past
/// `u32::MAX` dots yields in chunks; a scattered list yields singletons.
fn contiguous_runs(dots: &[Dot]) -> impl Iterator<Item = (u32, NonZeroU64, u32)> + '_ {
    let mut at = 0usize;
    core::iter::from_fn(move || {
        let first = *dots.get(at)?;
        let (station, base) = (first.station(), first.counter_nonzero());
        let mut len = 1u32;
        at += 1;
        while len < u32::MAX
            && dots.get(at)
                == base
                    .checked_add(u64::from(len))
                    .map(|next| Dot::new(station, next))
                    .as_ref()
        {
            len += 1;
            at += 1;
        }
        Some((station, base, len))
    })
}

/// The `k`-th member of a chain run whose head is `first`: the run's dots
/// are consecutive counters of one station, so every member is an identity
/// by construction (ruling R-91).
const fn run_member(first: Dot, k: u64) -> Dot {
    Dot::new(first.station(), first.counter_nonzero().saturating_add(k))
}

/// An anchor coordinate a placement verdict already proved woven, read
/// back as the identity it names. The verdict refuses a non-dot
/// coordinate, so this crossing cannot fail here (ruling R-91).
pub(in crate::metis::rhapsody) fn placed_anchor(raw: RawDot) -> Dot {
    Dot::try_from(raw).expect("a placed element's anchor is a woven dot")
}

/// A frame of the region-threading walk ([`Rhapsody::thread_region`]):
/// `Visit` schedules an element's reading order, `Emit` threads its slot.
/// The same two-frame shape as the traversal's own stack.
#[derive(Clone, Copy, Debug)]
enum RegionFrame {
    /// Expand this dot's children and schedule its own emit.
    Visit(Dot),
    /// Thread this dot's slot at the next consecutive position.
    Emit(Dot),
}

impl Rhapsody {
    /// Applies the in-place merge's delta-sized visibility flips to the
    /// maintained order thread and occupancy plane (the survivor fold's
    /// own residual arms, so both derived coordinates ride the same
    /// delta-sized settle).
    pub(in crate::metis::rhapsody) fn settle_thread_after_merge(
        &mut self,
        expelled: &[Dot],
        admitted: &[Dot],
    ) {
        // Both residual lists are ascending, so each settles the thread's
        // masks in maximal contiguous dot runs (one fragment-span write
        // instead of a descent per dot: the R-19 bulk arm; a scattered
        // residual degrades to singleton runs, the per-dot cost).
        for (station, first, len) in contiguous_runs(expelled) {
            self.thread.set_visible_run(station, first, len, false);
        }
        for (station, first, len) in contiguous_runs(admitted) {
            self.thread.set_visible_run(station, first, len, true);
        }
        // Each residual list is ascending, so each settles the occupancy
        // plane as one staged batch (a bulk expulsion is one linear
        // merge, never a row removal per dot: the gate review's
        // quadratic finding).
        self.visible_pages
            .apply_flips(expelled.iter().map(|&dot| (dot, false)));
        self.visible_pages
            .apply_flips(admitted.iter().map(|&dot| (dot, true)));
    }

    /// Flips one dot's visibility on the carried set and its occupancy
    /// shell together: the choke point every per-dot visibility mutation
    /// routes through (`weave`, the collation's flip and birth folds), so
    /// the two forms cannot drift.
    pub(in crate::metis::rhapsody) fn note_visibility(&mut self, dot: Dot, present: bool) {
        let (station, index) = (dot.station(), dot.counter());
        if present {
            let fresh = self.visible.insert(dot);
            // The plane never LEADS its carried set: a genuinely fresh
            // visibility may not find its bit already set (the collation's
            // batched settle excludes this call's births for exactly this;
            // the belt is what makes that hygiene observable).
            debug_assert!(
                !fresh || !self.visible_pages.contains(station, index),
                "the occupancy plane never leads its carried set"
            );
        } else {
            let _ = self.visible.remove(dot);
        }
        self.visible_pages.set(station, index, present);
    }

    /// Rebuilds the order thread from the walk (the contract form), the
    /// settle path for mutations the thread cannot absorb incrementally.
    pub(in crate::metis::rhapsody) fn rebuild_thread(&mut self) {
        let slots: Vec<(Dot, bool)> = {
            let mut walk = self.order_walk();
            let mut slots = Vec::with_capacity(self.skeleton.len());
            while let Some(entry) = walk.next_slot() {
                slots.push(entry);
            }
            slots
        };
        let mut thread = OrderThread::from_slots(slots.iter().copied());
        // Collect the last-child edges once, skipping unplaced components
        // (an unplaced last child has an unplaced parent by the placement
        // induction, so the whole edge sits outside the thread).
        let mut ends_edges: BTreeMap<Dot, Dot> = BTreeMap::new();
        let mut starts_edges: BTreeMap<Dot, Dot> = BTreeMap::new();
        for (anchor, last) in self.children.last_children(&self.skeleton) {
            if self.unplaced.contains(&last) {
                continue;
            }
            match anchor {
                Anchor::Origin => {}
                // A non-dot anchor coordinate names no placed dot, so the
                // rebuild's own placement filter would drop the edge in
                // any case: the funnel is the same verdict, said earlier
                // (ruling R-91).
                Anchor::Before(parent) => {
                    if let Ok(parent) = Dot::try_from(parent) {
                        let _ = starts_edges.insert(parent, last);
                    }
                }
                Anchor::After(parent) => {
                    if let Ok(parent) = Dot::try_from(parent) {
                        let _ = ends_edges.insert(parent, last);
                    }
                }
            }
        }
        thread.rebuild_regions(
            slots.iter().map(|&(dot, _)| dot),
            &ends_edges,
            &starts_edges,
        );
        self.thread = thread;
    }

    /// The placement verdict for `dot` landing on `anchor`: the origin is
    /// always placed; a coordinate places when it names a woven, placed
    /// dot other than `dot` itself (the self-anchor exclusion the S200
    /// review pinned; the target's own re-homing stays out of the
    /// verdict). A coordinate that is not a dot names no woven element,
    /// so it parks, exactly as the plane's own zero-index refusal made it
    /// park before the type carried the law (ruling R-91).
    pub(in crate::metis::rhapsody) fn anchor_places(&self, dot: Dot, anchor: Anchor) -> bool {
        anchor.dot().is_none_or(|raw| {
            Dot::try_from(raw).is_ok_and(|anchor| {
                anchor != dot && self.skeleton.contains(anchor) && !self.unplaced.contains(&anchor)
            })
        })
    }

    /// Materializes the placement set from a skeleton and its child plane: the
    /// structural walk from the origin over both side buckets reaches every
    /// placed dot, and whatever it cannot reach is dangling. `O(n log n)` for
    /// `n` woven elements, paid once per build-from-parts beside
    /// [`ChildPlane::build`]; the walk is iterative over a
    /// worklist, never recursive (skeleton depth is attacker-controllable).
    pub(in crate::metis::rhapsody) fn build_unplaced(
        skeleton: &IdentityPlane,
        children: &ChildPlane,
    ) -> alloc::collections::BTreeSet<Dot> {
        let mut reached: alloc::collections::BTreeSet<Dot> = alloc::collections::BTreeSet::new();
        let mut worklist: Vec<Dot> = children.iter(skeleton, Anchor::Origin).collect();
        while let Some(dot) = worklist.pop() {
            if !reached.insert(dot) {
                continue;
            }
            for anchor in [Anchor::After(dot.into()), Anchor::Before(dot.into())] {
                worklist.extend(children.iter(skeleton, anchor));
            }
        }
        skeleton
            .iter()
            .map(|(dot, _)| dot)
            .filter(|dot| !reached.contains(dot))
            .collect()
    }

    /// Records a woven element: its locus enters the skeleton and the dot
    /// becomes visible. Returns whether it was recorded.
    ///
    /// Refuses (returns `false`, no change) a dot already woven (a dot
    /// names one write; trait law 3, content is never reassigned). The
    /// non-dot `0` needs no arm: [`Dot`](crate::metis::Dot) carries that
    /// law (ruling R-91). A dangling anchor (a coordinate not yet woven
    /// here) is *accepted*: the element stays unreachable in
    /// [`order()`](Self::order) until the anchor's locus arrives (an
    /// out-of-order delta; the merge that carries the anchor makes it
    /// reachable, the holdback discipline living one layer down in the
    /// caller).
    #[must_use = "the write is refused (returns false) for an already-carried dot or a full plane"]
    pub fn weave(&mut self, dot: Dot, locus: Locus) -> bool {
        if !self.record_locus(dot, locus) {
            return false;
        }
        self.note_visibility(dot, true);
        let _ = self.thread.set_visible(dot, true);
        true
    }

    /// Weaves a whole chain run in one write: `run.len()` consecutive elements
    /// for dots `first_dot .. first_dot + run.len()`. The head sits at the
    /// caller's `anchor` and carries the reservation's first rank. Each
    /// interior is anchored `After` its predecessor with the successor rank.
    /// These are exactly the loci that many individual [`weave`](Self::weave)
    /// calls over the same reservation would record (the agreement law). The
    /// interior derivation is the codec's own chain law, so a woven run
    /// coalesces by construction in the identity plane, the order thread, the
    /// region segments, and the snapshot frame. The run-grade ingest door
    /// (ruling R-19): a paste, import, or file-open stops paying one skeleton
    /// probe, one bucket insert, and one thread descent per character.
    ///
    /// All-or-nothing: refuses (returns `false`, no change) a run passing
    /// the `u64::MAX` dot ceiling, a run past the identity plane's
    /// live-entry ceiling (refused in `O(1)`, before any run-sized
    /// allocation), and a run any
    /// of whose dots is already woven; on `true` every element was fresh
    /// and is now visible. A dangling `anchor` is accepted exactly as
    /// [`weave`](Self::weave) accepts it: the whole run parks unreachable
    /// until the anchor's locus arrives. The one contested corner, a
    /// parked dot already waiting on one of the run's own dots (a
    /// crossed out-of-order delivery), degrades to the per-dot path so
    /// its repair protocol runs unchanged; the honest bulk shapes never
    /// pay for that check beyond one emptiness read.
    ///
    /// All members share the reservation's station and kairotic (the
    /// chain law requires both), and the rank succession is the clock
    /// fold's own, so the run's guarantees are
    /// [`KairosRun`](crate::kairos::KairosRun)'s. The reservation is
    /// consumed BY VALUE, and the type is affine (no `Clone`), so spending
    /// one reservation under two identities is a move error rather than a
    /// discipline; a refused weave consumes it too (mint another, the
    /// clock only skips positions).
    // The by-value reservation is the API's affinity (a spend, not a cost):
    // the lint's advice would reopen exactly the reuse the type closes.
    #[allow(clippy::needless_pass_by_value)]
    #[must_use = "the write is refused (returns false) for a full plane, a ceiling-crossing run, or an already-woven dot"]
    pub fn weave_run(
        &mut self,
        first_dot: Dot,
        anchor: Anchor,
        run: crate::kairos::KairosRun,
    ) -> bool {
        let (station, first) = (first_dot.station(), first_dot.counter());
        let len = run.len();
        let Some(last) = first.checked_add(u64::from(len) - 1) else {
            return false;
        };
        // The capacity refusal comes BEFORE the locus derivation: a run
        // longer than the plane can ever hold must refuse in O(1), not
        // attempt a run-sized allocation first (the gate review's
        // round-three finding; `has_capacity_for` is the same
        // authoritative read the bulk insert refuses by).
        if !self.skeleton.has_capacity_for(len as usize) {
            return false;
        }
        // The run's loci, derived once: the head at the caller's anchor,
        // each interior one chain step (zero: the successor rank) past its
        // predecessor. This is the identity plane's own spelling and, by
        // the proven fold agreement, exactly the reservation's members.
        let head = Locus {
            anchor,
            rank: run.first(),
        };
        let mut loci = Vec::with_capacity(len as usize);
        loci.push(head);
        let (mut prev_dot, mut prev) = (RawDot::from(first_dot), head);
        for k in 1..u64::from(len) {
            let next = wire::apply_chain_step(prev_dot.into(), &prev, 0);
            loci.push(next);
            prev_dot = RawDot::new(station, first + k);
            prev = next;
        }
        // The contested corner: a parked dot waiting on one of the run's
        // own dots needs the per-dot repair protocol. Freshness is swept
        // here and capacity was swept above, so the degrade stays
        // all-or-nothing: without the capacity preflight, a run straddling
        // the identity plane's live-entry ceiling could weave its early
        // dots and refuse the rest (the gate review's round-two finding;
        // round three hoisted the check above the locus derivation).
        if !self.unplaced.is_empty() && self.a_parked_dot_waits_within(station, first, last) {
            for k in 0..u64::from(len) {
                if self.skeleton.contains(run_member(first_dot, k)) {
                    return false;
                }
            }
            for (k, &locus) in loci.iter().enumerate() {
                let woven = self.weave(run_member(first_dot, k as u64), locus);
                debug_assert!(woven, "the freshness and capacity sweeps admitted the run");
            }
            return true;
        }
        if !self.record_locus_run(first_dot, &loci, true) {
            return false;
        }
        // Visibility: the carried set first, then its occupancy shell (the
        // plane never leads its carried set), the same discipline the
        // per-dot choke point keeps. The thread's masks were threaded
        // visible at insertion.
        self.visible
            .insert_run(station, first_dot.counter_nonzero(), len);
        self.visible_pages.set_run(station, first, len);
        true
    }

    /// Whether any parked dot's anchor names a dot inside
    /// `(station, first ..= last)`: the [`weave_run`](Self::weave_run) and
    /// merge-fold degrade probe, `O(parked)` against a set that is empty on
    /// every honest in-order path.
    pub(in crate::metis::rhapsody) fn a_parked_dot_waits_within(
        &self,
        station: u32,
        first: u64,
        last: u64,
    ) -> bool {
        self.unplaced.iter().any(|dot| {
            self.skeleton
                .get(dot)
                .and_then(|locus| locus.anchor.dot())
                .is_some_and(|raw| {
                    raw.station == station && raw.counter >= first && raw.counter <= last
                })
        })
    }

    /// The ordering-state half of [`weave_run`](Self::weave_run): records a
    /// whole chain run into the skeleton and every maintained coordinate,
    /// in bulk. `loci` is the run's derived locus column (the head's anchor
    /// is the run's, each interior anchored `After` its predecessor: the
    /// caller guarantees the chain shape, which both callers derive by the
    /// chain law itself). Shared with the in-place causal merge's run arm,
    /// which is why visibility stays out: the survivor law owns that
    /// coordinate, and the thread masks are threaded from `visible`.
    ///
    /// All-or-nothing: the identity plane's bulk insert is the one
    /// freshness and ceiling authority, and nothing else moves on refusal.
    /// `thread_visible` is the mask the run's slots thread under: the local
    /// weave makes its run visible in the same stroke, while the merge arm
    /// threads invisible and lets the survivor fold's settle flip what it
    /// admits (the coordinate split the per-dot paths keep).
    pub(in crate::metis::rhapsody) fn record_locus_run(
        &mut self,
        first_dot: Dot,
        loci: &[Locus],
        thread_visible: bool,
    ) -> bool {
        let (station, first) = (first_dot.station(), first_dot.counter());
        if !self.skeleton.insert_run(first_dot, loci) {
            return false;
        }
        let len = u32::try_from(loci.len()).expect("a run is at most u32 long");
        let last = first + (u64::from(len) - 1);
        self.woven
            .insert_run(station, first_dot.counter_nonzero(), len);
        let anchor = loci[0].anchor;
        // Only the head enters a sibling bucket: each interior is its own
        // predecessor's sole chain child, the relation the coalesced child
        // plane derives from the identity plane per read (arc 11 phase
        // four), so storing it would only materialize what the plane
        // already spells.
        self.children.insert(&self.skeleton, anchor, first_dot);
        // The placement verdict is decided once for the whole run: the
        // interiors chain off the head, so they stand or park with it. An
        // anchor pointing into the run itself is the per-dot form's
        // self-reference refusal writ large: the head would have parked
        // before its anchor was woven, so the run parks.
        let anchor_in_run = anchor.dot().is_some_and(|raw| {
            raw.station == station && raw.counter >= first && raw.counter <= last
        });
        let placed = match anchor.dot().map(Dot::try_from) {
            None => true,
            Some(Ok(a)) => {
                !anchor_in_run && self.skeleton.contains(a) && !self.unplaced.contains(&a)
            }
            // A non-dot anchor coordinate names no woven element, so the
            // run parks exactly as it did before the type carried the law
            // (ruling R-91).
            Some(Err(_)) => false,
        };
        if placed {
            debug_assert!(
                !self.visible.contains(first_dot),
                "a fresh run's dots cannot be visible before the weave"
            );
            let region_begin = self.walk_slot_position(first_dot, anchor);
            self.thread.insert_run(
                region_begin,
                station,
                first_dot.counter_nonzero(),
                len,
                thread_visible,
            );
            let registered = self
                .thread
                .insert_region_run(first_dot, run_member(first_dot, u64::from(len) - 1));
            debug_assert!(registered, "a fresh run is uncovered by every segment");
            // The head landing as its anchor's stored tail moves the one
            // parent edge, exactly as the per-dot registration's tail arm;
            // the interiors' successor-shaped edges are the segment itself.
            let is_tail = self
                .children
                .bucket(&self.skeleton, anchor)
                .map(|bucket| bucket.last())
                == Some(first_dot);
            if is_tail {
                let linked = match anchor {
                    Anchor::Origin => true,
                    Anchor::Before(parent) => self
                        .thread
                        .replace_region_start(placed_anchor(parent), Some(first_dot)),
                    Anchor::After(parent) => self
                        .thread
                        .replace_region_end(placed_anchor(parent), Some(first_dot)),
                };
                debug_assert!(linked, "a placed anchor has a region node");
            }
        } else {
            self.unplaced
                .extend((0..u64::from(len)).map(|k| run_member(first_dot, k)));
        }
        true
    }

    /// Absorbs one maximal novel chain run from a merging store's skeleton
    /// (the in-place fold's run arm, R-19): the detector guarantees the
    /// dots are consecutive, same-station, absent here, and (past the
    /// head) each anchored `After` its predecessor, so the run takes the
    /// bulk machinery; the two contested corners (a parked dot waiting on
    /// a run dot, whose repair protocol is per-dot; a bulk refusal such as
    /// the live-entry ceiling) degrade to the per-dot fold, which absorbs
    /// whatever it lawfully can, exactly as before this arm existed.
    /// Visibility stays the survivor fold's: the run threads invisible and
    /// the settle flips what the fold admits.
    pub(in crate::metis::rhapsody) fn absorb_novel_run(&mut self, first_dot: Dot, loci: &[Locus]) {
        let (station, first) = (first_dot.station(), first_dot.counter());
        if loci.len() > 1 {
            let last = first + (loci.len() as u64 - 1);
            let contested =
                !self.unplaced.is_empty() && self.a_parked_dot_waits_within(station, first, last);
            if !contested && self.record_locus_run(first_dot, loci, false) {
                return;
            }
        }
        for (k, &locus) in loci.iter().enumerate() {
            let _ = self.record_locus(run_member(first_dot, k as u64), locus);
        }
    }

    /// Records one locus into the skeleton and the maintained child index,
    /// returning whether the dot was fresh (an already-woven dot is left
    /// untouched, trait law 3: a dot names one write). The ordering-state
    /// half of [`weave`](Self::weave), shared with the in-place causal merge
    /// (`causal_merge_from`, S182), which unions skeletons without touching
    /// visibility (the survivor law owns that coordinate).
    ///
    /// Maintains the child index incrementally: the dot enters its anchor's
    /// bucket at the one sorted position, an `O(log n)` search plus the vector
    /// shift, so neither hot write path rebuilds the whole index (S116, C8).
    /// `sibling_cmp` reads the just-inserted locus, so the skeleton insert
    /// precedes the bucket insert.
    pub(in crate::metis::rhapsody) fn record_locus(&mut self, dot: Dot, locus: Locus) -> bool {
        self.record_locus_transition(dot, locus).is_some()
    }

    /// Returns the newly placed region. An empty region is a fresh parked dot;
    /// `None` means the identity was already present.
    pub(in crate::metis::rhapsody) fn record_locus_transition(
        &mut self,
        dot: Dot,
        locus: Locus,
    ) -> Option<Vec<Dot>> {
        // The plane's insert refuses an occupied slot (trait law 3), the
        // non-dot zero, and its live-entry ceiling; a refusal must move no
        // derived coordinate, because `sibling_cmp` reads the just-inserted
        // locus (the S199 gate review's panic path at the ceiling).
        if !self.skeleton.insert(dot, locus) {
            return None;
        }
        let _ = self.woven.insert(dot);
        self.children.insert(&self.skeleton, locus.anchor, dot);
        // Maintained placement (S183): the fresh dot is placed iff its anchor
        // is the origin or a placed woven dot, decided locally off the
        // anchor's own placement (the chain verdict is inductive, so no walk).
        // A newly placed dot repairs the dangling subtree that was waiting on
        // it; a dangling one parks in `unplaced` until its repair arrives.
        let placed = match locus.anchor.dot().map(Dot::try_from) {
            None => true,
            Some(Ok(anchor)) => {
                anchor != dot && self.skeleton.contains(anchor) && !self.unplaced.contains(&anchor)
            }
            // A non-dot anchor coordinate names no woven element, so the
            // fresh dot parks exactly as it did before the type carried
            // the law (ruling R-91).
            Some(Err(_)) => false,
        };
        if placed {
            // Re-place parked descendants first. The fresh dot and every
            // repaired descendant form one contiguous reading region.
            let region = self.repair_placement_from(dot);
            let region_begin = self.walk_slot_position(dot, locus.anchor);
            self.thread_region(dot, region_begin);
            self.register_region_edges(dot, locus.anchor, &region);
            Some(region)
        } else {
            let _ = self.unplaced.insert(dot);
            Some(Vec::new())
        }
    }

    /// Threads the reading region rooted at a just-placed `root` (itself
    /// plus every descendant the repair sweep re-placed, reachable through
    /// the present-locus bucket edges) into the order thread at
    /// consecutive slots from `region_begin`: the region is one contiguous
    /// walk interval, and its elements arrive in reading order (Before
    /// subtrees, self, After subtrees, the walk's own schedule), so each
    /// insert is the next slot. The common weave (no parked descendants)
    /// is one visit, one emit, two empty bucket probes. Iterative, never
    /// recursive (skeleton depth is attacker-controllable), and each slot
    /// carries its current visibility (a parked dot may have been visible
    /// all along).
    pub(in crate::metis::rhapsody) fn thread_region(&mut self, root: Dot, region_begin: usize) {
        let mut position = region_begin;
        let mut stack: Vec<RegionFrame> = alloc::vec![RegionFrame::Visit(root)];
        while let Some(frame) = stack.pop() {
            match frame {
                RegionFrame::Visit(dot) => {
                    // Reading order per element, pushed LIFO-reversed:
                    // Before subtrees (stored order reversed), the element,
                    // After subtrees (stored order).
                    for kid in self
                        .children
                        .iter(&self.skeleton, Anchor::After(dot.into()))
                        .rev()
                    {
                        stack.push(RegionFrame::Visit(kid));
                    }
                    stack.push(RegionFrame::Emit(dot));
                    for kid in self
                        .children
                        .iter(&self.skeleton, Anchor::Before(dot.into()))
                    {
                        stack.push(RegionFrame::Visit(kid));
                    }
                }
                RegionFrame::Emit(dot) => {
                    let visible = self.visible.contains(dot);
                    self.thread.insert_slot(position, dot, visible);
                    position += 1;
                }
            }
        }
    }

    /// Adds the repaired region to the dynamic start/end forests, then
    /// updates the one parent edge whose last child may have changed.
    pub(in crate::metis::rhapsody) fn register_region_edges(
        &mut self,
        root: Dot,
        anchor: Anchor,
        region: &[Dot],
    ) {
        for &dot in region {
            let inserted = self.thread.insert_region_dot(dot);
            debug_assert!(inserted, "a repaired region was previously unplaced");
        }
        for &dot in region {
            if let Some(child) = self
                .children
                .bucket(&self.skeleton, Anchor::Before(dot.into()))
                .map(|bucket| bucket.last())
            {
                let linked = self.thread.replace_region_start(dot, Some(child));
                debug_assert!(linked);
            }
            if let Some(child) = self
                .children
                .bucket(&self.skeleton, Anchor::After(dot.into()))
                .map(|bucket| bucket.last())
            {
                let linked = self.thread.replace_region_end(dot, Some(child));
                debug_assert!(linked);
            }
        }
        let bucket = self
            .children
            .bucket(&self.skeleton, anchor)
            .expect("the fresh region root was bucketed");
        if bucket.last() != root {
            return;
        }
        let linked = match anchor {
            Anchor::Origin => return,
            Anchor::Before(parent) => self
                .thread
                .replace_region_start(placed_anchor(parent), Some(root)),
            Anchor::After(parent) => self
                .thread
                .replace_region_end(placed_anchor(parent), Some(root)),
        };
        debug_assert!(linked, "a placed anchor has a region node");
    }

    /// The walk's slot-space position for a just-bucketed placed element.
    /// The order semantics stay in `traversal.rs`; this reads the dynamic
    /// region endpoints maintained by the thread.
    ///
    /// * a bucket head reads adjacent to its anchor (right after an
    ///   `After` anchor's slot, right before a `Before` anchor's slot,
    ///   first of the document at the origin);
    /// * a non-head `After` (or origin) sibling reads right after its
    ///   stored predecessor's whole region;
    /// * a non-head `Before` sibling reads right before its stored
    ///   predecessor's region start (Before buckets read reversed).
    ///
    /// Region endpoints are link-cut forest roots. Lookup and last-child
    /// replacement are `O(log n)` amortized regardless of anchor depth, so
    /// distinct deep seams cannot turn split peer deltas into repeated
    /// whole-document work.
    pub(in crate::metis::rhapsody) fn walk_slot_position(
        &mut self,
        dot: Dot,
        anchor: Anchor,
    ) -> usize {
        let bucket = self
            .children
            .bucket(&self.skeleton, anchor)
            .expect("the dot was just bucketed under its anchor");
        let at = self
            .sibling_position(&bucket, dot)
            .expect("the dot is in its own bucket");
        let predecessor = (at > 0).then(|| {
            bucket
                .get(at - 1)
                .expect("a bucket position has an occupant")
        });
        match anchor {
            Anchor::Origin => match predecessor {
                None => 0,
                Some(prev) => {
                    let end = self
                        .thread
                        .region_end(prev)
                        .expect("a placed sibling has a region endpoint");
                    self.slot_of_placed(end) + 1
                }
            },
            Anchor::After(d) => match predecessor {
                None => self.slot_of_placed(placed_anchor(d)) + 1,
                Some(prev) => {
                    let end = self
                        .thread
                        .region_end(prev)
                        .expect("a placed sibling has a region endpoint");
                    self.slot_of_placed(end) + 1
                }
            },
            Anchor::Before(d) => match predecessor {
                None => self.slot_of_placed(placed_anchor(d)),
                Some(prev) => {
                    let start = self
                        .thread
                        .region_start(prev)
                        .expect("a placed sibling has a region endpoint");
                    self.slot_of_placed(start)
                }
            },
        }
    }

    /// The slot position of a placed element already carried by the thread.
    pub(in crate::metis::rhapsody) fn slot_of_placed(&self, target: Dot) -> usize {
        self.thread
            .position_of(target)
            .expect("a placed element holds a thread slot")
            .0
    }

    /// Collects the reading region rooted at a placed `root` (itself plus
    /// every descendant reachable through the present-locus bucket edges),
    /// in walk order: the same schedule
    /// [`thread_region`](Self::thread_region) threads, materialized so the
    /// re-placement path can detach exactly these slots. A placed dot's
    /// bucket children are placed (the placement induction), so the walk
    /// never meets an unplaced member. Iterative, never recursive.
    pub(in crate::metis::rhapsody) fn collect_region(&self, root: Dot) -> Vec<Dot> {
        let mut region: Vec<Dot> = Vec::new();
        let mut stack: Vec<RegionFrame> = alloc::vec![RegionFrame::Visit(root)];
        while let Some(frame) = stack.pop() {
            match frame {
                RegionFrame::Visit(dot) => {
                    for kid in self
                        .children
                        .iter(&self.skeleton, Anchor::After(dot.into()))
                        .rev()
                    {
                        stack.push(RegionFrame::Visit(kid));
                    }
                    stack.push(RegionFrame::Emit(dot));
                    for kid in self
                        .children
                        .iter(&self.skeleton, Anchor::Before(dot.into()))
                    {
                        stack.push(RegionFrame::Visit(kid));
                    }
                }
                RegionFrame::Emit(dot) => region.push(dot),
            }
        }
        region
    }

    /// Repairs placement downward from a just-placed dot: every dangling
    /// dot anchored (on either side) to a repaired dot becomes placed in
    /// turn.
    /// Iterative over a worklist, never recursive; each repaired dot leaves
    /// `unplaced` exactly once, so the sweep is linear in the repaired
    /// subtree over the map lookups, and the common weave (no dangling
    /// children waiting) costs two empty bucket probes. The repaired dots
    /// are exactly the placing dot's region members, which the caller
    /// threads in walk order right after this sweep
    /// ([`thread_region`](Self::thread_region)).
    pub(in crate::metis::rhapsody) fn repair_placement_from(&mut self, dot: Dot) -> Vec<Dot> {
        let mut region = alloc::vec![dot];
        let mut worklist: Vec<Dot> = alloc::vec![dot];
        while let Some(parent) = worklist.pop() {
            for anchor in [Anchor::After(parent.into()), Anchor::Before(parent.into())] {
                for kid in self.children.iter(&self.skeleton, anchor) {
                    if self.unplaced.remove(&kid) {
                        region.push(kid);
                        worklist.push(kid);
                    }
                }
            }
        }
        region
    }
}