minerva 0.2.0

//! The run-coalesced interned identity plane.
//!
//! The skeleton's carrier: per dot, its [`Locus`], stored as an intern table
//! rather than a `BTreeMap<Dot, Locus>`. Dots are dense per-station
//! counters by construction, so the plane exploits that density the way the
//! have-set's compact form already does: each station keys a *fiber* of
//! fixed-size pages, a page slot holds the dot's entry, and lookup is a
//! station probe plus page arithmetic. Ascending enumeration is the fibers in
//! station order, pages in index order.
//!
//! # What a slot may hold
//!
//! Honest editing traffic factors into *chain runs*: consecutive dots of
//! one station, each anchored after its predecessor, kairotic and minting
//! station constant, and ranks stepping by the clock. The snapshot frame uses
//! the same factoring for wire compression. A slot is one of two things:
//!
//! * a *handle* into the contiguous locus column, for chain heads and
//!   everything unchained; or
//! * an inline *rank step*, for an element whose locus is derivable from its
//!   predecessor slot. A stepped element pays no column entry at all: its
//!   locus materializes on read by folding steps forward from the nearest
//!   explicit slot.
//!
//! One law decides what may step: the snapshot codec's own chain law,
//! imported as [`chain_step`] / [`apply_chain_step`]. That interlock is the
//! point. The coalesced plane and the wire codec share one chain
//! derivation, so a plane run and a wire run can never name different chains,
//! and the rank column's agreement with the clock fold is already Kani-proven
//! at the codec.
//!
//! Two structural bounds keep every read cheap and total. Slot zero of a page
//! never steps, so a chain crossing a page boundary re-materializes and a
//! walk back to an explicit slot stays inside one page. And a step's
//! immediate predecessor slot is always occupied, because removal
//! re-materializes the follower before the predecessor leaves, so derivation
//! never dangles.
//!
//! # A shell, not a contract
//!
//! The contract form stays `BTreeMap<Dot, Locus>`; the plane
//! re-realizes it for cost, and the one law owed is agreement, pinned by a
//! twin property beside the value-level suites (`docs/metis-shell-discipline.adoc`).
//!
//! Layout is never part of a rhapsody's identity: equality and hashing read
//! the ascending enumeration. Thus, two planes with different handle
//! assignments, step-versus-handle spellings, or page residency compare equal
//! exactly when their skeletons
//! do.
//!
//! Three type rules hold that line:
//!
//! * [`RawHandle`] is `Copy + Eq` and deliberately **not** `Ord`, so no read
//!   can depend on handle order, and a sort attempt fails to compile at its
//!   own use site. Handles never escape this module tree.
//! * Fiber construction is **paged, never flat**: a decoded far dot buys
//!   exactly one [`PAGE_LEN`]-slot page parked in the fiber's exception map,
//!   never a vector sized by the dot value, so decode amplification stays
//!   bounded by the wire's existing per-record byte budget.
//! * Zero-cost assertions pin [`Slot`] at four bytes, the width the
//!   uncoalesced `Option<RawHandle>` niche already paid, so coalescing is
//!   pure column savings.
//!
//! # Excision
//!
//! `condense` removes in place: the slot clears, an emptied page frees, a
//! fiber's trailing empty prefix trims, and a retired handle joins a free
//! list the next intern reuses, so the locus column does not leak under
//! retention rounds.
//!
//! Two residues survive that, both layout rather than value, and the
//! accounting reads watch them: sparse survivors can strand
//! partially-occupied pages, and churn can strand explicit entries where a
//! rebuilt plane would step, since materialization is never undone in
//! place.

extern crate alloc;

use alloc::boxed::Box;
use alloc::collections::BTreeMap;
use alloc::vec::Vec;
use core::num::{NonZeroU32, NonZeroU64};

use super::placement::{Anchor, Dot, Locus};
use super::wire::{apply_chain_step, chain_step};
use crate::metis::dot::RawDot;

