vj_sim/sched/

wheel.rs

//! SCHED::WHEEL — the hierarchical time wheel (scheduler.md §2; ledger
//! SC-029, SC-030, SC-032).
//!
//! A 4-level, 256-slot-per-level hashed hierarchical timing wheel (the
//! classic "hashed and hierarchical timing wheels" construction) plus a
//! `BTreeMap` far-future overflow for delays beyond the four levels' combined
//! range (`256^4` ticks). Each level's slot spans `256^level` ticks; when the
//! wheel's own `current` position crosses a level's boundary, that level's
//! current-position bucket is cascaded — its entries are re-sorted into
//! whichever level/slot their *actual* absolute tick now resolves to (always
//! the same level or a strictly lower one, since the remaining delta to fire
//! only shrinks as `current` advances) — which is what lets a delay far
//! beyond level 0's 256-slot span still fire on the exact correct tick
//! without every intervening tick doing real work (SC-032's "advances now to
//! the next non-empty tick" driver relies on this).
//!
//! `TimerEntry` (SC-029) deliberately carries no `at`/tick field of its own —
//! the wheel tracks each entry's absolute tick internally (alongside it in
//! each slot) precisely so `arm_delay`'s zero-delay case can skip the wheel
//! entirely (return `Region::Inactive` for the caller to push directly onto
//! [`crate::sched::RegionQueues`]) rather than needing a wheel-internal
//! special case.
//!
//! FIFO order within one slot (`VecDeque`, SC-030) is preserved end to end —
//! including through a cascade, which drains a bucket in its own FIFO order
//! and re-inserts each entry via the same `push_back` path a fresh `arm`
//! would use — but that FIFO order is explicitly *not* the determinism
//! guarantee itself (SC-030's negative half): it is only ever an input to
//! [`crate::sched::sort_ready_set`]'s canonical ordering, which a
//! `DeterminismController` (Task 3) actually dispatches by.

use std::collections::{BTreeMap, VecDeque};

use vj_abi::Region;

use crate::sched::{PendingActivation, SimTick};
use crate::state::StateCellId;

const LEVELS: usize = 4;
const SLOTS: usize = 256;

// SC-026's own op-count instrumentation (test-only, zero cost in
// production): counts one "operation" per `insert_relative_at` call (an
// arm, a cascade's re-insert, or an overflow migration's re-insert) and one
// per element a `pop_due`/`cascade_boundary` drain actually visits -- the
// literal set the plan enumerates ("one insert, one cascade-check, one
// pop, one re-insert-on-non-match") -- so a test can assert total work is
// bounded by a small constant times the number of *entries*, never by the
// span of ticks advanced over.
#[cfg(test)]
thread_local! {
    static OP_COUNT: std::cell::Cell<u64> = const { std::cell::Cell::new(0) };
}

#[cfg(test)]
fn count_op() {
    OP_COUNT.with(|c| c.set(c.get() + 1));
}

#[cfg(test)]
fn reset_op_counter() {
    OP_COUNT.with(|c| c.set(0));
}

#[cfg(test)]
fn op_counter() -> u64 {
    OP_COUNT.with(|c| c.get())
}

/// What a fired [`TimerEntry`] actually does (B-sched Task 6, ledger
/// SC-069). Originally `TimerEntry` carried only `{activation, region}` --
/// "resume this activation" -- which had no way to express "trigger this
/// named event" at all: an `->>#5 my_event;` (a delayed nonblocking event
/// trigger, LRM 10.6.3) needs a wheel entry whose fired action wakes every
/// `@(my_event)` waiter, never resumes any particular process's own
/// continuation directly (there may be no continuation to resume at all --
/// the triggering process, if `->>`'s enclosing statement has already
/// returned control, might not even still exist as a live activation by the
/// time this fires).
#[derive(Clone, Copy, Debug)]
pub enum TimerAction {
    /// The original, only-ever-constructed-until-now shape: resume
    /// `activation` in `region` (`region` is almost always `Region::Active`;
    /// it exists so a `Suspend::Delay(0)` can be represented uniformly even
    /// though [`TimeWheel::arm_delay`]'s zero-delay case never actually
    /// constructs one -- it redirects to `Region::Inactive` before the wheel
    /// is touched at all).
    Resume(PendingActivation, Region),
    /// SC-069: fire the named-event trigger at `StateCellId` -- fans out
    /// through the exact same `ValueChangeSink::notify_change` path an
    /// ordinary committed write already uses (`Engine::advance_time_to`,
    /// called with `EdgeKind::None`, the sentinel `SensitivityTable::matches`
    /// already treats as "any change"/a trigger pulse), never conflated with
    /// resuming whoever originally armed the trigger.
    TriggerEvent(StateCellId),
}

/// One armed wheel entry (SC-029, SC-069). All fields are private outside
/// this crate's `sched` module -- nothing depends on this being a
/// single-field wrapper vs. some other internal shape, only on
/// `arm`/`arm_delay`/`arm_trigger`/`next_tick`/`advance_to`'s documented
/// behavior.
#[derive(Clone, Copy, Debug)]
pub struct TimerEntry {
    pub action: TimerAction,
}

/// The hierarchical time wheel (SC-029, SC-030, SC-032). See the module doc
/// comment for the level/cascade design. All fields are private — nothing
/// outside this module depends on the internal level/slot layout, only on
/// `arm`/`arm_delay`/`next_tick`/`advance_to`'s documented behavior.
pub struct TimeWheel {
    /// `levels[l][slot]` holds `(at, TimerEntry)` pairs currently assigned to
    /// level `l`'s slot `slot`, where `slot == (at >> (8 * l)) & 0xFF` at the
    /// time the entry was last placed (by `arm` or by a cascade).
    levels: [Vec<VecDeque<(SimTick, TimerEntry)>>; LEVELS],
    /// Entries whose delta from `current` exceeds every level's range
    /// (`>= 256^LEVELS` ticks out). Promoted into the levels as `current`
    /// advances close enough (see `cascade`).
    overflow: BTreeMap<SimTick, Vec<TimerEntry>>,
    /// The wheel's own notion of "now" — the last tick `advance_to` was
    /// called with (0 for a fresh wheel that has never advanced).
    current: SimTick,
    /// How many entries each level holds, so `next_tick` skips empty levels.
    counts: [usize; LEVELS],
    /// The earliest `at` across levels 1..LEVELS, when known (`None` inside
    /// the cell = must be recomputed): kept by `insert_relative_at`,
    /// invalidated when a cascade moves entries out of those levels, so a
    /// long-pending far timer does not make every `next_tick` rescan them.
    upper_min: std::cell::Cell<Option<Option<SimTick>>>,
}