/// Slots per fiber page. Sixty-four matches the honest-traffic run length
/// the S192 coalescing instrument measured (one maximal same-author run per
/// 64-character chunk), so a collaborating author's active region tends to
/// occupy whole pages.
pub(super) const PAGE_LEN: usize = 64;
const PAGE_LEN_U64: u64 = 64;

/// The live-entry ceiling: one below the handle payload's `2^31 - 1` cap.
/// Insert refuses at this count, which is also what keeps removal total:
/// the column holds one entry per explicit slot plus the free list, so
/// while live entries stay under the cap, a step slot can always be
/// re-materialized into an explicit handle without exhausting the payload
/// space (the removal path's intern never fails).
const LIVE_CAP: usize = Slot::PAYLOAD_MAX as usize - 1;

/// The ceiling, surfaced for the decoders: a frame declaring more loci
/// than any plane can hold is refused with a typed error, never folded
/// into a silently truncated value (the S199 gate review's finding; the
/// step tag halved the handle domain below the v2 count field's `u32`).
pub(in crate::metis::rhapsody) const PLANE_CAPACITY: usize = LIVE_CAP;

/// An interned element handle: an index into the locus column, plus one so
/// zero stays the empty slot. The payload never sets [`Slot`]'s step bit,
/// which [`RawHandle::from_index`] enforces by refusing indices at and
/// above `2^31 - 1`.
///
/// Deliberately `Copy + Eq` and **not** `Ord` or `Hash`: a handle is
/// arrival-order layout, never part of the value, so no read may compare,
/// sort, or key by it (the shell-discipline handle law). It is
/// crate-private and never crosses a store boundary; the merge paths
/// enumerate the other store in public `(dot, locus)` form.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
struct RawHandle(NonZeroU32);

impl RawHandle {
    /// The handle for a zero-based column index, refusing any index whose
    /// plus-one encoding would collide with the step bit.
    fn from_index(index: usize) -> Option<Self> {
        u32::try_from(index)
            .ok()
            .and_then(|raw| raw.checked_add(1))
            .filter(|&raw| raw <= Slot::PAYLOAD_MAX)
            .and_then(NonZeroU32::new)
            .map(Self)
    }

    /// The zero-based column index this handle names.
    const fn index(self) -> usize {
        self.0.get() as usize - 1
    }
}

/// One page slot: empty, an explicit handle into the locus column, or an
/// inline rank step deriving this dot's locus from its predecessor slot's.
///
/// Packed into four bytes (asserted below), the same width the phase-one
/// `Option<RawHandle>` paid: zero is empty, a clear high bit is a handle
/// (the plus-one column index), a set high bit is a step (the low 31 bits;
/// a genuine chain step past that width simply stays explicit, a layout
/// fallback the value never sees). Not `Ord`, not `Hash`: a slot is layout.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
struct Slot(u32);

const _: () = assert!(core::mem::size_of::<Slot>() == 4);
const _: () = assert!(core::mem::size_of::<Option<RawHandle>>() == 4);

impl Slot {
    /// The empty slot.
    const EMPTY: Self = Self(0);
    /// The step marker: the high bit.
    const STEP_BIT: u32 = 0x8000_0000;
    /// The widest payload either arm carries: 31 bits.
    const PAYLOAD_MAX: u32 = 0x7FFF_FFFF;

    /// Whether this slot holds nothing.
    const fn is_empty(self) -> bool {
        self.0 == 0
    }

    /// An explicit slot carrying `handle`.
    const fn explicit(handle: RawHandle) -> Self {
        Self(handle.0.get())
    }

    /// An inline step slot, refusing a step past the 31-bit payload (the
    /// caller falls back to an explicit entry).
    fn step(step: u64) -> Option<Self> {
        u32::try_from(step)
            .ok()
            .filter(|&raw| raw <= Self::PAYLOAD_MAX)
            .map(|raw| Self(Self::STEP_BIT | raw))
    }

    /// The handle, when this slot is explicit.
    fn as_explicit(self) -> Option<RawHandle> {
        if self.0 & Self::STEP_BIT == 0 {
            NonZeroU32::new(self.0).map(RawHandle)
        } else {
            None
        }
    }

    /// The rank step, when this slot is inline.
    const fn as_step(self) -> Option<u64> {
        if self.0 & Self::STEP_BIT == 0 {
            None
        } else {
            Some((self.0 & Self::PAYLOAD_MAX) as u64)
        }
    }
}

/// One fiber page: the slots for [`PAGE_LEN`] consecutive dot indices of
/// one station, plus an occupancy count so an emptied page frees itself.
#[derive(Clone, Debug)]
struct Page {
    /// Slot `s` holds the entry for dot index `page * PAGE_LEN + s + 1`.
    slots: [Slot; PAGE_LEN],
    /// How many slots are occupied; zero frees the page.
    occupancy: u16,
}

impl Page {
    const fn empty() -> Self {
        Self {
            slots: [Slot::EMPTY; PAGE_LEN],
            occupancy: 0,
        }
    }
}

/// One station's fiber: the dense low pages as a prefix vector (the floor
/// analog), the far pages as exceptions (the above analog). Every `above`
/// key is at least `prefix.len()`, so ascending enumeration is the prefix
/// then the exceptions.
#[derive(Clone, Debug, Default)]
struct StationFiber {
    /// Pages `0..prefix.len()`, present or freed in place.
    prefix: Vec<Option<Box<Page>>>,
    /// Exception pages past the prefix, keyed by page number.
    above: BTreeMap<u64, Box<Page>>,
}

impl StationFiber {
    /// Whether the fiber holds no pages at all.
    fn is_empty(&self) -> bool {
        self.above.is_empty() && self.prefix.iter().all(Option::is_none)
    }

    /// The page holding `page_number`, if present.
    fn page(&self, page_number: u64) -> Option<&Page> {
        match usize::try_from(page_number) {
            Ok(offset) if offset < self.prefix.len() => self.prefix[offset].as_deref(),
            _ => self.above.get(&page_number).map(Box::as_ref),
        }
    }

    /// The page holding `page_number` mutably, if present (never creates).
    fn page_mut(&mut self, page_number: u64) -> Option<&mut Page> {
        match usize::try_from(page_number) {
            Ok(offset) if offset < self.prefix.len() => self.prefix[offset].as_deref_mut(),
            _ => self.above.get_mut(&page_number).map(Box::as_mut),
        }
    }

    /// The page holding `page_number`, created if absent. Appending at the
    /// prefix boundary promotes any exception pages that became contiguous,
    /// keeping the `above`-keys-past-the-prefix invariant.
    fn page_mut_or_create(&mut self, page_number: u64) -> &mut Page {
        if let Ok(offset) = usize::try_from(page_number) {
            if offset < self.prefix.len() {
                return self.prefix[offset].get_or_insert_with(|| Box::new(Page::empty()));
            }
            if offset == self.prefix.len() {
                self.prefix.push(Some(Box::new(Page::empty())));
                self.promote_contiguous_exceptions();
                return self.prefix[offset]
                    .as_deref_mut()
                    .expect("the page was just pushed");
            }
        }
        self.above
            .entry(page_number)
            .or_insert_with(|| Box::new(Page::empty()))
    }

    /// Pulls exception pages down into the prefix while they are contiguous
    /// with its end.
    fn promote_contiguous_exceptions(&mut self) {
        while let Ok(next) = u64::try_from(self.prefix.len()) {
            let Some(page) = self.above.remove(&next) else {
                break;
            };
            self.prefix.push(Some(page));
        }
    }

    /// Drops an emptied page and trims the prefix's trailing freed pages.
    fn free_page(&mut self, page_number: u64) {
        match usize::try_from(page_number) {
            Ok(offset) if offset < self.prefix.len() => {
                self.prefix[offset] = None;
                while self.prefix.last().is_some_and(Option::is_none) {
                    let _ = self.prefix.pop();
                }
            }
            _ => {
                let _ = self.above.remove(&page_number);
            }
        }
    }
}