impl Default for TimeWheel {
    fn default() -> Self {
        Self {
            levels: std::array::from_fn(|_| (0..SLOTS).map(|_| VecDeque::new()).collect()),
            overflow: BTreeMap::new(),
            current: 0,
            counts: [0; LEVELS],
            upper_min: std::cell::Cell::new(Some(None)),
        }
    }
}

impl TimeWheel {
    pub fn new() -> Self {
        Self::default()
    }

    /// Which level a `delta`-ticks-from-`current` entry belongs in, or `None`
    /// for the far-future overflow map. Level `l`'s range is `[256^l,
    /// 256^(l+1))` ticks from `current`, so an entry's level only ever
    /// shrinks (never grows) as `current` advances and `delta` shrinks —
    /// this is what makes cascading a one-way, terminating redistribution.
    fn level_for(delta: SimTick) -> Option<usize> {
        const L0: u64 = SLOTS as u64;
        const L1: u64 = L0 * SLOTS as u64;
        const L2: u64 = L1 * SLOTS as u64;
        const L3: u64 = L2 * SLOTS as u64;
        if delta < L0 {
            Some(0)
        } else if delta < L1 {
            Some(1)
        } else if delta < L2 {
            Some(2)
        } else if delta < L3 {
            Some(3)
        } else {
            None
        }
    }

    /// Places `(at, entry)` into the correct level/slot (or overflow) *as
    /// of* `as_of` — the one insertion path shared by `arm` (fresh entries,
    /// `as_of == self.current`) and `cascade_boundary`/
    /// `migrate_overflow_due_by` (redistributed entries, `as_of` the exact
    /// tick the redistribution is happening at, which may be earlier than
    /// `self.current` if the caller hasn't advanced the field yet). Always
    /// recomputing fresh from `(at, as_of)` — rather than trusting an
    /// entry's *previous* level/slot — is what makes deferring a
    /// redistribution to a later, coarser-grained check (SC-026/031's
    /// bounded-cost requirement, see `advance_to`) still land every entry
    /// exactly where an eager per-tick walk would have: `level_for`'s
    /// output only ever shrinks as `as_of` grows, so computing it once at
    /// the final `as_of` is equivalent to recomputing it at every
    /// intermediate tick and always agrees with the last such recomputation.
    fn insert_relative_at(&mut self, at: SimTick, entry: TimerEntry, as_of: SimTick) {
        #[cfg(test)]
        count_op();
        let delta = at.saturating_sub(as_of);
        match Self::level_for(delta) {
            Some(l) => {
                let slot = ((at >> (8 * l)) & 0xFF) as usize;
                self.levels[l][slot].push_back((at, entry));
                self.counts[l] += 1;
                if l > 0
                    && let Some(known) = self.upper_min.get()
                {
                    self.upper_min
                        .set(Some(Some(known.map_or(at, |m| m.min(at)))));
                }
            }
            None => {
                self.overflow.entry(at).or_default().push(entry);
            }
        }
    }

    /// `insert_relative_at` as of the wheel's own current position — the
    /// path `arm` uses for a fresh entry.
    fn insert_relative(&mut self, at: SimTick, entry: TimerEntry) {
        self.insert_relative_at(at, entry, self.current);
    }

    /// Inserts `entry`, due at absolute tick `at`, into the correct
    /// level/slot or the far-future overflow map.
    pub fn arm(&mut self, at: SimTick, entry: TimerEntry) {
        self.insert_relative(at, entry);
    }

    /// The earliest tick with any pending entry across every level and the
    /// overflow map, or `None` if the wheel is entirely empty. Does not
    /// mutate the wheel (SC-032's driver calls this before deciding whether
    /// to advance at all).
    ///
    /// Must scan every entry in every slot, not just `front()`: at levels
    /// 1-3 a single slot legitimately holds entries with different absolute
    /// `at` values (any two ticks sharing the byte at bit-position `8*l`
    /// land in the same slot regardless of their other bytes), inserted in
    /// whatever order the caller armed them — so the FIFO-front entry is not
    /// necessarily the chronologically earliest one in that slot.
    /// Every tick with an entry, ascending and without duplicates.
    pub fn pending_ticks(&self) -> std::collections::BTreeSet<SimTick> {
        let mut out: std::collections::BTreeSet<SimTick> = self.overflow.keys().copied().collect();
        for level in &self.levels {
            for slot in level {
                out.extend(slot.iter().map(|(at, _)| *at));
            }
        }
        out
    }

    pub fn next_tick(&self) -> Option<SimTick> {
        let mut min: Option<SimTick> = None;
        // Level 0 holds only ticks in `[current, current + 256)` (an entry
        // goes there when it is less than 256 ticks out, and time never
        // advances past a pending tick), one tick per slot: the first
        // non-empty slot from `current`'s, circularly, is the level's earliest.
        if self.counts[0] > 0 {
            let start = (self.current & 0xFF) as usize;
            for k in 0..SLOTS {
                let slot = &self.levels[0][(start + k) % SLOTS];
                if let Some(at) = slot.iter().map(|(at, _)| *at).min() {
                    debug_assert!(at >= self.current, "a level-0 entry is in the past");
                    min = Some(at);
                    break;
                }
            }
        }
        let upper = match self.upper_min.get() {
            Some(known) => known,
            None => {
                let mut m: Option<SimTick> = None;
                for l in 1..LEVELS {
                    if self.counts[l] == 0 {
                        continue;
                    }
                    for slot in &self.levels[l] {
                        for (at, _) in slot.iter() {
                            m = Some(m.map_or(*at, |x| x.min(*at)));
                        }
                    }
                }
                self.upper_min.set(Some(m));
                m
            }
        };
        if let Some(at) = upper {
            min = Some(min.map_or(at, |m| m.min(at)));
        }
        if let Some((&at, _)) = self.overflow.iter().next() {
            min = Some(min.map_or(at, |m| m.min(at)));
        }
        min
    }