/// The page number and slot offset of a dot index (`index >= 1`).
fn page_slot(index: u64) -> (u64, usize) {
    (
        (index - 1) / PAGE_LEN_U64,
        usize::from(
            u8::try_from((index - 1) % PAGE_LEN_U64).expect("page slot modulo PAGE_LEN fits in u8"),
        ),
    )
}

/// The run-coalesced interned identity plane: the skeleton's `dot -> Locus`
/// carrier.
///
/// Reads are `&self`, writes are `&mut self`, and nothing here is interior
/// mutable (the C8 constraint the maintained coordinates already obey).
#[derive(Clone, Debug, Default)]
pub(super) struct IdentityPlane {
    /// Per station, its paged fiber of slots.
    stations: BTreeMap<u32, StationFiber>,
    /// The locus column, indexed by handle: chain heads and unchained
    /// entries only (a stepped element pays no column entry). A retired
    /// handle's slot is stale until the free list reuses it; no live
    /// explicit slot points at a stale entry.
    loci: Vec<Locus>,
    /// Retired handles, reused by the next intern so excision does not leak
    /// the column.
    free: Vec<RawHandle>,
    /// Live entry count, explicit and stepped alike.
    len: usize,
}

impl IdentityPlane {
    /// The empty plane.
    pub(super) const fn new() -> Self {
        Self {
            stations: BTreeMap::new(),
            loci: Vec::new(),
            free: Vec::new(),
            len: 0,
        }
    }

    /// Live entries.
    pub(super) const fn len(&self) -> usize {
        self.len
    }

    /// Whether the plane holds nothing.
    pub(super) const fn is_empty(&self) -> bool {
        self.len == 0
    }

    /// Whether `additional` more live entries fit under the live-entry
    /// ceiling: the one authoritative capacity read, shared by the bulk
    /// insert's refusal and the degrade path's all-or-nothing preflight
    /// (a per-dot fallback that outran the ceiling mid-run would weave a
    /// partial run; the S211 gate review's round-two finding).
    pub(super) const fn has_capacity_for(&self, additional: usize) -> bool {
        // Saturating: an absurd `additional` must refuse, not overflow.
        self.len.saturating_add(additional) <= LIVE_CAP
    }

    /// Whether `dot` has an entry.
    pub(super) fn contains(&self, dot: Dot) -> bool {
        let (station, index) = (dot.station(), dot.counter());
        let (page_number, slot) = page_slot(index);
        self.stations
            .get(&station)
            .and_then(|fiber| fiber.page(page_number))
            .is_some_and(|page| !page.slots[slot].is_empty())
    }

    /// The locus at `slot` of `page` (one fiber page of `station`), which
    /// must be occupied: an explicit slot reads the column; a stepped slot
    /// walks back to the nearest explicit slot and folds the chain law
    /// forward. The walk stays inside the page (slot zero never steps and a
    /// step's predecessor is always occupied, the two structural bounds),
    /// so materialization costs at most `PAGE_LEN - 1` step folds.
    fn materialize(&self, station: u32, page_number: u64, page: &Page, slot: usize) -> Locus {
        let mut explicit = slot;
        let handle = loop {
            if let Some(handle) = page.slots[explicit].as_explicit() {
                break handle;
            }
            debug_assert!(explicit > 0, "slot zero of a page is always explicit");
            explicit -= 1;
        };
        let mut locus = self.loci[handle.index()];
        // Dot index of slot `k` is `page_number * PAGE_LEN + k + 1`, so the
        // predecessor of slot `k`'s dot has index `page_number * PAGE_LEN + k`.
        for k in (explicit + 1)..=slot {
            let step = page.slots[k]
                .as_step()
                .expect("the walk back stopped at the nearest explicit slot");
            let prev_dot = (station, page_number * PAGE_LEN_U64 + k as u64);
            locus = apply_chain_step(prev_dot, &locus, step);
        }
        locus
    }

    /// The anchor of `dot`'s locus, if present, in constant time in the
    /// document (one station-map probe plus page arithmetic): a stepped
    /// slot is anchored `After` its predecessor by the chain law itself
    /// (`apply_chain_step` constructs exactly that anchor), and an explicit
    /// slot reads its column entry directly, so neither arm pays the
    /// walk-back materialization [`get`](Self::get) may. The read the
    /// coalesced child plane's chain-child probe rides (arc 11 phase four).
    pub(super) fn anchor_of(&self, dot: &Dot) -> Option<Anchor> {
        let (station, index) = (dot.station(), dot.counter());
        let (page_number, slot) = page_slot(index);
        let page = self.stations.get(&station)?.page(page_number)?;
        let entry = page.slots[slot];
        if let Some(handle) = entry.as_explicit() {
            return Some(self.loci[handle.index()].anchor);
        }
        if entry.as_step().is_some() {
            return Some(Anchor::After(RawDot {
                station,
                counter: index - 1,
            }));
        }
        None
    }

    /// The locus of `dot`, if present, materialized by value (a stepped
    /// entry has no stored `Locus` to lend a reference to).
    pub(super) fn get(&self, dot: &Dot) -> Option<Locus> {
        let (station, index) = (dot.station(), dot.counter());
        let (page_number, slot) = page_slot(index);
        let page = self.stations.get(&station)?.page(page_number)?;
        if page.slots[slot].is_empty() {
            return None;
        }
        Some(self.materialize(station, page_number, page, slot))
    }

    /// Interns one locus into the column, reusing a retired handle first.
    /// Total while live entries respect [`LIVE_CAP`]: the column holds one
    /// entry per explicit slot plus the free list, so its next index stays
    /// under the handle payload's ceiling.
    fn intern(&mut self, locus: Locus) -> RawHandle {
        if let Some(handle) = self.free.pop() {
            self.loci[handle.index()] = locus;
            handle
        } else {
            let handle = RawHandle::from_index(self.loci.len())
                .expect("the live-entry cap reserves column capacity for every explicit slot");
            self.loci.push(locus);
            handle
        }
    }

    /// Records `dot`'s locus, returning whether it was fresh. Refuses (no
    /// change) an occupied slot and the [`LIVE_CAP`] ceiling (which the
    /// wire's saturating count already treats as unrepresentable in
    /// memory, and which is what keeps removal's re-materialization
    /// total). The non-dot index `0` needs no arm: [`Dot`] carries that
    /// law (ruling R-91).
    ///
    /// The entry coalesces when it can: past slot zero, with the
    /// predecessor slot occupied and the locus continuing its chain under
    /// the codec's own law ([`chain_step`]), the slot stores the inline
    /// rank step and the column is untouched; everything else interns an
    /// explicit entry. A predecessor arriving later never re-compresses an
    /// explicit follower: the spelling is layout, not value.
    pub(super) fn insert(&mut self, dot: Dot, locus: Locus) -> bool {
        self.insert_inner(dot, locus, None)
    }

    /// Folds ascending `(dot, locus)` entries in, rolling the predecessor
    /// forward so a chain interior's spelling probe is `O(1)` instead of a
    /// walk back over its whole run: the build-from-parts paths (decode,
    /// the pure merge, restriction) stay linear in the skeleton. The hint
    /// rolls only across an entry this fold itself just recorded (a
    /// refused duplicate drops it), so it always equals the stored
    /// predecessor's materialization.
    pub(super) fn extend_ascending(&mut self, entries: impl IntoIterator<Item = (Dot, Locus)>) {
        let mut prev: Option<(Dot, Locus)> = None;
        for (dot, locus) in entries {
            let hint = match &prev {
                Some((prev_dot, prev_locus))
                    if prev_dot.station() == dot.station()
                        && prev_dot.counter().checked_add(1) == Some(dot.counter()) =>
                {
                    Some(prev_locus)
                }
                _ => None,
            };
            prev = if self.insert_inner(dot, locus, hint) {
                Some((dot, locus))
            } else {
                None
            };
        }
    }