    /// Cascades every level whose bucket for exactly `at_tick` must now be
    /// redistributed because `at_tick` is a multiple of that level's
    /// boundary (`256^level`) — the classic hierarchical-wheel rule. Does
    /// *not* touch the far-future overflow map (see `migrate_overflow_due_by`,
    /// called once per `advance_to`, not once per boundary — unlike the
    /// levels, an overflow entry's placement doesn't depend on which
    /// specific boundary tick it gets migrated at, only on the final `now`,
    /// so batching it is free; the levels genuinely do need each boundary
    /// visited, since an entry sitting in level `l`'s bucket for slot `s`
    /// only becomes reachable — and only gets a chance at a *further*
    /// cascade into a lower level — at the exact tick that bucket is due to
    /// roll over).
    ///
    /// A caller only ever needs to call this at ticks where `at_tick %
    /// SLOTS == 0` (see `advance_to`'s doc comment for why every other tick
    /// is a guaranteed no-op here: levels 2 and 3's boundaries are
    /// themselves multiples of level 1's `SLOTS`-tick boundary, so a tick
    /// that isn't can never satisfy any level's `is_multiple_of` check).
    fn cascade_boundary(&mut self, at_tick: SimTick) {
        for l in 1..LEVELS {
            let boundary = (SLOTS as u64).pow(l as u32);
            if at_tick.is_multiple_of(boundary) {
                let slot = ((at_tick >> (8 * l)) & 0xFF) as usize;
                if self.levels[l][slot].is_empty() {
                    continue;
                }
                let bucket: Vec<(SimTick, TimerEntry)> = self.levels[l][slot].drain(..).collect();
                self.counts[l] -= bucket.len();
                // Entries leave the upper levels: their minimum is stale.
                self.upper_min.set(None);
                for (at, entry) in bucket {
                    self.insert_relative_at(at, entry, at_tick);
                }
            }
        }
    }

    /// The smallest tick `m >= from` (a multiple of `SLOTS`) at which some
    /// level `l >= 1` has a non-empty bucket due to roll over (`m` a
    /// multiple of `256^l`, bucket `(m >> 8l) & 0xFF`), or `None` if no
    /// upper level holds an entry. At most `SLOTS` probes per occupied level.
    fn next_occupied_boundary(&self, from: SimTick) -> Option<SimTick> {
        let mut best: Option<SimTick> = None;
        for l in 1..LEVELS {
            if self.counts[l] == 0 {
                continue;
            }
            let step = (SLOTS as u64).pow(l as u32);
            let first = from.div_ceil(step).saturating_mul(step);
            for k in 0..SLOTS as u64 {
                let m = first.saturating_add(k.saturating_mul(step));
                if best.is_some_and(|b| m >= b) {
                    break;
                }
                if !self.levels[l][((m >> (8 * l)) & 0xFF) as usize].is_empty() {
                    best = Some(m);
                    break;
                }
            }
        }
        best
    }

    /// Promotes every far-future overflow entry now within the top level's
    /// range of `as_of` (i.e. `key < as_of + 256^LEVELS`) into the levels,
    /// placed exactly where `insert_relative_at` puts an entry that far out
    /// as of `as_of`. Called once per `advance_to` call with the wheel's
    /// *final* new `current` — SC-025/027/031's migration requirement is
    /// about an entry eventually becoming reachable at the correct tick,
    /// not about which specific intermediate tick performs the move, and
    /// `insert_relative_at` always recomputes fresh from `(at, as_of)`
    /// rather than trusting prior state, so migrating once at the final
    /// `as_of` instead of re-checking on every intervening tick is
    /// observably identical (delta only ever shrinks as `as_of` grows, so
    /// no entry can be placed too early, and any entry due at or before
    /// this call's target tick has `key <= as_of < threshold`, so it is
    /// always captured by this single check).
    fn migrate_overflow_due_by(&mut self, as_of: SimTick) {
        if self.overflow.is_empty() {
            return;
        }
        let threshold = as_of.saturating_add((SLOTS as u64).pow(LEVELS as u32));
        let due_keys: Vec<SimTick> = self.overflow.range(..threshold).map(|(k, _)| *k).collect();
        for key in due_keys {
            if let Some(entries) = self.overflow.remove(&key) {
                for entry in entries {
                    self.insert_relative_at(key, entry, as_of);
                }
            }
        }
    }

    /// Pops every entry stored for exactly `tick` out of level 0's slot for
    /// `tick` (plus a defensive overflow check, for a directly-`arm`ed
    /// far-future entry whose tick happens to be reached without ever
    /// cascading through a level-3 boundary — see `cascade`'s promotion
    /// step, which should already have moved it in by then, but this keeps
    /// `advance_to` correct even if `cascade`'s promotion window were ever
    /// narrowed). Preserves FIFO order (SC-030) — `VecDeque::drain` yields
    /// in push order, and `due` is built by iterating that drain in order.
    fn pop_due(&mut self, tick: SimTick) -> Vec<TimerEntry> {
        let slot = (tick & 0xFF) as usize;
        let bucket = &mut self.levels[0][slot];
        let mut due = Vec::with_capacity(bucket.len());
        // Level 0 holds one tick per slot (see `next_tick`), so normally
        // everything here is due; anything else goes back, in order. Draining
        // in place keeps the slot's buffer for the next time it fills.
        let mut later: Vec<(SimTick, TimerEntry)> = Vec::new();
        for (at, entry) in bucket.drain(..) {
            #[cfg(test)]
            count_op();
            if at == tick {
                due.push(entry);
            } else {
                later.push((at, entry));
            }
        }
        bucket.extend(later);
        self.counts[0] -= due.len();
        if let Some(entries) = self.overflow.remove(&tick) {
            due.extend(entries);
        }
        due
    }