    /// [`insert`](Self::insert)'s body. `prev_hint`, when supplied, must be
    /// the materialized locus of `dot`'s immediate predecessor as stored
    /// (the [`extend_ascending`](Self::extend_ascending) rolling duty); it
    /// replaces the walk-back materialization and nothing else, so a wrong
    /// hint could only mis-spell, never mis-value, and no caller can be
    /// wrong (the only hinting caller rolls the value it just stored).
    fn insert_inner(&mut self, dot: Dot, locus: Locus, prev_hint: Option<&Locus>) -> bool {
        let (station, index) = (dot.station(), dot.counter());
        if self.len >= LIVE_CAP {
            return false;
        }
        let (page_number, slot) = page_slot(index);
        // One immutable probe decides the refusal and the spelling.
        let mut inline: Option<Slot> = None;
        if let Some(page) = self
            .stations
            .get(&station)
            .and_then(|fiber| fiber.page(page_number))
        {
            if !page.slots[slot].is_empty() {
                return false;
            }
            if slot > 0 && !page.slots[slot - 1].is_empty() {
                let prev_dot = (station, index - 1);
                let step = prev_hint.map_or_else(
                    || {
                        let prev = self.materialize(station, page_number, page, slot - 1);
                        chain_step(prev_dot, &prev, (station, index), &locus)
                    },
                    |prev| chain_step(prev_dot, prev, (station, index), &locus),
                );
                inline = step.and_then(Slot::step);
            }
        }
        let entry = inline.unwrap_or_else(|| Slot::explicit(self.intern(locus)));
        let page = self
            .stations
            .entry(station)
            .or_default()
            .page_mut_or_create(page_number);
        page.slots[slot] = entry;
        page.occupancy += 1;
        self.len += 1;
        true
    }