    /// Advances the wheel's own `current` position up to `tick` (a no-op
    /// walk if `tick <= current` already), cascading any level-1+ buckets
    /// that boundary-cross along the way, then pops and returns every entry
    /// due at exactly `tick`. Preserves same-slot FIFO order (SC-030).
    ///
    /// SC-026/028/031's bounded-cost requirement: this deliberately does
    /// *not* walk every integer tick between the old and new `current` (an
    /// earlier revision did, at `O(tick - current)` cost — 27s wall-clock
    /// for a single `256^4`-tick jump on this machine, measured before this
    /// fix). Every level-1..3 cascade can only ever fire at a tick that is a
    /// multiple of `SLOTS` (level 1's own boundary — levels 2 and 3's
    /// boundaries, `SLOTS^2` and `SLOTS^3`, are themselves multiples of
    /// `SLOTS`), so every other integer tick is a *guaranteed* no-op for
    /// every level; this walks only the multiples of `SLOTS` in
    /// `(old current, tick]` (`cascade_boundary`), an `O((tick - current) /
    /// SLOTS)` bound — and does the far-future overflow migration exactly
    /// once per call (`migrate_overflow_due_by`, see its own doc comment
    /// for why once-per-call is equivalent to once-per-tick) rather than
    /// once per tick, which is what made the far-future overflow check
    /// itself `O(tick - current)` even though cascading alone wasn't.
    pub fn advance_to(&mut self, tick: SimTick) -> Vec<TimerEntry> {
        if tick > self.current {
            let slots = SLOTS as u64;
            let mut boundary = (self.current / slots + 1) * slots;
            // Visit only the boundaries whose bucket is non-empty
            // (`cascade_boundary` is a no-op on every other one), so a jump
            // of 10^12 ticks (`#1s` at a 1 ps precision, LRM 3.14.3) costs
            // O(occupied buckets), not O(tick / SLOTS).
            while boundary <= tick {
                match self.next_occupied_boundary(boundary) {
                    Some(b) if b <= tick => {
                        self.cascade_boundary(b);
                        boundary = b + slots;
                    }
                    _ => break,
                }
            }
            self.current = tick;
            self.migrate_overflow_due_by(tick);
        }
        self.pop_due(tick)
    }

    /// The one entry point a `Suspend::Delay` normalizes through (SC-029 —
    /// "without a special-cased wheel code path" is satisfied by the wheel
    /// simply never being touched for a same-delta redirect, not by an `if`
    /// inside `arm`). `delay_ticks == 0` returns `Region::Inactive` without
    /// calling `arm` at all; any other delay arms at `now + delay_ticks`
    /// with `region: Region::Active` and returns `Region::Active`.
    pub fn arm_delay(&mut self, now: SimTick, delay_ticks: u64, act: PendingActivation) -> Region {
        if delay_ticks == 0 {
            return Region::Inactive;
        }
        self.arm(
            now + delay_ticks,
            TimerEntry {
                action: TimerAction::Resume(act, Region::Active),
            },
        );
        Region::Active
    }

    /// SC-069: arms a delayed named-event trigger (`->>#N my_event;`'s
    /// scheduler-side shape) to fire at absolute tick `at`. Unlike
    /// `arm_delay`, there is no `delay_ticks == 0` special case here -- an
    /// undelayed `->` never suspends/touches the wheel at all (it fires the
    /// trigger synchronously through the same `notify_change` fan-out,
    /// scheduler.md §8), so every call to this method genuinely arms the
    /// wheel.
    pub fn arm_trigger(&mut self, at: SimTick, cell: StateCellId) {
        self.arm(
            at,
            TimerEntry {
                action: TimerAction::TriggerEvent(cell),
            },
        );
    }
}

/// Test-only structural accessors (SC-024/025/027/028/031's own closure
/// tests read internal layout directly — the doc comment on each call site
/// says which invariant it pins). Additive: no production caller uses
/// these, only `#[cfg(test)]` code, so they carry no run-time cost outside
/// tests.
#[cfg(test)]
impl TimeWheel {
    /// The number of slots at each of the 4 levels (pinned at `[256, 256,
    /// 256, 256]` by SC-024 — fails if `LEVELS`/`SLOTS` ever changes without
    /// updating that pinned test).
    fn level_slot_counts(&self) -> [usize; LEVELS] {
        std::array::from_fn(|l| self.levels[l].len())
    }

    /// Total entries currently sitting in the far-future overflow map
    /// (summed across every distinct key, not the key count).
    fn overflow_len(&self) -> usize {
        self.overflow.values().map(|v| v.len()).sum()
    }

    /// Total entries currently sitting anywhere in level `l` (summed across
    /// every slot at that level).
    fn level_len(&self, l: usize) -> usize {
        self.levels[l].iter().map(|slot| slot.len()).sum()
    }

    /// The number of entries in exactly `levels[level][slot]`.
    fn slot_len(&self, level: usize, slot: usize) -> usize {
        self.levels[level][slot].len()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::sched::{ResumeArg, SchedTime, sort_ready_set};

    fn pending_activation(id: u64, registration_order: u64) -> PendingActivation {
        PendingActivation {
            id,
            proc: id as u32,
            arg: ResumeArg::Start,
            reconv_iter: 0,
            logical_order: (SchedTime { ticks: 0 }, Region::Active, registration_order),
        }
    }

    fn timer_entry(id: u64) -> TimerEntry {
        TimerEntry {
            action: TimerAction::Resume(pending_activation(id, id), Region::Active),
        }
    }

    fn timer_entry_with_registration(id: u64, registration_order: u64) -> TimerEntry {
        TimerEntry {
            action: TimerAction::Resume(pending_activation(id, registration_order), Region::Active),
        }
    }

    /// Test-only unwrap: every existing (pre-SC-069) test only ever arms and
    /// fires `Resume` entries via `timer_entry`/`timer_entry_with_registration`
    /// above -- this panics loudly (rather than silently misreading a
    /// `TriggerEvent` entry's bytes) if that ever stops being true for one of
    /// them.
    fn resume_activation(entry: &TimerEntry) -> PendingActivation {
        match entry.action {
            TimerAction::Resume(act, _) => act,
            TimerAction::TriggerEvent(cell) => {
                panic!("expected a Resume timer entry, got TriggerEvent({cell:?})")
            }
        }
    }

    // SC-030 (positive half): same-tick entries fire in fixed intra-bucket
    // FIFO order.
    #[test]
    fn same_tick_entries_fire_in_fifo_insertion_order() {
        let mut wheel = TimeWheel::new();
        // Armed (inserted) in this order: id 3, then 1, then 2.
        wheel.arm(10, timer_entry(3));
        wheel.arm(10, timer_entry(1));
        wheel.arm(10, timer_entry(2));

        let fired = wheel.advance_to(10);
        let ids: Vec<u64> = fired.iter().map(|t| resume_activation(t).id).collect();
        assert_eq!(
            ids,
            vec![3, 1, 2],
            "same-tick entries must fire in FIFO arming order, never HashMap/id order"
        );
    }

    // SC-030 (negative half): wheel FIFO order feeds the canonical sort, but
    // is not itself the determinism guarantee.
    #[test]
    fn wheel_fifo_order_is_not_itself_the_determinism_guarantee() {
        let mut wheel = TimeWheel::new();
        // Arming order: id 1, then 2, then 3 — but each entry's
        // registration_order (logical_order.2, sort_ready_set's tiebreak)
        // deliberately disagrees with that arming order.
        wheel.arm(5, timer_entry_with_registration(1, 30));
        wheel.arm(5, timer_entry_with_registration(2, 10));
        wheel.arm(5, timer_entry_with_registration(3, 20));

        let fired = wheel.advance_to(5);
        let fifo_ids: Vec<u64> = fired.iter().map(|t| resume_activation(t).id).collect();
        assert_eq!(
            fifo_ids,
            vec![1, 2, 3],
            "the wheel itself must still report plain arming-order FIFO"
        );

        let mut acts: Vec<PendingActivation> = fired.iter().map(resume_activation).collect();
        sort_ready_set(&mut acts);
        let sorted_ids: Vec<u64> = acts.iter().map(|a| a.id).collect();

        assert_ne!(
            sorted_ids, fifo_ids,
            "canonical sort must be able to reorder relative to wheel FIFO order — \
             if this ever equals fifo_ids the test fixture's registration_order \
             disagreement stopped being exercised"
        );
        assert_eq!(
            sorted_ids,
            vec![2, 3, 1],
            "canonical order follows registration_order (logical_order.2), not wheel arming order"
        );
    }

    // SC-029: Delay(0) never touches the wheel, redirects to Inactive.
    #[test]
    fn arm_delay_zero_never_touches_the_wheel_and_redirects_to_inactive() {
        let mut wheel = TimeWheel::new();
        let before = wheel.next_tick();

        let region = wheel.arm_delay(100, 0, pending_activation(1, 0));

        assert_eq!(region, Region::Inactive);
        let after = wheel.next_tick();
        assert_eq!(
            before, after,
            "arm_delay(.., 0, ..) must not change wheel state at all"
        );
        assert_eq!(
            after, None,
            "a wheel that has only ever seen a delay-0 arm_delay call must remain empty"
        );
    }

    // Feeds SC-032: `run_until`'s "wheel empty -> terminate" check needs
    // next_tick() to actually report emptiness rather than e.g. panicking or
    // returning a sentinel tick.
    #[test]
    fn next_tick_returns_none_when_wheel_and_overflow_are_both_empty() {
        let wheel = TimeWheel::new();
        assert_eq!(wheel.next_tick(), None);
    }

    #[test]
    fn advance_to_cascades_a_higher_level_bucket_into_level_zero() {
        let mut wheel = TimeWheel::new();
        // 300 ticks is beyond level 0's 256-slot span, so this must land in
        // level 1 (or higher) and be cascaded down as `current` advances.
        wheel.arm(300, timer_entry(42));

        for tick in 1..300 {
            let fired = wheel.advance_to(tick);
            assert!(
                fired.is_empty(),
                "entry armed at tick 300 must not fire early, but fired at tick {tick}"
            );
        }

        let fired = wheel.advance_to(300);
        let ids: Vec<u64> = fired.iter().map(|t| resume_activation(t).id).collect();
        assert_eq!(
            ids,
            vec![42],
            "entry armed at tick 300 must fire exactly at tick 300 after cascading \
             from level 1 into level 0 — a dropped or double-fired cascade would fail this"
        );

        let fired_again = wheel.advance_to(301);
        assert!(
            fired_again.is_empty(),
            "an entry must never fire a second time after its tick has passed"
        );
    }

    // LRM 3.14.3: `#1s` at a 1 ps global precision is one 10^12-tick jump. The
    // cascade walk visits only occupied boundaries, so this returns at once
    // (it used to take 4*10^9 iterations) and still fires every entry on time.
    #[test]
    fn advance_to_a_huge_jump_cascades_only_occupied_boundaries_and_fires_on_time() {
        let mut wheel = TimeWheel::new();
        let far = 1_000_000_000_000u64;
        wheel.arm(far, timer_entry(1));
        wheel.arm(far + 300, timer_entry(2));
        wheel.arm(far - 5, timer_entry(3));
        wheel.arm(7, timer_entry(4));
        let start = std::time::Instant::now();
        assert_eq!(wheel.next_tick(), Some(7));
        let ids = |w: &mut TimeWheel, t| -> Vec<u64> {
            w.advance_to(t)
                .iter()
                .map(|e| resume_activation(e).id)
                .collect()
        };
        assert_eq!(ids(&mut wheel, 7), vec![4]);
        assert_eq!(wheel.next_tick(), Some(far - 5));
        assert_eq!(ids(&mut wheel, far - 5), vec![3]);
        assert_eq!(ids(&mut wheel, far), vec![1]);
        assert_eq!(wheel.next_tick(), Some(far + 300));
        assert_eq!(ids(&mut wheel, far + 300), vec![2]);
        assert_eq!(wheel.next_tick(), None);
        assert!(
            start.elapsed().as_secs() < 5,
            "the jump must not walk every boundary"
        );
    }

    // Regression: next_tick() must report the true minimum `at` across every
    // entry in a slot, not just the FIFO front. Two entries with different
    // `at` can legitimately collide in the same level>=1 slot (any two ticks
    // sharing the byte at bit-position `8*l` land in the same slot
    // regardless of their other bytes); arming the chronologically LATER one
    // first must not make next_tick() report it as the earliest.
    #[test]
    fn next_tick_finds_the_true_minimum_when_two_different_at_entries_collide_in_one_higher_level_slot()
     {
        let mut wheel = TimeWheel::new();
        // 1000 >> 8 == 3 and 800 >> 8 == 3: both land in level 1, slot 3.
        // Armed in descending `at` order (later-due first) so a naive
        // front()-only scan would report 1000 instead of the true min 800.
        wheel.arm(1000, timer_entry(100));
        wheel.arm(800, timer_entry(200));

        assert_eq!(
            wheel.next_tick(),
            Some(800),
            "next_tick() must return the chronologically earliest `at` in the \
             colliding slot (800), not the FIFO-front entry (1000)"
        );

        // Driving the wheel via the documented next_tick()/advance_to() loop
        // must fire entries in non-decreasing `at` order end-to-end, and the
        // wheel's `current` must never race past an unfired earlier entry.
        let mut fired_order: Vec<(SimTick, u64)> = Vec::new();
        while let Some(t) = wheel.next_tick() {
            for entry in wheel.advance_to(t) {
                fired_order.push((t, resume_activation(&entry).id));
            }
        }

        assert_eq!(
            fired_order,
            vec![(800, 200), (1000, 100)],
            "entries must be reported in non-decreasing `at` order: the tick-800 \
             entry (id 200) before the tick-1000 entry (id 100)"
        );
    }

    // Regression: arm_delay's nonzero-delay path (the branch every real
    // `#delay` statement will use) must actually arm the wheel and fire at
    // the correct absolute tick — not just return `Region::Active` without
    // scheduling anything.
    #[test]
    fn arm_delay_nonzero_arms_the_wheel_and_fires_at_the_correct_tick() {
        let mut wheel = TimeWheel::new();
        let now: SimTick = 42;
        let delay = 5;

        let region = wheel.arm_delay(now, delay, pending_activation(7, 0));

        assert_eq!(region, Region::Active);
        assert_eq!(
            wheel.next_tick(),
            Some(now + delay),
            "arm_delay(now, delay, ..) must arm the wheel to fire at now + delay"
        );

        let fired = wheel.advance_to(now + delay);
        let ids: Vec<u64> = fired.iter().map(|t| resume_activation(t).id).collect();
        assert_eq!(
            ids,
            vec![7],
            "the activation armed via arm_delay must actually fire at the \
             correct tick — a forgotten self.arm() call would leave this empty"
        );
    }

    // SC-024: struct layout is exactly 4 levels of 256 slots each.
    #[test]
    fn sc024_wheel_has_exactly_four_levels_of_256_slots_each() {
        let wheel = TimeWheel::new();
        assert_eq!(
            wheel.level_slot_counts(),
            [256, 256, 256, 256],
            "the wheel must have exactly 4 hierarchical levels of 256 \
             slots each — a change to LEVELS/SLOTS without updating this \
             pinned test must fail here first"
        );
    }

    // SC-025: a delay beyond every level's range (past 256^4 ticks, level
    // 3's own top) lands in the far-future BTreeMap overflow, is visible
    // from there via next_tick() (never silently dropped/truncated), and
    // fires exactly once when advanced to directly from a fresh wheel —
    // proving advance_to's cost is not an O(delay) walk (see advance_to's
    // own doc comment: an earlier, naive per-tick-walk implementation took
    // 27s wall-clock for exactly this jump; this call must complete in a
    // small fraction of a second).
    #[test]
    fn sc025_a_delay_beyond_every_level_lands_in_overflow_and_fires_exactly_once() {
        let mut wheel = TimeWheel::new();
        let far: SimTick = 256u64.pow(4) + 1000;
        wheel.arm(far, timer_entry(99));

        assert_eq!(
            wheel.overflow_len(),
            1,
            "must be sitting in the overflow map"
        );
        for l in 0..4 {
            assert_eq!(
                wheel.level_len(l),
                0,
                "must not be in any level {l} bucket before any advance"
            );
        }
        assert_eq!(
            wheel.next_tick(),
            Some(far),
            "an entry beyond every level's range must still be visible via \
             next_tick() from the overflow map, never silently dropped or truncated"
        );

        let fired = wheel.advance_to(far);
        let ids: Vec<u64> = fired.iter().map(|t| resume_activation(t).id).collect();
        assert_eq!(
            ids,
            vec![99],
            "must fire exactly once, at the exact far-future tick, when \
             advanced to directly from a fresh wheel with no intervening steps"
        );

        let fired_again = wheel.advance_to(far + 1);
        assert!(
            fired_again.is_empty(),
            "must never fire a second time after its tick has passed"
        );
    }

    // SC-027: an entry starting in the far-future overflow map migrates
    // into the level structure as `current` (driven forward in coarse
    // strides, not one tick at a time) approaches within the wheel's total
    // span of it, and fires at the exact correct tick afterward. Also
    // proves the migration is a real, observable state transition (not
    // already-there-by-construction): the entry is confirmed to be in
    // overflow and NOT in any level before the threshold, then confirmed to
    // be out of overflow and IN a level after it.
    #[test]
    fn sc027_overflow_entry_migrates_into_the_wheel_as_now_approaches_it() {
        let mut wheel = TimeWheel::new();
        // 100_000 ticks past the wheel's total span (256^4) -- comfortably
        // in overflow (SC-025's own boundary), with a migration threshold
        // (`far - 256^4 == 100_000`) far enough out that reaching it takes
        // many coarse strides, not a lucky single step.
        let far: SimTick = 256u64.pow(4) + 100_000;
        let migration_point: SimTick = far - 256u64.pow(4); // == 100_000
        wheel.arm(far, timer_entry(7));

        assert_eq!(wheel.overflow_len(), 1);
        for l in 0..4 {
            assert_eq!(wheel.level_len(l), 0);
        }

        let stride: SimTick = 256;
        let mut current: SimTick = 0;
        while current <= migration_point {
            current += stride;
            let fired = wheel.advance_to(current);
            assert!(fired.is_empty(), "must not fire before its tick");
            if current <= migration_point {
                assert_eq!(
                    wheel.overflow_len(),
                    1,
                    "must not migrate out of overflow before `current` is \
                     within the wheel's total span of it (current={current}, \
                     migration_point={migration_point})"
                );
                for l in 0..4 {
                    assert_eq!(
                        wheel.level_len(l),
                        0,
                        "must not appear in level {l} before the migration point"
                    );
                }
            }
        }

        // `current` has now crossed the migration point: the entry must be
        // physically out of overflow and sitting in exactly one level.
        assert_eq!(
            wheel.overflow_len(),
            0,
            "must have migrated out of overflow once `current` is within \
             the wheel's total span of its tick"
        );
        let total_in_levels: usize = (0..4).map(|l| wheel.level_len(l)).sum();
        assert_eq!(
            total_in_levels, 1,
            "must be sitting in exactly one level bucket after migration"
        );
        assert_eq!(
            wheel.next_tick(),
            Some(far),
            "must remain discoverable at its exact original tick after migration"
        );

        let fired = wheel.advance_to(far);
        let ids: Vec<u64> = fired.iter().map(|t| resume_activation(t).id).collect();
        assert_eq!(
            ids,
            vec![7],
            "must fire at the exact correct tick after migrating in from overflow"
        );
    }

    // SC-026: bounded insert/advance cost, expressed as a deterministic
    // operation-count assertion (never a wall-clock benchmark — this
    // codebase's own memory notes flag wall-clock thresholds as a gate that
    // cannot fail/pass reliably). 100_000 entries at 100_000 distinct
    // near-term ticks, drained one tick at a time; total instrumented
    // `insert_relative_at`/`pop_due`-visit count must stay linear in the
    // number of *entries*, never quadratic and never proportional to some
    // unrelated span of ticks. A `BinaryHeap`/re-sorted-`Vec` reimplementation
    // (O(log n) or O(n) per operation, scaling with total resident entries)
    // would blow past this bound at N=100_000.
    #[test]
    fn sc026_dense_near_term_insert_and_advance_cost_is_bounded_per_operation() {
        reset_op_counter();
        let mut wheel = TimeWheel::new();
        const N: u64 = 100_000;
        for t in 1..=N {
            wheel.arm(t, timer_entry(t));
        }

        let mut total_fired = 0usize;
        for t in 1..=N {
            total_fired += wheel.advance_to(t).len();
        }
        assert_eq!(
            total_fired, N as usize,
            "every armed entry must eventually fire exactly once"
        );

        // K=4: one insert (arm), at most one cascade re-insert as the entry
        // migrates down through the levels it started in, and one pop-scan
        // visit when it fires -- an upper bound per entry, not a tight one.
        const K: u64 = 4;
        let ops = op_counter();
        assert!(
            ops <= N * K,
            "instrumented op count {ops} exceeded the O(1)-amortized-per-entry \
             bound of {N}*{K}={}; a wrong implementation whose per-operation \
             cost scales with the total resident entry count would blow past \
             this bound at N={N}",
            N * K
        );
    }

    // SC-028: level-0 bucket membership is exact `tick mod 256` placement
    // (O(1) direct index), never a scan that could collide or leave a slot
    // empty. 256 entries at ticks 0..256 (all within level 0's own range)
    // must land one-per-slot, with slot index equal to the tick itself.
    #[test]
    fn sc028_level_zero_bucket_membership_is_exact_tick_mod_256_not_a_scan() {
        let mut wheel = TimeWheel::new();
        for t in 0..256u64 {
            wheel.arm(t, timer_entry(t));
        }
        for slot in 0..256usize {
            assert_eq!(
                wheel.slot_len(0, slot),
                1,
                "level 0 slot {slot} must hold exactly the one entry armed \
                 at that exact tick — direct index placement, not a scan \
                 that could double up one slot and leave another empty"
            );
        }
    }

    // SC-031: a mixed near-future/far-future workload (a `#1`-style clock
    // running right alongside a single far-future timeout) shares one wheel
    // with no unbounded preallocation and no migration duplication. The
    // outer level arrays are fixed-size Vecs by construction; this asserts
    // that invariant holds under 10_000 iterations of sustained near-term
    // load, not just by inspection, and that the far-future entry is never
    // duplicated by a buggy migration path.
    #[test]
    fn sc031_mixed_near_and_far_future_delays_share_the_wheel_with_no_unbounded_preallocation() {
        let mut wheel = TimeWheel::new();
        let far: SimTick = 256u64.pow(4) + 100_000_000;
        wheel.arm(far, timer_entry(1));
        assert_eq!(
            wheel.overflow_len(),
            1,
            "an entry beyond the wheel's total span must start in overflow"
        );

        let mut now: SimTick = 0;
        for _ in 0..10_000u64 {
            now += 1;
            wheel.arm(now, timer_entry(2));
            let fired = wheel.advance_to(now);
            assert_eq!(
                fired.len(),
                1,
                "the near-term entry armed one tick ahead must fire exactly \
                 on schedule every iteration"
            );

            assert_eq!(
                wheel.level_slot_counts(),
                [256, 256, 256, 256],
                "the outer level arrays must never grow or reallocate under \
                 sustained near-term load next to a far-future entry"
            );
            assert!(
                wheel.overflow_len() <= 1,
                "the far-future entry must never be duplicated by a buggy \
                 migration path"
            );
        }
    }

    // SC-081/082 (B-sched Task 2, scheduler-level proof): two independent
    // re-arming chains -- chain A (id 0) re-arms itself every 5 ticks,
    // chain B (id 1) every 7 ticks -- driven for 70 ticks (2 full periods
    // of the 35-tick LCM beat) via the exact `next_tick()`/`advance_to()`
    // loop `Engine::run_until` itself uses. Chain A's fire set must be
    // exactly {5,10,...,70} and chain B's exactly {7,14,...,70}, with BOTH
    // firing at the beat coincidences (tick 35 and tick 70) -- a same-slot
    // collision bug that merges or drops one chain's entry at a shared tick
    // would show up here as a missing or duplicated fire at exactly 35/70.
    #[test]
    fn two_coprime_period_chains_interleave_with_no_missed_edges_at_their_beat_frequency() {
        const PERIOD_A: u64 = 5;
        const PERIOD_B: u64 = 7;
        const HORIZON: u64 = 70;

        let mut wheel = TimeWheel::new();
        wheel.arm(PERIOD_A, timer_entry(0));
        wheel.arm(PERIOD_B, timer_entry(1));

        let mut fired_a: Vec<u64> = Vec::new();
        let mut fired_b: Vec<u64> = Vec::new();
        let mut fired_pairs: Vec<(u64, u64)> = Vec::new();

        while let Some(t) = wheel.next_tick() {
            if t > HORIZON {
                break;
            }
            for entry in wheel.advance_to(t) {
                fired_pairs.push((t, resume_activation(&entry).id));
                match resume_activation(&entry).id {
                    0 => {
                        fired_a.push(t);
                        if t + PERIOD_A <= HORIZON {
                            wheel.arm(t + PERIOD_A, timer_entry(0));
                        }
                    }
                    1 => {
                        fired_b.push(t);
                        if t + PERIOD_B <= HORIZON {
                            wheel.arm(t + PERIOD_B, timer_entry(1));
                        }
                    }
                    other => panic!("unexpected chain id {other}"),
                }
            }
        }

        let expected_a: Vec<u64> = (1..=14).map(|k| k * PERIOD_A).collect();
        let expected_b: Vec<u64> = (1..=10).map(|k| k * PERIOD_B).collect();
        assert_eq!(
            fired_a, expected_a,
            "chain A (period {PERIOD_A}) must fire at exactly every multiple \
             of its own period up to the horizon, with no missed or \
             duplicated edge"
        );
        assert_eq!(
            fired_b, expected_b,
            "chain B (period {PERIOD_B}) must fire at exactly every multiple \
             of its own period up to the horizon, with no missed or \
             duplicated edge"
        );
        assert!(
            fired_pairs.contains(&(35, 0)) && fired_pairs.contains(&(35, 1)),
            "both chains must fire at their first beat coincidence (tick 35) \
             -- a same-tick cross-chain collision that merges or drops one \
             chain's entry would fail this: got {fired_pairs:?}"
        );
        assert!(
            fired_pairs.contains(&(70, 0)) && fired_pairs.contains(&(70, 1)),
            "both chains must fire at their second beat coincidence (tick \
             70) -- got {fired_pairs:?}"
        );
    }

    // SC-082 (construction proof at the wheel level): two chains with
    // periods 2500 and 3500 ticks -- not integer multiples of each other
    // at a coarse (e.g. 1ns) tick unit, but exact integers at the actual
    // tick precision (representing 2.5ns/3.5ns at 1ps precision) -- driven
    // for 4 periods each. Every fired tick for each chain must be an exact
    // multiple of its own period with zero cumulative drift: a
    // rational-timescale-arithmetic bug (float accumulation, integer
    // division truncation) would show up as e.g. `9999` instead of
    // `10000`, which a flat-`u64`-tick-counter implementation (SC-023)
    // cannot exhibit by construction -- this test is the proof, not a
    // hand-wave.
    #[test]
    fn sc082_tick_precision_represents_two_non_integer_multiple_periods_exactly() {
        const PERIOD_A: u64 = 2500;
        const PERIOD_B: u64 = 3500;
        const REPS: usize = 4;

        let mut wheel = TimeWheel::new();
        wheel.arm(PERIOD_A, timer_entry(0));
        wheel.arm(PERIOD_B, timer_entry(1));

        let mut fired_a: Vec<u64> = Vec::new();
        let mut fired_b: Vec<u64> = Vec::new();

        while fired_a.len() < REPS || fired_b.len() < REPS {
            let t = wheel
                .next_tick()
                .expect("wheel must not go empty before both chains complete their 4 periods");
            for entry in wheel.advance_to(t) {
                match resume_activation(&entry).id {
                    0 => {
                        fired_a.push(t);
                        if fired_a.len() < REPS {
                            wheel.arm(t + PERIOD_A, timer_entry(0));
                        }
                    }
                    1 => {
                        fired_b.push(t);
                        if fired_b.len() < REPS {
                            wheel.arm(t + PERIOD_B, timer_entry(1));
                        }
                    }
                    other => panic!("unexpected chain id {other}"),
                }
            }
        }

        assert_eq!(
            fired_a,
            vec![2500, 5000, 7500, 10000],
            "chain A's fired ticks must be exact integer multiples of its \
             own 2500-tick period with zero cumulative drift"
        );
        assert_eq!(
            fired_b,
            vec![3500, 7000, 10500, 14000],
            "chain B's fired ticks must be exact integer multiples of its \
             own 3500-tick period with zero cumulative drift"
        );
    }

    // --- Task 6 (B-sched, ledger SC-069): TimerAction::TriggerEvent -----

    /// SC-069's wheel-level half: a `Resume` and a `TriggerEvent` entry armed
    /// at the same tick must both come back from `advance_to`, each with its
    /// own `action` variant intact, in FIFO arming order (SC-030's existing
    /// guarantee, now proven to hold across the two `TimerAction` variants
    /// too -- a wrong implementation that collapsed `TimerAction` back into
    /// a single shape, or that dropped/miscounted one of the two arms, would
    /// fail this).
    #[test]
    fn resume_and_trigger_entries_at_the_same_tick_both_fire_correctly() {
        let mut wheel = TimeWheel::new();
        let cell = StateCellId(77);
        // Armed in this order: the Resume entry first, then the trigger.
        wheel.arm(10, timer_entry(1));
        wheel.arm_trigger(10, cell);

        let fired = wheel.advance_to(10);
        assert_eq!(
            fired.len(),
            2,
            "both entries armed at the same tick must fire together, neither \
             dropped nor merged"
        );
        match fired[0].action {
            TimerAction::Resume(act, region) => {
                assert_eq!(act.id, 1);
                assert_eq!(region, Region::Active);
            }
            TimerAction::TriggerEvent(c) => panic!(
                "expected the first fired entry (FIFO arming order) to be the \
                 Resume entry, got TriggerEvent({c:?})"
            ),
        }
        match fired[1].action {
            TimerAction::TriggerEvent(c) => assert_eq!(c, cell),
            TimerAction::Resume(act, _) => panic!(
                "expected the second fired entry to be the TriggerEvent entry \
                 (FIFO arming order preserved across the two TimerAction \
                 variants), got Resume(id={})",
                act.id
            ),
        }

        let fired_again = wheel.advance_to(11);
        assert!(
            fired_again.is_empty(),
            "neither entry may fire a second time"
        );
    }
}