    /// Records a contiguous run of loci for dots
    /// `(station, first) .. (station, first + loci.len())` in one pass:
    /// the bulk-ingest form of [`insert`](Self::insert), agreeing with the
    /// per-dot fold exactly (same refusals, same spellings). All-or
    /// -nothing: refuses (no change) when the run is empty, would pass the
    /// `u64::MAX` dot ceiling, would exceed the live-entry ceiling, or
    /// lands on ANY occupied slot; the caller gets one verdict for the
    /// whole run instead of a partial weave. The non-dot `0` needs no arm:
    /// [`Dot`] carries that law (ruling R-91).
    ///
    /// The cost leaves the per-dot form's station-map probe per element
    /// behind: one fiber traversal, each page created once and filled
    /// slot-wise, the chain law folded forward exactly as
    /// [`extend_ascending`](Self::extend_ascending) rolls it (chainable
    /// interiors spell as inline steps, page-boundary slots and
    /// unchainable entries intern explicit column entries).
    pub(super) fn insert_run(&mut self, first_dot: Dot, loci: &[Locus]) -> bool {
        let (station, first) = (first_dot.station(), first_dot.counter());
        let len = loci.len();
        if len == 0 {
            return false;
        }
        let Some(last) = first.checked_add(len as u64 - 1) else {
            return false;
        };
        if !self.has_capacity_for(len) {
            return false;
        }
        // The freshness sweep: every slot in the run's page range must be
        // empty. Absent pages are trivially free; present pages are
        // checked slot-wise within the run's span.
        if let Some(fiber) = self.stations.get(&station) {
            let (first_page, _) = page_slot(first);
            let (last_page, _) = page_slot(last);
            for page_number in first_page..=last_page {
                let Some(page) = fiber.page(page_number) else {
                    continue;
                };
                let from = if page_number == first_page {
                    page_slot(first).1
                } else {
                    0
                };
                let to = if page_number == last_page {
                    page_slot(last).1
                } else {
                    PAGE_LEN - 1
                };
                if page.slots[from..=to].iter().any(|slot| !slot.is_empty()) {
                    return false;
                }
            }
        }
        // The head takes the per-dot path whole: it may chain off a
        // predecessor slot already in the plane (the paste-at-own-tail
        // shape), and `insert_inner`'s probe is the one authority on that
        // spelling.
        if !self.insert_inner(first_dot, loci[0], None) {
            debug_assert!(false, "the freshness sweep admitted the head");
            return false;
        }
        // The interiors fill page-wise: one `page_mut_or_create` per page,
        // the chain law rolled forward off the predecessor locus in hand.
        let mut prev = loci[0];
        let mut at = 1usize;
        while at < len {
            let index = first + at as u64;
            let (page_number, slot_from) = page_slot(index);
            let take = (PAGE_LEN - slot_from).min(len - at);
            // Interned entries are decided against `prev` BEFORE the page
            // borrow (the intern owns the column).
            let mut entries: [Slot; PAGE_LEN] = [Slot::EMPTY; PAGE_LEN];
            for (k, entry) in entries.iter_mut().enumerate().take(take) {
                let locus = loci[at + k];
                let prev_dot = (station, index + k as u64 - 1);
                let inline = if slot_from + k > 0 {
                    chain_step(prev_dot, &prev, (station, index + k as u64), &locus)
                        .and_then(Slot::step)
                } else {
                    // Slot zero of a page never steps (the structural
                    // bound every read relies on).
                    None
                };
                *entry = inline.unwrap_or_else(|| Slot::explicit(self.intern(locus)));
                prev = locus;
            }
            let page = self
                .stations
                .entry(station)
                .or_default()
                .page_mut_or_create(page_number);
            for (slot, &entry) in page.slots[slot_from..slot_from + take]
                .iter_mut()
                .zip(&entries)
            {
                debug_assert!(slot.is_empty(), "the sweep verified");
                *slot = entry;
            }
            page.occupancy += u16::try_from(take).expect("a page holds at most PAGE_LEN slots");
            self.len += take;
            at += take;
        }
        true
    }

    /// Removes `dot`'s entry, returning its locus. The slot clears, an
    /// emptied page frees, the fiber's trailing freed prefix trims (an
    /// emptied fiber leaves the station map), and an explicit entry's
    /// handle retires to the free list. A stepped follower is
    /// re-materialized into an explicit entry first, so no surviving slot
    /// ever derives from a hole (the second structural bound; total under
    /// [`LIVE_CAP`]'s column reservation).
    pub(super) fn remove(&mut self, dot: Dot) -> Option<Locus> {
        let (station, index) = (dot.station(), dot.counter());
        let (page_number, slot) = page_slot(index);
        let page = self.stations.get(&station)?.page(page_number)?;
        if page.slots[slot].is_empty() {
            return None;
        }
        let locus = self.materialize(station, page_number, page, slot);
        // The follower's re-materialization stays in-page: slot PAGE_LEN - 1
        // has no in-page follower, and the next page's slot zero is already
        // explicit by the structural bound.
        let follower = (slot + 1 < PAGE_LEN && page.slots[slot + 1].as_step().is_some())
            .then(|| self.materialize(station, page_number, page, slot + 1));
        let follower_entry =
            follower.map(|follower_locus| Slot::explicit(self.intern(follower_locus)));
        let fiber = self
            .stations
            .get_mut(&station)
            .expect("the fiber was probed above");
        let page = fiber
            .page_mut(page_number)
            .expect("the page was probed above");
        if let Some(entry) = follower_entry {
            page.slots[slot + 1] = entry;
        }
        let departing = page.slots[slot];
        page.slots[slot] = Slot::EMPTY;
        page.occupancy -= 1;
        let emptied = page.occupancy == 0;
        if let Some(handle) = departing.as_explicit() {
            self.free.push(handle);
        }
        if emptied {
            fiber.free_page(page_number);
        }
        if fiber.is_empty() {
            let _ = self.stations.remove(&station);
        }
        self.len -= 1;
        Some(locus)
    }

    /// Every entry, strictly ascending by `(station, index)`: stations in
    /// map order, each fiber's prefix pages then its exception pages (the
    /// exceptions all key past the prefix by invariant), slots in index
    /// order. Stepped entries materialize in rolling `O(1)`: ascending
    /// order visits a step's predecessor immediately before it, so the fold
    /// carries the last materialized locus instead of walking back.
    pub(super) fn iter(&self) -> impl Iterator<Item = (Dot, Locus)> + '_ {
        self.stations.iter().flat_map(move |(&station, fiber)| {
            let prefix = fiber
                .prefix
                .iter()
                .enumerate()
                .filter_map(|(number, page)| page.as_deref().map(|page| (number as u64, page)));
            let above = fiber
                .above
                .iter()
                .map(|(&number, page)| (number, page.as_ref()));
            prefix.chain(above).flat_map(move |(number, page)| {
                let mut last: Option<Locus> = None;
                page.slots
                    .iter()
                    .enumerate()
                    .filter_map(move |(slot, entry)| {
                        if entry.is_empty() {
                            last = None;
                            return None;
                        }
                        let index = number * PAGE_LEN as u64 + slot as u64 + 1;
                        let locus = entry.as_explicit().map_or_else(
                            || {
                                let step = entry
                                    .as_step()
                                    .expect("an occupied slot is explicit or stepped");
                                let prev =
                                    last.expect("a stepped slot's predecessor is always occupied");
                                apply_chain_step((station, index - 1), &prev, step)
                            },
                            |handle| self.loci[handle.index()],
                        );
                        last = Some(locus);
                        // One-based page arithmetic: slot `s` of page `p`
                        // is index `p * 64 + s + 1`, so the counter is
                        // structurally at least one (ruling R-91).
                        let counter = NonZeroU64::new(index)
                            .expect("a page slot's one-based index is at least one");
                        Some((Dot::new(station, counter), locus))
                    })
            })
        })
    }

    /// Allocated pages across every fiber, the decode-amplification and
    /// sparse-survivor accounting read (deterministic counts, the
    /// measurements module's discipline).
    #[cfg(test)]
    fn pages_allocated(&self) -> usize {
        self.stations
            .values()
            .map(|fiber| {
                fiber.prefix.iter().filter(|page| page.is_some()).count() + fiber.above.len()
            })
            .sum()
    }

    /// The locus column's length (live explicit entries plus
    /// retired-but-unreused slots), the free-list-reuse accounting read.
    #[cfg(test)]
    const fn column_len(&self) -> usize {
        self.loci.len()
    }

    /// Live explicit column entries: chain heads plus everything unchained,
    /// the run-coalescing accounting read the closed-forms probe asserts
    /// (on honest chained traffic this is the run count, not the element
    /// count).
    #[cfg(any(test, feature = "instrumentation"))]
    pub(super) const fn explicit_entries(&self) -> usize {
        self.loci.len() - self.free.len()
    }
}

// The layout (handles, step-versus-explicit spellings, page residency,
// free-list state) is never part of the value: equality and hashing read
// the ascending enumeration, exactly the map form's own semantics.
impl PartialEq for IdentityPlane {
    fn eq(&self, other: &Self) -> bool {
        self.len == other.len && self.iter().eq(other.iter())
    }
}
impl Eq for IdentityPlane {}
impl core::hash::Hash for IdentityPlane {
    fn hash<H: core::hash::Hasher>(&self, state: &mut H) {
        state.write_usize(self.len);
        for entry in self.iter() {
            entry.hash(state);
        }
    }
}

#[cfg(test)]
mod tests;