sv_sim_sim/sched/

control.rs

//! SCHED::CONTROL — the `SchedulerControl` trait, the `Engine` type that
//! assembles every earlier `sched` task's pieces, and the ABI entry-point
//! mapping (Task 18, `docs/superpowers/plans/2026-09-23-wave-A-sched.md`;
//! ledger SC-083, SC-116).
//!
//! **Deviations from the plan's literal text (recorded here and in
//! `docs/CONTRACTS.md`'s r3 Deviations section, item 8):**
//!
//! 1. **Resolved (integration, SC-055/SC-056):** `Engine` owns the
//!    [`QueuedGate`] Task 10 introduced. `EngineNotifySink::notify_change`
//!    enqueues through [`sensitivity_match_to_enqueue`] (still Task 9's one
//!    canonical [`SensitivityTable::matches`] underneath, so SC-042's "no
//!    duplicate implementation" holds), and `drain_and_sort` calls
//!    [`QueuedGate::reset_on_dequeue`] for every activation the instant it is
//!    popped off a region queue, strictly before any `resume()`. A process
//!    sensitive to several cells that change in one pass is queued once; a
//!    re-trigger landing after its dequeue queues it again.
//! 2. **`register_process`'s body does more than the plan's literal two
//!    statements** (`table.register` + the conditional `install_static`
//!    call — that half is unchanged and still the crate's one and only
//!    production call site of `install_static`, SC-045). It additionally
//!    branches on `kind`: a `ProcessKind::Final` process is queued into
//!    `self.finals` (SC-077/SC-078 — `final` blocks run only in the
//!    dedicated post-`run_until` pass, Task 14, never dispatched at t=0);
//!    every other kind gets one `ResumeArg::Start` `PendingActivation` pushed
//!    onto `Region::Active` for the current time. Nothing in this task's own
//!    ledger scope or the plan's prose performs "initial dispatch of
//!    `initial`/`always*` processes at t=0" (scheduler.md §1 item 1c)
//!    anywhere else, and no other piece assembled so far in this crate does
//!    either — so without this addition `run_until`/`step_region` would have
//!    no path to ever call `resume(Start)` on anything, making SC-083's own
//!    "single-step through regions" acceptance unsatisfiable. The plan's own
//!    text pins only the `install_static` call count ("No other line...may
//!    call `install_static` a second time"), not the body's total statement
//!    count, so this is additive, not a violation of that specific pin.
//! 3. **`Engine::run_until`/`step_region` are real, working implementations
//!    this task ships (not stubs), but intentionally simplified relative to
//!    the eventual full LRM-faithful loop**: `Suspend::WaitEvents` is wired
//!    directly through [`SensitivityTable::install_dynamic`] (which takes an
//!    `EventWaitSet` directly, exactly what `Suspend::WaitEvents` carries, so
//!    no `WaitSpec` bridge is needed) — **and** `dispatch_ready_set` calls
//!    [`SensitivityTable::remove_dynamic`] for exactly the two `ResumeArg`
//!    variants (`EventFired`/`Disabled`) that terminate a dynamic wait,
//!    before `apply_suspend` processes that same call's resulting `Suspend`
//!    (review fix — an earlier draft installed but never removed, leaking
//!    every dynamic wait's sensitivity-table row forever); `Suspend::ForkJoin`
//!    is wired through [`ResumeDriver::handle_fork`] with every spawned child
//!    backed by [`UnmintedForkChild`], an immediately-`Done` placeholder body
//!    — no codegen exists this wave to mint a fork branch's real compiled
//!    kernel (`SpawnSpec::kernel` indexes a `SV_SIM_PROCESS_TABLE` that
//!    doesn't exist yet), so this is D1a's already-established "interim v1
//!    shadow, not a substitute" posture, extended from `ProcessDescriptor`
//!    bookkeeping (Task 11/12) to the `Engine` dispatch loop itself. **Review
//!    fix:** `apply_suspend`'s `Suspend::Done` arm now calls
//!    [`ResumeDriver::on_child_done`] when the completing process is itself a
//!    fork child (`ProcessDescriptor::fork_parent.is_some()`), and queues the
//!    parent's `ResumeArg::ForkChildDone` activation onto `Region::Active`
//!    when the join condition is satisfied — an earlier draft left this
//!    entirely unwired, so `fork`/`join`/`join_any` silently hung forever
//!    (the parent was never reactivated) despite the bookkeeping fields
//!    themselves being real. **Resolved (B-sched wave Task 7, ledger
//!    SC-062/SC-115):** two gaps this deviation note previously left open,
//!    now closed:
//!    - **Fork-child slot reclaim.** `Suspend::Done` is unconditionally
//!      terminal for the completing process (scheduler.md §5): a fork
//!      child's own `ProcessTable` slot is now reclaimed by
//!      [`ResumeDriver::on_child_done`] itself (`resume.rs`'s own doc
//!      comment) the instant it is called for that child — regardless of
//!      whether that particular completion happens to satisfy/
//!      partially-satisfy the parent's join — never left dangling in the
//!      table forever the way it was before this task.
//!    - **`JoinKind::None`'s "parent proceeds immediately, no waiting"
//!      continuation is now wired**, *without* inventing a new
//!      cross-track `ResumeArg` variant meaning "just continue"
//!      (`ResumeArg` remains A2-process/codegen-owned,
//!      `docs/CONTRACTS.md`'s ownership table, untouched by this task): the
//!      `Suspend::ForkJoin` arm, for `JoinKind::None` specifically,
//!      synchronously calls [`ResumeDriver::resume_for`] again for the
//!      same `proc` (never [`ResumeDriver::start`] — `proc` already went
//!      through `Start` once to reach this very `Suspend::ForkJoin` in the
//!      first place, so a second `start()` would be rejected by its own
//!      once-only CAS guard) with the existing `ResumeArg::TimeElapsed`
//!      variant reused as the least-surprising "no special payload" choice,
//!      right in the same `apply_suspend` call that registered the
//!      children — a direct re-entry, never delivered as a resume-with-a-
//!      reason, matching LRM's "proceeds without waiting" (there is
//!      nothing to wait *for*). See `apply_suspend`'s own `Suspend::
//!      ForkJoin` arm for the exact mechanism and
//!      `sc115_join_none_children_run_detached_while_the_parent_proceeds_
//!      without_waiting` for the closing proof. Task 21 (ledger SC-001/
//!      SC-007, its own separate scope, "`Engine` stops being a skeleton")
//!      remains the task chartered to replace `UnmintedForkChild` with real
//!      fork-child kernels once codegen mints them; this task's own tests
//!      exercise `Suspend::Done`/`Suspend::Delay`/`Suspend::WaitEvents`/
//!      `Suspend::ForkJoin` (`JoinKind::All` and `JoinKind::None`)
//!      end-to-end through `run_until`/`step_region`.
//! 4. **`EngineOps::request_finish` writes directly into `engine.
//!    finish_requested` (a `Mutex<Option<i32>>`), not through Task 16's
//!    [`crate::sched::request_finish_as_deferred_effect`]/`DeferIoDispatcher`
//!    path.** SC-075's literal text asks for `request_finish` to be "queued
//!    as a defer_io-class effect gated on gvt()"; routing through the actual
//!    `DeferIoDispatcher` here would require constructing a *fresh* one per
//!    call (`Engine` has no persistent `DeferIoDispatcher` field — Task 18's
//!    own `defer_io` deliberately bypasses it for the same reason, see this
//!    task's own spec text on `EngineOps::defer_io`), which would silently
//!    reset that fresh dispatcher's own `seq` counter to 0 on every call —
//!    exactly the bug `EngineOps::defer_io` was designed to avoid by writing
//!    straight against `Engine`'s own persistent `io_seq`. Task 21's plan
//!    text explicitly reconciles this: for v1's `SequentialController`
//!    (`gvt() == now` always), a per-timestep `finish_requested` check *is*
//!    SC-075's GVT-gated position requirement, since gvt never lags `now` —
//!    "the reduction is documented here rather than asserted silently".
//!    Task 16's own `request_finish_as_deferred_effect` remains a separate,
//!    independently-tested primitive; it is simply not what `EngineOps`
//!    (the one concrete, production `SchedulerOps` this task adds) calls.
//! 5. **New `SensitivityTable::entry_count`** (`sched/sensitivity.rs`, Task
//!    9's file) and **new `Engine::test_sensitivity_entry_count`** — the
//!    plan's own text for the SC-045 exactly-once test names a direct field
//!    read (`self.sensitivity.table.values()...`) that cannot compile from
//!    this module (`SensitivityTable::table` is private to `sensitivity.rs`);
//!    `entry_count` is the minimal, in-track (same `sched` track, additive,
//!    non-breaking) accessor that makes the literal test possible.
//! 6. **New `DeltaGuard::count` getter (`sched/guard.rs`, Task 15's file)
//!    and new, non-`#[cfg(test)]`, public `Engine::
//!    max_delta_iterations_in_a_timestep`** (Task 21, ledger SC-007) — the
//!    plan's own `region_order_fixture.rs` test text asks to assert "the
//!    delta-loop guard's count never exceeded 3 in any one timestep", but
//!    that integration test is a separate compilation unit under `tests/`
//!    and cannot reach `Engine`'s private `guard` field or any `#[cfg(test)]`
//!    accessor (those only exist in this crate's own unit-test build, per
//!    this module's own established precedent for every other private-field
//!    accessor above). `run_until` now records `DeltaGuard::count()`'s value
//!    into a running high-water mark (`max_delta_iterations_seen`) once per
//!    timestep, right before `guard.reset()` — additive, non-breaking, and
//!    changes no existing behavior or contract type.

use std::sync::Mutex;
use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU64, Ordering};

use sv_sim_abi::{Region, SimHandle};

use crate::determinism::{DeterminismController, SequentialController};
use crate::sched::{
    CompiledProcess, DEFAULT_MAX_DELTA_ITERATIONS, DeferredIoCallback, DeltaGuard, FinalProcess,
    IoSink, JoinKind, NbaLog, PendingActivation, ProcessId, ProcessRequest, ProcessState,
    ProcessTable, QueuedGate, RegionCallback, RegionCallbackRegionError, RegionQueues,
    RepeatCounters, RepeatOutcome, ResumeArg, ResumeDriver, SchedTime, SchedulerCtx, SchedulerOps,
    SensitivityTable, SimTick, SpawnSpec, StaticSensitivity, Suspend, TimeWheel, TimerAction,
    form_ready_set, mailbox, semaphore, sort_ready_set,
};
use crate::state::{
    ArenaSnapshot, ChangeDelta, EdgeKind, FourStateValue, SignalAccess, SnapshotRing, StateCellId,
    ValueChangeSink,
};

/// Bound on queued synchronous continuations in one `resume_synchronously`
/// drain before it is reported as a livelock.
/// Label `$exit` hands `sv_rt_disable_named` (LRM 24.7): not a block name but
/// "every program initial thread of the caller's program".
pub const EXIT_PROGRAM_LABEL: &str = "$exit";

const MAX_SYNC_RESUMES: u32 = 30_000;

/// Fork branches one time step may spawn before it is reported as a livelock:
/// each live branch is kept until it finishes, so a zero-delay loop that
/// forks forever (`always fork #0; #1; join_any`) would otherwise exhaust
/// memory long before the delta guard trips.
const MAX_FORKS_PER_STEP: u32 = 50_000;

/// `run_until`'s outcome (scheduler.md's `run_until`/`step_region` return
/// types, refined only as much as this wave's tests need).
#[derive(Clone, Debug)]
pub enum RunResult {
    /// The wheel emptied naturally with no `finish_requested`/deadline hit.
    Completed,
    /// `SchedulerOps::request_finish(code)` was observed at a timestep
    /// boundary (SC-075).
    Finished { code: i32 },
    /// `deadline` was reached before the run otherwise completed.
    Deadline,
    /// The delta-loop guard tripped (SC-080/DT-050): a zero-delay loop that
    /// never lets time advance. A named runtime error for the caller to
    /// report, never a panic; the payload is the guard's message.
    Livelock(String),
}

/// `step_region`'s outcome.
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
pub enum StepResult {
    /// The region cursor advanced (work was dispatched, or the wheel
    /// advanced `now`).
    Advanced,
    /// Nothing is queued anywhere and the wheel is empty — the simulation is
    /// quiescent.
    Blocked,
}

/// Mirrors `slang::ast::ProceduralBlockSymbol::procedureKind`'s 6 relevant
/// values 1:1 (scheduler.md's `ProceduralBlockKindMirror`, never previously
/// typed — `docs/CONTRACTS.md` r3 item 4).
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
pub enum ProcessKind {
    Initial,
    Always,
    AlwaysComb,
    AlwaysLatch,
    AlwaysFF,
    Final,
}

/// The debugger/VPI-facing control surface (SC-083): run-to-completion,
/// run-until(time), single-region-step, and process registration.
pub trait SchedulerControl {
    /// **Task 9 (A2-part2, CG-102) added `region_id`:** the process's own
    /// `ProcessSideTable::region_id` (`sv-sim-codegen::process_lower`'s own
    /// field), threaded through to `ProcessTable::set_region_id` so
    /// `dispatch_ready_set` can later look it up per-activation via
    /// `ProcessTable::region_id`. v1's one real production call site
    /// (`sv-sim-bin::run::run_design`) always passes `0` (ARCHITECTURE §4
    /// item 1's single-conservative-partition value -- the same "correct
    /// degenerate v1 value, not a stub" posture `region_id`'s own producing-
    /// side doc comment already documents).
    fn register_process(
        &mut self,
        proc: Box<dyn CompiledProcess>,
        kind: ProcessKind,
        sensitivity: Option<StaticSensitivity>,
        region_id: u32,
    ) -> ProcessId;
    fn run_until(&mut self, deadline: Option<SchedTime>) -> RunResult;
    fn step_region(&mut self) -> StepResult;
    fn run_finals(&mut self);
    fn now(&self) -> SchedTime;
    fn set_determinism_controller(&mut self, ctrl: Box<dyn DeterminismController>);
}

/// **Task 9 (A2-part2, CG-102).** The scheduler's own partition identity --
/// distinct from `sv-sim-sim::analysis::RegionId` (the WOCB static-analysis
/// pass's own, unimplemented-this-wave partition-assignment output type):
/// this is the *dispatch-side* value `partition_of` computes FROM a
/// process's already-assigned `region_id`, never itself assigned by
/// anything upstream.
#[derive(Clone, Copy, PartialEq, Eq, Hash, Debug)]
pub struct PartitionId(pub u32);

/// **Task 9 (A2-part2, CG-102).** `PartitionId(region_id % partition_count.
/// max(1))` -- a real, general function (directly unit-testable with
/// `partition_count > 1` to prove it actually reads `region_id`, not a
/// hardcoded-`0` stub). `dispatch_ready_set`'s one real v1 call site always
/// passes `partition_count = 1` (ARCHITECTURE §4 item 1's single-
/// conservative-partition value -- the same citation `region_id`'s own
/// doc comment uses), so every real v1 dispatch still yields
/// `PartitionId(0)` for everything -- the same "correct degenerate v1
/// value, not a stub" posture applied to the *consuming* side, not only
/// the *producing* side. `.max(1)` makes `partition_count == 0` a defined,
/// non-panicking degenerate case (one partition) rather than a divide-by-
/// zero, since nothing about this function's own contract requires a
/// caller to have already proven `partition_count >= 1`.
pub fn partition_of(region_id: u32, partition_count: u32) -> PartitionId {
    PartitionId(region_id % partition_count.max(1))
}

/// One not-yet-fired [`Engine::region_callbacks`] registration (clippy
/// `type_complexity`: named here rather than spelled out inline at every
/// use site).
type RegionCallbackEntry = (SchedTime, Region, Box<dyn RegionCallback>);

/// A fork child whose real compiled body codegen has not minted yet this
/// wave (see this module's doc comment, deviation 3): resumes and
/// immediately reports `Suspend::Done`.
struct UnmintedForkChild;

impl CompiledProcess for UnmintedForkChild {
    fn resume(&mut self, _state: SimHandle, _arg: ResumeArg, _ops: &dyn SchedulerOps) -> Suspend {
        Suspend::Done
    }
}

/// sva T1 (§1.3/§1.4): the Preponed-region snapshot ring, plus the closure
/// that takes a fresh [`ArenaSnapshot`] of whatever concrete, STATE-owned
/// arena this `Engine`'s own `arena: Box<dyn SignalAccess>` is really
/// backed by. `sched` deliberately carries no concrete STATE dependency of
/// its own (this module's doc comment, "sched borrows the SignalAccess
/// contract but does not implement it") and `SignalAccess` itself exposes
/// no snapshot/restore capability, so the snapshot-taking closure is
/// supplied by whoever constructs a real, `StateEngine`-backed `Engine`
/// (`sv-sim-runtime`, once a later task wires a live design's ring depth in
/// -- see `state::snapshot`'s own module doc comment, deviation note) via
/// [`Engine::configure_preponed_ring`], never derived from `dyn
/// SignalAccess` itself.
struct PreponedRing {
    ring: SnapshotRing,
    source: Box<dyn Fn() -> ArenaSnapshot>,
}

/// A nonblocking write waiting out its intra-assignment delay
/// (`Engine::delayed_nbas`).
struct DelayedNba {
    cell: StateCellId,
    row: Option<u64>,
    value: FourStateValue,
    mask: Vec<u64>,
    window: Option<u32>,
    source: ProcessId,
    /// A delayed nonblocking event trigger (`->> #d e`): staged as a pulse.
    pulse: bool,
    /// A delayed collection-element write (`MaskedNbaUpdate::coll`).
    coll: Option<super::nba::CollWrite>,
    /// A delayed class-property write (`MaskedNbaUpdate::obj`).
    obj: Option<super::nba::ObjWrite>,
    /// Staged by the Re-NBA region's commit (a Reactive process or a clocking
    /// drive, LRM 14.16) rather than the NBA region's.
    reactive: bool,
}

/// A clocking drive waiting for an edge of its clock signal (`output negedge
/// #d q`, LRM 1800-2023 14.4): staged `after` ticks past the next `edge` of
/// `cell`.
struct EdgeNba {
    cell: StateCellId,
    edge: EdgeKind,
    after: u64,
    nba: DelayedNba,
}

impl DelayedNba {
    /// Stages this write into the NBA or Re-NBA log it belongs to.
    fn stage_into(self, engine: &Engine) {
        let mut nba = if self.reactive {
            &engine.re_nba
        } else {
            &engine.nba
        }
        .lock()
        .unwrap();
        if let Some(w) = self.obj {
            nba.stage_obj(w, self.source);
        } else if self.pulse {
            nba.stage_pulse(self.cell, self.row, self.source);
        } else if let Some(coll) = self.coll {
            nba.stage_coll(self.cell, self.row, self.mask, coll, self.source);
        } else {
            nba.stage_masked(
                self.cell,
                self.row,
                self.value,
                self.mask,
                self.window,
                self.source,
            );
        }
    }
}

impl EngineOps<'_> {
    /// Parks a delayed nonblocking write until `delay` ticks from now, or, for
    /// the [`EDGE_DELAY`] marker, until the edge the running clocking drive
    /// named (LRM 14.4 edge output skew).
    fn defer_nba(&self, delay: u64, d: DelayedNba) {
        if delay == super::kernel_adapter::EDGE_DELAY {
            let w = super::kernel_adapter::edge_drive()
                .expect("an edge-delayed write outside an edge-skewed clocking drive");
            // 14.16: a drive to the block's own event that executes in the
            // time step the event was triggered in is coincident with it.
            if w.on_event
                && self.engine.clocking_events.lock().unwrap().get(&w.cell) == Some(&self.now.ticks)
            {
                if w.after == 0 {
                    d.stage_into(self.engine);
                } else {
                    self.engine
                        .delayed_nbas
                        .lock()
                        .unwrap()
                        .entry(self.now.ticks + w.after)
                        .or_default()
                        .push(d);
                }
                return;
            }
            self.engine.edge_nbas.lock().unwrap().push(EdgeNba {
                cell: w.cell,
                edge: w.edge,
                after: w.after,
                nba: d,
            });
            self.engine.has_edge_nbas.store(true, Ordering::Relaxed);
            return;
        }
        self.engine
            .delayed_nbas
            .lock()
            .unwrap()
            .entry(self.now.ticks + delay)
            .or_default()
            .push(d);
    }
}

/// The first type in the crate that assembles every earlier task's pieces
/// together.
pub struct Engine {
    queues: Mutex<RegionQueues>,
    wheel: TimeWheel,
    nba: Mutex<NbaLog>,
    /// The Re-NBA region's log (LRM 4.4.2.6 / 4.5): nonblocking assignments
    /// staged by Reactive-region processes (program blocks, checkers), committed
    /// only once Re-Active has drained ([`Self::commit_re_nba_once`]). Empty and
    /// untouched for a design without them.
    re_nba: Mutex<NbaLog>,
    /// Whether any Reactive process is registered; gates the per-stage lookup
    /// that routes an NBA to `re_nba`.
    has_reactive: std::sync::atomic::AtomicBool,
    /// Intra-assignment-delayed nonblocking writes (`q <= #d v`), by the
    /// tick whose time step stages them into `nba` (in staging order).
    delayed_nbas: Mutex<std::collections::BTreeMap<SimTick, Vec<DelayedNba>>>,
    /// Edge-skewed clocking drives (`output negedge q`, LRM 14.4) waiting for
    /// their clock edge; empty (and `has_edge_nbas` false) without any.
    edge_nbas: Mutex<Vec<EdgeNba>>,
    /// The tick each clocking block's event counter (by cell) was last bumped
    /// at ([`SchedulerOps::note_clocking_event`]); empty without clocking blocks.
    clocking_events: Mutex<std::collections::HashMap<StateCellId, u64>>,
    has_edge_nbas: std::sync::atomic::AtomicBool,
    sensitivity: SensitivityTable,
    /// Per-pass "already queued" dedup (SC-055/SC-056): set when a
    /// sensitivity match enqueues a process, cleared when that activation is
    /// popped off its region queue.
    gate: QueuedGate,
    /// **NEW (Task 3, ledger SC-052).** Live decrementing counts for every
    /// `(ProcessId, term_index)` pair mid-`repeat(N)@(...)` wait — see
    /// `RepeatCounters`'s own doc comment (`sched/repeat.rs`).
    /// `EngineNotifySink::notify_change` consults it before ever enqueueing
    /// a `Repeat`-sourced match, and `dispatch_ready_set`'s dynamic-wait-
    /// resumed cleanup calls `RepeatCounters::clear` alongside
    /// `SensitivityTable::remove_dynamic`.
    repeats: RepeatCounters,
    table: ProcessTable,
    driver: ResumeDriver,
    finals: Vec<FinalProcess>,
    guard: DeltaGuard,
    /// `sched` borrows the `SignalAccess` *contract* but does not implement
    /// it — the real arena is STATE-owned and does not exist this wave.
    arena: Mutex<Box<dyn SignalAccess>>,
    io_sink: Box<dyn IoSink>,
    /// Persists for `Engine`'s whole lifetime — `EngineOps::defer_io`
    /// deliberately writes here directly rather than through a
    /// freshly-constructed `DeferIoDispatcher` per pass (see this module's
    /// doc comment, deviation 4).
    io_seq: AtomicU64,
    /// Ids for `PendingActivation`s this `Engine` mints (both
    /// `register_process`'s own initial-dispatch activation and every
    /// `EngineNotifySink`-minted wakeup).
    activation_seq: AtomicU64,
    /// `resume_synchronously`'s trampoline: `Some(queue)` while a drain is
    /// running (a nested request is queued, not recursed into, so a process
    /// that forks `join_none` in a zero-delay loop cannot overflow the stack).
    sync_queue: Option<std::collections::VecDeque<(ProcessId, ResumeArg)>>,
    /// Set when one trampoline drain exceeded `MAX_SYNC_RESUMES`: a process
    /// that never blocks (`always fork ... join_none join`), reported by
    /// `run_until` as `RunResult::Livelock`.
    sync_livelock: Option<String>,
    /// Fork branches spawned since time last advanced (`MAX_FORKS_PER_STEP`).
    forks_this_step: u32,
    /// `request_finish`'s target; checked once per timestep by `run_until`,
    /// never mid-pass (SC-075).
    finish_requested: Mutex<Option<i32>>,
    /// LRM 24.3: the program blocks' `initial` threads that have not ended
    /// (see [`Engine::mark_program_initial`]); `None` when the design has no
    /// program initial, so such designs pay nothing.
    program_roots: Option<std::collections::HashSet<ProcessId>>,
    controller: Box<dyn DeterminismController>,
    now: SchedTime,
    current_region: Region,
    /// Called once at the end of every timestep, after the Postponed region
    /// (waveform dumping samples signal values here -- `$dumpvars`).
    timestep_end_hook: Option<Box<dyn FnMut(SchedTime)>>,
    /// The embedder's scheduling hooks (`super::host`), if any.
    host: Option<Box<dyn super::host::Host>>,
    /// The tick of the last slot the host's slot-begin hook ran for.
    host_slot: Option<SimTick>,
    /// **NEW (Task 21, ledger SC-007) — a monotonically-nondecreasing
    /// high-water mark of [`DeltaGuard::count`], captured once per timestep
    /// (in `run_until`, just before `guard.reset()`), never per delta
    /// iteration.** See [`Engine::max_delta_iterations_in_a_timestep`]'s own
    /// doc comment for why this exists: SC-007's acceptance requires
    /// evidence that the Active/Inactive/NBA delta loop actually converges
    /// (a bounded pass count) for a real design, and `DeltaGuard::count`
    /// itself was private with no getter before this task, so nothing
    /// outside this crate's own `#[cfg(test)]` unit tests could observe that
    /// bound at all — `region_order_fixture.rs` (a separate compilation
    /// unit under `tests/`) needs exactly this.
    max_delta_iterations_seen: AtomicU32,
    /// **NEW (Task 4, ledger SC-090/SC-123/SC-020).** Every not-yet-fired
    /// `register_region_callback` registration, as `(fire_at, region, cb)`,
    /// in registration order (a plain `Vec`, not a priority structure — per-
    /// timestep callback volume is small, and `region` already partitions
    /// `Reactive` from `Postponed`, the same rationale `RegionQueues` itself
    /// uses). [`Self::drain_region_callbacks`] removes+fires every entry
    /// whose `region` matches and whose `fire_at <= self.now`, in the order
    /// they appear here (registration order) — never resorted.
    region_callbacks: Mutex<Vec<RegionCallbackEntry>>,
    /// **NEW (Task 9, B-sched wave, ledger SC-125).** `ProcessId`s
    /// `EngineOps::disable_fork` queued mid-`resume()` -- `SchedulerOps`
    /// methods only ever get `&self`, so the actual recursive
    /// `ProcessTable` mutation is deferred to
    /// [`Self::drain_pending_disable_fork`], called at the top of
    /// [`Self::apply_suspend`] (see that method's own doc comment).
    pending_disable_fork: Mutex<Vec<ProcessId>>,
    /// The processes `SchedulerOps::spawn_detached` queued (16.11): registered,
    /// to start in Reactive, by [`Self::drain_pending_spawns`].
    pending_spawns: Mutex<Vec<Box<dyn CompiledProcess>>>,
    /// `disable f;` requests from a branch of fork `f` (see
    /// `SchedulerOps::disable_fork_block`), drained with
    /// `pending_disable_fork`.
    pending_disable_block: Mutex<Vec<ProcessId>>,
    /// `SchedulerOps::disable_named` requests: `(requester, targets)`, where
    /// `targets` is the already-resolved (at queue time, synchronously --
    /// see that method's own doc comment) list of live `ProcessId`s a
    /// cross-process/fork-sibling `disable <label>;` reached. Drained by
    /// [`Self::drain_pending_disable_named`], the same `&self`-queue /
    /// `&mut self`-drain split every other `pending_disable_*` field uses.
    pending_disable_named: Mutex<Vec<(ProcessId, Vec<ProcessId>)>>,
    /// `SchedulerOps::disable_named` requests for the block or task activation
    /// `label` of the requester's instance (LRM 9.6.2): `(requester, label)`,
    /// applied by [`Self::drain_pending_disable_scope`].
    pending_disable_scope: Mutex<Vec<(ProcessId, String)>>,
    /// For a process a `disable` redirected (`Self::redirect_process`): the
    /// id of that redirect's activation. Activations of the process queued
    /// before it (a delay's timer, an event wake) are stale and dropped at
    /// dispatch.
    activation_floor: rustc_hash::FxHashMap<ProcessId, u64>,
    /// For a process a dispatched `EventFired` activation resumed: the next
    /// activation id at that moment. Another `EventFired` of the process with
    /// a smaller id was queued while the wait that activation ended was still
    /// installed -- a write by an earlier item of the same pass to a second
    /// signal the wait names (`@(a[j])` also waits on `j`), after the dequeue
    /// cleared the dedup flag -- so it is the same trigger (LRM 9.4.2: the
    /// process was already scheduled) and is dropped at dispatch; delivered, it
    /// would resume the process's NEXT wait early or hit a finished process.
    /// A wake with an id at or above the floor was queued for a later wait.
    /// The dispatch also clears the process's dedup flag (see the call site):
    /// the dropped wake held it set, which would otherwise swallow the first
    /// real wake of the next wait.
    event_floor: rustc_hash::FxHashMap<ProcessId, u64>,
    /// Processes `disable fork;` terminated -- their still-queued
    /// activations are dropped at dispatch instead of treated as unknown ids.
    killed: std::collections::HashSet<ProcessId>,
    /// Classes plan Task 13 (LRM 9.7): `process` class requests queued by
    /// `EngineOps::process_request` (`(requester, request)`, in call order),
    /// applied by [`Self::drain_pending_process_requests`] at the top of
    /// [`Self::apply_suspend`] -- the same `&self`-queue / `&mut self`-drain
    /// split `pending_disable_fork` uses.
    pending_process_requests: Mutex<Vec<(ProcessId, ProcessRequest)>>,
    /// Processes a `process` handle exists for (`SchedulerOps::process_track`).
    /// Only these get an [`Self::ended`] entry when they end.
    tracked: Mutex<std::collections::HashSet<ProcessId>>,
    /// Set by the first `process_track`; gates the per-suspend `waiting`
    /// bookkeeping so a program that never uses `process` pays nothing.
    any_tracked: AtomicBool,
    /// How a tracked process ended. Fork children are reclaimed from the
    /// table the moment they finish, so this map is the only record that
    /// answers `status()` and `await()` for them afterwards.
    ended: std::collections::HashMap<ProcessId, ProcessState>,
    /// `awaiters[t]`: processes blocked in `t.await()`, woken when `t` ends.
    awaiters: std::collections::HashMap<ProcessId, Vec<ProcessId>>,
    /// The active `$monitor`'s argument cells and its changed-since-last-
    /// print flag (see `SchedulerOps::set_monitor_watch`).
    monitor_watch: Mutex<std::collections::HashSet<StateCellId>>,
    /// Mirrors `!monitor_watch.is_empty()` so `notify_change` skips the lock.
    any_monitor_watch: AtomicBool,
    monitor_changed: AtomicBool,
    /// **NEW (sva T1, LRM 16.5.1/16.5.2).** `None` for a design with no
    /// assertion/clocking-block/`$past` site at all (the "pays nothing"
    /// guard; see [`PreponedRing`]'s own doc comment and
    /// `state::snapshot::SnapshotRing`'s). Pushed to once per time step by
    /// [`Self::push_preponed_snapshot`].
    preponed: Mutex<Option<PreponedRing>>,
    /// Classes plan Task 14: the instance scope [`Self::register_process`]
    /// seeds a new process from (see [`Engine::set_registration_scope`]).
    registration_scope: Option<u32>,
    /// Whether the process being registered under `registration_scope` runs a
    /// static declaration initializer rather than being a thread (see
    /// [`Engine::set_registration_static_initializer`]).
    registration_static_initializer: bool,
    /// **Test-only spy (Task 9, A2-part2, CG-102).** Records every
    /// `(ProcessId, region_id)` pair `dispatch_ready_set` actually looked up
    /// via `ProcessTable::region_id` during one dispatch pass --
    /// `dispatch_ready_set_looks_up_partition_of_using_each_readied_
    /// processs_own_stored_region_id`'s own seam, proving the field flows
    /// from registration through to dispatch rather than being
    /// stored-but-never-read. Only exists in `cfg(test)` builds (this
    /// crate's own unit tests, which have `&mut Engine`/private-field
    /// access an external `tests/` crate does not) -- never part of this
    /// type's real, non-test shape.
    #[cfg(test)]
    region_id_queries_seen: Vec<(ProcessId, u32)>,
}

/// The concrete `SchedulerOps` implementation, constructed once per
/// region-pass (see this module's doc comment / `dispatch_ready_set`). `now`/
/// `region`/`reconv_iter` are genuinely constant for every `WorkItem` one
/// region-pass dispatches.
struct EngineOps<'e> {
    engine: &'e Engine,
    now: SchedTime,
    region: Region,
    reconv_iter: u32,
}

impl SchedulerOps for EngineOps<'_> {
    fn now(&self) -> SchedTime {
        self.now
    }

    fn pending_ticks(&self) -> Vec<u64> {
        let mut ticks = self.engine.wheel.pending_ticks();
        ticks.extend(self.engine.delayed_nbas.lock().unwrap().keys().copied());
        ticks.into_iter().filter(|&t| t > self.now.ticks).collect()
    }

    fn waiting_processes(&self, cell: StateCellId) -> Vec<ProcessId> {
        self.engine.sensitivity.waiting_on(cell)
    }

    fn current_region(&self) -> Region {
        self.region
    }

    fn note_reactive_drive(&self) {
        self.engine.has_reactive.store(true, Ordering::Relaxed);
    }

    fn note_clocking_event(&self, counter: StateCellId) {
        self.engine
            .clocking_events
            .lock()
            .unwrap()
            .insert(counter, self.now.ticks);
    }

    fn stage_nba(&self, cell: StateCellId, value: FourStateValue, source: ProcessId) {
        self.engine
            .nba_for(source)
            .lock()
            .unwrap()
            .stage(cell, value, source);
    }

    fn stage_nba_masked(
        &self,
        cell: StateCellId,
        row: Option<u64>,
        value: FourStateValue,
        mask: Vec<u64>,
        source: ProcessId,
    ) {
        self.engine
            .nba_for(source)
            .lock()
            .unwrap()
            .stage_masked(cell, row, value, mask, None, source);
    }

    fn stage_nba_window(
        &self,
        cell: StateCellId,
        window: u32,
        value: FourStateValue,
        mask: Vec<u64>,
        source: ProcessId,
        delay: u64,
    ) {
        if delay == 0 {
            self.engine.nba_for(source).lock().unwrap().stage_masked(
                cell,
                None,
                value,
                mask,
                Some(window),
                source,
            );
            return;
        }
        self.defer_nba(
            delay,
            DelayedNba {
                cell,
                row: None,
                value,
                mask,
                window: Some(window),
                source,
                pulse: false,
                coll: None,
                obj: None,
                reactive: self.engine.stages_reactive(source),
            },
        );
    }

    fn stage_nba_masked_delayed(
        &self,
        cell: StateCellId,
        row: Option<u64>,
        value: FourStateValue,
        mask: Vec<u64>,
        source: ProcessId,
        delay: u64,
    ) {
        if delay == 0 {
            return self.stage_nba_masked(cell, row, value, mask, source);
        }
        self.defer_nba(
            delay,
            DelayedNba {
                cell,
                row,
                value,
                mask,
                window: None,
                source,
                pulse: false,
                coll: None,
                obj: None,
                reactive: self.engine.stages_reactive(source),
            },
        );
    }

    fn stage_nba_coll(
        &self,
        cell: StateCellId,
        row: Option<u64>,
        elem: Vec<u64>,
        coll: super::nba::CollWrite,
        source: ProcessId,
        delay: u64,
    ) {
        if delay == 0 {
            self.engine
                .nba_for(source)
                .lock()
                .unwrap()
                .stage_coll(cell, row, elem, coll, source);
            return;
        }
        self.defer_nba(
            delay,
            DelayedNba {
                cell,
                row,
                value: FourStateValue::Narrow(0),
                mask: elem,
                window: None,
                source,
                pulse: false,
                coll: Some(coll),
                obj: None,
                reactive: self.engine.stages_reactive(source),
            },
        );
    }

    fn stage_nba_obj(&self, w: super::nba::ObjWrite, source: ProcessId, delay: u64) {
        if delay == 0 {
            self.engine
                .nba_for(source)
                .lock()
                .unwrap()
                .stage_obj(w, source);
            return;
        }
        self.defer_nba(
            delay,
            DelayedNba {
                cell: StateCellId(0),
                row: None,
                value: FourStateValue::Narrow(0),
                mask: Vec::new(),
                window: None,
                source,
                pulse: false,
                coll: None,
                obj: Some(w),
                reactive: self.engine.stages_reactive(source),
            },
        );
    }

    fn stage_nba_pulse(&self, cell: StateCellId, row: Option<u64>, source: ProcessId, delay: u64) {
        if delay == 0 {
            self.engine
                .nba_for(source)
                .lock()
                .unwrap()
                .stage_pulse(cell, row, source);
            return;
        }
        self.defer_nba(
            delay,
            DelayedNba {
                cell,
                row,
                value: FourStateValue::Narrow(0),
                mask: Vec::new(),
                window: None,
                source,
                pulse: true,
                coll: None,
                obj: None,
                reactive: self.engine.stages_reactive(source),
            },
        );
    }

    fn commit_blocking_write(&self, cell: StateCellId, value: FourStateValue) {
        let mut arena = self.engine.arena.lock().unwrap();
        let mut sink = EngineNotifySink {
            engine: self.engine,
            at: self.now,
            reconv_iter: self.reconv_iter,
            delta: None,
        };
        let _ = crate::sched::commit_blocking_write(cell, value, &mut **arena, self.now, &mut sink);
    }

    fn notify_blocking_row_change(&self, cell: StateCellId) {
        let mut sink = EngineNotifySink {
            engine: self.engine,
            at: self.now,
            reconv_iter: self.reconv_iter,
            delta: None,
        };
        crate::sched::notify_blocking_change(cell, EdgeKind::None, self.now, &mut sink);
        let mut arena = self.engine.arena.lock().unwrap();
        crate::sched::notify_alias_changes(&mut **arena, self.now, &mut sink);
    }

    fn notify_blocking_wide_change(
        &self,
        cell: StateCellId,
        edge: EdgeKind,
        delta: Option<ChangeDelta>,
    ) {
        let mut sink = EngineNotifySink {
            engine: self.engine,
            at: self.now,
            reconv_iter: self.reconv_iter,
            delta,
        };
        crate::sched::notify_blocking_change(cell, edge, self.now, &mut sink);
    }

    fn wants_delta(&self, cell: StateCellId) -> bool {
        self.engine.sensitivity.has_select(cell)
    }

    fn request_finish(&self, code: i32) {
        // See this module's doc comment, deviation 4.
        *self.engine.finish_requested.lock().unwrap() = Some(code);
    }

    fn defer_io(&self, at: SchedTime, region: Region, cb: DeferredIoCallback) {
        // Textually identical to `DeferIoDispatcher::defer_io`'s body,
        // deliberately not reused through one (see this module's doc
        // comment / the plan's own text): a fresh `DeferIoDispatcher`
        // reconstructed per pass would reset `seq` to 0 every pass.
        let seq = self.engine.io_seq.fetch_add(1, Ordering::SeqCst);
        self.engine.io_sink.enqueue(at, region, seq, cb);
    }

    /// **NEW (Task 4, ledger SC-090/SC-123).** The one real production
    /// implementation — forwards straight to `Engine::push_region_callback`
    /// (see that method's own doc comment for why it is a plain, always-
    /// available `Engine` method rather than `#[cfg(test)]`-gated: this impl
    /// is its one production call site, and this module's own test setup
    /// code calls it directly too, for registrations made before `run_until`
    /// starts rather than from inside a dispatched process's `resume()`).
    ///
    /// **Review fix (Task 4 r1):** the `SchedulerOps::register_region_callback`
    /// trait signature is a cross-track contract (many `#[cfg(test)]` mock
    /// implementations outside this crate, per that trait method's own doc
    /// comment) and returns `()`, so it cannot forward
    /// `push_region_callback`'s new typed `Result` to its caller. `.expect`
    /// turns `RegionCallbackRegionError` into an immediate, loud panic here
    /// instead — never a silent, forever-unfired registration — at this
    /// trait's one production implementation.
    fn register_region_callback(&self, region: Region, at: SchedTime, cb: Box<dyn RegionCallback>) {
        self.engine.push_region_callback(at, region, cb).expect(
            "SchedulerOps::register_region_callback: region must be Reactive, Postponed, or \
                 Observed -- Active/Inactive/Nba/Preponed are never drained by \
                 drain_region_callbacks and a callback registered for one would never fire",
        );
    }

    /// **NEW (Task 9, ledger SC-125).** The one real override of
    /// `SchedulerOps::disable_fork`'s default no-op (see that trait
    /// method's own doc comment for why it is a default, not a required
    /// method, this wave). Only queues `proc` into `Engine::
    /// pending_disable_fork` -- this call only ever has `&self` (`EngineOps`
    /// holds a shared `&'e Engine`), so the real recursive `ProcessTable`
    /// mutation happens in `Engine::drain_pending_disable_fork`, called at
    /// the top of `Engine::apply_suspend`, which always runs with `&mut
    /// self` immediately after the `resume()` call (this one included)
    /// that might have queued it.
    fn disable_fork(&self, proc: ProcessId) {
        self.engine.pending_disable_fork.lock().unwrap().push(proc);
    }

    fn disable_fork_block(&self, proc: ProcessId) {
        self.engine.pending_disable_block.lock().unwrap().push(proc);
    }

    fn spawn_detached(&self, proc: Box<dyn CompiledProcess>) {
        self.engine.pending_spawns.lock().unwrap().push(proc);
    }

    /// `SchedulerOps::disable_named`'s one real override: resolves the
    /// target set and the `requester`-ends-too question synchronously (a
    /// read of the still-untouched `ProcessTable`, exactly like
    /// `process_request`'s own `ProcessRequest::Kill` arm), then queues the
    /// resolved targets for `Engine::drain_pending_disable_named` to kill.
    fn disable_named(&self, requester: ProcessId, label: &str) -> bool {
        let root = self.engine.fork_root(requester);
        let mut targets = Vec::new();
        if label == EXIT_PROGRAM_LABEL {
            // `$exit` (LRM 24.7) is only lowered inside a program, so the
            // caller is itself a program thread and ends; the other threads
            // of its program are found in `drain_pending_disable_scope` (the
            // caller's compiled process is locked while it runs).
            self.engine
                .pending_disable_scope
                .lock()
                .unwrap()
                .push((requester, label.to_string()));
            return true;
        }
        self.engine.collect_disable_named(root, label, &mut targets);
        let ends_caller = targets
            .iter()
            .any(|&t| t == requester || self.engine.is_in_fork_subtree(requester, t));
        self.engine
            .pending_disable_named
            .lock()
            .unwrap()
            .push((requester, targets));
        // ... and the processes of the instance suspended inside the block or
        // task `label` (LRM 9.6.2) resume after it.
        self.engine
            .pending_disable_scope
            .lock()
            .unwrap()
            .push((requester, label.to_string()));
        ends_caller
    }

    fn wake_process(&self, proc: ProcessId, term: usize) {
        let act = PendingActivation {
            id: self.engine.activation_seq.fetch_add(1, Ordering::SeqCst),
            proc,
            arg: ResumeArg::EventFired { term_index: term },
            reconv_iter: 0,
            logical_order: (self.engine.now, Region::Active, proc as u64),
        };
        self.engine
            .queues
            .lock()
            .unwrap()
            .push(Region::Active, act)
            .expect("Region::Active is never rejected by RegionQueues::push");
    }

    fn set_monitor_watch(&self, cells: Vec<StateCellId>) {
        let mut watch = self.engine.monitor_watch.lock().unwrap();
        *watch = cells.into_iter().collect();
        self.engine
            .any_monitor_watch
            .store(!watch.is_empty(), Ordering::Relaxed);
        drop(watch);
        self.engine.monitor_changed.store(true, Ordering::SeqCst);
    }

    /// Classes plan Task 13: queues `req` (applied at the requester's next
    /// `apply_suspend`) and reports whether it ends the requester itself.
    fn process_request(&self, requester: ProcessId, req: ProcessRequest) -> bool {
        let kills_requester = match req {
            ProcessRequest::Kill(target) => self.engine.is_in_fork_subtree(requester, target),
            ProcessRequest::Suspend(_)
            | ProcessRequest::Resume(_)
            | ProcessRequest::Await(_)
            | ProcessRequest::SyncBlock
            | ProcessRequest::SyncWake(_) => false,
        };
        self.engine
            .pending_process_requests
            .lock()
            .unwrap()
            .push((requester, req));
        kills_requester
    }

    fn process_state(&self, target: ProcessId) -> ProcessState {
        self.engine.effective_process_state(target)
    }

    fn process_track(&self, proc: ProcessId) {
        self.engine.tracked.lock().unwrap().insert(proc);
        self.engine.any_tracked.store(true, Ordering::SeqCst);
    }
}

/// The concrete [`SchedulerCtx`] implementation
/// [`Engine::drain_region_callbacks`] constructs to invoke each due
/// [`RegionCallback`] — read-only, backed by `Engine`'s own arena `Mutex`
/// (mirrors [`EngineNotifySink`]'s identical shape/rationale).
struct EngineSchedulerCtx<'e> {
    engine: &'e Engine,
    now: SchedTime,
}

impl SchedulerCtx for EngineSchedulerCtx<'_> {
    fn now(&self) -> SchedTime {
        self.now
    }

    fn read(&self, cell: StateCellId) -> FourStateValue {
        self.engine.arena.lock().unwrap().read(cell)
    }

    fn rearm(&self, region: Region, cb: Box<dyn RegionCallback>) {
        self.engine
            .push_region_callback(self.now, region, cb)
            .expect("RegionCallback::invoke re-armed for a region Engine never drains");
    }

    fn take_monitor_changed(&self) -> bool {
        self.engine.monitor_changed.swap(false, Ordering::SeqCst)
    }
}

/// The concrete `ValueChangeSink` implementation `EngineOps::
/// commit_blocking_write` and `Engine::commit_nba_once` construct to receive
/// `notify_change` calls during a commit.
struct EngineNotifySink<'e> {
    engine: &'e Engine,
    at: SchedTime,
    reconv_iter: u32,
    /// The old/new words of the update the next `notify_change` reports, when
    /// a select waiter needs them (`ValueChangeSink::note_delta`).
    delta: Option<ChangeDelta>,
}

impl ValueChangeSink for EngineNotifySink<'_> {
    fn wants_delta(&self, cell: StateCellId) -> bool {
        self.engine.sensitivity.has_select(cell)
    }

    fn note_delta(&mut self, _cell: StateCellId, delta: ChangeDelta) {
        self.delta = Some(delta);
    }

    /// **Task 3 (ledger SC-052) rewiring:** `SensitivityTable::matches` now
    /// yields `(ProcessId, term_index, repeat_total, iff)`, not just
    /// `ProcessId` -- this is the one production call site that needs the
    /// extra fields (SC-055's dedup gate does not, so this no longer
    /// delegates to `dedup::sensitivity_match_to_enqueue`, which only ever
    /// returns a bare `Vec<ProcessId>`; `SensitivityTable::matches` itself
    /// is still Task 9's one canonical matching implementation underneath --
    /// SC-042's "no duplicate implementation" is unaffected).
    ///
    /// **Task 5 (ledger SC-068) addition:** a match whose row carries
    /// `iff: Some(pred)` is filtered FIRST, before repeat-counting -- `pred`
    /// is invoked (via the same `SimHandle(null)` placeholder every other
    /// `CompiledProcess::resume()`/`RegionCallback` call site in this module
    /// already uses, since this crate holds no real ABI-facing state
    /// pointer of its own) and a `false` result drops the firing entirely,
    /// exactly as if the edge/cell condition had never matched at all: no
    /// `RepeatCounters::on_term_fired` call (an `iff`-false edge must not
    /// count toward a `repeat(N)@(... iff ...)`'s count either), no
    /// dedup-gate mark, no enqueue, and the sensitivity entry stays
    /// installed. This is `PredicateFn`'s settled, binding calling
    /// convention (`docs/CONTRACTS.md` D8) -- a bare
    /// `extern "C" fn(SimHandle) -> bool` produced per-callsite by codegen's
    /// `compile_iff_guard`, which reads whatever cell(s) it needs itself
    /// through the handle; this call site never passes a `StateCellId`
    /// alongside it.
    ///
    /// For each `iff`-passing (or `iff`-less) match whose row carries
    /// `repeat_total: Some(n)` (an `EventTerm::Repeat`-sourced row), the
    /// firing is routed through `RepeatCounters::on_term_fired` next: on
    /// `StillWaiting` this firing is swallowed -- no dedup-gate mark, no
    /// enqueue, and the sensitivity entry stays installed (nothing here or
    /// on any other path removes it for a non-terminal firing). Only a
    /// `repeat_total: None` row, or a `Some` row that just reported
    /// `Satisfied`, proceeds to the existing SC-055 dedup-gate-then-enqueue
    /// step.
    fn notify_change(
        &mut self,
        cell: StateCellId,
        edge: EdgeKind,
        _at: sv_sim_abi::SimTime,
        _region: Region,
    ) {
        // Lock-free fast path: the watch set is empty unless a `$monitor` is
        // active, which is the common case for the ~10^7 calls of a long run.
        if self.engine.any_monitor_watch.load(Ordering::Relaxed)
            && self.engine.monitor_watch.lock().unwrap().contains(&cell)
        {
            self.engine.monitor_changed.store(true, Ordering::SeqCst);
            // Under LRM 21.2.3's expression-value test, re-evaluate the
            // monitor's arguments at this update (a glitch that restores the
            // value within the step still owes a display).
            crate::sched::kernel_adapter::monitor_probe();
        }
        if self.engine.has_edge_nbas.load(Ordering::Relaxed) {
            self.engine.fire_edge_nbas(cell, edge);
        }
        let handle = SimHandle(std::ptr::null_mut());
        // The process making this write is executing, not waiting: it is
        // never woken by its own change (LRM 1800-2023 4.7, 9.2.2.2: e.g. an
        // `always_comb` writing an array element it also reads must not
        // re-trigger itself through its static sensitivity).
        let writer = crate::sched::current_process();
        // Each woken process with the index of the wait term that woke it
        // (`wait_order` reads it, `sv_rt_fired_term`).
        let delta = self.delta.take();
        let matched: smallvec::SmallVec<[(ProcessId, usize); 8]> = self
            .engine
            .sensitivity
            .matches_delta(cell, edge, delta.as_ref())
            .filter(|(proc, _, _, _)| Some(*proc) != writer)
            .filter(|(_, _, _, iff)| match iff {
                Some(pred) => (pred.0)(handle),
                None => true,
            })
            // Every matched term is recorded for the resume guard of an event
            // list (`sv_rt_fired_mask`), including those of a process this
            // pass finds already queued: several members can fire in one wake
            // (LRM 9.4.2.3) and the guard must see them all.
            .inspect(|(proc, term_index, _, _)| {
                crate::sched::kernel_adapter::note_fired_term(*proc, *term_index);
            })
            .filter_map(
                |(proc, term_index, repeat_total, _iff)| match repeat_total {
                    Some(declared_total) => {
                        match self
                            .engine
                            .repeats
                            .on_term_fired(proc, term_index, declared_total)
                        {
                            RepeatOutcome::StillWaiting => None,
                            RepeatOutcome::Satisfied => Some((proc, term_index)),
                        }
                    }
                    None => Some((proc, term_index)),
                },
            )
            // SC-055: only processes not already queued this pass come back.
            .filter(|&(proc, _)| self.engine.gate.mark_queued_if_not_already(proc))
            .collect();
        if matched.is_empty() {
            return;
        }
        let mut queues = self.engine.queues.lock().unwrap();
        for (proc, term_index) in matched {
            let id = self.engine.activation_seq.fetch_add(1, Ordering::SeqCst);
            // LRM 17.7.3: a checker's processes are sensitive "in the Reactive
            // region" -- their wake-ups wait for Active/Inactive/NBA (and the
            // Observed region) to settle, so they see only settled values.
            let home = if self.engine.table.is_reactive(proc) {
                Region::Reactive
            } else {
                Region::Active
            };
            let act = PendingActivation {
                id,
                proc,
                arg: ResumeArg::EventFired { term_index },
                reconv_iter: self.reconv_iter,
                logical_order: (self.at, home, proc as u64),
            };
            // Active and Reactive are regions `RegionQueues::push` never
            // rejects (`RegionQueuePushError::NbaHasNoQueue` is the only
            // variant that can fire for them, and only for `Region::Nba`).
            queues.push(home, act).expect(
                "Region::Active and Region::Reactive are never rejected by RegionQueues::push",
            );
        }
    }
}

impl Engine {
    /// `sched` borrows the `SignalAccess`/`IoSink` *contracts* but does not
    /// implement them this wave — dependency-injecting both is what lets
    /// `Engine` be genuinely constructible and testable without this crate
    /// depending on a STATE/RUNTIME implementation. `controller` starts as
    /// `SequentialController::default_boxed()` (Task 3), overridable via
    /// `set_determinism_controller`.
    pub fn new(arena: Box<dyn SignalAccess>, io: Box<dyn IoSink>) -> Self {
        Engine {
            queues: Mutex::new(RegionQueues::new()),
            wheel: TimeWheel::new(),
            nba: Mutex::new(NbaLog::new()),
            re_nba: Mutex::new(NbaLog::new()),
            has_reactive: std::sync::atomic::AtomicBool::new(false),
            delayed_nbas: Mutex::new(std::collections::BTreeMap::new()),
            edge_nbas: Mutex::new(Vec::new()),
            clocking_events: Mutex::new(std::collections::HashMap::new()),
            has_edge_nbas: std::sync::atomic::AtomicBool::new(false),
            sensitivity: SensitivityTable::new(),
            gate: QueuedGate::new(),
            repeats: RepeatCounters::new(),
            table: ProcessTable::new(),
            driver: ResumeDriver,
            finals: Vec::new(),
            guard: DeltaGuard::new(DEFAULT_MAX_DELTA_ITERATIONS),
            arena: Mutex::new(arena),
            io_sink: io,
            io_seq: AtomicU64::new(0),
            activation_seq: AtomicU64::new(0),
            sync_queue: None,
            sync_livelock: None,
            forks_this_step: 0,
            finish_requested: Mutex::new(None),
            program_roots: None,
            controller: SequentialController::default_boxed(),
            now: SchedTime { ticks: 0 },
            current_region: Region::Active,
            timestep_end_hook: None,
            host: None,
            host_slot: None,
            max_delta_iterations_seen: AtomicU32::new(0),
            region_callbacks: Mutex::new(Vec::new()),
            pending_disable_fork: Mutex::new(Vec::new()),
            pending_spawns: Mutex::new(Vec::new()),
            pending_disable_block: Mutex::new(Vec::new()),
            pending_disable_named: Mutex::new(Vec::new()),
            pending_disable_scope: Mutex::new(Vec::new()),
            activation_floor: rustc_hash::FxHashMap::default(),
            event_floor: rustc_hash::FxHashMap::default(),
            killed: std::collections::HashSet::new(),
            pending_process_requests: Mutex::new(Vec::new()),
            tracked: Mutex::new(std::collections::HashSet::new()),
            any_tracked: AtomicBool::new(false),
            ended: std::collections::HashMap::new(),
            awaiters: std::collections::HashMap::new(),
            monitor_watch: Mutex::new(std::collections::HashSet::new()),
            any_monitor_watch: AtomicBool::new(false),
            monitor_changed: AtomicBool::new(false),
            preponed: Mutex::new(None),
            registration_scope: None,
            registration_static_initializer: false,
            #[cfg(test)]
            region_id_queries_seen: Vec::new(),
        }
    }

    /// Classes plan Task 14 (LRM 18.14.1): names the instance scope whose
    /// initialization RNG seeds every process registered from now on
    /// (`None`, the default, seeds nothing: the process draws from the
    /// unseeded default like any process with no known parent). The
    /// production caller sets it around the user's procedural blocks only,
    /// so the infrastructure processes registered elsewhere (net resolvers,
    /// samplers, ...) never advance an instance's initialization RNG and so
    /// never shift the seed any user process gets.
    pub fn set_registration_scope(&mut self, scope: Option<u32>) {
        self.registration_scope = scope;
    }

    /// While set (and a [`Self::set_registration_scope`] scope is active), a
    /// process registered runs a static declaration initializer (`int x =
    /// f();`, `C h = new;`) rather than being a thread of its own: LRM 18.14.1
    /// gives such code no thread, so it draws from the instance's
    /// initialization RNG directly ([`process_rng::alias_init`]) instead of
    /// being seeded from it.
    pub fn set_registration_static_initializer(&mut self, on: bool) {
        self.registration_static_initializer = on;
    }

    /// Installs the end-of-timestep hook (see the field's doc comment),
    /// replacing any previous one.
    pub fn set_timestep_end_hook(&mut self, hook: Box<dyn FnMut(SchedTime)>) {
        self.timestep_end_hook = Some(hook);
    }

    /// **NEW (sva T1, LRM 16.5.1/16.5.2).** Installs a Preponed-region
    /// snapshot ring of the given `depth` (`state::snapshot::SnapshotRing`'s
    /// own doc comment), sourced by calling `source` once per time step.
    /// Replaces any previously configured ring. Never called at all for a
    /// design with no assertion/clocking-block/`$past` site -- [`Self::
    /// push_preponed_snapshot`] stays a no-op (no allocation, `source`
    /// never invoked) until this is.
    pub fn configure_preponed_ring(
        &mut self,
        depth: usize,
        source: impl Fn() -> ArenaSnapshot + 'static,
    ) {
        *self.preponed.lock().unwrap() = Some(PreponedRing {
            ring: SnapshotRing::new(depth),
            source: Box::new(source),
        });
    }

    /// Pushes this time step's Preponed snapshot (16.5.1) by invoking the
    /// closure [`Self::configure_preponed_ring`] installed -- a no-op when
    /// no ring is configured at all (see [`Self::preponed`]'s own doc
    /// comment). Called exactly once for every genuinely new nonempty time
    /// slot the wheel advances into -- never for the engine's initial
    /// `Region::Active` entry at time 0, which neither driver API precedes
    /// with a push (`Engine::new` sets `current_region = Active` directly).
    /// `run_until`'s outer loop calls this right after `advance_time_to`,
    /// in the `Some(t)` arm of its own `next_event_tick()` match (sva T1
    /// §1.4: "right where it currently detects a new nonempty time slot");
    /// the single-stepping API folds the identical call into
    /// [`Self::step_region`]'s `Postponed -> Active` transition, on the same
    /// `next_event_tick() == Some` branch (see that method's own doc
    /// comment) -- the two call sites are kept in lock-step so a caller
    /// driving a given `Engine` through either API (never both interleaved,
    /// the same assumption [`Self::current_region`]'s own doc comment
    /// already makes for the single-stepper) ends up with the same Preponed
    /// ring for the same simulated history. See
    /// `run_until_and_step_region_push_the_same_preponed_ring` for the
    /// regression proof.
    fn push_preponed_snapshot(&self) {
        let mut guard = self.preponed.lock().unwrap();
        if let Some(p) = guard.as_mut() {
            let snap = (p.source)();
            p.ring.push(snap);
        }
    }

    /// Test-only read access into the configured Preponed ring (see
    /// [`Self::configure_preponed_ring`]) -- `f` receives `None` when no
    /// ring is configured at all, or when `ticks_back` is out of the
    /// configured ring's own retained range (`SnapshotRing::get`'s own doc
    /// comment). Takes a closure rather than returning `Option<&ArenaSnapshot>`
    /// directly since the `Mutex` guard cannot outlive this call.
    #[cfg(test)]
    fn with_preponed_entry<R>(
        &self,
        ticks_back: usize,
        f: impl FnOnce(Option<&ArenaSnapshot>) -> R,
    ) -> R {
        let guard = self.preponed.lock().unwrap();
        f(guard.as_ref().and_then(|p| p.ring.get(ticks_back)))
    }

    /// Test-only accessor for [`Self::region_id_queries_seen`] -- see that
    /// field's own doc comment.
    #[cfg(test)]
    fn region_id_queries_seen(&self) -> &[(ProcessId, u32)] {
        &self.region_id_queries_seen
    }

    /// **NEW (Task 21, ledger SC-007).** The maximum [`DeltaGuard::count`]
    /// this `Engine` has observed at the settled end of any one timestep's
    /// inner Active/Inactive/NBA convergence loop, across its whole run so
    /// far — a high-water mark, not the current timestep's own count (which
    /// is always 0 immediately after `run_until` returns, since `guard.
    /// reset()` runs every timestep). Existence-proof surface for SC-007's
    /// "region-stratified delta-cycle convergence" acceptance: a caller can
    /// assert this stayed low for a design that should settle in a handful
    /// of passes, catching an accidental unbounded-oscillation regression
    /// well before it would ever hit `DEFAULT_MAX_DELTA_ITERATIONS`.
    pub fn max_delta_iterations_in_a_timestep(&self) -> u32 {
        self.max_delta_iterations_seen.load(Ordering::SeqCst)
    }

    /// Testability getter (DT-007's "selected by default when no other
    /// controller is configured", made externally observable): see
    /// `DeterminismController::kind`.
    pub fn controller_kind(&self) -> &'static str {
        self.controller.kind()
    }

    /// **NEW (A2-part2 Task 5, ledger CG-162):** exposes the region
    /// [`Self::step_region`] is about to process next -- needed so the ABI's
    /// `sim_eval_region(requested_region, ..)` (`sv-sim-runtime::
    /// abi_entry`) can single-step exactly up to (but never past)
    /// `requested_region`, and, for `Region::Nba` specifically, stop there
    /// WITHOUT calling `step_region`/[`Self::advance_nba_region`] at all:
    /// RHS evaluation for any pending NBA writes already happened during
    /// the prior Active/Inactive dispatch passes (`nba_enqueue` only ever
    /// stages into `NbaLog`, never touches the arena -- see
    /// `EngineOps::stage_nba`), so observing "we have arrived at the Nba
    /// boundary" is genuinely a read-only query, and
    /// [`Self::advance_nba_region`] stays the one, separate, explicit call
    /// that applies the staged log (CG-162's "never collapsed into one
    /// call").
    pub fn current_region(&self) -> Region {
        self.current_region
    }

    /// Drains `region`'s queue into a canonically-sorted ready-set (Task 2's
    /// `sort_ready_set`, applied here since `form_ready_set`'s own contract
    /// is purely "collect what's ready").
    fn drain_and_sort(&self, region: Region) -> Vec<PendingActivation> {
        let mut ready = {
            let mut queues = self.queues.lock().unwrap();
            form_ready_set(&mut queues, region)
                .expect("Active/Inactive/Reactive/Postponed are never Region::Nba")
        };
        self.reset_dequeued(&ready);
        sort_ready_set(&mut ready);
        ready
    }

    /// SC-056: the dequeue point. Every activation just popped off a region
    /// queue clears its process's queued flag here, before any `resume()`, so
    /// a change landing between this dequeue and the process's next
    /// sensitivity registration re-queues it instead of being swallowed.
    fn reset_dequeued(&self, popped: &[PendingActivation]) {
        for act in popped {
            self.gate.reset_on_dequeue(act.proc);
        }
    }

    /// Partitions `ready` via the installed `DeterminismController`, then
    /// dispatches every resulting `WorkItem` through `ResumeDriver`, using
    /// one `EngineOps` shared for the whole pass (constructed via a
    /// temporary shared reborrow of `self`, dropped before any per-item
    /// `&mut self` post-processing — see this module's doc comment).
    /// Returns whether anything was actually dispatched.
    ///
    /// **Task 9 (A2-part2, CG-102):** before running the per-item dispatch
    /// body below, `ready` is grouped by each activation's own STORED
    /// `region_id` (`ProcessTable::region_id`, via `partition_of`) -- a
    /// real, if v1-degenerate, partition-aware dispatch pass, not a
    /// placeholder: with `partition_count` fixed at `1` (v1's one real
    /// call site, ARCHITECTURE §4 item 1), every activation maps to
    /// `PartitionId(0)`, so `groups` always has exactly one entry holding
    /// the whole `ready` set in its original order, and the loop below runs
    /// the (otherwise unchanged) per-partition dispatch body exactly once
    /// -- bit-for-bit the same observable behavior as before this task.
    fn dispatch_ready_set(&mut self, region: Region, ready: Vec<PendingActivation>) -> bool {
        if ready.is_empty() {
            return false;
        }
        let mut groups: std::collections::BTreeMap<u32, Vec<PendingActivation>> =
            std::collections::BTreeMap::new();
        for act in ready {
            let region_id = self.table.region_id(act.proc);
            #[cfg(test)]
            self.region_id_queries_seen.push((act.proc, region_id));
            let partition = partition_of(region_id, 1);
            groups.entry(partition.0).or_default().push(act);
        }
        for (_, partition_ready) in groups {
            self.dispatch_one_partition(region, partition_ready);
        }
        true
    }

    /// The per-partition dispatch body [`Self::dispatch_ready_set`]'s new
    /// grouping loop calls once per distinct `PartitionId` among `ready`'s
    /// activations (always exactly once, for v1's real `partition_count ==
    /// 1`) -- unchanged from what `dispatch_ready_set` itself did, in full,
    /// before this task.
    fn dispatch_one_partition(&mut self, region: Region, ready: Vec<PendingActivation>) {
        let work = self.controller.partition(&ready, region, self.now);
        // Each activation's suspend is applied as soon as it returns, before
        // the next one runs: a process that has just started waiting (`@(u)`)
        // is woken by a write a later process makes in the same pass, as in
        // any sequential order of the two (LRM 1800-2023 §4.7) -- a wait
        // installed only after the whole pass missed it.
        for item in &work {
            let proc = item.activation.proc;
            // An activation queued before a `disable` redirected the process
            // (its timer, an event wake) is stale: the wait it ended is gone.
            if self
                .activation_floor
                .get(&proc)
                .is_some_and(|&floor| item.activation.id < floor)
            {
                continue;
            }
            // A wake for a process that already finished is stale (LRM 9.4.2:
            // a finished thread is no longer waiting). It arises inside one
            // dispatched batch: the gate is cleared at dequeue (SC-056), so
            // an earlier member's write re-matches a later member's
            // still-installed wait (`fork @(a[j]) ... join`, a frame
            // variable in the event expression); that member then runs on
            // its first activation, completes and is reclaimed, and its
            // second activation finds no process.
            if matches!(item.activation.arg, ResumeArg::EventFired { .. })
                && self.table.get(proc).is_none()
            {
                continue;
            }
            if matches!(item.activation.arg, ResumeArg::EventFired { .. }) {
                if self
                    .event_floor
                    .get(&proc)
                    .is_some_and(|&floor| item.activation.id < floor)
                {
                    continue;
                }
                self.event_floor
                    .insert(proc, self.activation_seq.load(Ordering::SeqCst));
                // A stale wake queued earlier in this pass still holds the
                // process's dedup flag; clear it now (the process stops
                // waiting here) so the first write to its next wait queues a
                // wake instead of being deduplicated against the dropped one.
                self.gate.reset_on_dequeue(proc);
            }
            // Processes resumed via `ResumeArg::EventFired`/`Disabled` -- the
            // only two variants that terminate a `Suspend::WaitEvents`
            // dynamic wait -- lose their dynamic sensitivity entries
            // (`sensitivity.rs`: entries "are gone from the table the
            // instant the owning process resumes"). Removed BEFORE the
            // process runs: a process is not waiting while it executes, so
            // its own blocking writes must not wake it again through the
            // wait it was resumed from (an `always @*` that reads a variable
            // it also writes would otherwise re-trigger itself forever;
            // LRM 1800-2023 §4.7, §9.4.2). Its new `Suspend` is installed by
            // `apply_suspend` below, after this removal.
            if matches!(
                item.activation.arg,
                ResumeArg::EventFired { .. } | ResumeArg::Disabled
            ) {
                self.sensitivity.remove_dynamic(proc);
                // Task 3 (ledger SC-052): belt-and-suspenders -- a process
                // that resumes for any reason while a `Repeat` term's count is
                // only partially satisfied (e.g. it was `Disabled` mid-wait)
                // must not leave a stale partial count behind for a future,
                // unrelated dynamic wait reusing the same `term_index`.
                self.repeats.clear(proc);
            }
            // Classes plan Task 13 (LRM 9.7): a suspended process does not
            // run. The activation that would have run it is parked and
            // replayed by `p.resume()`; a running one is no longer waiting.
            if let Some(desc) = self.table.get_mut(proc) {
                if desc.suspended {
                    desc.parked = Some(item.activation.arg);
                    continue;
                }
                desc.waiting = false;
                desc.wait_set = None;
            }
            // 38.36.1 `cbStartOfThread`/`cbEnterThread`: a static process is
            // created when it first runs (a fork branch was announced when
            // its `fork` executed); any later activation is a resume.
            if super::thread_hook_installed()
                && let Some(desc) = self.table.get(proc)
            {
                let event = if matches!(item.activation.arg, ResumeArg::Start) {
                    desc.fork_parent
                        .is_none()
                        .then_some(super::ThreadEvent::Start {
                            pid: proc,
                            parent: None,
                            spawn: None,
                        })
                } else {
                    Some(super::ThreadEvent::Enter { pid: proc })
                };
                if let Some(event) = event {
                    self.fire_thread_event_now(event);
                }
            }
            let suspend = {
                let ops = EngineOps {
                    engine: &*self,
                    now: self.now,
                    region,
                    reconv_iter: 0,
                };
                let handle = SimHandle(std::ptr::null_mut());
                // A process terminated by `disable fork;` (SC-125,
                // `terminate_fork_descendants`) may still have activations in
                // flight -- a pending `#delay` timer or a signal-change wake.
                // A killed process never runs again (LRM 9.6.3), so they are
                // dropped here. Every OTHER unknown id is still a bookkeeping
                // bug and panics below.
                if self.table.get(proc).is_none() && self.killed.contains(&proc) {
                    continue;
                }
                if matches!(item.activation.arg, ResumeArg::Start) {
                    self.driver.start(&self.table, proc, handle, &ops)
                } else {
                    self.driver
                        .resume_for(&self.table, proc, item.activation.arg, handle, &ops)
                }
                .expect(
                    "every dispatched ProcessId came from this Engine's own \
                     register_process/handle_fork, so UnknownProcess/AlreadyStarted \
                     here would mean a bug in Engine's own bookkeeping",
                )
                // `ops` (and the shared reborrow of `self` it holds) goes out
                // of scope here, before the mutable calls below.
            };
            self.apply_suspend(proc, suspend);
        }
    }

    /// The NBA log a nonblocking assignment staged by `source` belongs to: the
    /// Re-NBA log for a Reactive-region process, the NBA log otherwise.
    /// Whether a nonblocking write `source` stages now belongs to the Re-NBA
    /// region: `source` is a Reactive-region process, or the running kernel
    /// is in a clocking-block synchronous drive (LRM 14.16: drives update in
    /// the Re-NBA region).
    fn stages_reactive(&self, source: ProcessId) -> bool {
        self.has_reactive.load(Ordering::Relaxed)
            && (self.table.is_reactive(source) || super::kernel_adapter::reactive_drive())
    }

    /// An `edge` of `cell` just happened: stages the edge-skewed clocking
    /// drives (LRM 1800-2023 14.4) waiting on it, `after` ticks later when they
    /// carry a post-edge delay.
    fn fire_edge_nbas(&self, cell: StateCellId, edge: EdgeKind) {
        if edge == EdgeKind::None {
            return;
        }
        let fired: Vec<EdgeNba> = {
            let mut list = self.edge_nbas.lock().unwrap();
            let (hit, keep): (Vec<_>, Vec<_>) = std::mem::take(&mut *list)
                .into_iter()
                .partition(|w| w.cell == cell && (w.edge == edge || w.edge == EdgeKind::BothEdges));
            *list = keep;
            if list.is_empty() {
                self.has_edge_nbas.store(false, Ordering::Relaxed);
            }
            hit
        };
        for w in fired {
            if w.after == 0 {
                w.nba.stage_into(self);
            } else {
                self.delayed_nbas
                    .lock()
                    .unwrap()
                    .entry(self.now.ticks + w.after)
                    .or_default()
                    .push(w.nba);
            }
        }
    }

    fn nba_for(&self, source: ProcessId) -> &Mutex<NbaLog> {
        if self.stages_reactive(source) {
            &self.re_nba
        } else {
            &self.nba
        }
    }

    /// The Re-NBA region commit pass (LRM 4.5): applies the nonblocking
    /// assignments Reactive-region processes staged. Returns whether anything
    /// changed.
    fn commit_re_nba_once(&self) -> bool {
        let mut nba = self.re_nba.lock().unwrap();
        if nba.is_empty() {
            return false;
        }
        let mut arena = self.arena.lock().unwrap();
        let mut sink = EngineNotifySink {
            engine: self,
            at: self.now,
            reconv_iter: 0,
            delta: None,
        };
        crate::sched::commit_nba(&mut nba, &mut **arena, self.now, &mut sink) > 0
    }

    /// LRM 4.5/4.6: once the Reactive region set (Re-Active, Re-Inactive,
    /// Re-NBA) is entered, iteration over the design regions does not resume
    /// until all of it is empty. Drains Re-Active to empty, then one
    /// Re-Inactive batch (back to Re-Active), then Re-NBA, repeating.
    fn drain_reactive_set(&mut self) {
        loop {
            loop {
                let ready = self.drain_and_sort(Region::Reactive);
                if ready.is_empty() {
                    break;
                }
                self.dispatch_ready_set(Region::Reactive, ready);
            }
            self.drain_region_callbacks(Region::Reactive);
            self.drain_pending_spawns();
            let re_inactive = {
                let mut queues = self.queues.lock().unwrap();
                queues.drain_re_inactive()
            };
            if !re_inactive.is_empty() {
                self.reset_dequeued(&re_inactive);
                self.dispatch_ready_set(Region::Reactive, re_inactive);
            } else {
                self.maybe_commit_re_nba();
                self.apply_class_event_wakes();
                let done = {
                    let queues = self.queues.lock().unwrap();
                    queues.reactive_is_empty() && queues.re_inactive_is_empty()
                };
                if done {
                    break;
                }
            }
            self.guard.tick(self.now).expect(
                "SC-080/DT-050: delta-loop livelock guard exceeded -- hard error, \
                 never warn-and-truncate",
            );
        }
    }

    /// Runs the Re-NBA region when Re-Active and Re-Inactive are empty
    /// (LRM 4.5).
    fn maybe_commit_re_nba(&self) {
        if !self.has_reactive.load(Ordering::Relaxed) {
            return;
        }
        let drained = {
            let queues = self.queues.lock().unwrap();
            queues.reactive_is_empty() && queues.re_inactive_is_empty()
        };
        if drained {
            self.commit_re_nba_once();
        }
    }

    /// The SC-039 NBA commit pass for the current timestep. Returns whether
    /// anything actually changed.
    fn commit_nba_once(&self) -> bool {
        let mut nba = self.nba.lock().unwrap();
        let mut arena = self.arena.lock().unwrap();
        let mut sink = EngineNotifySink {
            engine: self,
            at: self.now,
            reconv_iter: 0,
            delta: None,
        };
        crate::sched::commit_nba(&mut nba, &mut **arena, self.now, &mut sink) > 0
    }

    /// **NEW (A2-part2 Task 5, ledger CG-162):** the LRM's NBA
    /// update-APPLICATION half, extracted verbatim from `step_region`'s own
    /// `Region::Nba` arm (a pure refactor -- `step_region`'s own observable
    /// behavior, for every existing caller including debugger/VPI
    /// single-stepping, is unchanged bit-for-bit) so it can also be called
    /// directly as its own, separately-named operation. The ABI's
    /// `sim_commit_nba` entry point (`sv-sim-runtime::abi_entry`) needs
    /// exactly this, decoupled from `sim_eval_region(Nba)` (see
    /// [`Self::current_region`]'s own doc comment for that half) --
    /// ARCHITECTURE §5's "`sim_commit_nba` kept distinct from
    /// `eval_region(Nba)` to mirror the LRM's RHS-eval vs update-apply
    /// split," never collapsed into one call.
    pub fn advance_nba_region(&mut self) -> StepResult {
        let changed = self.commit_nba_once();
        self.apply_class_event_wakes();
        let mut settled = {
            let queues = self.queues.lock().unwrap();
            queues.active_and_inactive_both_empty()
        };
        // sva T2 (§1.4): Active/Inactive/Nba have genuinely settled this
        // round -- drain Observed (the automaton-engine step hook) before
        // committing to Reactive, and re-check: a matured action could hand
        // new work to Active/Inactive, in which case the delta loop must
        // re-run rather than fall through. See
        // `Self::drain_observed_and_check_settled`'s own doc comment.
        if !changed && settled {
            settled = self.drain_observed_and_check_settled();
        }
        if changed || !settled {
            self.guard.tick(self.now).expect(
                "SC-080/DT-050: delta-loop livelock guard exceeded -- hard error, \
                 never warn-and-truncate",
            );
            self.current_region = Region::Active;
        } else {
            self.guard.reset();
            self.forks_this_step = 0;
            self.current_region = Region::Reactive;
        }
        StepResult::Advanced
    }

    /// Advances the wheel to `tick`, dispatching every popped `TimerEntry`
    /// per its own `TimerAction` (SC-069): a `Resume` entry pushes onto its
    /// target region's queue exactly as before this task; a `TriggerEvent`
    /// entry is fanned out through [`EngineNotifySink::notify_change`] --
    /// the same path an ordinary committed write already uses -- with
    /// `EdgeKind::None` (the sentinel `SensitivityTable::matches` already
    /// treats as "any change"/a trigger pulse, `sensitivity.rs`'s own doc
    /// comment). Then advances `self.now`.
    ///
    /// Triggers are collected and fired only after every `Resume` entry has
    /// already been pushed and `self.now` updated: `notify_change` reads
    /// `self.now` (via the `EngineNotifySink` it constructs) for the
    /// `PendingActivation`s it enqueues, so a trigger fired at the new tick
    /// must see that tick as `self.now`, not the pre-advance value.
    /// The earliest tick with anything scheduled: a wheel entry or a
    /// delayed nonblocking write.
    /// Installs the embedder's scheduling hooks (`super::host`).
    pub fn set_host(&mut self, host: Box<dyn super::host::Host>) {
        self.host = Some(host);
        self.host_slot = None;
    }

    /// Runs `f` on the installed host, if any, with this engine's ops (also
    /// installed as the current ops, so code the host calls can write).
    fn with_host<R>(
        &mut self,
        region: Region,
        f: impl FnOnce(&mut dyn super::host::Host, &dyn SchedulerOps) -> R,
    ) -> Option<R> {
        let mut host = self.host.take()?;
        let r = {
            let ops = EngineOps {
                engine: &*self,
                now: self.now,
                region,
                reconv_iter: 0,
            };
            crate::sched::with_scheduler_ops(&ops, || f(&mut *host, &ops))
        };
        self.host = Some(host);
        Some(r)
    }

    /// Reports a thread event to the installed thread hook with this
    /// engine's ops (`super::thread_hooks`).
    fn fire_thread_event_now(&self, event: super::ThreadEvent) {
        let ops = EngineOps {
            engine: self,
            now: self.now,
            region: Region::Active,
            reconv_iter: 0,
        };
        super::fire_thread_event(&ops, event);
    }

    /// The host's slot-begin hook, once per time slot.
    fn host_slot_begin(&mut self) {
        if self.host.is_some() && self.host_slot != Some(self.now.ticks) {
            self.host_slot = Some(self.now.ticks);
            self.with_host(Region::Active, |h, ops| h.slot_begin(ops));
        }
    }

    fn next_event_tick(&self) -> Option<SimTick> {
        let delayed = self.delayed_nbas.lock().unwrap().keys().next().copied();
        let host = self.host.as_ref().and_then(|h| h.next_wakeup());
        [self.wheel.next_tick(), delayed, host]
            .into_iter()
            .flatten()
            .min()
    }

    fn advance_time_to(&mut self, tick: SimTick) {
        let due = self.wheel.advance_to(tick);
        let mut triggers: Vec<StateCellId> = Vec::new();
        {
            let mut queues = self.queues.lock().unwrap();
            for entry in due {
                match entry.action {
                    TimerAction::Resume(act, region) => {
                        queues
                            .push(region, act)
                            .expect("TimeWheel::arm_delay only ever arms Region::Active entries");
                    }
                    TimerAction::TriggerEvent(cell) => triggers.push(cell),
                }
            }
        }
        self.now = SchedTime { ticks: tick };
        // Delayed nonblocking writes due now join this step's NBA log, ahead
        // of any the step's own processes stage (they were scheduled first).
        let due_nbas: Vec<DelayedNba> = {
            let mut delayed = self.delayed_nbas.lock().unwrap();
            let later = delayed.split_off(&(tick + 1));
            std::mem::replace(&mut *delayed, later)
                .into_values()
                .flatten()
                .collect()
        };
        if !due_nbas.is_empty() {
            for d in due_nbas {
                d.stage_into(self);
            }
        }
        for cell in triggers {
            let mut sink = EngineNotifySink {
                engine: self,
                at: self.now,
                reconv_iter: 0,
                delta: None,
            };
            sink.notify_change(
                cell,
                EdgeKind::None,
                self.now.to_abi_display(0),
                Region::Active,
            );
        }
    }

    /// **NEW (Task 4, ledger SC-090/SC-123).** Pushes `cb` onto
    /// [`Self::region_callbacks`] to fire once `region` is next drained
    /// (via [`Self::drain_region_callbacks`]) with `self.now >= at`. Not
    /// `#[cfg(test)]`-gated: `EngineOps::register_region_callback` (the
    /// `SchedulerOps` trait method's one production implementation) is a
    /// thin forward to this, and this module's own test setup calls it
    /// directly for callbacks registered before `run_until` starts (where
    /// no `SchedulerOps`/dispatched-process context exists yet).
    ///
    /// **Review fix (Task 4 r1):** exhaustive `match region`, mirroring
    /// [`RegionQueues::push`]'s own established `Result` convention:
    /// `Reactive`/`Postponed` are the only two regions
    /// [`Self::drain_region_callbacks`] ever drains (see that method's own
    /// doc comment), so only those two push and return `Ok(())`;
    /// `Active`/`Inactive`/`Nba` return
    /// `Err(RegionCallbackRegionError::NotDrained(region))` instead of
    /// silently accepting a registration that would sit in
    /// `region_callbacks` forever and never fire. **sva T1/T2:**
    /// `Preponed` was added to the rejecting side (T1); `Observed` was
    /// added to the ACCEPTING side (T2) once `drain_region_callbacks` grew
    /// a real caller for it -- see the match's own per-arm comments below.
    pub fn push_region_callback(
        &self,
        at: SchedTime,
        region: Region,
        cb: Box<dyn RegionCallback>,
    ) -> Result<(), RegionCallbackRegionError> {
        match region {
            // sva T2 (§1.4, review fix, blocking finding 2), Observed's
            // half: `Observed` joins the ACCEPTING side, unlike `Preponed`
            // below -- this is the literal mechanism by which the
            // Observed-region automaton-engine step (a later task) gets
            // invoked: its engine step registers itself as one more region
            // callback here, fired by `drain_region_callbacks(Region::
            // Observed)` (see `Self::drain_observed_and_check_settled`).
            Region::Reactive | Region::Postponed | Region::Observed => {
                self.region_callbacks.lock().unwrap().push((at, region, cb));
                Ok(())
            }
            // sva T1 (§1.4, review fix, blocking finding 2): `Preponed`
            // joins the REJECTING side, never the accepting one -- the
            // Preponed snapshot push is inline code in `run_until`'s own
            // loop / `step_region`'s `Postponed` arm (see
            // `Engine::push_preponed_snapshot`), never a registered
            // callback, so nothing may register one for it.
            Region::Active | Region::Inactive | Region::Nba | Region::Preponed => {
                Err(RegionCallbackRegionError::NotDrained(region))
            }
        }
    }

    /// **NEW (Task 4, ledger SC-090/SC-123/SC-020's scheduler-owned half).**
    /// Removes and fires every registered callback for `region` whose
    /// `fire_at <= self.now`, in registration order (never resorted — see
    /// [`Self::region_callbacks`]'s own doc comment). Called from
    /// `run_until`/`step_region` immediately after each
    /// `dispatch_ready_set(Region::Reactive, ..)`/`dispatch_ready_set(
    /// Region::Postponed, ..)` call — i.e. strictly after that region's own
    /// process dispatch, and (for `Postponed`) strictly after the whole
    /// Active/Inactive/NBA delta loop has already converged this timestep.
    ///
    /// **sva T2:** also called for `Region::Observed`, but not from a
    /// `dispatch_ready_set` call site like Reactive/Postponed -- `Observed`
    /// has no `RegionQueues` queue (§1.4), so it is drained instead from
    /// [`Self::drain_observed_and_check_settled`], itself called only once
    /// Active/Inactive/Nba have genuinely settled (see that method's own
    /// doc comment).
    ///
    /// This is SC-090's own literal acceptance ("fires exactly once per
    /// settled time step") and the pinning `region_callback`'s own module
    /// doc describes for `1step` (SC-123): a callback registered for
    /// `Region::Postponed` fires strictly after every other
    /// `Region::Postponed` callback registered before it, in the same
    /// drain pass, never in a separate/later timestep.
    fn drain_region_callbacks(&self, region: Region) {
        let due: Vec<Box<dyn RegionCallback>> = {
            let mut callbacks = self.region_callbacks.lock().unwrap();
            let mut due = Vec::new();
            let mut i = 0;
            while i < callbacks.len() {
                if callbacks[i].1 == region && callbacks[i].0 <= self.now {
                    due.push(callbacks.remove(i).2);
                } else {
                    i += 1;
                }
            }
            due
        };
        if due.is_empty() {
            return;
        }
        let ctx = EngineSchedulerCtx {
            engine: self,
            now: self.now,
        };
        // A callback may run compiled code (a `$strobe`/`$monitor` argument
        // thunk) that calls `$time` etc. -- make this Engine's ops reachable
        // to those runtime entry points exactly as during a kernel dispatch.
        let ops = EngineOps {
            engine: self,
            now: self.now,
            region,
            reconv_iter: 0,
        };
        for mut cb in due {
            super::kernel_adapter::with_scheduler_ops(&ops, || cb.invoke(&ctx));
        }
    }

    /// Whether the Reactive region, which has just run, left work for the rest
    /// of this time step (LRM 4.5: after the Reactive region the simulator
    /// goes back to the Active region if anything is scheduled there): a
    /// process it woke in Active or Inactive, a nonblocking assignment it
    /// staged (the Re-NBA region, collapsed into the single NBA region), a
    /// process or callback it left in Reactive for its next round.
    fn reactive_left_work(&self) -> bool {
        {
            let queues = self.queues.lock().unwrap();
            if !queues.active_and_inactive_both_empty()
                || !queues.reactive_is_empty()
                || !queues.re_inactive_is_empty()
            {
                return true;
            }
        }
        if !self.nba.lock().unwrap().is_empty() {
            return true;
        }
        if self.has_reactive.load(Ordering::Relaxed) && !self.re_nba.lock().unwrap().is_empty() {
            return true;
        }
        // A procedural concurrent assertion a Reactive process reached
        // matures in the next Observed region (16.14.6), which the delta loop
        // reaches again only if sent round.
        if !crate::sched::proc_assert::is_empty() {
            return true;
        }
        self.region_callbacks
            .lock()
            .unwrap()
            .iter()
            .any(|(at, region, _)| *region == Region::Reactive && *at <= self.now)
    }

    /// **NEW (sva T2, §1.4).** Drains any registered `Region::Observed`
    /// callback -- the automaton-engine step hook a later task wires in;
    /// none is registered yet this task, so this is a plumbing-only no-op
    /// in every design today, proved by
    /// `step_region_never_observes_current_region_eq_observed` and the
    /// zero-SVA-cost guard the T1 ring already established -- and reports
    /// whether `Active`/`Inactive` are (still) both empty afterward.
    ///
    /// Called from both `run_until`'s inlined delta loop and
    /// [`Self::advance_nba_region`] (the single-stepper's `Region::Nba`
    /// arm) at the identical point: only once NBA committed no further
    /// change AND Active/Inactive are already empty -- i.e. only at the
    /// genuine idle point the LRM's own Observed region is pinned to
    /// (§1.4: "fires inside the existing inner loop every time Active/
    /// Inactive/Nba all empty out"), never speculatively on every delta
    /// iteration. A matured report an Observed callback schedules into
    /// `Region::Reactive` (via the generalized `sched::maturation`, through
    /// `SchedulerOps::register_region_callback`) is consequently already
    /// sitting in `region_callbacks` by the time this same timestep's own
    /// Reactive dispatch runs -- see
    /// `observed_region_maturation_fires_before_reactive_same_timestep`
    /// and `observed_engine_step_result_visible_by_next_reactive_step`.
    ///
    /// If an Observed callback's own effect (not implemented by any task
    /// yet) handed new work to Active/Inactive, the caller must re-enter
    /// the delta loop rather than proceed to Reactive -- the reference
    /// algorithm's own "re-fires on each outer re-iteration, not just
    /// once" structure (Clause 4.5) -- which is exactly why this returns
    /// the re-checked `bool` rather than deciding the transition itself.
    fn drain_observed_and_check_settled(&self) -> bool {
        self.drain_region_callbacks(Region::Observed);
        let queues = self.queues.lock().unwrap();
        queues.active_and_inactive_both_empty()
    }

    /// Applies one dispatched process's returned `Suspend`, exhaustive over
    /// all 7 variants (see this module's doc comment, deviation 3, for
    /// `WaitEvents`/`ForkJoin`'s scope this task; `WaitFork`/`WaitOrder` are
    /// B-sched Task 9's addition, ledger SC-125; `WaitForGvt` is abi Task 8's,
    /// AB-085).
    ///
    /// **Task 8 correction (AB-085):** the `abi`-track plan this arm was
    /// written against claimed `Suspend` had no exhaustive-match consumer in
    /// this file at the time `WaitForGvt` was added; that was wrong -- this
    /// very function already matched all 4 pre-existing variants with no
    /// wildcard arm, so adding a 5th variant is a real, necessary edit here,
    /// not an optional one. The fix is minimal and additive: `WaitForGvt`
    /// delegates to the exact same code the `Delay(ticks)` arm below already
    /// runs, with `ticks = 0` -- i.e. it shares that variant's dispatch path
    /// verbatim rather than introducing any new execution mechanism, which
    /// is AB-085's literal acceptance criterion. Real GVT-barrier semantics
    /// (gating resumption on every parallel lane reaching the same logical
    /// time, not just the next delta step) are the determinism/parallel-
    /// engine track's job and are not implemented here.
    fn apply_suspend(&mut self, proc: ProcessId, suspend: Suspend) {
        // Task 9 (B-sched wave, SC-125): drains every `disable fork;`
        // request `EngineOps::disable_fork` queued mid-`resume()` (that
        // trait method only ever gets `&self`, so it cannot mutate
        // `self.table` itself). `apply_suspend` always runs with `&mut
        // self` immediately after the `resume()` call(s) that might have
        // queued a request -- see `Self::drain_pending_disable_fork`'s own
        // doc comment for why this placement is sufficient.
        self.drain_pending_disable_fork();
        self.drain_pending_disable_named();
        self.drain_pending_disable_scope();
        self.drain_pending_process_requests();
        // A process that `process::kill`ed itself (or whose ancestor it
        // killed) is already gone from the table, with its parent notified
        // and its exit recorded; whatever it returned is moot.
        if self.killed.contains(&proc) && self.table.get(proc).is_none() {
            return;
        }
        if self.any_tracked.load(Ordering::SeqCst)
            && !matches!(suspend, Suspend::Done)
            && let Some(desc) = self.table.get_mut(proc)
        {
            desc.waiting = true;
            desc.wait_set = match &suspend {
                Suspend::WaitEvents(set) => Some(set.clone()),
                _ => None,
            };
        }
        match suspend {
            Suspend::WaitForGvt => self.apply_suspend(proc, Suspend::Delay(0)),
            Suspend::ByRequest => self.apply_by_request(proc),
            Suspend::Done => {
                // Review fix: report this completion to ResumeDriver::
                // on_child_done when `proc` is itself a fork child
                // (`resume.rs`'s own doc comment: on_child_done "reports one
                // fork child's completion to its parent, returning the
                // parent's own next resume() argument exactly when the join
                // condition is satisfied"). `fork_parent` is read from
                // `proc`'s own table entry (set once, by handle_fork, and
                // never for a non-child process), never a linear scan.
                if let Some(parent) = self.table.get(proc).and_then(|d| d.fork_parent) {
                    // Task 9 (SC-125): `on_child_done` now returns
                    // `Result<Option<ResumeArg>, ResumeError>` --
                    // `Err(WaitOrderViolated(..))` means a `wait_order(...)`-
                    // suspended parent observed an out-of-order completion.
                    // There is no real-SV reporting path for this yet this
                    // wave (no track's codegen emits `Suspend::WaitOrder`
                    // this wave -- this task's own e2e test is gated on a
                    // currently-unscheduled future wave), so the only
                    // correct scheduler-side behavior available today is
                    // exactly what a satisfied-`None` result already does:
                    // never resume the parent for this wait. Never a panic
                    // that would take down an otherwise-unrelated run over a
                    // still-undeliverable SV-level diagnostic.
                    if let Ok(Some(arg)) = self.driver.on_child_done(&mut self.table, proc) {
                        let act = PendingActivation {
                            id: self.activation_seq.fetch_add(1, Ordering::SeqCst),
                            proc: parent,
                            arg,
                            reconv_iter: 0,
                            logical_order: (self.now, Region::Active, parent as u64),
                        };
                        self.queues
                            .lock()
                            .unwrap()
                            .push(Region::Active, act)
                            .expect("Region::Active is never rejected by RegionQueues::push");
                    }
                }
                self.note_exit(proc, ProcessState::Finished);
            }
            Suspend::Delay(ticks) => {
                let act = PendingActivation {
                    id: self.activation_seq.fetch_add(1, Ordering::SeqCst),
                    proc,
                    arg: ResumeArg::TimeElapsed,
                    reconv_iter: 0,
                    logical_order: (
                        SchedTime {
                            ticks: self.now.ticks + ticks,
                        },
                        Region::Active,
                        proc as u64,
                    ),
                };
                // The one entry point a Suspend::Delay normalizes through
                // (wheel.rs's own doc comment).
                let dest = self.wheel.arm_delay(self.now.ticks, ticks, act);
                if dest == Region::Inactive && self.table.is_reactive(proc) {
                    // LRM 4.4.2.7, 4.5: `#0` taken from the Reactive region
                    // resumes in Re-Inactive, once Reactive has drained.
                    let act = PendingActivation {
                        logical_order: (act.logical_order.0, Region::Reactive, act.logical_order.2),
                        ..act
                    };
                    self.queues.lock().unwrap().push_re_inactive(act);
                } else if dest == Region::Inactive {
                    // Review fix (mirrors subscribe.rs::register_wait's
                    // identical delay_ticks==0 case): `act` was built above
                    // with logical_order.1 == Region::Active (the common
                    // case, where arm_delay arms the wheel instead). When
                    // arm_delay instead reports Region::Inactive
                    // (delay_ticks==0), that tag must agree with the queue
                    // the activation actually lands in -- order.rs's
                    // canonical_key_of/sort_ready_set read logical_order.1
                    // directly to build the canonical key (SC-010/DT-005/
                    // DT-010) -- so re-tag it to Region::Inactive before
                    // pushing.
                    let act = PendingActivation {
                        logical_order: (act.logical_order.0, Region::Inactive, act.logical_order.2),
                        ..act
                    };
                    self.queues
                        .lock()
                        .unwrap()
                        .push(Region::Inactive, act)
                        .expect("Region::Inactive is never rejected by RegionQueues::push");
                }
            }
            Suspend::WaitEvents(set) => {
                self.sensitivity.install_dynamic(&set, proc);
            }
            Suspend::ForkJoin(kind, spawns) => {
                // Each child is built by the PARENT (`CompiledProcess::
                // mint_fork_child`): it alone knows its compiled branch
                // kernels and its instance's `SimHandle`. The parent's own
                // `resume` has already returned, so its mutex is free here.
                // A parent with no branch kernels (hand-written test
                // processes) keeps the inert `UnmintedForkChild` placeholder.
                let mut minted: Vec<Box<dyn CompiledProcess>> = {
                    let parent = self.table.get(proc).map(|d| d.compiled.lock().unwrap());
                    spawns
                        .iter()
                        .map(|spec| {
                            parent
                                .as_ref()
                                .and_then(|p| p.mint_fork_child(spec))
                                .unwrap_or_else(|| Box::new(UnmintedForkChild))
                        })
                        .collect()
                };
                minted.reverse();
                let mut mint_child = |_spec: &SpawnSpec| -> Box<dyn CompiledProcess> {
                    minted
                        .pop()
                        .expect("handle_fork mints exactly one child per SpawnSpec")
                };
                let children =
                    self.driver
                        .handle_fork(&mut self.table, proc, kind, &spawns, &mut mint_child);
                self.forks_this_step = self.forks_this_step.saturating_add(children.len() as u32);
                if self.forks_this_step > MAX_FORKS_PER_STEP && self.sync_livelock.is_none() {
                    self.sync_livelock = Some(format!(
                        "more than {MAX_FORKS_PER_STEP} fork branches spawned at {:?} without \
                         time advancing (a zero-delay fork loop; hard error per SC-080/DT-050 \
                         policy)",
                        self.now
                    ));
                }
                // Tags each newly minted child with its own `SpawnSpec::
                // disable_name` (the fork's own label and/or this branch's
                // named outermost block, LRM 9.6.2) -- `EngineOps::
                // disable_named`'s `collect_disable_named` is the one reader.
                for (&child, spec) in children.iter().zip(spawns.iter()) {
                    if !spec.disable_name.is_empty()
                        && let Some(child_desc) = self.table.get_mut(child)
                    {
                        child_desc
                            .disable_names
                            .extend(spec.disable_name.iter().cloned());
                    }
                }
                // Task 14 (LRM 18.14.2): each child's generator is seeded
                // with the next value of the forking process's, one draw per
                // child in source order, before any child can run.
                for &child in &children {
                    super::process_rng::preseed(child, super::process_rng::draw_from_process(proc));
                }
                // 38.36.1 `cbStartOfThread`: each branch is a new thread,
                // created here, before it (or the continuing parent) runs.
                if super::thread_hook_installed() {
                    for (&child, spec) in children.iter().zip(&spawns) {
                        self.fire_thread_event_now(super::ThreadEvent::Start {
                            pid: child,
                            parent: Some(proc),
                            spawn: Some(spec.clone()),
                        });
                    }
                }
                {
                    let mut queues = self.queues.lock().unwrap();
                    for &child in &children {
                        let act = PendingActivation {
                            id: self.activation_seq.fetch_add(1, Ordering::SeqCst),
                            proc: child,
                            arg: ResumeArg::Start,
                            reconv_iter: 0,
                            logical_order: (self.now, Region::Active, child as u64),
                        };
                        queues
                            .push(Region::Active, act)
                            .expect("Region::Active is never rejected by RegionQueues::push");
                    }
                }
                // Task 7 (ledger SC-115): `JoinKind::None`'s "the parent
                // proceeds without waiting" -- resolved here, per this
                // module's own doc comment (deviation 3), *without* a new
                // cross-track `ResumeArg` variant meaning "just continue"
                // (`ResumeArg` is A2-process/codegen-owned,
                // `docs/CONTRACTS.md`'s ownership table): the scheduler
                // itself immediately re-invokes the parent's own next
                // segment, synchronously, in this same `apply_suspend`
                // call, right after registering (and enqueueing the Start
                // activation for) every spawned child above -- "proceeds
                // without waiting" because there genuinely is nothing to
                // wait *for*, never delivered as a resume-with-a-reason.
                //
                // **Deviation from the plan's literal `self.driver.start(
                // ..)` text:** that cannot be `ResumeDriver::start` --
                // `proc` already went through `Start` once to reach this
                // very `Suspend::ForkJoin` return in the first place, so a
                // second `start()` call would be rejected outright by its
                // own once-only `AlreadyStarted` CAS guard (`resume.rs`).
                // This calls `resume_for` instead, which is the driver's
                // own "continue an already-started process" entry point --
                // the same one `dispatch_one_partition`'s own ordinary loop
                // uses for every non-`Start` resume. It is reused here with
                // `ResumeArg::TimeElapsed` -- the least-surprising "no
                // special payload" choice among the 5 variants that already
                // exist, matching how an ordinary `Suspend::Delay`
                // continuation already carries no information beyond "keep
                // going". The resulting `Suspend` is fed back into this
                // same function recursively, exactly as `dispatch_one_
                // partition`'s own loop would have done for it.
                // LRM 9.3.2: a fork with no branch statements has nothing to
                // wait for under ANY join keyword (`fork join` is a no-op);
                // without this the parent parked forever on a join no child
                // could ever satisfy.
                if kind == JoinKind::None || spawns.is_empty() {
                    self.resume_synchronously(proc, ResumeArg::TimeElapsed);
                }
            }
            // --- Task 9 (B-sched wave, ledger SC-125) ---
            Suspend::WaitFork => {
                self.apply_wait_fork(proc);
            }
            Suspend::WaitOrder(order) => {
                self.apply_wait_order(proc, order);
            }
        }
    }

    /// **NEW (Task 9, B-sched wave).** Synchronously re-invokes `proc`'s own
    /// next segment with `arg` and immediately applies the resulting
    /// `Suspend`, in this same call -- the "the caller does not suspend, it
    /// just keeps going" pattern `Suspend::ForkJoin`'s own `JoinKind::None`
    /// continuation established (this module's doc comment, deviation 3).
    /// Factored out of that arm so `Suspend::WaitFork`/`Suspend::WaitOrder`'s
    /// own "already satisfied, nothing left to actually wait for" cases
    /// (`apply_wait_fork`/`apply_wait_order` below) can reuse it verbatim
    /// instead of re-deriving the same `resume_for`/`apply_suspend`
    /// recursion.
    fn resume_synchronously(&mut self, proc: ProcessId, arg: ResumeArg) {
        // Trampoline: every caller is in tail position of `apply_suspend`, so
        // a request made while a drain is running runs right after the
        // current one returns, with the same effect and bounded stack.
        if let Some(queue) = self.sync_queue.as_mut() {
            queue.push_back((proc, arg));
            return;
        }
        self.sync_queue = Some(std::collections::VecDeque::new());
        self.resume_synchronously_once(proc, arg);
        let mut drained = 0u32;
        while self.sync_livelock.is_none()
            && let Some((p, a)) = self.sync_queue.as_mut().and_then(|q| q.pop_front())
        {
            drained += 1;
            if drained > MAX_SYNC_RESUMES {
                self.sync_livelock = Some(format!(
                    "a process continued without ever blocking {MAX_SYNC_RESUMES} times at \
                     {:?} (a zero-delay `fork ... join_none` loop never lets time advance; \
                     hard error per SC-080/DT-050 policy)",
                    self.now
                ));
                break;
            }
            self.resume_synchronously_once(p, a);
        }
        self.sync_queue = None;
    }

    fn resume_synchronously_once(&mut self, proc: ProcessId, arg: ResumeArg) {
        if let Some(desc) = self.table.get_mut(proc) {
            desc.waiting = false;
        }
        let suspend = {
            let ops = EngineOps {
                engine: &*self,
                now: self.now,
                region: Region::Active,
                reconv_iter: 0,
            };
            let handle = SimHandle(std::ptr::null_mut());
            self.driver
                .resume_for(&self.table, proc, arg, handle, &ops)
                .expect(
                    "proc is a known, already-started ProcessId in this Engine's own \
                     table (it just returned a Suspend to reach this call) -- \
                     UnknownProcess/AlreadyStarted here would mean a bug in Engine's \
                     own bookkeeping",
                )
        };
        self.apply_suspend(proc, suspend);
    }

    /// **NEW (Task 9, ledger SC-125).** `Suspend::WaitFork`'s scheduler-side
    /// installation: counts how many of `proc`'s own `fork_children` are
    /// still live (not yet `Suspend::Done` + reclaimed) right now. If none
    /// are (the common "every child already finished before `wait fork;`
    /// itself ran" case, or a degenerate `wait fork;` with no live
    /// children at all), `proc` resumes immediately and synchronously --
    /// mirroring `Suspend::ForkJoin`'s own `JoinKind::None` "nothing to
    /// actually wait for" continuation just above -- rather than
    /// round-tripping through the scheduler loop for an already-satisfied
    /// wait. Otherwise, installs the live count onto `proc`'s own
    /// descriptor for `ResumeDriver::on_child_done` to decrement as each
    /// child later completes.
    fn apply_wait_fork(&mut self, proc: ProcessId) {
        let children = self
            .table
            .get(proc)
            .map(|d| d.fork_children.clone())
            .unwrap_or_default();
        let live = children
            .iter()
            .filter(|&&child| self.table.get(child).is_some())
            .count() as u32;
        if live == 0 {
            self.resume_synchronously(proc, ResumeArg::AllForkChildrenDone);
        } else if let Some(desc) = self.table.get_mut(proc) {
            desc.wait_fork_remaining = Some(live);
        }
    }

    /// **NEW (Task 9, ledger SC-125).** `Suspend::WaitOrder(order)`'s
    /// scheduler-side installation. An empty `order` (a degenerate
    /// `wait_order()`) is vacuously satisfied, so `proc` resumes
    /// immediately and synchronously (mirroring `apply_wait_fork`'s
    /// identical "nothing left to wait for" case); otherwise installs
    /// `proc`'s own ordered-completion expectation for `ResumeDriver::
    /// on_child_done` to advance/validate as each declared child completes.
    fn apply_wait_order(&mut self, proc: ProcessId, order: Vec<ProcessId>) {
        if order.is_empty() {
            self.resume_synchronously(proc, ResumeArg::WaitOrderSatisfied);
            return;
        }
        if let Some(desc) = self.table.get_mut(proc) {
            desc.wait_order = Some((order, 0));
            desc.wait_order_violated = false;
        }
    }

    /// **NEW (Task 9, ledger SC-125).** Drains every `ProcessId`
    /// `EngineOps::disable_fork` queued (see that method's own doc comment
    /// for why the actual mutation cannot happen synchronously inside that
    /// `&self` call) and, for each, recursively terminates its live
    /// descendants via [`Self::terminate_fork_descendants`]. Called at the
    /// top of [`Self::apply_suspend`], which always runs with `&mut self`
    /// immediately after the `resume()` call(s) that might have queued a
    /// request -- so by the time any caller's own dispatch step (a
    /// `run_until`/`step_region` call, or a direct `apply_suspend` call in
    /// a test) returns, every requested descendant is already gone.
    fn drain_pending_disable_fork(&mut self) {
        let requested: Vec<ProcessId> =
            std::mem::take(&mut *self.pending_disable_fork.lock().unwrap());
        for proc in requested {
            self.terminate_fork_descendants(proc);
        }
        let blocks: Vec<ProcessId> =
            std::mem::take(&mut *self.pending_disable_block.lock().unwrap());
        for proc in blocks {
            self.disable_enclosing_fork(proc);
        }
    }

    /// `disable f;` run by `proc`, a branch of fork `f`: `proc`'s own
    /// descendants and every sibling branch (with theirs) end now; `proc`
    /// returns `Done` right after the call, and that completion alone is
    /// what the parent's join still waits for.
    fn disable_enclosing_fork(&mut self, proc: ProcessId) {
        self.terminate_fork_descendants(proc);
        let Some(parent) = self.table.get(proc).and_then(|d| d.fork_parent) else {
            return;
        };
        // Siblings are the children of THIS fork only (same `fork_group`):
        // `disable f` names one fork block, never the parent's other forks.
        let group = self.table.get(proc).map_or(0, |d| d.fork_group);
        let siblings: Vec<ProcessId> = self
            .table
            .get(parent)
            .map(|d| d.fork_children.clone())
            .unwrap_or_default()
            .into_iter()
            .filter(|&s| self.table.get(s).is_some_and(|d| d.fork_group == group))
            .collect();
        for s in siblings {
            if s != proc {
                self.terminate_fork_descendants(s);
                self.drop_waits(s);
                self.table.remove(s);
                self.killed.insert(s);
                self.note_exit(s, ProcessState::Killed);
            }
        }
        if let Some(parent_desc) = self.table.get_mut(parent)
            && parent_desc.fork_gen == group
            && parent_desc.fork_remaining > 0
        {
            parent_desc.fork_remaining = 1;
        }
    }

    /// **NEW (Task 9, ledger SC-125).** `disable fork;`'s actual recursive
    /// termination: removes every live entry in `proc`'s own
    /// `fork_children`, and, for each, its own descendants first
    /// (post-order) -- never `proc` itself. A child already reclaimed
    /// (e.g. it had already completed normally before `disable fork;` ran)
    /// is silently skipped, not an error: `disable fork;`'s own LRM
    /// semantics only ever terminate what is still live.
    fn terminate_fork_descendants(&mut self, proc: ProcessId) {
        let children = self
            .table
            .get(proc)
            .map(|d| d.fork_children.clone())
            .unwrap_or_default();
        for child in children {
            if self.table.get(child).is_some() {
                self.terminate_fork_descendants(child);
                self.drop_waits(child);
                self.table.remove(child);
                self.killed.insert(child);
                self.note_exit(child, ProcessState::Killed);
            }
        }
    }

    /// `EngineOps::disable_named`'s deferred half: for each queued
    /// `(requester, targets)`, force-kills every target's subtree except
    /// `requester` itself (see [`Self::kill_subtree_except`] for why
    /// `requester` is always excluded: its own end, when it is one of the
    /// targets or descends from one, already happened through the abort
    /// flag the synchronous half of `disable_named` told it to set, the
    /// same mechanism `process::kill()` uses on itself -- killing its table
    /// entry a second time here, out from under its own still-returning
    /// `resume()` call, is exactly the hazard that mechanism exists to
    /// avoid).
    fn drain_pending_disable_named(&mut self) {
        let requests: Vec<(ProcessId, Vec<ProcessId>)> =
            std::mem::take(&mut *self.pending_disable_named.lock().unwrap());
        for (requester, targets) in requests {
            for target in targets {
                self.kill_subtree_except(target, requester);
            }
        }
    }

    /// `EngineOps::disable_named`'s other deferred half (LRM 9.6.2): for each
    /// queued `(requester, label)`, every other process of the requester's
    /// instance that is suspended inside the named block or task activation
    /// `label` (`CompiledProcess::disable_exit`) ends that activity and
    /// resumes after it ([`Self::redirect_process`]). The requester's own
    /// jump out of `label` is part of its compiled control flow.
    fn drain_pending_disable_scope(&mut self) {
        let requests: Vec<(ProcessId, String)> =
            std::mem::take(&mut *self.pending_disable_scope.lock().unwrap());
        for (requester, label) in requests {
            let root = self.fork_root(requester);
            let instance = self
                .table
                .get(root)
                .and_then(|d| d.compiled.lock().unwrap().disable_instance());
            let Some(instance) = instance else { continue };
            if label == EXIT_PROGRAM_LABEL {
                // Every program `initial` thread of the caller's program
                // instance ends, fork descendants included.
                for t in 0..self.table.count() as ProcessId {
                    let Some(d) = self.table.get(t) else { continue };
                    if d.program_thread
                        && d.fork_parent.is_none()
                        && (t == root
                            || d.compiled.lock().unwrap().disable_instance() == Some(instance))
                    {
                        self.kill_subtree_except(t, requester);
                    }
                }
                continue;
            }
            // `disable u1.blk;` (LRM 9.6.2, an instance path): the compiled
            // kernels of the descendant instance carry the scope under
            // `<requester instance>/<path>` and `/<absolute path>` (run.rs).
            if label.contains('.') {
                let keys = [format!("{instance}/{label}"), format!("/{label}")];
                for target in 0..self.table.count() as ProcessId {
                    if target == requester || self.ended.contains_key(&target) {
                        continue;
                    }
                    let exit = self.table.get(target).and_then(|d| {
                        let compiled = d.compiled.lock().unwrap();
                        keys.iter().find_map(|k| compiled.disable_exit(k))
                    });
                    if let Some(pc) = exit {
                        self.redirect_process(target, pc);
                    }
                }
                // ... and the fork branches of the block's process left
                // running (the process may be past the block, or finished):
                // activities enabled within the block end with it.
                let leaf = label.rsplit('.').next().unwrap_or(&label);
                for root in 0..self.table.count() as ProcessId {
                    if root == requester
                        || self.table.get(root).is_none_or(|d| d.fork_parent.is_some())
                    {
                        continue;
                    }
                    let owns = self.table.get(root).is_some_and(|d| {
                        let compiled = d.compiled.lock().unwrap();
                        keys.iter().any(|k| compiled.owns_disable_scope(k))
                    });
                    if owns {
                        let mut branches = Vec::new();
                        self.collect_disable_named(root, leaf, &mut branches);
                        for t in branches {
                            self.kill_subtree_except(t, requester);
                        }
                    }
                }
                continue;
            }
            // A fork branch of ANOTHER process of the instance named `label`
            // ends too. Done here, not in the synchronous half: the requester's
            // compiled process is locked while it runs.
            let mut branches = Vec::new();
            self.collect_disable_named_in_peers(root, &label, &mut branches);
            for t in branches {
                self.kill_subtree_except(t, requester);
            }
            for target in 0..self.table.count() as ProcessId {
                if target == requester || self.ended.contains_key(&target) {
                    continue;
                }
                let exit = self.table.get(target).and_then(|d| {
                    let compiled = d.compiled.lock().unwrap();
                    (compiled.disable_instance() == Some(instance))
                        .then(|| compiled.disable_exit(&label))
                        .flatten()
                });
                if let Some(pc) = exit {
                    self.redirect_process(target, pc);
                }
            }
        }
    }

    /// Ends what `target` is waiting for and makes it run again at `pc` in
    /// this time step (a `disable` of the block or task it is inside, LRM
    /// 9.6.2): its timer, event wait, `wait`, semaphore or mailbox wait and the
    /// processes it joins are dropped; activations queued for it before this
    /// point are stale ([`Self::activation_floor`]).
    fn redirect_process(&mut self, target: ProcessId, pc: u32) {
        self.drop_waits(target);
        let was_joined = self
            .table
            .get(target)
            .map(|d| d.compiled.lock().unwrap().redirect_resume(pc))
            .unwrap_or(false);
        let children = self
            .table
            .get(target)
            .map(|d| d.fork_children.clone())
            .unwrap_or_default();
        if was_joined {
            // LRM 9.6.2: the activities enabled within the block end too.
            for child in children {
                if self.table.get(child).is_some() {
                    self.kill_subtree_except(child, target);
                }
            }
        }
        if let Some(desc) = self.table.get_mut(target) {
            desc.waiting = false;
            desc.wait_set = None;
            desc.awaiting = None;
            desc.parked = None;
            desc.desensitized = false;
            if was_joined {
                desc.fork_remaining = 0;
                desc.wait_fork_remaining = None;
                desc.wait_order = None;
            }
        }
        let id = self.activation_seq.fetch_add(1, Ordering::SeqCst);
        self.activation_floor.insert(target, id);
        let act = PendingActivation {
            id,
            proc: target,
            arg: ResumeArg::Disabled,
            reconv_iter: 0,
            logical_order: (self.now, Region::Active, target as u64),
        };
        self.queues
            .lock()
            .unwrap()
            .push(Region::Active, act)
            .expect("Region::Active is never rejected by RegionQueues::push");
    }

    /// `node` and every live descendant of it, post-order, EXCEPT `exclude`
    /// itself (`exclude`'s own table entry is left untouched -- it ends
    /// through its own CFG/abort-flag path, if at all; this call still
    /// recurses INTO `exclude`'s own children, since whatever `exclude`
    /// itself forked has no such self-ending mechanism and must be force-
    /// killed like any other descendant). Mirrors [`Self::kill_process`]
    /// exactly (including its "already reclaimed is silently skipped, not
    /// an error" posture) with that one exclusion added.
    fn kill_subtree_except(&mut self, node: ProcessId, exclude: ProcessId) {
        let children = self
            .table
            .get(node)
            .map(|d| d.fork_children.clone())
            .unwrap_or_default();
        for child in children {
            if self.table.get(child).is_some() {
                self.kill_subtree_except(child, exclude);
            }
        }
        if node != exclude && self.table.get(node).is_some() && !self.ended.contains_key(&node) {
            self.terminate_victim(node);
        }
    }

    /// The top-level ancestor of `node` (follows `fork_parent` to its end) --
    /// [`EngineOps::disable_named`]'s search scope, so a `disable <label>;`
    /// run by one process never reaches a same-named branch spawned by a
    /// wholly unrelated top-level process of the same instance.
    fn fork_root(&self, node: ProcessId) -> ProcessId {
        let mut cur = node;
        while let Some(parent) = self.table.get(cur).and_then(|d| d.fork_parent) {
            cur = parent;
        }
        cur
    }

    /// [`Self::collect_disable_named`] over the fork trees of the other
    /// top-level processes of `root`'s instance (the processes that share its
    /// [`CompiledProcess::disable_instance`]), for a `disable <name>;` of a
    /// fork branch's block that lives in another process (LRM 9.6.2).
    fn collect_disable_named_in_peers(
        &self,
        root: ProcessId,
        label: &str,
        out: &mut Vec<ProcessId>,
    ) {
        let instance = self
            .table
            .get(root)
            .and_then(|d| d.compiled.lock().unwrap().disable_instance());
        let Some(instance) = instance else { return };
        for other in 0..self.table.count() as ProcessId {
            if other == root || self.ended.contains_key(&other) {
                continue;
            }
            let Some(d) = self.table.get(other) else {
                continue;
            };
            if d.fork_parent.is_some()
                || d.compiled.lock().unwrap().disable_instance() != Some(instance)
            {
                continue;
            }
            self.collect_disable_named(other, label, out);
        }
    }

    /// Every live process in `node`'s own fork subtree (`node` included)
    /// whose [`super::ProcessDescriptor::disable_names`] contains `label`,
    /// appended to `out` in no particular order -- `EngineOps::
    /// disable_named`'s match set. Recurses into every match too: a label
    /// reused by several concurrently live occurrences (a loop or a
    /// reentrant task forking the same named branch repeatedly) is not
    /// assumed unique, matching every one of them, never just the first.
    fn collect_disable_named(&self, node: ProcessId, label: &str, out: &mut Vec<ProcessId>) {
        let Some(desc) = self.table.get(node) else {
            return;
        };
        if desc.disable_names.iter().any(|n| n == label) {
            out.push(node);
        }
        for &child in &desc.fork_children {
            self.collect_disable_named(child, label, out);
        }
    }

    // --- Classes plan Task 13 (LRM 9.7): the `process` class ---

    /// Pushes an Active-region activation for `proc` (the shape the fork
    /// completion path above uses).
    fn push_active(&mut self, proc: ProcessId, arg: ResumeArg) {
        let act = PendingActivation {
            id: self.activation_seq.fetch_add(1, Ordering::SeqCst),
            proc,
            arg,
            reconv_iter: 0,
            logical_order: (self.now, Region::Active, proc as u64),
        };
        self.queues
            .lock()
            .unwrap()
            .push(Region::Active, act)
            .expect("Region::Active is never rejected by RegionQueues::push");
    }

    /// Whether `node` is `ancestor` itself or one of its fork descendants
    /// (the `fork_parent` chain, which reclaimed ancestors break -- a
    /// descendant of a reclaimed process is not found, see `kill_process`).
    fn is_in_fork_subtree(&self, node: ProcessId, ancestor: ProcessId) -> bool {
        let mut cur = Some(node);
        while let Some(p) = cur {
            if p == ancestor {
                return true;
            }
            cur = self.table.get(p).and_then(|d| d.fork_parent);
        }
        false
    }

    /// `process::status()`'s answer before any queued request is overlaid.
    fn base_process_state(&self, target: ProcessId) -> ProcessState {
        if let Some(&state) = self.ended.get(&target) {
            return state;
        }
        match self.table.get(target) {
            None if self.killed.contains(&target) => ProcessState::Killed,
            None => ProcessState::Finished,
            Some(d) if d.suspended => ProcessState::Suspended,
            Some(d) if d.waiting || d.awaiting.is_some() || d.sync_blocked => ProcessState::Waiting,
            Some(_) => ProcessState::Running,
        }
    }

    /// `process::status()` (LRM 9.7): the base state with every request the
    /// calling process has queued but the engine has not yet applied laid
    /// over it, in call order -- so `p.kill(); p.status()` reads `KILLED`
    /// in the same segment.
    fn effective_process_state(&self, target: ProcessId) -> ProcessState {
        let mut state = self.base_process_state(target);
        for &(_, req) in self.pending_process_requests.lock().unwrap().iter() {
            match req {
                ProcessRequest::Kill(root) => {
                    if !matches!(state, ProcessState::Finished | ProcessState::Killed)
                        && self.is_in_fork_subtree(target, root)
                    {
                        state = ProcessState::Killed;
                    }
                }
                ProcessRequest::Suspend(t) => {
                    if t == target && matches!(state, ProcessState::Running | ProcessState::Waiting)
                    {
                        state = ProcessState::Suspended;
                    }
                }
                ProcessRequest::Resume(t) => {
                    if t == target && state == ProcessState::Suspended {
                        state = ProcessState::Waiting;
                    }
                }
                // Blocking in / waking from a synchronisation object is not
                // something the caller can observe through `status()` before
                // the engine applies it.
                ProcessRequest::Await(_)
                | ProcessRequest::SyncBlock
                | ProcessRequest::SyncWake(_) => {}
            }
        }
        state
    }

    /// Applies every queued `process` request, in call order. Called at the
    /// top of [`Self::apply_suspend`].
    fn drain_pending_process_requests(&mut self) {
        let requests: Vec<(ProcessId, ProcessRequest)> =
            std::mem::take(&mut *self.pending_process_requests.lock().unwrap());
        for (requester, req) in requests {
            match req {
                ProcessRequest::Kill(target) => self.kill_process(target),
                ProcessRequest::Suspend(target) => self.suspend_process(target),
                ProcessRequest::Resume(target) => self.resume_process(target),
                ProcessRequest::Await(target) => self.register_await(requester, target),
                ProcessRequest::SyncBlock => self.apply_sync_block(requester),
                ProcessRequest::SyncWake(target) => self.apply_sync_wake(target),
            }
        }
    }

    /// `p.kill()` (LRM 9.7.3): `target` and every live fork descendant end
    /// before this returns -- descendants first (post-order), `target` last.
    /// A `target` that already ended is skipped but its live descendants are
    /// still terminated (the LRM requires it). A descendant of a *reclaimed*
    /// process is not reachable (its `fork_parent` link died with the
    /// table entry); that gap is documented in the classes plan.
    fn kill_process(&mut self, target: ProcessId) {
        let children = self
            .table
            .get(target)
            .map(|d| d.fork_children.clone())
            .unwrap_or_default();
        for child in children {
            if self.table.get(child).is_some() {
                self.kill_process(child);
            }
        }
        if self.table.get(target).is_some() && !self.ended.contains_key(&target) {
            self.terminate_victim(target);
        }
    }

    /// Ends one live process: drops its waits, records `KILLED`, reports it
    /// to its fork parent's join exactly like a normal completion
    /// (`ResumeDriver::on_child_done`, which also reclaims the table slot),
    /// and wakes its awaiters. Activations already queued for it are
    /// dropped at dispatch (`killed`).
    fn terminate_victim(&mut self, victim: ProcessId) {
        self.drop_waits(victim);
        let parent = self.table.get(victim).and_then(|d| d.fork_parent);
        // `Err(WaitOrderViolated)` has no reporting path (see the `Done`
        // arm of `apply_suspend`); the slot is reclaimed either way.
        let join = self.driver.on_child_done(&mut self.table, victim);
        if let (Some(parent), Ok(Some(arg))) = (parent, join) {
            self.push_active(parent, arg);
        }
        self.killed.insert(victim);
        self.note_exit(victim, ProcessState::Killed);
    }

    /// Drops everything a process about to end is parked on. EVERY path that
    /// ends a live process (`process::kill`, `disable fork`, `disable` of an
    /// enclosing fork) goes through here before the table slot is reclaimed: a
    /// victim left in a semaphore or mailbox queue would be handed the next key
    /// or message and swallow it (the timeout idiom `fork s.get(1); #5;
    /// join_any disable fork;` would lose the following `put`).
    fn drop_waits(&mut self, victim: ProcessId) {
        self.sensitivity.remove_dynamic(victim);
        self.repeats.clear(victim);
        // A process killed while blocked in `semaphore.get` leaves the queue,
        // which can let the waiters behind it through.
        let mut released = Vec::new();
        crate::sched::class_event::cancel(victim);
        semaphore::cancel(victim, &mut |req| released.push(req));
        for req in released {
            if let ProcessRequest::SyncWake(target) = req {
                self.apply_sync_wake(target);
            }
        }
        // Likewise in a mailbox call: a blocked put/get/peek leaves its queue
        // and gives back what it held (a woken waiter's undelivered message).
        mailbox::cancel(victim);
    }

    /// Records how a tracked process ended and wakes whoever is blocked in
    /// `await()` on it.
    fn note_exit(&mut self, proc: ProcessId, state: ProcessState) {
        if let Some(roots) = self.program_roots.as_mut() {
            roots.remove(&proc);
        }
        // 38.36.1 `cbEndOfThread`: the thread is deleted.
        if super::thread_hook_installed() {
            self.fire_thread_event_now(super::ThreadEvent::End {
                pid: proc,
                killed: state == ProcessState::Killed,
            });
        }
        if self.tracked.lock().unwrap().contains(&proc) {
            self.ended.entry(proc).or_insert(state);
        }
        for awaiter in self.awaiters.remove(&proc).unwrap_or_default() {
            if let Some(desc) = self.table.get_mut(awaiter)
                && desc.awaiting == Some(proc)
            {
                desc.awaiting = None;
                self.push_active(awaiter, ResumeArg::TimeElapsed);
            }
        }
        // `wait (p.status() == process::FINISHED)` sleeps on object state: a
        // process ending (finished or killed) is a status change no signal
        // reports (LRM 1800-2023 9.7). Only a tracked process has a status
        // handle to wait on: a statically sensitive process that finishes each
        // wake (a class covergroup's watcher, which object state wakes) is not
        // an ending, and notifying for it would wake itself forever.
        if crate::sched::object_state_watched() && self.tracked.lock().unwrap().contains(&proc) {
            let mut sink = EngineNotifySink {
                engine: self,
                at: self.now,
                reconv_iter: 0,
                delta: None,
            };
            crate::sched::notify_blocking_change(
                crate::sched::OBJECT_STATE_CELL,
                EdgeKind::None,
                self.now,
                &mut sink,
            );
        }
    }

    /// `p.suspend()` (LRM 9.7.4): no effect on a process that already ended
    /// or is already suspended. A process blocked in an `@(...)` event
    /// control is desensitised: its dynamic sensitivity is removed, so an
    /// event that fires while it is suspended is missed, not replayed. A
    /// `wait(expr)` waiter keeps its sensitivity (its resumed kernel
    /// re-checks the condition, so a parked wake is harmless), and a delay
    /// or other wait keeps its timer/join, whose firing is parked and
    /// replayed by [`Self::resume_process`].
    fn suspend_process(&mut self, target: ProcessId) {
        if self.ended.contains_key(&target) {
            return;
        }
        let Some(desc) = self.table.get_mut(target) else {
            return;
        };
        if desc.suspended {
            return;
        }
        desc.suspended = true;
        let desensitize = desc.waiting
            && desc
                .wait_set
                .as_ref()
                .is_some_and(|set| !set.rechecks_on_wake);
        if desensitize {
            desc.desensitized = true;
            self.sensitivity.remove_dynamic(target);
        }
    }

    /// `p.resume()` (LRM 9.7.5): a no-op unless `target` is suspended;
    /// otherwise it runs again. An activation parked while it was suspended
    /// is replayed (it was already runnable, or is a re-checking wait or a
    /// delay that expired); a desensitised `@(...)` waiter that nothing woke
    /// is resensitised to the same event expression.
    fn resume_process(&mut self, target: ProcessId) {
        let Some(desc) = self.table.get_mut(target) else {
            return;
        };
        if !desc.suspended {
            return;
        }
        desc.suspended = false;
        let was_desensitized = std::mem::take(&mut desc.desensitized);
        if let Some(arg) = desc.parked.take() {
            self.push_active(target, arg);
        } else if was_desensitized && let Some(set) = desc.wait_set.clone() {
            self.sensitivity.install_dynamic(&set, target);
        }
    }

    /// `p.await()` (LRM 9.7): blocks `requester` until `target` ends. A
    /// `target` that already ended (or is `requester` itself, which the
    /// runtime shim reports as an error) blocks nothing.
    fn register_await(&mut self, requester: ProcessId, target: ProcessId) {
        if requester == target
            || self.ended.contains_key(&target)
            || self.table.get(target).is_none()
        {
            return;
        }
        if let Some(desc) = self.table.get_mut(requester) {
            desc.awaiting = Some(target);
            self.awaiters.entry(target).or_default().push(requester);
        }
    }

    /// [`ProcessRequest::SyncBlock`]: the requester's `get` queued it on a
    /// semaphore. [`Self::apply_by_request`] then leaves it yielded.
    fn apply_sync_block(&mut self, requester: ProcessId) {
        if let Some(desc) = self.table.get_mut(requester) {
            desc.sync_blocked = true;
        }
    }

    /// Applies the wakes an NBA/Re-NBA commit pass raised for the waiters of a
    /// class event a `->> e` triggered (LRM 15.5.1; `sched::class_event`): the
    /// pass has no process to carry the requests, so it parks them.
    fn apply_class_event_wakes(&mut self) {
        for req in super::class_event::take_deferred_wakes() {
            if let ProcessRequest::SyncWake(target) = req {
                self.apply_sync_wake(target);
            }
        }
    }

    /// [`ProcessRequest::SyncWake`]: `target`'s `get` was granted its keys, so
    /// it runs again (parked instead if `p.suspend()` suspended it meanwhile,
    /// like every other activation). A target that no longer exists was killed
    /// after being granted; its keys are gone with it. A live target that is
    /// not blocked would be a block/wake ordering bug -- `semaphore` emits
    /// `SyncBlock` before any `SyncWake` of the same process, and the queue
    /// preserves that order -- so it is reported loudly instead of resuming a
    /// running process a second time.
    fn apply_sync_wake(&mut self, target: ProcessId) {
        let Some(desc) = self.table.get_mut(target) else {
            return;
        };
        if !desc.sync_blocked && super::class_event::is_dual(target) {
            // A class event woke a process in a mixed `@(class_event or
            // signal ...)` wait (LRM 1800-2023 9.4.2.1): it also waits on
            // signals, so the signal side is dropped first (and an activation
            // a signal already queued is stale: `event_floor`), and the
            // process resumes as if an event fired (term `-1`: no signal term).
            self.sensitivity.remove_dynamic(target);
            self.repeats.clear(target);
            let floor = self.activation_seq.load(Ordering::SeqCst);
            self.event_floor.insert(target, floor);
            self.push_active(
                target,
                ResumeArg::EventFired {
                    term_index: usize::MAX,
                },
            );
            return;
        }
        assert!(
            desc.sync_blocked,
            "internal: semaphore woke process {target}, which is not blocked in a get()"
        );
        desc.sync_blocked = false;
        self.push_active(target, ResumeArg::TimeElapsed);
    }

    /// [`Suspend::ByRequest`]: the process yielded right after queueing a
    /// request, which [`Self::drain_pending_process_requests`] has just
    /// applied. It keeps going at once unless the request left it
    /// self-suspended (the activation is parked for `resume()`) or blocked
    /// in `await()` (`note_exit` wakes it).
    fn apply_by_request(&mut self, proc: ProcessId) {
        let Some(desc) = self.table.get_mut(proc) else {
            return;
        };
        if desc.suspended {
            desc.parked = Some(ResumeArg::TimeElapsed);
        } else if desc.awaiting.is_none() && !desc.sync_blocked {
            self.resume_synchronously(proc, ResumeArg::TimeElapsed);
        }
    }

    // --- #[cfg(test)]-only accessors, added for exactly the tests below ---

    #[cfg(test)]
    fn test_dispatch_ops(&self, now: SchedTime, region: Region, reconv_iter: u32) -> EngineOps<'_> {
        EngineOps {
            engine: self,
            now,
            region,
            reconv_iter,
        }
    }

    #[cfg(test)]
    fn test_read_cell(&self, cell: StateCellId) -> FourStateValue {
        self.arena.lock().unwrap().read(cell)
    }

    #[cfg(test)]
    fn test_nba_len(&self) -> usize {
        self.nba.lock().unwrap().len() + self.re_nba.lock().unwrap().len()
    }

    #[cfg(test)]
    fn test_drain_active(&mut self) -> Vec<PendingActivation> {
        let popped = self.queues.get_mut().unwrap().drain_active();
        self.reset_dequeued(&popped);
        popped
    }

    #[cfg(test)]
    fn test_drain_inactive(&mut self) -> Vec<PendingActivation> {
        let popped = self.queues.get_mut().unwrap().drain_inactive();
        self.reset_dequeued(&popped);
        popped
    }

    #[cfg(test)]
    fn test_sensitivity_entry_count(&self) -> usize {
        self.sensitivity.entry_count()
    }

    /// **Task 7 (B-sched wave, ledger SC-062, SC-115).** Whether `id` is
    /// still a live entry in this `Engine`'s own `ProcessTable` -- the
    /// literal accessor Task 7's own plan text names, used to prove a
    /// completed fork child's slot really was reclaimed (`SC-062`) and,
    /// conversely, that a `JoinKind::None` child that has NOT yet been
    /// dispatched is still live (`SC-115`).
    #[cfg(test)]
    fn test_process_exists(&self, id: ProcessId) -> bool {
        self.table.get(id).is_some()
    }
}

impl Engine {
    /// [`SchedulerControl::register_process`]'s body, with the process's home
    /// region: `reactive` processes (LRM 17.7.3: a checker's procedures and
    /// continuous assignments) start and wake in the Reactive region, every
    /// other process in Active. See [`Self::register_reactive_process`].
    fn register_process_in(
        &mut self,
        proc: Box<dyn CompiledProcess>,
        kind: ProcessKind,
        sensitivity: Option<StaticSensitivity>,
        region_id: u32,
        reactive: bool,
        procedure: bool,
    ) -> ProcessId {
        let id = self.table.register(proc);
        // Task 14 (LRM 18.14.1): a static process is seeded from its
        // containing instance's initialization RNG, before it can ever draw.
        match self.registration_scope {
            Some(scope) if self.registration_static_initializer => {
                super::process_rng::alias_init(id, scope);
            }
            Some(scope) => {
                super::process_rng::preseed(id, super::process_rng::draw_from_init(scope));
            }
            // An id a finished initializer process left behind must not keep
            // its alias.
            None => super::process_rng::clear_alias(id),
        }
        if reactive {
            self.table.set_reactive(id);
            self.has_reactive.store(true, Ordering::Relaxed);
        }
        let home = if reactive {
            Region::Reactive
        } else {
            Region::Active
        };
        // LRM 9.2.2.2: an `always_comb` / `always_latch` "is automatically
        // triggered once at time zero, after all initial and always
        // procedures have been started". Its start activation therefore
        // goes to Inactive, which the time step drains only once Active has
        // quiesced (every other process has run to its first suspension),
        // so a sibling `always @(ev)` is already waiting for an event the
        // comb block triggers on that first run (ivtest br_gh508a).
        let start_region = match kind {
            ProcessKind::AlwaysComb | ProcessKind::AlwaysLatch if procedure => Region::Inactive,
            _ => home,
        };
        // Task 9 (A2-part2, CG-102): stores `region_id` for
        // `dispatch_ready_set` to look up later via `ProcessTable::
        // region_id` -- see `set_region_id`'s own doc comment for why this
        // is a separate call rather than a wider `register` signature.
        self.table.set_region_id(id, region_id);
        // The crate's one and only production call site of `install_static`
        // (SC-045) — called at most once per registration, never a second
        // time for the same `id` anywhere in this body.
        if let Some(sens) = sensitivity.as_ref() {
            self.sensitivity.install_static(sens, id);
        }
        match kind {
            ProcessKind::Final => {
                self.finals.push(FinalProcess {
                    proc: id,
                    registration_order: id as u64,
                });
            }
            ProcessKind::Initial
            | ProcessKind::Always
            | ProcessKind::AlwaysComb
            | ProcessKind::AlwaysLatch
            | ProcessKind::AlwaysFF => {
                let act = PendingActivation {
                    id: self.activation_seq.fetch_add(1, Ordering::SeqCst),
                    proc: id,
                    arg: ResumeArg::Start,
                    reconv_iter: 0,
                    logical_order: (self.now, home, id as u64),
                };
                if start_region != home {
                    // The pending start already is the process's first run: a
                    // change landing before it must not queue a second one.
                    self.gate.mark_queued_if_not_already(id);
                }
                self.queues.lock().unwrap().push(start_region, act).expect(
                    "Region::Active and Region::Reactive are never rejected by RegionQueues::push",
                );
            }
        }
        id
    }

    /// Registers the processes [`SchedulerOps::spawn_detached`] queued, each a
    /// one-shot Reactive-region process (16.11) that starts in the iteration
    /// that queued it.
    fn drain_pending_spawns(&mut self) {
        let spawned = std::mem::take(&mut *self.pending_spawns.lock().unwrap());
        for proc in spawned {
            self.register_process_in(proc, ProcessKind::Initial, None, 0, true, false);
        }
    }

    /// Registers a process that runs in the Reactive region (LRM 17.7.3): a
    /// checker's procedures and continuous assignments "are scheduled in the
    /// Reactive region, similarly to programs". Its start activation and
    /// every wake-up (a static or dynamic sensitivity match) are queued in
    /// Reactive, which the time step drains after Active, Inactive, NBA and
    /// the Observed region have settled and re-enters the delta loop from when
    /// it leaves work behind (see `run_until`). Otherwise identical to
    /// [`SchedulerControl::register_process`].
    /// Marks `id` (just registered) as a program block's `initial` thread
    /// (LRM 24.3: "the simulation ends implicitly when all the program
    /// threads end"). Its fork descendants inherit the mark.
    pub fn mark_program_initial(&mut self, id: ProcessId) {
        if let Some(d) = self.table.get_mut(id) {
            d.program_thread = true;
        }
        self.program_roots
            .get_or_insert_with(Default::default)
            .insert(id);
    }

    /// Whether the design has program `initial` threads and every one of them,
    /// with every fork descendant, has ended (LRM 24.3: implicit `$finish`).
    fn program_threads_all_ended(&self) -> bool {
        let Some(roots) = self.program_roots.as_ref() else {
            return false;
        };
        roots.is_empty() && !self.table.any_live_program_child()
    }

    pub fn register_reactive_process(
        &mut self,
        proc: Box<dyn CompiledProcess>,
        kind: ProcessKind,
        sensitivity: Option<StaticSensitivity>,
        region_id: u32,
    ) -> ProcessId {
        self.register_process_in(proc, kind, sensitivity, region_id, true, false)
    }

    /// [`SchedulerControl::register_process`] for a user-written procedure
    /// (`initial`/`always*`): an `always_comb`/`always_latch` starts after every
    /// other process of the time-zero Active region has started, LRM 1800-2023
    /// 9.2.2.2 ("automatically triggered once at time zero, after all initial
    /// and always procedures have been started"). A continuous assignment
    /// (also an `AlwaysComb` kernel) keeps `register_process`: it starts in
    /// registration order like any Active process.
    pub fn register_procedure(
        &mut self,
        proc: Box<dyn CompiledProcess>,
        kind: ProcessKind,
        sensitivity: Option<StaticSensitivity>,
        region_id: u32,
    ) -> ProcessId {
        self.register_process_in(proc, kind, sensitivity, region_id, false, true)
    }

    /// Narrows `proc`'s `always_comb`/`always_latch` sensitivity on `cell` to
    /// the constant bit windows it reads (LRM 9.2.2.2.1, longest static
    /// prefix); see [`SensitivityTable::narrow_static`].
    pub fn narrow_static_sensitivity(
        &mut self,
        proc: ProcessId,
        cell: StateCellId,
        windows: &[(u32, u32)],
    ) {
        self.sensitivity.narrow_static(proc, cell, windows);
    }
}

impl SchedulerControl for Engine {
    /// See this module's doc comment, deviation 2, for the additional
    /// `kind`-based branching beyond the plan's literal two statements.
    fn register_process(
        &mut self,
        proc: Box<dyn CompiledProcess>,
        kind: ProcessKind,
        sensitivity: Option<StaticSensitivity>,
        region_id: u32,
    ) -> ProcessId {
        self.register_process_in(proc, kind, sensitivity, region_id, false, false)
    }

    /// scheduler.md §1's pseudocode, concretely: per timestep, an inner
    /// Active/Inactive/NBA delta-convergence loop (guarded by `DeltaGuard`,
    /// SC-079/SC-080), then Reactive/Postponed drained once each,
    /// `finish_requested` checked once per timestep (SC-075 — see this
    /// module's doc comment, deviation 4), then the wheel is asked for the
    /// next tick.
    fn run_until(&mut self, deadline: Option<SchedTime>) -> RunResult {
        loop {
            // The host's start of this time slot (`super::host`).
            self.host_slot_begin();
            loop {
                // LRM 1800-2023 §4.5: the Active region runs until it is
                // EMPTY -- including everything woken while it runs (a
                // continuous assignment reacting to `a = 10;`) -- before
                // the Inactive region (`#0`) moves in; each Inactive batch
                // then hands back to Active the same way.
                loop {
                    let active_ready = self.drain_and_sort(Region::Active);
                    if active_ready.is_empty() {
                        let inactive_ready = self.drain_and_sort(Region::Inactive);
                        if inactive_ready.is_empty() {
                            break;
                        }
                        self.dispatch_ready_set(Region::Inactive, inactive_ready);
                    } else {
                        self.dispatch_ready_set(Region::Active, active_ready);
                    }
                    self.with_host(Region::Active, |h, ops| h.activity(ops));
                    if let Some(msg) = self.sync_livelock.take() {
                        return RunResult::Livelock(msg);
                    }
                    if let Err(e) = self.guard.tick(self.now) {
                        return RunResult::Livelock(e.to_string());
                    }
                }
                self.with_host(Region::Active, |h, ops| h.nba_synch(ops));
                let nba_changed = self.commit_nba_once();
                self.apply_class_event_wakes();
                self.with_host(Region::Active, |h, ops| h.activity(ops));
                let mut settled = {
                    let queues = self.queues.lock().unwrap();
                    queues.active_and_inactive_both_empty()
                };
                // sva T2 (§1.4): the identical Observed-drain-and-recheck
                // `advance_nba_region` (the single-stepper's own Nba arm)
                // applies, inlined here for `run_until`'s own delta loop --
                // see `Self::drain_observed_and_check_settled`'s doc
                // comment for why this is a re-check, not a one-shot drain.
                if !nba_changed && settled {
                    settled = self.drain_observed_and_check_settled();
                    // The host's read-write synch point: a write there runs
                    // the loop again.
                    if settled
                        && self
                            .with_host(Region::Active, |h, ops| h.rw_synch(ops))
                            .unwrap_or(false)
                    {
                        settled = false;
                    }
                    if settled {
                        break;
                    }
                }
                if let Err(e) = self.guard.tick(self.now) {
                    return RunResult::Livelock(e.to_string());
                }
            }
            // Task 21/SC-007: record this timestep's settled iteration count
            // into the running high-water mark before the guard resets it.
            self.max_delta_iterations_seen
                .fetch_max(self.guard.count(), Ordering::SeqCst);
            // LRM 4.5: the Reactive region (checker processes, assertion
            // action blocks) runs after Active/Inactive/NBA/Observed have
            // settled, and whatever it leaves behind -- a design process it
            // woke, a nonblocking assignment it staged (Re-NBA, collapsed
            // into the one NBA region), more Reactive work -- sends the time
            // step round the delta loop again. The delta guard keeps counting
            // across these rounds: a livelock through Reactive is as fatal as
            // one inside Active.
            self.drain_reactive_set();
            if self.reactive_left_work() {
                if let Err(e) = self.guard.tick(self.now) {
                    return RunResult::Livelock(e.to_string());
                }
                continue;
            }
            self.guard.reset();
            self.forks_this_step = 0;
            let postponed_ready = self.drain_and_sort(Region::Postponed);
            self.dispatch_ready_set(Region::Postponed, postponed_ready);
            self.drain_region_callbacks(Region::Postponed);
            self.with_host(Region::Postponed, |h, ops| h.slot_end(ops));
            if let Some(hook) = self.timestep_end_hook.as_mut() {
                hook(self.now);
            }

            if self.program_threads_all_ended() {
                self.finish_requested.lock().unwrap().get_or_insert(0);
            }
            if let Some(code) = *self.finish_requested.lock().unwrap() {
                return RunResult::Finished { code };
            }
            if let Some(d) = deadline
                && self.now >= d
            {
                return RunResult::Deadline;
            }
            match self.next_event_tick() {
                Some(t) => {
                    self.advance_time_to(t);
                    // sva T1 (§1.4, review fix r2, blocking finding 1): the
                    // Preponed snapshot push fires here -- right where
                    // `next_event_tick()` has just detected a genuinely new
                    // nonempty time slot and `advance_time_to` actually
                    // moved `self.now` -- mirroring `step_region`'s own
                    // Postponed->Active transition (`Self::step_region`'s
                    // `Region::Postponed` arm, `Some(t)` branch) exactly, so
                    // both driver APIs push the ring the same number of
                    // times for the same simulated history. An earlier
                    // draft pushed unconditionally at the top of this outer
                    // loop, which also fired on the very first iteration --
                    // before `Region::Active` had ever been dispatched at
                    // time 0 -- with no counterpart in `step_region` (`Engine
                    // ::new` enters `Region::Active` directly, no Preponed
                    // push precedes it there either), so the two APIs
                    // produced differently-sized rings for the identical
                    // history. See
                    // `run_until_and_step_region_push_the_same_preponed_ring`
                    // below for the regression proof.
                    self.push_preponed_snapshot();
                }
                None => return RunResult::Completed,
            }
        }
    }

    /// Single-steps through the 5-region cycle, one call per region (see
    /// this module's doc comment, deviation 3, for scope): `Active` ->
    /// `Inactive` -> `Nba` -> (back to `Active` if not yet settled, else)
    /// `Reactive` -> `Postponed` -> (wheel advance, back to `Active`, or
    /// `Blocked` if the wheel is empty too). A debugger-mock harness driving
    /// this repeatedly observes only committed state between calls: every
    /// write reaching STATE goes through `commit_blocking_write`/`commit_nba`
    /// synchronously, inside this call, never buffered across calls.
    fn step_region(&mut self) -> StepResult {
        match self.current_region {
            Region::Active | Region::Inactive => {
                let region = self.current_region;
                let ready = self.drain_and_sort(region);
                self.dispatch_ready_set(region, ready);
                self.current_region = if region == Region::Active {
                    Region::Inactive
                } else {
                    Region::Nba
                };
                StepResult::Advanced
            }
            Region::Nba => self.advance_nba_region(),
            Region::Reactive => {
                self.drain_reactive_set();
                // `run_until`'s loop, one region at a time: work the Reactive
                // region left behind sends the time step back to Active.
                if self.reactive_left_work() {
                    self.guard.tick(self.now).expect(
                        "SC-080/DT-050: delta-loop livelock guard exceeded -- hard error, \
                         never warn-and-truncate",
                    );
                    self.current_region = Region::Active;
                } else {
                    self.current_region = Region::Postponed;
                }
                StepResult::Advanced
            }
            Region::Postponed => {
                let ready = self.drain_and_sort(Region::Postponed);
                self.dispatch_ready_set(Region::Postponed, ready);
                self.drain_region_callbacks(Region::Postponed);
                match self.next_event_tick() {
                    Some(t) => {
                        self.advance_time_to(t);
                        // sva T1 (§1.4, review fix r1, blocking finding 1):
                        // the snapshot push is folded into this
                        // Postponed->Active transition, but ONLY on the
                        // branch that actually starts a new time step
                        // (`advance_time_to` just ran, `self.now` genuinely
                        // moved) -- the single-stepper's own counterpart of
                        // `run_until`'s outer-loop-top push, which likewise
                        // only ever runs again after `next_event_tick()`
                        // returned `Some` and `advance_time_to` was called
                        // (`run_until` returns `RunResult::Completed`
                        // instead of looping again on `None`, so it never
                        // re-pushes for a quiescent wheel either). Moving
                        // this call out of the `None` arm below is the fix
                        // itself -- see that arm's own doc comment.
                        self.push_preponed_snapshot();
                        self.current_region = Region::Active;
                        StepResult::Advanced
                    }
                    None => {
                        // sva T1 (§1.4, review fix r1, blocking finding 1):
                        // **no** snapshot push here. `next_event_tick() ==
                        // None` means the wheel is empty and nothing is
                        // scheduled -- `self.now` does not advance and no
                        // new time slot begins (LRM 16.5.1's sampled value
                        // is defined per "time slot"; a repeated
                        // `step_region()` call on an already-quiescent/
                        // completed simulation re-enters this exact arm
                        // without any new time slot ever starting). An
                        // earlier draft pushed here unconditionally, so a
                        // debugger-style caller single-stepping a
                        // completed design re-pushed the SAME unchanged
                        // snapshot on every call, silently inflating a
                        // bounded ring with duplicate entries and evicting
                        // genuinely distinct older ticks that a future
                        // `$past` consumer needs intact. See
                        // `step_region_blocked_does_not_repush_the_
                        // preponed_ring` below for the regression proof.
                        self.current_region = Region::Active;
                        StepResult::Blocked
                    }
                }
            }
            // sva T1 (§1.4, review fix, blocking finding 2): `self.
            // current_region` is never set to `Region::Preponed` anywhere
            // in this file -- the Preponed snapshot push is the
            // unconditional inline side effect above (and `run_until`'s own
            // outer-loop-top push for the full-run driver), never a state
            // the single-stepper transitions into. **Deviation from the
            // plan's literal text:** the plan's own prose claims
            // `step_region`'s match "keeps its original 5 arms and compiles
            // unchanged" -- not literally possible once `Region` gains a
            // 6th variant with no wildcard arm (`cargo build` confirms:
            // E0004 non-exhaustive patterns without this arm). This
            // defensive `unreachable!` arm is the minimal fix that keeps
            // the plan's INTENDED behavior (no real transition into this
            // region) while satisfying the compiler; the dedicated
            // regression test below proves it is in fact never reached.
            //
            // sva T2: `Region::Observed` joins this same defensive arm for
            // the identical reason -- per §1.4, "that re-check is exactly
            // the decision point already inside step_region's existing
            // Region::Nba => self.advance_nba_region() arm", so `Observed`
            // is drained from inside `advance_nba_region` (folded into the
            // Nba->Reactive transition, see
            // `Self::drain_observed_and_check_settled`), never a state
            // `current_region` itself transitions into either. The plan's
            // own claim that this task "gains no new arm at all" means no
            // new TRANSITION logic, not that the match's pattern list is
            // unchanged -- the same E0004 constraint T1 already hit applies
            // again for the 7th variant, so `Observed` is folded into this
            // identical defensive arm rather than a second one.
            Region::Preponed | Region::Observed => unreachable!(
                "step_region: current_region must never be Region::Preponed or \
                 Region::Observed -- the Preponed snapshot push is an inline side effect \
                 folded into the Postponed->Active transition, and the Observed engine-step \
                 drain is folded into the Nba->Reactive transition; neither is ever an \
                 observable single-stepper state"
            ),
        }
    }

    fn run_finals(&mut self) {
        // `self.finals` is taken out (and put back afterward) so `ops`'s
        // shared reborrow of `self` below doesn't conflict with the `&mut
        // Vec<FinalProcess>` `finals::run_finals` needs.
        let mut finals = std::mem::take(&mut self.finals);
        let handle = SimHandle(std::ptr::null_mut());
        {
            let ops = EngineOps {
                engine: &*self,
                now: self.now,
                region: Region::Active,
                reconv_iter: 0,
            };
            // LRM 9.2.3: `$finish` from within a final procedure ends the
            // simulation immediately, so no later final runs. The request
            // that ended the simulation is set aside while they run.
            let ended_by = self.finish_requested.lock().unwrap().take();
            finals.sort_by_key(|f| f.registration_order);
            for f in finals.chunks_mut(1) {
                crate::sched::run_finals(f, &self.table, handle, &ops, &self.driver);
                if self.finish_requested.lock().unwrap().is_some() {
                    break;
                }
            }
            let mut slot = self.finish_requested.lock().unwrap();
            if slot.is_none() {
                *slot = ended_by;
            }
        }
        self.finals = finals;
    }

    fn now(&self) -> SchedTime {
        self.now
    }

    fn set_determinism_controller(&mut self, ctrl: Box<dyn DeterminismController>) {
        self.controller = ctrl;
    }
}

/// The literal SC-116 deliverable: pairs each of the 9 ABI entry points
/// (`sim_abi_version` excluded — versioning is `abi`-owned) with the
/// `Engine`/`SchedulerControl` method or internal pass it calls.
/// Documentation-as-data (a table, not prose) so it's `assert_eq!`-testable
/// and can't silently go stale.
///
/// **`SimHandle` concurrent-call threading contract (workspace-abi.md's own
/// flagged open question):** one `Engine` instance is never called
/// concurrently for two different ABI entry points on the same `SimHandle`
/// — the engine itself serializes ABI entry (a `Mutex<Engine>` at the
/// `sv-sim-runtime` loader boundary, outside this crate); concurrency inside
/// a run happens only *below* the ABI boundary, inside
/// `DeterminismController::partition`/`drain_and_resolve`.
pub fn abi_entry_point_map() -> &'static [(&'static str, &'static str)] {
    &[
        (
            sv_sim_abi::entry::SIM_INIT_SYMBOL,
            "Engine::new (+ SchedulerControl::register_process for each elaborated process)",
        ),
        (
            sv_sim_abi::entry::SIM_SETTLE_SYMBOL,
            "Engine::run_until(Some(now)) -- settle all zero-delay activity at the current time",
        ),
        (
            sv_sim_abi::entry::SIM_EVAL_REGION_SYMBOL,
            "Engine::step_region",
        ),
        (
            sv_sim_abi::entry::SIM_COMMIT_NBA_SYMBOL,
            "Engine::<internal NBA commit pass -- commit_nba, Task 7/13>",
        ),
        (
            sv_sim_abi::entry::SIM_SAVE_SYMBOL,
            "Engine::<checkpoint -- DeterminismController-owned, not materialized this wave>",
        ),
        (
            sv_sim_abi::entry::SIM_RESTORE_SYMBOL,
            "Engine::<restore -- DeterminismController-owned, not materialized this wave>",
        ),
        (
            sv_sim_abi::entry::SIM_STEP_TIME_SYMBOL,
            "Engine::run_until(Some(deadline)) -- advance exactly one timestep",
        ),
        (sv_sim_abi::entry::SIM_FINAL_SYMBOL, "Engine::run_finals"),
        (
            sv_sim_abi::entry::SIM_PROCESS_COUNT_SYMBOL,
            "Engine::<ProcessTable length accessor>",
        ),
    ]
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::sched::{ActivationId, EventTerm, EventWaitSet, TimerEntry};
    use std::collections::HashMap;
    use std::sync::{Arc, Mutex as StdMutex};

    // --- shared test doubles (independently declared -- see this module's
    // doc comment: neither `notify.rs`'s/`ops.rs`'s `FakeArena` nor Task 8's
    // `MockSchedulerOps` is reachable from this module) ---

    struct TestArena(HashMap<StateCellId, FourStateValue>);

    impl TestArena {
        fn new() -> Self {
            TestArena(HashMap::new())
        }

        fn seeded(cell: StateCellId, value: FourStateValue) -> Self {
            let mut m = HashMap::new();
            m.insert(cell, value);
            TestArena(m)
        }
    }

    impl SignalAccess for TestArena {
        fn read(&self, cell: StateCellId) -> FourStateValue {
            self.0
                .get(&cell)
                .cloned()
                .unwrap_or(FourStateValue::Narrow(0))
        }

        fn write(&mut self, cell: StateCellId, value: FourStateValue) -> bool {
            let changed = match self.0.get(&cell) {
                Some(FourStateValue::Narrow(existing)) => {
                    if let FourStateValue::Narrow(new) = &value {
                        *existing != *new
                    } else {
                        true
                    }
                }
                Some(FourStateValue::Wide { .. }) => true,
                None => true,
            };
            self.0.insert(cell, value);
            changed
        }

        fn subscribe(
            &mut self,
            _cell: StateCellId,
            _role: crate::state::ListenerRole,
            _sink: Box<dyn ValueChangeSink + Send>,
        ) {
            // Not exercised by this task's tests.
        }
    }

    #[derive(Default)]
    struct TestIoSink {
        calls: StdMutex<Vec<(SchedTime, Region, u64)>>,
    }

    impl IoSink for TestIoSink {
        fn enqueue(&self, at: SchedTime, region: Region, seq: u64, cb: DeferredIoCallback) {
            drop(cb);
            self.calls.lock().unwrap().push((at, region, seq));
        }
    }

    fn narrow(v: u128) -> FourStateValue {
        FourStateValue::Narrow(v)
    }

    fn new_engine() -> Engine {
        Engine::new(Box::new(TestArena::new()), Box::new(TestIoSink::default()))
    }

    // --- register_process / run_until / run_finals process fixtures ---

    /// Logs `(ProcessId, ResumeArg)` for every `resume()` call, always
    /// returning `Suspend::Done` (never re-arms). `id_for_log` is filled in
    /// (via the shared cell) only *after* `register_process` returns the
    /// real id -- `resume()` is never called before registration completes,
    /// so it is always `Some` by the time any log entry is pushed.
    struct LoggingProcess {
        id_for_log: Arc<StdMutex<Option<ProcessId>>>,
        log: Arc<StdMutex<Vec<(ProcessId, &'static str)>>>,
    }

    impl CompiledProcess for LoggingProcess {
        fn resume(
            &mut self,
            _state: SimHandle,
            arg: ResumeArg,
            _ops: &dyn SchedulerOps,
        ) -> Suspend {
            let tag = match arg {
                ResumeArg::Start => "Start",
                ResumeArg::TimeElapsed => "TimeElapsed",
                ResumeArg::EventFired { .. } => "EventFired",
                ResumeArg::ForkChildDone { .. } => "ForkChildDone",
                ResumeArg::Disabled => "Disabled",
                ResumeArg::AllForkChildrenDone => "AllForkChildrenDone",
                ResumeArg::WaitOrderSatisfied => "WaitOrderSatisfied",
            };
            let id = self
                .id_for_log
                .lock()
                .unwrap()
                .expect("id_for_log must be set before resume() is ever called");
            self.log.lock().unwrap().push((id, tag));
            Suspend::Done
        }
    }

    fn logging_process(
        log: &Arc<StdMutex<Vec<(ProcessId, &'static str)>>>,
    ) -> (Box<LoggingProcess>, Arc<StdMutex<Option<ProcessId>>>) {
        let id_cell = Arc::new(StdMutex::new(None));
        (
            Box::new(LoggingProcess {
                id_for_log: id_cell.clone(),
                log: log.clone(),
            }),
            id_cell,
        )
    }

    // --- SC-083 ---

    #[test]
    fn sequential_controller_is_selected_when_none_is_ever_set() {
        let mut engine = new_engine();
        let log = Arc::new(StdMutex::new(Vec::new()));
        let (proc, id_cell) = logging_process(&log);
        let id = engine.register_process(proc, ProcessKind::Initial, None, 0);
        *id_cell.lock().unwrap() = Some(id);

        let result = engine.run_until(Some(SchedTime { ticks: 100 }));
        assert!(matches!(result, RunResult::Completed));
        assert_eq!(engine.controller_kind(), "sequential");
    }

    /// Mutation-test kill: `run_until`'s deadline check must fire on
    /// `self.now >= deadline`, not `self.now > deadline` -- the two agree
    /// for every test elsewhere in this file that names a deadline strictly
    /// later than where the run would otherwise settle, so only an exact
    /// `now == deadline` boundary distinguishes them. A process that settles
    /// immediately with no delay leaves `now` at its initial `SchedTime{
    /// ticks: 0}`; a deadline of exactly `0` must therefore report
    /// `RunResult::Deadline` on the very first pass, not silently continue to
    /// `RunResult::Completed`.
    #[test]
    fn run_until_reports_deadline_when_now_exactly_equals_it() {
        let mut engine = new_engine();
        engine.register_process(Box::new(UnmintedForkChild), ProcessKind::Initial, None, 0);

        let result = engine.run_until(Some(SchedTime { ticks: 0 }));
        assert!(
            matches!(result, RunResult::Deadline),
            "now has already reached the deadline exactly (both 0) -- must report \
             RunResult::Deadline, not RunResult::Completed; got {result:?}"
        );
    }

    #[test]
    fn set_determinism_controller_overrides_the_default() {
        struct FakeController;
        impl DeterminismController for FakeController {
            fn partition(
                &self,
                _ready: &[PendingActivation],
                _region: Region,
                _now: SchedTime,
            ) -> Vec<crate::determinism::WorkItem> {
                vec![]
            }
            fn on_read(&self, _act: ActivationId, _cell: StateCellId, _version: u64) {}
            fn on_write(
                &self,
                _act: ActivationId,
                _cell: StateCellId,
                _prior: FourStateValue,
                _version: u64,
            ) {
            }
            fn gvt(&self) -> SchedTime {
                SchedTime { ticks: 0 }
            }
            fn drain_and_resolve(
                &self,
                _work: Vec<crate::determinism::WorkItem>,
            ) -> crate::determinism::ResolvedBatch {
                crate::determinism::ResolvedBatch::default()
            }
            fn kind(&self) -> &'static str {
                "fake-tagged"
            }
        }

        let mut engine = new_engine();
        assert_eq!(engine.controller_kind(), "sequential");
        engine.set_determinism_controller(Box::new(FakeController));
        assert_eq!(engine.controller_kind(), "fake-tagged");
    }

    /// SC-083 — a debugger-mock harness single-steps through regions via
    /// `step_region()`, observing only committed state between steps: `p1`
    /// (Initial) blocking-writes `cell_a`; `p2` (AlwaysComb, statically
    /// sensitive to `cell_a`) is woken by that write and blocking-writes
    /// `cell_b`. Every read via the engine's arena accessor, at every point
    /// during the multi-call sequence, shows either the pre-write default or
    /// the fully-committed written value -- never anything else, since
    /// `commit_blocking_write` only ever applies synchronously.
    #[test]
    fn step_region_single_steps_exposing_only_committed_state() {
        struct P1 {
            cell_a: StateCellId,
        }
        impl CompiledProcess for P1 {
            fn resume(
                &mut self,
                _state: SimHandle,
                arg: ResumeArg,
                ops: &dyn SchedulerOps,
            ) -> Suspend {
                if matches!(arg, ResumeArg::Start) {
                    ops.commit_blocking_write(self.cell_a, narrow(1));
                }
                Suspend::Done
            }
        }
        struct P2 {
            cell_b: StateCellId,
        }
        impl CompiledProcess for P2 {
            fn resume(
                &mut self,
                _state: SimHandle,
                arg: ResumeArg,
                ops: &dyn SchedulerOps,
            ) -> Suspend {
                if matches!(arg, ResumeArg::EventFired { .. }) {
                    ops.commit_blocking_write(self.cell_b, narrow(1));
                }
                Suspend::Done
            }
        }

        let cell_a = StateCellId(1);
        let cell_b = StateCellId(2);
        let mut engine = new_engine();
        engine.register_process(Box::new(P1 { cell_a }), ProcessKind::Initial, None, 0);
        engine.register_process(
            Box::new(P2 { cell_b }),
            ProcessKind::AlwaysComb,
            Some(StaticSensitivity {
                edges: vec![(cell_a, None)],
            }),
            0,
        );

        // Before anything runs: both cells read the pre-write default.
        assert!(matches!(
            engine.test_read_cell(cell_a),
            FourStateValue::Narrow(0)
        ));
        assert!(matches!(
            engine.test_read_cell(cell_b),
            FourStateValue::Narrow(0)
        ));

        let mut saw_a_committed = false;
        let mut saw_b_committed = false;
        for _ in 0..16 {
            engine.step_region();
            let a = engine.test_read_cell(cell_a);
            let b = engine.test_read_cell(cell_b);
            // At every single point, each cell is exactly its pre-write
            // default or its one fully-committed value -- never a third,
            // "partially applied" value.
            assert!(matches!(
                a,
                FourStateValue::Narrow(0) | FourStateValue::Narrow(1)
            ));
            assert!(matches!(
                b,
                FourStateValue::Narrow(0) | FourStateValue::Narrow(1)
            ));
            if matches!(a, FourStateValue::Narrow(1)) {
                saw_a_committed = true;
            }
            if matches!(b, FourStateValue::Narrow(1)) {
                // cell_b only ever becomes 1 after cell_a already did --
                // p2's write is causally downstream of p1's.
                assert!(saw_a_committed, "cell_b committed before cell_a");
                saw_b_committed = true;
            }
        }
        assert!(
            saw_a_committed,
            "p1's write was never observed as committed"
        );
        assert!(
            saw_b_committed,
            "p2's woken write was never observed as committed"
        );
    }

    /// sva SV-D (LRM 17.7.3, 4.5): a Reactive process (a checker procedure) runs only after
    /// the Active/Inactive/NBA queues have emptied, however many Active processes the
    /// time step's earlier regions chain through, and the work it leaves behind (here an Active
    /// process woken by its write) is run in the same time step -- the Reactive-to-Active
    /// feedback of the delta loop -- not deferred to the next `run_until` call.
    #[test]
    fn reactive_process_runs_after_active_settles_and_feeds_back_to_active() {
        struct Tracer {
            name: &'static str,
            log: Arc<StdMutex<Vec<String>>>,
            write: Option<(StateCellId, u128)>,
        }
        impl CompiledProcess for Tracer {
            fn resume(
                &mut self,
                _state: SimHandle,
                arg: ResumeArg,
                ops: &dyn SchedulerOps,
            ) -> Suspend {
                let tag = match arg {
                    ResumeArg::Start => "start",
                    ResumeArg::EventFired { .. } => "event",
                    _ => "other",
                };
                self.log
                    .lock()
                    .unwrap()
                    .push(format!("{}:{tag}", self.name));
                if let Some((cell, v)) = self.write {
                    ops.commit_blocking_write(cell, narrow(v));
                }
                Suspend::Done
            }
        }
        let (a, b, c) = (StateCellId(1), StateCellId(2), StateCellId(3));
        let log = Arc::new(StdMutex::new(Vec::new()));
        let tracer = |name, write| {
            Box::new(Tracer {
                name,
                log: log.clone(),
                write,
            })
        };
        let sens = |cell| {
            Some(StaticSensitivity {
                edges: vec![(cell, None)],
            })
        };
        let mut engine = new_engine();
        // Registered Reactive-process-first: the id order must not matter.
        // `r` (Reactive) is woken by `b`, which the Active `b_w` writes after `a`
        // changed; it writes `c`, which wakes the Active `c_w`.
        engine.register_reactive_process(
            tracer("r", Some((c, 1))),
            ProcessKind::AlwaysComb,
            sens(b),
            0,
        );
        engine.register_process(tracer("w", Some((a, 1))), ProcessKind::Initial, None, 0);
        engine.register_process(
            tracer("b_w", Some((b, 1))),
            ProcessKind::AlwaysComb,
            sens(a),
            0,
        );
        engine.register_process(tracer("c_w", None), ProcessKind::AlwaysComb, sens(c), 0);

        let result = engine.run_until(None);
        assert!(matches!(result, RunResult::Completed));
        let log = log.lock().unwrap().clone();
        let pos = |s: &str| {
            log.iter()
                .position(|e| e == s)
                .unwrap_or_else(|| panic!("{s} never ran: {log:?}"))
        };
        // The Reactive process's start activation is not run in Active: both Active
        // chains (w -> b_w, and c_w's own start) complete before any `r` entry.
        let first_r = log.iter().position(|e| e.starts_with("r:")).unwrap();
        assert!(
            pos("w:start") < first_r && pos("b_w:event") < first_r && pos("c_w:start") < first_r,
            "a Reactive process ran before the Active region settled: {log:?}"
        );
        // Feedback: the Active process woken by `r`'s write ran in the same time step.
        assert!(
            pos("c_w:event") > pos("r:event"),
            "the Active wake-up of a Reactive write was not run: {log:?}"
        );
    }

    /// Mutation-test kill: `step_region`'s `Active`/`Inactive` arm advances
    /// `current_region` to `Inactive` from `Active` and to `Nba` from
    /// `Inactive` -- a mutant that swapped those two targets passes every
    /// other test in this file, since none of them observe `current_region`
    /// directly (they only observe committed cell state or dispatch logs,
    /// which happen to still converge to the same eventual outcome for a
    /// design with no `Inactive`-region-specific behavior). This test reads
    /// `Engine::current_region` directly -- a private field, reachable here
    /// since this test module is a descendant of `control`.
    #[test]
    fn step_region_advances_active_to_inactive_to_nba_in_order() {
        let mut engine = new_engine();
        assert_eq!(
            engine.current_region,
            Region::Active,
            "a fresh Engine starts in Active"
        );

        engine.step_region();
        assert_eq!(
            engine.current_region,
            Region::Inactive,
            "Active must step to Inactive next, never Nba"
        );

        engine.step_region();
        assert_eq!(
            engine.current_region,
            Region::Nba,
            "Inactive must step to Nba next, never back to Inactive"
        );
    }

    /// sva T1 (§1.4, review fix, blocking finding 2): drives the
    /// single-stepper through many full 5-region cycles (well past the
    /// `register_wait`-style edge cases) and asserts `current_region` is
    /// never observed to be `Region::Preponed` at any point -- the
    /// behavioral half of `step_region`'s own defensive `unreachable!`
    /// arm: if that arm were ever actually reached, this test would panic
    /// (via the `unreachable!` itself) rather than silently pass.
    #[test]
    fn step_region_never_observes_current_region_eq_preponed() {
        let mut engine = new_engine();
        engine.register_process(Box::new(UnmintedForkChild), ProcessKind::Initial, None, 0);

        for _ in 0..30 {
            engine.step_region();
            assert_ne!(
                engine.current_region(),
                Region::Preponed,
                "the single-stepper must never transition into Region::Preponed -- the \
                 Preponed snapshot push is an inline side effect, never an observable region"
            );
        }
    }

    /// sva T2 (§1.4, review fix, blocking finding 2): the identical
    /// behavioral proof as `step_region_never_observes_current_region_eq_
    /// preponed`, now for `Region::Observed` -- if `step_region`'s
    /// defensive `unreachable!` arm (now covering both variants) were ever
    /// actually reached for `Observed`, this test would panic rather than
    /// silently pass.
    #[test]
    fn step_region_never_observes_current_region_eq_observed() {
        let mut engine = new_engine();
        engine.register_process(Box::new(UnmintedForkChild), ProcessKind::Initial, None, 0);

        for _ in 0..30 {
            engine.step_region();
            assert_ne!(
                engine.current_region(),
                Region::Observed,
                "the single-stepper must never transition into Region::Observed -- the \
                 Observed engine-step drain is folded into advance_nba_region, never an \
                 observable region"
            );
        }
    }

    /// sva T1 r1 (review fix, blocking finding 1): `step_region`'s
    /// `Region::Postponed` arm must push the configured Preponed ring
    /// exactly once per time step that genuinely BEGINS -- never again
    /// merely because a caller keeps single-stepping an already-quiescent
    /// simulation. A one-shot process (`Suspend::Delay(1)` on its first
    /// `Start` resume, then `Suspend::Done` on its second) arms the wheel
    /// for exactly one real tick transition and then goes permanently
    /// quiet, so `next_event_tick()` returns `Some` exactly once across the
    /// whole run and `None` (`StepResult::Blocked`) on every Postponed arm
    /// after that -- precisely the "ordinary debugger-driver" shape the
    /// finding's own live repro used (30 `step_region()` calls on a
    /// completed/quiescent design hit `StepResult::Blocked` repeatedly).
    /// This proves both halves of the fix: the push DOES fire on the one
    /// real `Some(t)` transition, and it does NOT fire again on any of the
    /// many subsequent `None`/`Blocked` cycles -- a regression to the
    /// unconditional-push bug (pushing in both match arms) would fail the
    /// second assertion by inflating `push_count` past `1`.
    #[test]
    fn step_region_blocked_does_not_repush_the_preponed_ring() {
        use crate::state::{StateEngine, TemplateRegistry};
        use std::cell::Cell;
        use std::rc::Rc;

        struct OnceDelayed {
            started: bool,
        }
        impl CompiledProcess for OnceDelayed {
            fn resume(
                &mut self,
                _state: SimHandle,
                arg: ResumeArg,
                _ops: &dyn SchedulerOps,
            ) -> Suspend {
                match (self.started, arg) {
                    (false, ResumeArg::Start) => {
                        self.started = true;
                        Suspend::Delay(1)
                    }
                    _ => Suspend::Done,
                }
            }
        }

        let mut engine = new_engine();
        let push_count = Rc::new(Cell::new(0u32));
        {
            let push_count = push_count.clone();
            // A fresh StateEngine purely to mint real `ArenaSnapshot`
            // values -- this test counts invocations of the source
            // closure, never the snapshotted content itself (already
            // covered by `sva_preponed_sampling_matches_lrm_16_5_1` above).
            let state_engine = StateEngine::new(TemplateRegistry::new());
            engine.configure_preponed_ring(0, move || {
                push_count.set(push_count.get() + 1);
                state_engine.snapshot(None)
            });
        }
        engine.register_process(
            Box::new(OnceDelayed { started: false }),
            ProcessKind::Initial,
            None,
            0,
        );

        let mut blocked_seen = 0;
        for _ in 0..30 {
            if engine.step_region() == StepResult::Blocked {
                blocked_seen += 1;
            }
        }
        assert!(
            blocked_seen > 0,
            "this 30-call sequence must actually reach StepResult::Blocked at least once \
             (it does, repeatedly, once the one armed tick has fired and gone quiet), or \
             the regression assertion below would be vacuous"
        );
        assert_eq!(
            push_count.get(),
            1,
            "exactly one real time-step transition occurred (the single Delay(1)) -- the \
             ring must have been pushed exactly once, never once per StepResult::Blocked \
             call (that was the bug: a quiescent/completed single-stepper re-pushed the \
             same unchanged snapshot on every call, silently inflating a bounded ring and \
             evicting genuinely distinct older ticks a future $past consumer needs intact)"
        );
    }

    /// sva T1 r2 (review fix, blocking finding 1): `run_until` and
    /// `step_region` are two separate driver entry points for the exact
    /// same `Engine`/`push_preponed_snapshot` mechanism (`Self::
    /// push_preponed_snapshot`'s own doc comment), so the identical
    /// simulated history must push the Preponed ring the same number of
    /// times regardless of which one drives it. This drives the SAME
    /// one-shot process (`Suspend::Delay(1)` then `Suspend::Done` -- the
    /// exact repro the finding's own live check used) through two freshly
    /// constructed engines, one via `run_until(None)` and one via 30
    /// `step_region()` calls (mirroring `step_region_blocked_does_not_
    /// repush_the_preponed_ring` immediately above), and asserts both
    /// counted exactly one push. Before the fix, `run_until`'s outer loop
    /// pushed unconditionally at the top -- including on its very first
    /// iteration, before `Region::Active` had ever been dispatched at time
    /// 0 -- so this process's `run_until` push count was 2 while its
    /// `step_region` push count was 1 for the identical history; this test
    /// fails on that regression by catching the mismatch directly, not by
    /// hard-coding either side's count in isolation.
    #[test]
    fn run_until_and_step_region_push_the_same_preponed_ring() {
        use crate::state::{StateEngine, TemplateRegistry};
        use std::cell::Cell;
        use std::rc::Rc;

        struct OnceDelayed {
            started: bool,
        }
        impl CompiledProcess for OnceDelayed {
            fn resume(
                &mut self,
                _state: SimHandle,
                arg: ResumeArg,
                _ops: &dyn SchedulerOps,
            ) -> Suspend {
                match (self.started, arg) {
                    (false, ResumeArg::Start) => {
                        self.started = true;
                        Suspend::Delay(1)
                    }
                    _ => Suspend::Done,
                }
            }
        }

        fn count_pushes_via(driver: impl FnOnce(&mut Engine)) -> u32 {
            let mut engine = new_engine();
            let push_count = Rc::new(Cell::new(0u32));
            {
                let push_count = push_count.clone();
                let state_engine = StateEngine::new(TemplateRegistry::new());
                engine.configure_preponed_ring(0, move || {
                    push_count.set(push_count.get() + 1);
                    state_engine.snapshot(None)
                });
            }
            engine.register_process(
                Box::new(OnceDelayed { started: false }),
                ProcessKind::Initial,
                None,
                0,
            );
            driver(&mut engine);
            push_count.get()
        }

        let run_until_pushes = count_pushes_via(|engine| {
            let result = engine.run_until(None);
            assert!(matches!(result, RunResult::Completed));
        });
        let step_region_pushes = count_pushes_via(|engine| {
            for _ in 0..30 {
                engine.step_region();
            }
        });

        assert_eq!(
            run_until_pushes, step_region_pushes,
            "run_until and step_region must push the configured Preponed ring the same \
             number of times for the identical simulated history (one real Delay(1) \
             transition) -- a mismatch here means the two driver APIs disagree about what \
             'once per time step' means, which is exactly the asymmetry this regression \
             test exists to catch"
        );
        assert_eq!(
            run_until_pushes, 1,
            "exactly one real time-step transition occurred (the single Delay(1)); neither \
             driver may push at the engine's initial Region::Active entry at time 0 (no \
             Preponed push precedes it for either API -- Engine::new sets current_region = \
             Active directly)"
        );
    }

    /// sva T1 end-to-end (§1.3/§1.4): a signal written in `Region::Active`
    /// is read back through a configured Preponed ring at the START of the
    /// following time steps -- the "stub consumer" this task's own plan
    /// describes (no `$past`/assertion lowering exists yet to read the ring
    /// from compiled SV source; `Engine::with_preponed_entry` is this
    /// task's own test-only substitute). Driven via [`Engine::step_region`]
    /// (never `run_until`'s deadline-based API) specifically so each
    /// Postponed->Active transition -- and so each ring push -- is counted
    /// exactly once (5 `step_region` calls == one full Active/Inactive/
    /// Nba/Reactive/Postponed cycle, the same convention
    /// `suspend_delay_logical_order_uses_absolute_time_across_chained_delays`
    /// above already establishes).
    ///
    /// Verified by hand against LRM 16.5.1: a depth-1 ring's `get(0)`
    /// ("current") always holds the value as of the end of the time step
    /// that just finished; `get(1)` holds the value as of the end of the
    /// time step before that. `cell` is written to 10 during time step 0's
    /// `Active` region and to 20 during time step 1's -- after time step 0
    /// finishes (one 5-step cycle), `get(0)` must read 10 and `get(1)` must
    /// be unretained (`None`, nothing pushed before entry 0 yet); after
    /// time step 1 finishes (a second 5-step cycle), `get(0)` must read 20
    /// and `get(1)` must read 10.
    #[test]
    fn sva_preponed_sampling_matches_lrm_16_5_1() {
        use crate::state::{
            SlotClass, SlotLocation, StateCellId as RealStateCellId, StateEngine, TemplateRegistry,
        };
        use std::cell::RefCell;
        use std::rc::Rc;

        // A real, StateEngine-backed signal, mirroring `state::snapshot`'s
        // own test fixtures -- `Engine`'s own `arena: Box<dyn SignalAccess>`
        // stays a `TestArena` double (sched's established, STATE-decoupled
        // test convention), while the Preponed ring's `source` closure
        // reads the REAL StateEngine `MirroringArena::write` below also
        // writes into, proving the ring genuinely snapshots live arena
        // state, not the test double. `Rc<RefCell<_>>`, not `Arc<Mutex<_>>`:
        // `StateEngine` holds `Rc` internals of its own (`state::instance`'s
        // own doc comment) and is neither `Send` nor `Sync` -- this is
        // single-threaded test code, so there is no reason to reach for a
        // thread-safe wrapper clippy's `arc_with_non_send_sync` correctly
        // flags as misleading here.
        let state_engine = Rc::new(RefCell::new(StateEngine::new(TemplateRegistry::new())));
        let real_cell: RealStateCellId = {
            let mut se = state_engine.borrow_mut();
            // SAFETY: local, single-owned engine; no other live borrow of
            // the inline plane at this point.
            let word_offset = unsafe { se.arena.alloc_inline_word() };
            se.alloc_cell(SlotLocation {
                class: SlotClass::Inline,
                word_offset,
                bit_width: 8,
                elem_word_count: 1,
                row_stride_words: 0,
                num_rows: 0,
            })
        };

        // `Engine`'s own test-double cell (what the scripted process
        // actually addresses through `SchedulerOps`) maps 1:1 onto
        // `real_cell` by sharing the same numeric id -- `MirroringArena`
        // mirrors every write into `state_engine` too, so both views stay
        // in lockstep without `sched` needing a concrete STATE dependency.
        let cell = StateCellId(real_cell.0);

        struct MirroringArena {
            inner: TestArena,
            mirror_to: Rc<RefCell<StateEngine>>,
        }
        impl SignalAccess for MirroringArena {
            fn read(&self, cell: StateCellId) -> FourStateValue {
                self.inner.read(cell)
            }
            fn write(&mut self, cell: StateCellId, value: FourStateValue) -> bool {
                self.mirror_to.borrow_mut().commit_write(
                    RealStateCellId(cell.0),
                    value.clone(),
                    &mut (),
                );
                self.inner.write(cell, value)
            }
            fn subscribe(
                &mut self,
                cell: StateCellId,
                role: crate::state::ListenerRole,
                sink: Box<dyn ValueChangeSink + Send>,
            ) {
                self.inner.subscribe(cell, role, sink);
            }
        }

        /// Writes 10 at `Start` (time step 0), 20 at the first
        /// `TimeElapsed` (time step 1), then goes quiet (`step == 2`'s
        /// `TimeElapsed` writes nothing and stays done) -- `Suspend::
        /// Delay(1)` after each write is what makes the wheel advance
        /// exactly one tick per time step.
        struct SequencedWriter {
            cell: StateCellId,
            step: u32,
        }
        impl CompiledProcess for SequencedWriter {
            fn resume(
                &mut self,
                _state: SimHandle,
                arg: ResumeArg,
                ops: &dyn SchedulerOps,
            ) -> Suspend {
                match (self.step, arg) {
                    (0, ResumeArg::Start) => {
                        ops.commit_blocking_write(self.cell, narrow(10));
                        self.step = 1;
                        Suspend::Delay(1)
                    }
                    (1, ResumeArg::TimeElapsed) => {
                        ops.commit_blocking_write(self.cell, narrow(20));
                        self.step = 2;
                        Suspend::Delay(1)
                    }
                    _ => Suspend::Done,
                }
            }
        }

        let mut engine = Engine::new(
            Box::new(MirroringArena {
                inner: TestArena::new(),
                mirror_to: state_engine.clone(),
            }),
            Box::new(TestIoSink::default()),
        );
        {
            let src = state_engine.clone();
            engine.configure_preponed_ring(1, move || src.borrow().snapshot(None));
        }
        engine.register_process(
            Box::new(SequencedWriter { cell, step: 0 }),
            ProcessKind::Initial,
            None,
            0,
        );

        // `FourStateValue` carries no `PartialEq` (reference/introspection
        // shape only, `state::mod.rs`'s own doc comment on it) -- this
        // narrows down to the one `u128` lane this test's own single
        // `SlotClass::Inline` cell ever uses, mirroring `state::snapshot`'s
        // own `read_narrow` test helper.
        let read_through_ring = |engine: &Engine, ticks_back: usize| -> Option<u128> {
            engine.with_preponed_entry(ticks_back, |snap| {
                snap.map(|snap| {
                    let mut se = state_engine.borrow_mut();
                    se.restore(snap)
                        .expect("same StateEngine the ring sampled from, same layout");
                    match se.read(real_cell) {
                        FourStateValue::Narrow(w) => w,
                        FourStateValue::Wide { .. } => panic!("expected Narrow"),
                    }
                })
            })
        };

        // One full 5-region cycle completes time step 0 (Start dispatched
        // during Active, writing 10 and arming Delay(1); the Postponed step
        // then advances the wheel to tick 1 and pushes the Preponed
        // snapshot for the start of tick 1).
        for _ in 0..5 {
            engine.step_region();
        }
        assert_eq!(
            read_through_ring(&engine, 0),
            Some(10u128),
            "16.5.1: after time step 0 finishes, the ring's current entry must read the \
             value committed during time step 0's own Active region (10)"
        );
        assert_eq!(
            read_through_ring(&engine, 1),
            None,
            "only one push has happened so far -- ticks_back=1 must not be retained yet"
        );

        // A second 5-region cycle completes time step 1 (the resumed
        // TimeElapsed writes 20 and arms another Delay(1); Postponed pushes
        // the Preponed snapshot for the start of tick 2).
        for _ in 0..5 {
            engine.step_region();
        }
        assert_eq!(
            read_through_ring(&engine, 0),
            Some(20u128),
            "16.5.1: after time step 1 finishes, the ring's current entry must read the \
             value committed during time step 1's own Active region (20)"
        );
        assert_eq!(
            read_through_ring(&engine, 1),
            Some(10u128),
            "16.5.1: ticks_back=1 must still read time step 0's own committed value (10)"
        );
    }

    #[test]
    fn register_process_returns_the_ids_run_until_and_run_finals_operate_on() {
        let log: Arc<StdMutex<Vec<(ProcessId, &'static str)>>> =
            Arc::new(StdMutex::new(Vec::new()));
        let mut engine = new_engine();

        let mut regular_ids = Vec::new();
        for _ in 0..3 {
            let (proc, id_cell) = logging_process(&log);
            let id = engine.register_process(proc, ProcessKind::Initial, None, 0);
            *id_cell.lock().unwrap() = Some(id);
            regular_ids.push(id);
        }

        let mut final_ids = Vec::new();
        for _ in 0..3 {
            let (proc, id_cell) = logging_process(&log);
            let id = engine.register_process(proc, ProcessKind::Final, None, 0);
            *id_cell.lock().unwrap() = Some(id);
            final_ids.push(id);
        }

        let result = engine.run_until(None);
        assert!(matches!(result, RunResult::Completed));
        engine.run_finals();

        let entries = log.lock().unwrap().clone();
        // Every regular process ran (via run_until's initial dispatch) with
        // exactly one Start; no final-kind process was dispatched by
        // run_until (Final processes are never queued onto Active).
        for &id in &regular_ids {
            let starts: Vec<_> = entries
                .iter()
                .filter(|(p, tag)| *p == id && *tag == "Start")
                .collect();
            assert_eq!(
                starts.len(),
                1,
                "process {id} must have gotten exactly one Start"
            );
        }
        // The finals ran (via run_finals), in registration order.
        let final_start_order: Vec<ProcessId> = entries
            .iter()
            .filter(|(p, _)| final_ids.contains(p))
            .map(|(p, _)| *p)
            .collect();
        assert_eq!(
            final_start_order, final_ids,
            "finals must run in registration order"
        );
        for &id in &final_ids {
            let starts: Vec<_> = entries
                .iter()
                .filter(|(p, tag)| *p == id && *tag == "Start")
                .collect();
            assert_eq!(
                starts.len(),
                1,
                "final process {id} must have gotten exactly one Start"
            );
        }
    }

    /// Classes plan Task 14 (LRM 18.14.1): a process registered while
    /// `registration_scope` names an instance is seeded from that instance's
    /// initialization RNG, one draw per process in registration order; two
    /// instances have independent streams; and a process registered with no
    /// scope (infrastructure) does not advance any instance's RNG.
    #[test]
    fn registered_processes_are_seeded_from_their_instances_init_rng() {
        use crate::sched::process_rng::{self, with_process_state};
        let nop = || -> Box<dyn CompiledProcess> { script(|_, _| Suspend::Done) };
        process_rng::reset();
        let mut engine = new_engine();
        engine.set_registration_scope(Some(0));
        let a = engine.register_process(nop(), ProcessKind::Initial, None, 0);
        engine.set_registration_scope(None);
        let infra = engine.register_process(nop(), ProcessKind::Initial, None, 0);
        engine.set_registration_scope(Some(0));
        let b = engine.register_process(nop(), ProcessKind::Initial, None, 0);
        engine.set_registration_scope(Some(1));
        let c = engine.register_process(nop(), ProcessKind::Initial, None, 0);
        engine.set_registration_scope(None);
        let seen: Vec<u64> = [a, infra, b, c]
            .iter()
            .map(|&p| with_process_state(Some(p), |s| *s))
            .collect();

        process_rng::reset();
        let (e0, e1, e2) = (
            process_rng::draw_from_init(0),
            process_rng::draw_from_init(0),
            process_rng::draw_from_init(1),
        );
        assert_eq!(seen[0], e0, "first process of instance 0");
        assert_eq!(
            seen[2], e1,
            "second process of instance 0 (the infra one skipped)"
        );
        assert_eq!(seen[3], e2, "first process of instance 1");
        assert_eq!(seen[1], process_rng::DEFAULT_SEED, "unscoped: never seeded");
        assert_ne!(seen[0], seen[2]);
    }

    /// Classes plan Task 14 (LRM 18.14.2): `fork` children are seeded when
    /// they are created, one draw each in source order from the parent's own
    /// generator (so the parent advances by exactly one step per child), no
    /// matter when they later run.
    #[test]
    fn fork_children_are_seeded_from_the_forking_process_in_source_order() {
        use crate::sched::process_rng::{self, with_process_state};
        process_rng::reset();
        let mut engine = new_engine();
        let spawn = || SpawnSpec {
            kernel: 0,
            entry_pc: 0,
            parent: 0,
            disable_name: Vec::new(),
        };
        let parent = engine.register_process(
            script(move |arg, _| match arg {
                ResumeArg::Start => Suspend::ForkJoin(JoinKind::None, vec![spawn(), spawn()]),
                _ => Suspend::Done,
            }),
            ProcessKind::Initial,
            None,
            0,
        );
        assert_eq!(parent, 0);
        // The parent's stream, before the fork.
        with_process_state(Some(parent), |s| *s = 99);
        let ready = engine.drain_and_sort(Region::Active);
        engine.dispatch_ready_set(Region::Active, ready);
        let seen: Vec<u64> = [1, 2]
            .iter()
            .map(|&c| with_process_state(Some(c), |s| *s))
            .collect();
        let parent_after = with_process_state(Some(parent), |s| *s);

        let mut p = 99u64;
        let want = [
            process_rng::splitmix64_next(&mut p),
            process_rng::splitmix64_next(&mut p),
        ];
        assert_eq!(seen, want, "children seeded by successive parent draws");
        assert_eq!(parent_after, p, "one parent step per child, no more");
        assert_ne!(seen[0], seen[1]);
    }

    /// SC-045's "installation happens exactly once, at registration" proof:
    /// no `commit_nba`/`commit_blocking_write`/`notify_change` call anywhere
    /// in this test, so downstream dedup cannot mask a double-install.
    #[test]
    fn register_process_installs_static_sensitivity_exactly_once() {
        let a = StateCellId(1);
        let b = StateCellId(2);
        let c = StateCellId(3);
        let mut engine = new_engine();
        engine.register_process(
            Box::new(UnmintedForkChild),
            ProcessKind::AlwaysComb,
            Some(StaticSensitivity {
                edges: vec![(a, None), (b, None), (c, None)],
            }),
            0,
        );
        assert_eq!(engine.test_sensitivity_entry_count(), 3);
    }

    #[test]
    fn abi_entry_point_map_names_all_nine_entry_points_exactly_once() {
        let map = abi_entry_point_map();
        let names: std::collections::HashSet<&str> = map.iter().map(|(name, _)| *name).collect();
        assert_eq!(names.len(), map.len(), "no duplicate entry-point names");
        let expected: std::collections::HashSet<&str> = [
            "sim_init",
            "sim_settle",
            "sim_eval_region",
            "sim_commit_nba",
            "sim_save",
            "sim_restore",
            "sim_step_time",
            "sim_final",
            "sim_process_count",
        ]
        .into_iter()
        .collect();
        assert_eq!(names, expected);
    }

    // --- EngineOps / EngineNotifySink wiring against Engine's own state ---

    #[test]
    fn engine_ops_commit_blocking_write_writes_through_to_the_engines_own_arena() {
        let cell = StateCellId(7);
        let engine = Engine::new(
            Box::new(TestArena::seeded(cell, narrow(0))),
            Box::new(TestIoSink::default()),
        );
        let ops = engine.test_dispatch_ops(SchedTime { ticks: 0 }, Region::Active, 0);
        ops.commit_blocking_write(cell, narrow(1));
        assert!(matches!(
            engine.test_read_cell(cell),
            FourStateValue::Narrow(1)
        ));
    }

    #[test]
    fn engine_ops_commit_blocking_write_wakes_a_statically_sensitive_process_via_the_real_queues() {
        let cell = StateCellId(9);
        let mut engine = Engine::new(
            Box::new(TestArena::seeded(cell, narrow(0))),
            Box::new(TestIoSink::default()),
        );
        let p1 =
            engine.register_process(Box::new(UnmintedForkChild), ProcessKind::Initial, None, 0);
        let p2 = engine.register_process(
            Box::new(UnmintedForkChild),
            ProcessKind::AlwaysFF,
            Some(StaticSensitivity {
                edges: vec![(cell, Some(EdgeKind::Posedge))],
            }),
            0,
        );
        // Both p1/p2 auto-queued a Start activation at registration; drain
        // that away so the assertion below only counts the wakeup this test
        // itself triggers.
        let _ = engine.test_drain_active();

        {
            let ops = engine.test_dispatch_ops(SchedTime { ticks: 0 }, Region::Active, 0);
            // 0 -> 1 on bit 0 is a Posedge (notify.rs's infer_edge table).
            ops.commit_blocking_write(cell, narrow(1));
        }

        let queued = engine.test_drain_active();
        assert_eq!(queued.len(), 1, "exactly one new activation queued");
        assert_eq!(
            queued[0].proc, p2,
            "the statically-sensitive process p2 was woken"
        );
        let _ = p1;
    }

    /// SC-055 through the real `Engine`: a process statically sensitive to
    /// two cells that both change in the same pass is queued exactly once.
    /// Without the gate `notify_change` pushes one activation per matching
    /// change (2 here).
    #[test]
    fn engine_queues_a_process_once_when_two_of_its_cells_change_in_one_pass() {
        let a = StateCellId(20);
        let b = StateCellId(21);
        let mut engine = new_engine();
        let p = engine.register_process(
            Box::new(UnmintedForkChild),
            ProcessKind::AlwaysComb,
            Some(StaticSensitivity {
                edges: vec![(a, None), (b, None)],
            }),
            0,
        );
        let _ = engine.test_drain_active();

        {
            let ops = engine.test_dispatch_ops(SchedTime { ticks: 0 }, Region::Active, 0);
            ops.commit_blocking_write(a, narrow(1));
            ops.commit_blocking_write(b, narrow(1));
        }

        let queued = engine.test_drain_active();
        assert_eq!(
            queued.iter().map(|q| q.proc).collect::<Vec<_>>(),
            vec![p],
            "two changes in one pass must queue the process once, not twice"
        );
    }

    /// SC-056 through the real `Engine`: once the queued activation has been
    /// popped (the dequeue point resets the flag), a new change queues the
    /// process again. A gate that never reset would swallow the second wakeup.
    #[test]
    fn engine_requeues_a_process_changed_again_after_its_activation_was_dequeued() {
        let a = StateCellId(22);
        let mut engine = new_engine();
        let p = engine.register_process(
            Box::new(UnmintedForkChild),
            ProcessKind::AlwaysComb,
            Some(StaticSensitivity {
                edges: vec![(a, None)],
            }),
            0,
        );
        let _ = engine.test_drain_active();

        {
            let ops = engine.test_dispatch_ops(SchedTime { ticks: 0 }, Region::Active, 0);
            ops.commit_blocking_write(a, narrow(1));
        }
        assert_eq!(
            engine.test_drain_active().len(),
            1,
            "first change queues the process"
        );

        {
            let ops = engine.test_dispatch_ops(SchedTime { ticks: 0 }, Region::Active, 1);
            ops.commit_blocking_write(a, narrow(0));
        }
        let queued = engine.test_drain_active();
        assert_eq!(
            queued.iter().map(|q| q.proc).collect::<Vec<_>>(),
            vec![p],
            "a change after the dequeue must queue the process again"
        );
    }

    #[test]
    fn engine_ops_stage_nba_appends_to_the_engines_own_nba_log() {
        let engine = new_engine();
        assert_eq!(engine.test_nba_len(), 0);
        let ops = engine.test_dispatch_ops(SchedTime { ticks: 0 }, Region::Active, 0);
        ops.stage_nba(StateCellId(1), narrow(1), 0);
        assert_eq!(engine.test_nba_len(), 1);
    }

    #[test]
    fn engine_ops_defer_io_forwards_with_a_seq_counter_that_persists_across_passes() {
        // `io_sink: Box<dyn IoSink>` is not downcastable, so this test uses a
        // shared-state sink to observe the recorded seqs from outside.
        struct SharedIoSink(Arc<StdMutex<Vec<u64>>>);
        impl IoSink for SharedIoSink {
            fn enqueue(&self, _at: SchedTime, _region: Region, seq: u64, cb: DeferredIoCallback) {
                drop(cb);
                self.0.lock().unwrap().push(seq);
            }
        }
        let seqs = Arc::new(StdMutex::new(Vec::new()));
        let engine = Engine::new(
            Box::new(TestArena::new()),
            Box::new(SharedIoSink(seqs.clone())),
        );
        let ops1 = engine.test_dispatch_ops(SchedTime { ticks: 0 }, Region::Active, 0);
        ops1.defer_io(SchedTime { ticks: 0 }, Region::Active, Box::new(|_| {}));
        let ops2 = engine.test_dispatch_ops(SchedTime { ticks: 5 }, Region::Inactive, 0);
        ops2.defer_io(SchedTime { ticks: 5 }, Region::Inactive, Box::new(|_| {}));

        assert_eq!(
            *seqs.lock().unwrap(),
            vec![0, 1],
            "seq must strictly increase across two independently-constructed EngineOps"
        );
    }

    // --- review fixes (r1) ---

    /// Review fix — `apply_suspend`'s `Suspend::Delay` arm used to always
    /// build `act` with `logical_order.1 == Region::Active`, even when
    /// `wheel.arm_delay` reports `Region::Inactive` (a zero-delay `#0`
    /// reactivation) and the activation is pushed straight into
    /// `Region::Inactive` instead. Mirrors `subscribe.rs::register_wait`'s
    /// own `delay_zero_lands_directly_in_inactive_not_the_wheel` test for
    /// the identical hazard, one file over.
    #[test]
    fn apply_suspend_delay_zero_lands_directly_in_inactive_with_the_correct_region_tag() {
        struct DelayZeroProcess;
        impl CompiledProcess for DelayZeroProcess {
            fn resume(
                &mut self,
                _state: SimHandle,
                arg: ResumeArg,
                _ops: &dyn SchedulerOps,
            ) -> Suspend {
                match arg {
                    ResumeArg::Start => Suspend::Delay(0),
                    _ => Suspend::Done,
                }
            }
        }

        let mut engine = new_engine();
        let proc =
            engine.register_process(Box::new(DelayZeroProcess), ProcessKind::Initial, None, 0);

        // Single-steps exactly the Active region: dispatches the process's
        // Start, which returns Suspend::Delay(0) -- apply_suspend's
        // wheel.arm_delay call reports Region::Inactive (delay_ticks == 0,
        // per wheel.rs's own doc comment) without ever touching the wheel.
        engine.step_region();

        let inactive = engine.test_drain_inactive();
        assert_eq!(
            inactive.len(),
            1,
            "the zero-delay reactivation must land directly in Region::Inactive"
        );
        assert_eq!(inactive[0].proc, proc);
        assert_eq!(
            inactive[0].logical_order.1,
            Region::Inactive,
            "an activation landed in the Inactive queue must carry Region::Inactive as its \
             own logical_order.1 tag, agreeing with the queue it actually lives in -- \
             order.rs's canonical_key_of/sort_ready_set read logical_order.1 directly to \
             build the canonical key that SC-010/DT-005/DT-010 depend on"
        );
    }

    /// Whole-track review fix (AB-085): the only prior test for
    /// `Suspend::WaitForGvt` (`wait_for_gvt_shares_the_suspend_dispatch_path`
    /// in `sched::mod`) never called `apply_suspend` at all -- it matched a
    /// test-local, hand-written `label` function against the `Suspend` enum,
    /// which only proves that function's own match arms are exhaustive, not
    /// that `apply_suspend`'s `Suspend::WaitForGvt => self.apply_suspend(proc,
    /// Suspend::Delay(0))` arm actually delegates anywhere. This test drives
    /// a real process through the real `Engine`/`apply_suspend` with
    /// `Suspend::WaitForGvt` and asserts the resulting scheduler state is
    /// identical (region, queue length, logical_order tag) to what
    /// `apply_suspend_delay_zero_lands_directly_in_inactive_with_the_correct_
    /// region_tag` (immediately above) observes for `Suspend::Delay(0)` --
    /// proving the delegation for real. Before this test existed, changing
    /// the `WaitForGvt` arm to `Suspend::Delay(1)`, to a silent no-op, or to
    /// removing the delegation entirely (stranding the process) would still
    /// have passed the full test suite.
    #[test]
    fn apply_suspend_wait_for_gvt_delegates_to_the_same_path_as_delay_zero() {
        struct WaitForGvtProcess;
        impl CompiledProcess for WaitForGvtProcess {
            fn resume(
                &mut self,
                _state: SimHandle,
                arg: ResumeArg,
                _ops: &dyn SchedulerOps,
            ) -> Suspend {
                match arg {
                    ResumeArg::Start => Suspend::WaitForGvt,
                    _ => Suspend::Done,
                }
            }
        }

        let mut engine = new_engine();
        let proc =
            engine.register_process(Box::new(WaitForGvtProcess), ProcessKind::Initial, None, 0);

        // Single-steps exactly the Active region: dispatches the process's
        // Start, which returns Suspend::WaitForGvt. If apply_suspend's
        // `Suspend::WaitForGvt` arm really delegates to
        // `self.apply_suspend(proc, Suspend::Delay(0))`, the observable
        // outcome must be pixel-for-pixel identical to the Delay(0) test
        // above: one reactivation, landed directly in Region::Inactive
        // (never touching the wheel), tagged with Region::Inactive.
        engine.step_region();

        let inactive = engine.test_drain_inactive();
        assert_eq!(
            inactive.len(),
            1,
            "WaitForGvt must delegate to Delay(0)'s dispatch, landing its reactivation \
             directly in Region::Inactive exactly like Delay(0) does -- a broken or removed \
             delegation would leave this queue empty (process stranded) or route it \
             elsewhere"
        );
        assert_eq!(inactive[0].proc, proc);
        assert_eq!(
            inactive[0].logical_order.1,
            Region::Inactive,
            "the delegated activation must carry the same Region::Inactive logical_order \
             tag Delay(0) produces, agreeing with the queue it actually lives in -- a \
             WaitForGvt arm changed to Delay(1) or to any wheel-touching path would instead \
             land here with Region::Active, or not land here at all this step"
        );
        assert_eq!(
            inactive[0].logical_order.0,
            SchedTime { ticks: 0 },
            "delegating to Delay(0) from now == 0 must produce the same absolute fire tick \
             Delay(0) produces (0 + 0 == 0), not some other tick a differently-parameterized \
             delegation (e.g. Delay(1)) would produce"
        );
    }

    /// Mutation-test kill: `apply_suspend`'s `Suspend::Delay` arm must build
    /// the reactivation's `logical_order.0` from `self.now.ticks + ticks`
    /// (the real absolute fire tick), never from `ticks` alone. The bug is
    /// invisible at `now == 0` (where the two computations coincide, exactly
    /// the case every other `Delay` test in this file exercises), so this
    /// test chains two delays -- `Delay(3)` fired from `now == 0`, then
    /// `Delay(4)` fired from the resulting `now == 3` -- and inspects the
    /// *second* reactivation's own `logical_order.0` after the wheel fires
    /// it: it must be `SchedTime{ticks: 7}` (3+4), never `SchedTime{ticks:
    /// 4}`. The wheel's own real fire time is unaffected by this bug (it
    /// computes `now + delay_ticks` internally, independent of the
    /// activation's `logical_order` field), so `engine.now()` alone cannot
    /// catch it -- only inspecting the queued activation's own field can.
    #[test]
    fn suspend_delay_logical_order_uses_absolute_time_across_chained_delays() {
        struct TwoStageDelay;
        impl CompiledProcess for TwoStageDelay {
            fn resume(
                &mut self,
                _state: SimHandle,
                arg: ResumeArg,
                _ops: &dyn SchedulerOps,
            ) -> Suspend {
                match arg {
                    ResumeArg::Start => Suspend::Delay(3),
                    ResumeArg::TimeElapsed => Suspend::Delay(4),
                    _ => Suspend::Done,
                }
            }
        }

        let mut engine = new_engine();
        engine.register_process(Box::new(TwoStageDelay), ProcessKind::Initial, None, 0);

        // Exactly 10 step_region() calls: 5 regions (Active/Inactive/Nba/
        // Reactive/Postponed) x 2 timesteps -- the first timestep dispatches
        // Start (arming Delay(3)) and advances the wheel to tick 3 without
        // dispatching the resulting reactivation; the second timestep
        // dispatches that TimeElapsed (arming Delay(4)) and advances the
        // wheel to tick 7, again landing the reactivation in Region::Active
        // without dispatching it.
        for _ in 0..10 {
            engine.step_region();
        }

        assert_eq!(
            engine.now(),
            SchedTime { ticks: 7 },
            "the wheel's own real fire time (unaffected by this bug) must be 0+3, then 3+4 = 7"
        );

        let active = engine.test_drain_active();
        assert_eq!(
            active.len(),
            1,
            "the second delay's reactivation must be sitting in Region::Active, undispatched"
        );
        assert_eq!(
            active[0].logical_order.0,
            SchedTime { ticks: 7 },
            "logical_order.0 must be the absolute fire tick (now=3 + delay=4 = 7), not \
             the raw delay amount (4) alone"
        );
    }

    /// Review fix — `apply_suspend`'s `Suspend::WaitEvents` arm installed a
    /// dynamic sensitivity entry on suspend but nothing in `Engine`'s
    /// dispatch path ever called `SensitivityTable::remove_dynamic` when the
    /// waiting process later resumed, leaking the entry forever and risking
    /// a phantom wakeup of an already-resumed process on a later write to
    /// the same cell.
    #[test]
    fn wait_events_dynamic_sensitivity_entry_is_removed_when_the_process_resumes() {
        struct WaiterProcess {
            cell: StateCellId,
        }
        impl CompiledProcess for WaiterProcess {
            fn resume(
                &mut self,
                _state: SimHandle,
                arg: ResumeArg,
                _ops: &dyn SchedulerOps,
            ) -> Suspend {
                match arg {
                    ResumeArg::Start => Suspend::WaitEvents(EventWaitSet {
                        terms: vec![EventTerm::Edge {
                            cell: self.cell,
                            edge: EdgeKind::Posedge,
                            iff: None,
                        }],
                        ..Default::default()
                    }),
                    _ => Suspend::Done,
                }
            }
        }
        struct WriterProcess {
            cell: StateCellId,
        }
        impl CompiledProcess for WriterProcess {
            fn resume(
                &mut self,
                _state: SimHandle,
                arg: ResumeArg,
                ops: &dyn SchedulerOps,
            ) -> Suspend {
                match arg {
                    // Delay past t=0 so this write lands in its own,
                    // later region pass -- one full pass after
                    // WaiterProcess's own Start has already been applied
                    // (apply_suspend, and therefore install_dynamic, only
                    // runs once every item dispatched *in the same pass*
                    // has itself returned its Suspend), so the dynamic
                    // entry genuinely exists by the time this write fires.
                    ResumeArg::Start => Suspend::Delay(1),
                    ResumeArg::TimeElapsed => {
                        ops.commit_blocking_write(self.cell, narrow(1));
                        Suspend::Done
                    }
                    other => panic!("WriterProcess must only see Start/TimeElapsed, got {other:?}"),
                }
            }
        }

        let cell = StateCellId(42);
        let mut engine = Engine::new(
            Box::new(TestArena::seeded(cell, narrow(0))),
            Box::new(TestIoSink::default()),
        );
        engine.register_process(
            Box::new(WaiterProcess { cell }),
            ProcessKind::Initial,
            None,
            0,
        );
        engine.register_process(
            Box::new(WriterProcess { cell }),
            ProcessKind::Initial,
            None,
            0,
        );

        let result = engine.run_until(None);
        assert!(matches!(result, RunResult::Completed));

        assert_eq!(
            engine.test_sensitivity_entry_count(),
            0,
            "the dynamic entry must be removed once the waiting process resumes via \
             EventFired -- sensitivity.rs's own doc comment: dynamic entries \"are gone \
             from the table the instant the owning process resumes\" -- an earlier draft \
             never called SensitivityTable::remove_dynamic anywhere in Engine's dispatch \
             path, leaking this row forever"
        );
    }

    /// Mutation-test kill: `dispatch_ready_set`'s dynamic-wait-resumed
    /// removal set names two `ResumeArg` variants, `EventFired` and
    /// `Disabled` -- but no other test in this file ever drives a
    /// `ResumeArg::Disabled` activation through `dispatch_ready_set` (no
    /// production path in this crate constructs one yet), so a mutant that
    /// narrowed the removal check to `EventFired` alone would pass every
    /// other test. This test dispatches `Start` normally (installing one
    /// dynamic sensitivity entry via `Suspend::WaitEvents`), then hand-builds
    /// a `ResumeArg::Disabled` activation for the same process and dispatches
    /// it directly -- `dispatch_ready_set`/`drain_and_sort` are private
    /// methods on `Engine`, reachable here since this test module is a
    /// descendant of `control`.
    #[test]
    fn dispatch_ready_set_removes_dynamic_sensitivity_entry_for_disabled_resume_too() {
        struct WaiterProcess {
            cell: StateCellId,
        }
        impl CompiledProcess for WaiterProcess {
            fn resume(
                &mut self,
                _state: SimHandle,
                arg: ResumeArg,
                _ops: &dyn SchedulerOps,
            ) -> Suspend {
                match arg {
                    ResumeArg::Start => Suspend::WaitEvents(EventWaitSet {
                        terms: vec![EventTerm::Edge {
                            cell: self.cell,
                            edge: EdgeKind::Posedge,
                            iff: None,
                        }],
                        ..Default::default()
                    }),
                    _ => Suspend::Done,
                }
            }
        }

        let cell = StateCellId(99);
        let mut engine = new_engine();
        let proc = engine.register_process(
            Box::new(WaiterProcess { cell }),
            ProcessKind::Initial,
            None,
            0,
        );

        let ready = engine.drain_and_sort(Region::Active);
        engine.dispatch_ready_set(Region::Active, ready);
        assert_eq!(
            engine.test_sensitivity_entry_count(),
            1,
            "Suspend::WaitEvents must install exactly one dynamic sensitivity entry"
        );

        let disabled_act = PendingActivation {
            id: 999,
            proc,
            arg: ResumeArg::Disabled,
            reconv_iter: 0,
            logical_order: (engine.now(), Region::Active, 0),
        };
        engine.dispatch_ready_set(Region::Active, vec![disabled_act]);

        assert_eq!(
            engine.test_sensitivity_entry_count(),
            0,
            "a ResumeArg::Disabled resume must also remove its process's dynamic \
             sensitivity entry, exactly like EventFired does -- a mutant narrowing the \
             removal check to EventFired alone would leave this row (and every future \
             Disabled-terminated wait's row) leaked forever"
        );
    }

    /// Review fix — `apply_suspend`'s `Suspend::ForkJoin` arm spawned
    /// children via `ResumeDriver::handle_fork` but nothing in `Engine`
    /// ever called `ResumeDriver::on_child_done` when a spawned child later
    /// completed, so the parent was never reactivated: fork/join was
    /// completely nonfunctional (any forking process silently hung forever
    /// while `run_until` still reported normal completion). Drives a real
    /// `JoinKind::All` fork with one child through `Engine::run_until` and
    /// asserts the parent actually resumes via `ResumeArg::ForkChildDone`.
    #[test]
    fn fork_join_all_actually_resumes_the_parent_when_its_one_child_completes() {
        struct ParentProcess {
            log: Arc<StdMutex<Vec<&'static str>>>,
        }
        impl CompiledProcess for ParentProcess {
            fn resume(
                &mut self,
                _state: SimHandle,
                arg: ResumeArg,
                _ops: &dyn SchedulerOps,
            ) -> Suspend {
                match arg {
                    ResumeArg::Start => {
                        self.log.lock().unwrap().push("forked");
                        Suspend::ForkJoin(
                            JoinKind::All,
                            vec![SpawnSpec {
                                kernel: 0,
                                entry_pc: 0,
                                parent: 0,
                                disable_name: Vec::new(),
                            }],
                        )
                    }
                    ResumeArg::ForkChildDone { .. } => {
                        self.log.lock().unwrap().push("joined");
                        Suspend::Done
                    }
                    other => panic!(
                        "ParentProcess must only ever be resumed with Start or \
                         ForkChildDone, got {other:?}"
                    ),
                }
            }
        }

        let log = Arc::new(StdMutex::new(Vec::new()));
        let mut engine = new_engine();
        engine.register_process(
            Box::new(ParentProcess { log: log.clone() }),
            ProcessKind::Initial,
            None,
            0,
        );

        let result = engine.run_until(None);
        assert!(matches!(result, RunResult::Completed));

        assert_eq!(
            log.lock().unwrap().as_slice(),
            &["forked", "joined"],
            "the parent must actually be reactivated via ResumeArg::ForkChildDone once its \
             one child (an UnmintedForkChild, immediately Done) completes -- an earlier \
             draft never wired ResumeDriver::on_child_done into Engine at all, so the \
             parent would silently hang forever (never reach \"joined\") despite run_until \
             still reporting RunResult::Completed"
        );
    }

    // --- Task 7 (B-sched wave, ledger SC-062, SC-115) ---

    /// SC-062's own literal acceptance, proven through the real `Engine`
    /// dispatch loop (not just `ResumeDriver` in isolation, which
    /// `resume.rs`'s own `join_all_resumes_parent_only_after_every_child_
    /// reports_done`/`on_child_done_reclaims_the_childs_slot_when_the_join_
    /// is_satisfied` already cover): a `JoinAll` fork of 3 children (each an
    /// `UnmintedForkChild`, immediately `Done` on its own first dispatch --
    /// this wave's own documented interim placeholder, see this module's
    /// doc comment deviation 3) completing one after another *within the
    /// same Active-region dispatch pass*. Asserts the parent resumes (logs
    /// "joined") only once, after all 3, and that every one of the 3
    /// forked children's own `ProcessId`s is unreachable via
    /// `Engine::test_process_exists` by the time it does. Fails if slot
    /// reclaim is never wired (every child id would still report `true`),
    /// or if it reclaims the wrong process (e.g. the parent).
    #[test]
    fn sc062_joinall_child_returning_done_satisfies_the_parents_join_and_reclaims_its_slot() {
        struct ParentProcess {
            log: Arc<StdMutex<Vec<&'static str>>>,
        }
        impl CompiledProcess for ParentProcess {
            fn resume(
                &mut self,
                _state: SimHandle,
                arg: ResumeArg,
                _ops: &dyn SchedulerOps,
            ) -> Suspend {
                match arg {
                    ResumeArg::Start => {
                        self.log.lock().unwrap().push("forked");
                        Suspend::ForkJoin(
                            JoinKind::All,
                            vec![
                                SpawnSpec {
                                    kernel: 0,
                                    entry_pc: 0,
                                    parent: 0,
                                    disable_name: Vec::new(),
                                },
                                SpawnSpec {
                                    kernel: 1,
                                    entry_pc: 0,
                                    parent: 0,
                                    disable_name: Vec::new(),
                                },
                                SpawnSpec {
                                    kernel: 2,
                                    entry_pc: 0,
                                    parent: 0,
                                    disable_name: Vec::new(),
                                },
                            ],
                        )
                    }
                    ResumeArg::ForkChildDone { .. } => {
                        self.log.lock().unwrap().push("joined");
                        Suspend::Done
                    }
                    other => panic!(
                        "ParentProcess must only ever be resumed with Start or \
                         ForkChildDone, got {other:?}"
                    ),
                }
            }
        }

        let log = Arc::new(StdMutex::new(Vec::new()));
        let mut engine = new_engine();
        let parent = engine.register_process(
            Box::new(ParentProcess { log: log.clone() }),
            ProcessKind::Initial,
            None,
            0,
        );
        assert_eq!(parent, 0, "test assumes a fresh engine's first id is 0");

        let result = engine.run_until(None);
        assert!(matches!(result, RunResult::Completed));
        assert_eq!(
            log.lock().unwrap().as_slice(),
            &["forked", "joined"],
            "the parent must resume via ForkChildDone exactly once, only after all 3 \
             children have completed"
        );

        for child in (parent + 1)..=(parent + 3) {
            assert!(
                !engine.test_process_exists(child),
                "fork child {child} must be unreachable via the process table once its \
                 Suspend::Done has been processed and the JoinAll it belongs to has \
                 been satisfied -- SC-062's own literal 'the child's slot is reclaimed'"
            );
        }
    }

    /// SC-115's own literal acceptance ("a process can be spawned via fork
    /// after sim_init has run, with its own scheduled entry in the
    /// ready-set machinery") combined with this task's own `JoinKind::None`
    /// resolution (this module's doc comment, deviation 3): the parent's
    /// own next segment must run *synchronously*, inside the very same
    /// `apply_suspend` call that registered the fork's children -- no
    /// second `dispatch_ready_set`/`run_until` pass needed -- while the
    /// spawned child is left genuinely live (not dispatched, not
    /// reclaimed) at that exact point, proving `JoinKind::None` neither
    /// blocks on nor otherwise touches its child.
    #[test]
    fn sc115_join_none_children_run_detached_while_the_parent_proceeds_without_waiting() {
        struct ParentProcess {
            log: Arc<StdMutex<Vec<&'static str>>>,
        }
        impl CompiledProcess for ParentProcess {
            fn resume(
                &mut self,
                _state: SimHandle,
                arg: ResumeArg,
                _ops: &dyn SchedulerOps,
            ) -> Suspend {
                match arg {
                    ResumeArg::Start => {
                        self.log.lock().unwrap().push("forked");
                        Suspend::ForkJoin(
                            JoinKind::None,
                            vec![SpawnSpec {
                                kernel: 0,
                                entry_pc: 0,
                                parent: 0,
                                disable_name: Vec::new(),
                            }],
                        )
                    }
                    ResumeArg::TimeElapsed => {
                        self.log.lock().unwrap().push("proceeded");
                        Suspend::Done
                    }
                    other => panic!(
                        "ParentProcess must only ever be resumed with Start or \
                         TimeElapsed (JoinKind::None's own chosen continuation arg, \
                         see this module's doc comment), got {other:?}"
                    ),
                }
            }
        }

        let log = Arc::new(StdMutex::new(Vec::new()));
        let mut engine = new_engine();
        let parent = engine.register_process(
            Box::new(ParentProcess { log: log.clone() }),
            ProcessKind::Initial,
            None,
            0,
        );
        assert_eq!(parent, 0, "test assumes a fresh engine's first id is 0");
        let child = parent + 1;

        // `register_process` already queued the parent's own Start
        // activation onto Region::Active. Dispatching exactly that one
        // ready-set (a single `dispatch_ready_set` call) must be enough to
        // see BOTH "forked" and "proceeded" in the log -- proving
        // JoinKind::None's continuation happened inside this same call,
        // never requiring a second dispatch pass the way waiting for a
        // real ForkChildDone would.
        let ready = engine.drain_and_sort(Region::Active);
        engine.dispatch_ready_set(Region::Active, ready);

        assert_eq!(
            log.lock().unwrap().as_slice(),
            &["forked", "proceeded"],
            "JoinKind::None must resume the parent's own next segment synchronously, \
             within the very same apply_suspend call that registered its child -- \
             never leaving it suspended waiting for a ForkChildDone that will never \
             come (the pre-Task-7 bug this row closes)"
        );
        assert!(
            engine.test_process_exists(child),
            "the spawned child must still be a live, registered process at this exact \
             point -- it was only just enqueued (Region::Active, ResumeArg::Start) by \
             this same dispatch_ready_set call, never itself dispatched by it, so \
             JoinKind::None resolving the parent must not have touched the child at \
             all (neither reclaiming it nor otherwise depending on its completion)"
        );
    }

    /// R-run Task 17 review fix (r1) -- blocking finding: `apply_suspend`'s
    /// `Suspend::ForkJoin` arm's `mint_child` closure ignores its own
    /// `SpawnSpec` argument entirely and unconditionally builds an
    /// `UnmintedForkChild` (this module's own doc comment deviation 3;
    /// `sv-sim-codegen`'s R-run Task 17 now produces real, distinct
    /// `SpawnSpec::kernel` indices per branch, but nothing in `Engine`
    /// consumes that field yet -- Task 21 remains chartered to fix that).
    /// No prior test pinned this AS a tested property -- every existing
    /// `Suspend::ForkJoin` test above happens to use small/repeating
    /// `kernel` values (0, 1, 2) that a partial, buggy future minter could
    /// special-case and still pass. This test spawns children with
    /// deliberately large, mutually distinct `kernel` indices (a shape no
    /// real per-program kernel table plausibly has entries for yet) and
    /// proves every single one still completes via a bare `Suspend::Done`
    /// on its own very first `Start` dispatch -- the literal, observable
    /// signature of `UnmintedForkChild` (a real compiled branch kernel
    /// could legitimately suspend, loop, or otherwise not complete in one
    /// dispatch). Fails loudly the instant a future minter starts
    /// consulting `SpawnSpec::kernel` for any subset of values without this
    /// test being updated alongside it.
    #[test]
    fn apply_suspend_fork_join_still_mints_unmintedforkchild_regardless_of_spawnspec_kernel() {
        struct ParentProcess;
        impl CompiledProcess for ParentProcess {
            fn resume(
                &mut self,
                _state: SimHandle,
                arg: ResumeArg,
                _ops: &dyn SchedulerOps,
            ) -> Suspend {
                match arg {
                    ResumeArg::Start => Suspend::ForkJoin(
                        JoinKind::All,
                        vec![
                            SpawnSpec {
                                kernel: 424_242,
                                entry_pc: 0,
                                parent: 0,
                                disable_name: Vec::new(),
                            },
                            SpawnSpec {
                                kernel: 7,
                                entry_pc: 3,
                                parent: 0,
                                disable_name: Vec::new(),
                            },
                            SpawnSpec {
                                kernel: u32::MAX,
                                entry_pc: 0,
                                parent: 0,
                                disable_name: Vec::new(),
                            },
                        ],
                    ),
                    // Every child completes in one dispatch (proven below),
                    // so the join is satisfied by the time this fires; the
                    // real assertions are about the children, not this arm.
                    ResumeArg::ForkChildDone { .. } => Suspend::Done,
                    other => panic!(
                        "ParentProcess must only ever see Start or ForkChildDone, got {other:?}"
                    ),
                }
            }
        }

        let mut engine = new_engine();
        let parent =
            engine.register_process(Box::new(ParentProcess), ProcessKind::Initial, None, 0);
        assert_eq!(parent, 0, "test assumes a fresh engine's first id is 0");

        // Dispatch the parent's own Start -- this is the call that runs
        // apply_suspend's Suspend::ForkJoin arm and mints the 3 children.
        let ready = engine.drain_and_sort(Region::Active);
        engine.dispatch_ready_set(Region::Active, ready);

        let children = [parent + 1, parent + 2, parent + 3];
        for &child in &children {
            assert!(
                engine.test_process_exists(child),
                "child {child} must exist right after being minted, before its own first \
                 dispatch"
            );
        }

        // Dispatch every child's own queued Start activation in one pass.
        let ready = engine.drain_and_sort(Region::Active);
        assert_eq!(
            ready.len(),
            3,
            "exactly the 3 freshly-minted children's Start activations must be ready, got {}",
            ready.len()
        );
        engine.dispatch_ready_set(Region::Active, ready);

        for &child in &children {
            assert!(
                !engine.test_process_exists(child),
                "child {child} (SpawnSpec::kernel ignored by today's mint_child) must have \
                 completed via a bare Suspend::Done on its own very first Start dispatch and \
                 been reclaimed -- exactly UnmintedForkChild's own documented behavior, \
                 regardless of how large or distinct its own SpawnSpec::kernel index was. A \
                 real per-branch compiled kernel is not guaranteed to complete in one dispatch \
                 at all, so this failing is the signal that mint_child has started consulting \
                 SpawnSpec::kernel for at least one of these values -- update this test (and, \
                 per Task 21's own charter, the CLI-level fork/join gate) alongside that change."
            );
        }
    }

    // --- Task 9 (A2-part2, CG-102): region_id-consuming partition-aware
    // dispatch ---------------------------------------------------------

    /// `partition_of` genuinely reads `region_id`, not a hardcoded `0` --
    /// real modulo arithmetic against `partition_count`.
    #[test]
    fn partition_of_is_a_real_modulo_not_a_hardcoded_zero() {
        assert_eq!(partition_of(5, 4), PartitionId(1));
        assert_eq!(partition_of(6, 4), PartitionId(2));
        assert_eq!(partition_of(0, 4), PartitionId(0));
    }

    /// v1's one real call site always passes `partition_count = 1`
    /// (ARCHITECTURE §4 item 1) -- pinned here as a property of the
    /// function itself, for several distinct `region_id` values, not an
    /// assumption left implicit at the call site.
    #[test]
    fn partition_of_is_the_documented_v1_degenerate_single_partition_value() {
        for region_id in [0, 3, 9] {
            assert_eq!(
                partition_of(region_id, 1),
                PartitionId(0),
                "region_id={region_id}: v1's real partition_count=1 call site must always \
                 yield PartitionId(0), for any region_id"
            );
        }
    }

    /// `partition_of` never divides by zero -- `partition_count == 0`
    /// degrades to the same single-partition behavior as `1`, rather than
    /// panicking.
    #[test]
    fn partition_of_with_zero_partition_count_does_not_panic() {
        assert_eq!(partition_of(7, 0), PartitionId(0));
    }

    /// CG-102's scheduler-side literal acceptance ("consumed by the
    /// scheduler for partition-aware dispatch"): register 3 processes with
    /// distinct `region_id`s (0, 3, 9) via the real `SchedulerControl::
    /// register_process` API, dispatch their initial-registration
    /// activations, and confirm the spy hook saw every one of their
    /// `ProcessId`s queried against exactly the `region_id` each was
    /// registered with -- proving the field flows from registration
    /// through to dispatch rather than being stored-but-never-read (the
    /// literal gap the plan's own "Verified live" review finding named),
    /// and would fail if `dispatch_ready_set` were refactored to default
    /// every lookup to `0` regardless of what was registered.
    #[test]
    fn dispatch_ready_set_looks_up_partition_of_using_each_readied_processs_own_stored_region_id() {
        let mut engine = new_engine();
        let p0 =
            engine.register_process(Box::new(UnmintedForkChild), ProcessKind::Initial, None, 0);
        let p3 =
            engine.register_process(Box::new(UnmintedForkChild), ProcessKind::Initial, None, 3);
        let p9 =
            engine.register_process(Box::new(UnmintedForkChild), ProcessKind::Initial, None, 9);

        // Each `register_process` above already queued its own Start
        // activation into Region::Active -- drive exactly one dispatch pass
        // over all 3 so `dispatch_ready_set` genuinely runs its grouping
        // loop once, over all 3 activations together.
        let ready = engine.drain_and_sort(Region::Active);
        assert_eq!(
            ready.len(),
            3,
            "all 3 registrations' Start activations must be pending"
        );
        engine.dispatch_ready_set(Region::Active, ready);

        let mut seen = engine.region_id_queries_seen().to_vec();
        seen.sort_unstable();
        let mut want = vec![(p0, 0u32), (p3, 3u32), (p9, 9u32)];
        want.sort_unstable();
        assert_eq!(
            seen, want,
            "dispatch_ready_set must query ProcessTable::region_id for every readied \
             process, and every value returned must exactly match what register_process \
             stored for it"
        );
    }

    // --- Task 3 (B-sched, ledger SC-004, SC-005, SC-052, SC-053):
    // EventTerm::Repeat engine semantics + ##N/clocking-block runtime
    // resolution ---

    /// SC-052's own literal acceptance, verified against the production
    /// `Engine` dispatch loop (not a hand-simulated one): a
    /// `repeat(3)@(posedge clk)`-shaped mock process must resume only after
    /// the 3rd matching posedge, never the 1st or 2nd, and exactly once
    /// after that (not once per remaining posedge -- which would mean a
    /// `RepeatOutcome::StillWaiting` firing wrongly fell through to
    /// enqueue).
    #[test]
    fn engine_resumes_a_repeat_3_process_only_on_the_third_matching_edge_not_the_first() {
        struct Repeat3Waiter {
            cell: StateCellId,
            resume_log: Arc<StdMutex<Vec<ResumeArg>>>,
        }
        impl CompiledProcess for Repeat3Waiter {
            fn resume(
                &mut self,
                _state: SimHandle,
                arg: ResumeArg,
                _ops: &dyn SchedulerOps,
            ) -> Suspend {
                self.resume_log.lock().unwrap().push(arg);
                match arg {
                    ResumeArg::Start => Suspend::WaitEvents(EventWaitSet {
                        terms: vec![EventTerm::Repeat {
                            count: 3,
                            inner: Box::new(EventTerm::Edge {
                                cell: self.cell,
                                edge: EdgeKind::Posedge,
                                iff: None,
                            }),
                        }],
                        ..Default::default()
                    }),
                    _ => Suspend::Done,
                }
            }
        }

        let cell = StateCellId(200);
        let mut engine = Engine::new(
            Box::new(TestArena::seeded(cell, narrow(0))),
            Box::new(TestIoSink::default()),
        );
        let log: Arc<StdMutex<Vec<ResumeArg>>> = Arc::new(StdMutex::new(Vec::new()));
        engine.register_process(
            Box::new(Repeat3Waiter {
                cell,
                resume_log: log.clone(),
            }),
            ProcessKind::Initial,
            None,
            0,
        );

        // Dispatch Start: installs the Repeat{count:3, inner: Edge} dynamic
        // wait.
        let ready = engine.drain_and_sort(Region::Active);
        engine.dispatch_ready_set(Region::Active, ready);
        assert_eq!(
            log.lock().unwrap().len(),
            1,
            "only Start's own resume() call must have happened so far"
        );

        let write = |engine: &Engine, value: FourStateValue| {
            let ops = engine.test_dispatch_ops(engine.now(), Region::Active, 0);
            ops.commit_blocking_write(cell, value);
        };

        // Posedge 1 (0 -> 1): StillWaiting -- must NOT enqueue/resume.
        write(&engine, narrow(1));
        let ready = engine.drain_and_sort(Region::Active);
        engine.dispatch_ready_set(Region::Active, ready);
        assert_eq!(
            log.lock().unwrap().len(),
            1,
            "the 1st matching posedge must not resume a repeat(3) waiter"
        );

        // Negedge (1 -> 0): does not match a Posedge-qualified dynamic entry
        // at all -- irrelevant filler, mirroring an SV clock's real toggle
        // pattern.
        write(&engine, narrow(0));

        // Posedge 2 (0 -> 1): StillWaiting again -- must still NOT resume.
        write(&engine, narrow(1));
        let ready = engine.drain_and_sort(Region::Active);
        engine.dispatch_ready_set(Region::Active, ready);
        assert_eq!(
            log.lock().unwrap().len(),
            1,
            "the 2nd matching posedge must not resume a repeat(3) waiter either"
        );

        write(&engine, narrow(0));

        // Posedge 3 (0 -> 1): Satisfied -- must resume exactly now.
        write(&engine, narrow(1));
        let ready = engine.drain_and_sort(Region::Active);
        engine.dispatch_ready_set(Region::Active, ready);
        {
            let log = log.lock().unwrap();
            assert_eq!(
                log.len(),
                2,
                "the 3rd matching posedge must resume the waiter exactly once"
            );
            assert!(
                matches!(log[1], ResumeArg::EventFired { .. }),
                "the resuming call's ResumeArg must be EventFired, got {:?}",
                log[1]
            );
        }
        assert_eq!(
            engine.test_sensitivity_entry_count(),
            0,
            "the dynamic entry (and its RepeatCounters row) must be gone once satisfied"
        );

        // Two more posedges after the process has already resumed and
        // returned Suspend::Done (no re-arm): must produce no further
        // resume() calls at all -- proves a StillWaiting firing never falls
        // through to enqueue, and that a satisfied repeat's row is not
        // somehow still alive to double-fire.
        write(&engine, narrow(0));
        write(&engine, narrow(1));
        write(&engine, narrow(0));
        write(&engine, narrow(1));
        let ready = engine.drain_and_sort(Region::Active);
        engine.dispatch_ready_set(Region::Active, ready);
        assert_eq!(
            log.lock().unwrap().len(),
            2,
            "no resume() call must happen for any posedge after the repeat(3) waiter \
             has already resumed and completed -- exactly once total after the 3rd \
             posedge, never once per remaining posedge"
        );
    }

    /// SC-053's own differentiating acceptance clause: "the same compiled
    /// `##N` delay produces different absolute tick timing when its
    /// associated clock's period changes dynamically at runtime" -- proven
    /// here at the scheduler level by driving a deliberately IRREGULAR
    /// (non-periodic) clock through a `repeat(2)@(posedge clk)`-shaped mock
    /// process and asserting it resumes at the clock's real 2nd posedge tick
    /// (12), never a precomputed `3 + 2*period`-style absolute tick a wrong
    /// "optimize by caching a tick offset" implementation would produce.
    #[test]
    fn sc053_repeat_based_wait_re_reads_the_clocks_live_period_not_a_cached_tick_offset() {
        #[derive(Clone, Copy, Debug, PartialEq, Eq)]
        enum LogTag {
            Start,
            EventFired,
        }

        struct RepeatWaiter {
            cell: StateCellId,
            log: Arc<StdMutex<Vec<(LogTag, SchedTime)>>>,
        }
        impl CompiledProcess for RepeatWaiter {
            fn resume(
                &mut self,
                _state: SimHandle,
                arg: ResumeArg,
                ops: &dyn SchedulerOps,
            ) -> Suspend {
                match arg {
                    ResumeArg::Start => {
                        self.log.lock().unwrap().push((LogTag::Start, ops.now()));
                        Suspend::WaitEvents(EventWaitSet {
                            terms: vec![EventTerm::Repeat {
                                count: 2,
                                inner: Box::new(EventTerm::Edge {
                                    cell: self.cell,
                                    edge: EdgeKind::Posedge,
                                    iff: None,
                                }),
                            }],
                            ..Default::default()
                        })
                    }
                    ResumeArg::EventFired { .. } => {
                        self.log
                            .lock()
                            .unwrap()
                            .push((LogTag::EventFired, ops.now()));
                        Suspend::Done
                    }
                    other => panic!("RepeatWaiter must only see Start/EventFired, got {other:?}"),
                }
            }
        }

        // Irregular clock: a real signal (not a "##N" literal) whose edges
        // land at ticks 3 (posedge), 5 (negedge, irrelevant filler), and 12
        // (posedge) -- deliberately never landing on a fixed period, so any
        // implementation that tried to precompute
        // `first_posedge_tick + (count - 1) * observed_period` would land on
        // a wrong tick (3 + 1*2 = 5, which is a negedge, not even a 2nd
        // posedge) instead of the real 2nd posedge at 12.
        struct IrregularClockWriter {
            cell: StateCellId,
            step: u32,
        }
        impl CompiledProcess for IrregularClockWriter {
            fn resume(
                &mut self,
                _state: SimHandle,
                arg: ResumeArg,
                ops: &dyn SchedulerOps,
            ) -> Suspend {
                match (self.step, arg) {
                    (0, ResumeArg::Start) => {
                        self.step = 1;
                        Suspend::Delay(3) // fires at tick 3
                    }
                    (1, ResumeArg::TimeElapsed) => {
                        ops.commit_blocking_write(self.cell, narrow(1)); // posedge @ 3
                        self.step = 2;
                        Suspend::Delay(2) // fires at tick 5
                    }
                    (2, ResumeArg::TimeElapsed) => {
                        ops.commit_blocking_write(self.cell, narrow(0)); // negedge @ 5
                        self.step = 3;
                        Suspend::Delay(7) // fires at tick 12
                    }
                    (3, ResumeArg::TimeElapsed) => {
                        ops.commit_blocking_write(self.cell, narrow(1)); // posedge @ 12
                        self.step = 4;
                        Suspend::Done
                    }
                    (step, other) => {
                        panic!("IrregularClockWriter unexpected (step={step}, arg={other:?})")
                    }
                }
            }
        }

        let cell = StateCellId(201);
        let mut engine = Engine::new(
            Box::new(TestArena::seeded(cell, narrow(0))),
            Box::new(TestIoSink::default()),
        );
        let log: Arc<StdMutex<Vec<(LogTag, SchedTime)>>> = Arc::new(StdMutex::new(Vec::new()));
        engine.register_process(
            Box::new(RepeatWaiter {
                cell,
                log: log.clone(),
            }),
            ProcessKind::Initial,
            None,
            0,
        );
        engine.register_process(
            Box::new(IrregularClockWriter { cell, step: 0 }),
            ProcessKind::Initial,
            None,
            0,
        );

        let result = engine.run_until(None);
        assert!(matches!(result, RunResult::Completed));

        let log = log.lock().unwrap();
        assert_eq!(
            *log,
            vec![
                (LogTag::Start, SchedTime { ticks: 0 }),
                (LogTag::EventFired, SchedTime { ticks: 12 }),
            ],
            "the repeat(2) waiter must resume exactly at tick 12 (the clock's real 2nd \
             posedge), never a precomputed offset from a cached period"
        );
    }

    // --- Task 4 (B-sched, ledger SC-090, SC-123, SC-020 scheduler-owned
    // half): Postponed/Reactive region-boundary placement: region
    // callbacks + `1step` ---

    /// SC-090's own literal acceptance: a `$strobe`/`$monitor`-shaped
    /// `Region::Postponed` callback sampling a cell written twice within
    /// one timestep's delta loop must read only the FINAL, post-delta-
    /// convergence value, and must fire exactly once for that timestep --
    /// never once per intermediate write (which would mean it ran
    /// interleaved with the delta loop, not strictly after it converged).
    #[test]
    fn postponed_callback_fires_exactly_once_and_only_after_delta_convergence() {
        struct TwiceWritingProcess {
            cell: StateCellId,
        }
        impl CompiledProcess for TwiceWritingProcess {
            fn resume(
                &mut self,
                _state: SimHandle,
                arg: ResumeArg,
                ops: &dyn SchedulerOps,
            ) -> Suspend {
                match arg {
                    ResumeArg::Start => {
                        ops.commit_blocking_write(self.cell, narrow(0));
                        Suspend::Delay(0)
                    }
                    ResumeArg::TimeElapsed => {
                        ops.commit_blocking_write(self.cell, narrow(1));
                        Suspend::Done
                    }
                    other => panic!("TwiceWritingProcess unexpected arg {other:?}"),
                }
            }
        }

        struct ReadCellCallback {
            cell: StateCellId,
            log: Arc<StdMutex<Vec<FourStateValue>>>,
        }
        impl RegionCallback for ReadCellCallback {
            fn invoke(&mut self, ctx: &dyn SchedulerCtx) {
                self.log.lock().unwrap().push(ctx.read(self.cell));
            }
        }

        let cell = StateCellId(300);
        let mut engine = Engine::new(
            Box::new(TestArena::seeded(cell, narrow(9))),
            Box::new(TestIoSink::default()),
        );
        let seen: Arc<StdMutex<Vec<FourStateValue>>> = Arc::new(StdMutex::new(Vec::new()));
        {
            let ops = engine.test_dispatch_ops(engine.now(), Region::Active, 0);
            ops.register_region_callback(
                Region::Postponed,
                engine.now(),
                Box::new(ReadCellCallback {
                    cell,
                    log: seen.clone(),
                }),
            );
        }
        engine.register_process(
            Box::new(TwiceWritingProcess { cell }),
            ProcessKind::Initial,
            None,
            0,
        );

        let result = engine.run_until(None);
        assert!(matches!(result, RunResult::Completed));

        let recorded = seen.lock().unwrap();
        assert_eq!(
            recorded.len(),
            1,
            "the Postponed callback must fire exactly once for the one timestep this \
             run has, not once per intermediate write inside the delta loop"
        );
        assert!(
            matches!(recorded[0], FourStateValue::Narrow(1)),
            "must read the FINAL, post-delta-convergence value (1), never the \
             intermediate write (0) or the pre-run seed (9); got {:?}",
            recorded[0]
        );
    }

    /// Region ordering plus registration-order-per-region: a `Region::
    /// Reactive` callback and a `Region::Postponed` callback registered for
    /// the same `now` must both fire exactly once, Reactive strictly before
    /// Postponed (the same region ordering `run_until` already enforces for
    /// ordinary process dispatch), each observing the same final settled
    /// value -- proving the two regions' callback drains don't share state
    /// that lets one suppress or corrupt the other.
    #[test]
    fn reactive_and_postponed_callbacks_at_the_same_time_do_not_interfere() {
        struct OneShotWriter {
            cell: StateCellId,
        }
        impl CompiledProcess for OneShotWriter {
            fn resume(
                &mut self,
                _state: SimHandle,
                _arg: ResumeArg,
                ops: &dyn SchedulerOps,
            ) -> Suspend {
                ops.commit_blocking_write(self.cell, narrow(7));
                Suspend::Done
            }
        }

        struct TaggedReadCallback {
            tag: &'static str,
            cell: StateCellId,
            log: Arc<StdMutex<Vec<(&'static str, FourStateValue)>>>,
        }
        impl RegionCallback for TaggedReadCallback {
            fn invoke(&mut self, ctx: &dyn SchedulerCtx) {
                self.log
                    .lock()
                    .unwrap()
                    .push((self.tag, ctx.read(self.cell)));
            }
        }

        let cell = StateCellId(301);
        let mut engine = Engine::new(
            Box::new(TestArena::seeded(cell, narrow(0))),
            Box::new(TestIoSink::default()),
        );
        let log: Arc<StdMutex<Vec<(&'static str, FourStateValue)>>> =
            Arc::new(StdMutex::new(Vec::new()));
        {
            let ops = engine.test_dispatch_ops(engine.now(), Region::Active, 0);
            ops.register_region_callback(
                Region::Reactive,
                engine.now(),
                Box::new(TaggedReadCallback {
                    tag: "Reactive",
                    cell,
                    log: log.clone(),
                }),
            );
            ops.register_region_callback(
                Region::Postponed,
                engine.now(),
                Box::new(TaggedReadCallback {
                    tag: "Postponed",
                    cell,
                    log: log.clone(),
                }),
            );
        }
        engine.register_process(
            Box::new(OneShotWriter { cell }),
            ProcessKind::Initial,
            None,
            0,
        );

        let result = engine.run_until(None);
        assert!(matches!(result, RunResult::Completed));

        let recorded = log.lock().unwrap();
        assert_eq!(
            recorded.len(),
            2,
            "both callbacks must fire exactly once each"
        );
        assert_eq!(
            recorded[0].0, "Reactive",
            "Reactive's own callback must fire before Postponed's"
        );
        assert_eq!(recorded[1].0, "Postponed");
        assert!(matches!(recorded[0].1, FourStateValue::Narrow(7)));
        assert!(matches!(recorded[1].1, FourStateValue::Narrow(7)));
    }

    /// sva T2 (§1.4): `Region::Observed` is drained (via
    /// `drain_observed_and_check_settled`) strictly before `Region::
    /// Reactive`'s own dispatch/callback drain, within the SAME timestep --
    /// the reference algorithm's own stratified-region ordering (Clause
    /// 4.5), not merely "eventually, on some later timestep". Registers one
    /// tagged callback at each region (no process at all is needed: with
    /// nothing registered, Active/Inactive/Nba settle on `run_until`'s very
    /// first delta-loop pass, which is exactly the "genuinely idle" point
    /// `drain_observed_and_check_settled` is pinned to).
    #[test]
    fn observed_region_maturation_fires_before_reactive_same_timestep() {
        struct TagCallback {
            tag: &'static str,
            log: Arc<StdMutex<Vec<&'static str>>>,
        }
        impl RegionCallback for TagCallback {
            fn invoke(&mut self, _ctx: &dyn SchedulerCtx) {
                self.log.lock().unwrap().push(self.tag);
            }
        }

        let mut engine = new_engine();
        let log: Arc<StdMutex<Vec<&'static str>>> = Arc::new(StdMutex::new(Vec::new()));
        {
            let ops = engine.test_dispatch_ops(engine.now(), Region::Active, 0);
            ops.register_region_callback(
                Region::Reactive,
                engine.now(),
                Box::new(TagCallback {
                    tag: "Reactive",
                    log: log.clone(),
                }),
            );
            ops.register_region_callback(
                Region::Observed,
                engine.now(),
                Box::new(TagCallback {
                    tag: "Observed",
                    log: log.clone(),
                }),
            );
        }

        let result = engine.run_until(None);
        assert!(matches!(result, RunResult::Completed));

        let recorded = log.lock().unwrap();
        assert_eq!(
            *recorded,
            vec!["Observed", "Reactive"],
            "Observed must mature before Reactive within the same settled timestep"
        );
    }

    /// sva T2 (§1.4, review fix, blocking finding 2): a matured report an
    /// Observed callback schedules into `Region::Reactive` mid-drain (via
    /// `SchedulerOps::register_region_callback`, exactly as `sched::
    /// maturation::report` will for a real SVA pass/fail action, §1.7)
    /// must already be sitting in `region_callbacks` by the time the
    /// single-stepping debugger's OWN NEXT `step_region()` call reaches
    /// `Region::Reactive` -- not delayed to a later timestep, and not
    /// fired early from inside the Observed drain itself.
    #[test]
    fn observed_engine_step_result_visible_by_next_reactive_step() {
        struct TagCallback {
            tag: &'static str,
            log: Arc<StdMutex<Vec<&'static str>>>,
        }
        impl RegionCallback for TagCallback {
            fn invoke(&mut self, _ctx: &dyn SchedulerCtx) {
                self.log.lock().unwrap().push(self.tag);
            }
        }

        struct ObservedReportsIntoReactive {
            log: Arc<StdMutex<Vec<&'static str>>>,
        }
        impl RegionCallback for ObservedReportsIntoReactive {
            fn invoke(&mut self, _ctx: &dyn SchedulerCtx) {
                // Stands in for a matured SVA thread's `sv_rt_sva_report_
                // pass` (§1.7): the Observed-region engine step schedules a
                // Reactive action from inside its own callback, exactly as
                // `sched::maturation::report` does for every report kind.
                let log = self.log.clone();
                crate::sched::with_current_ops(move |ops| {
                    ops.register_region_callback(
                        Region::Reactive,
                        ops.now(),
                        Box::new(TagCallback {
                            tag: "matured",
                            log: log.clone(),
                        }),
                    );
                });
            }
        }

        let mut engine = new_engine();
        let log: Arc<StdMutex<Vec<&'static str>>> = Arc::new(StdMutex::new(Vec::new()));
        {
            let ops = engine.test_dispatch_ops(engine.now(), Region::Active, 0);
            ops.register_region_callback(
                Region::Observed,
                engine.now(),
                Box::new(ObservedReportsIntoReactive { log: log.clone() }),
            );
        }

        assert_eq!(engine.current_region(), Region::Active);
        engine.step_region(); // Active -> Inactive
        assert_eq!(engine.current_region(), Region::Inactive);
        engine.step_region(); // Inactive -> Nba
        assert_eq!(engine.current_region(), Region::Nba);
        engine.step_region(); // Nba settles, drains Observed (schedules
        // "matured" into Reactive), -> Reactive.
        assert_eq!(engine.current_region(), Region::Reactive);
        assert!(
            log.lock().unwrap().is_empty(),
            "the matured report must not fire before Reactive's own step runs"
        );

        engine.step_region(); // Reactive drains region callbacks -> Postponed.
        assert_eq!(engine.current_region(), Region::Postponed);
        assert_eq!(
            *log.lock().unwrap(),
            vec!["matured"],
            "the report registered mid-Observed-drain must already be visible to the very \
             next Reactive step"
        );
    }

    /// SC-123's pinned `1step` placement: a process suspended via `1step`
    /// timing registers itself as a `Region::Postponed` callback at the
    /// current `now` (this test's own mock process does so directly via
    /// `ops.register_region_callback`, standing in for the codegen
    /// lowering no track emits yet this wave -- see `region_callback`'s
    /// own module doc comment). Its continuation must run strictly after
    /// every ORDINARY `Region::Postponed` callback registered before it,
    /// never interleaved before them -- the exact placement a wrongly-
    /// modeled `Suspend::Delay(0)` (which lands in `Region::Inactive`,
    /// strictly *before* Postponed even runs) could never produce.
    #[test]
    fn sc123_1step_continuation_runs_after_every_ordinary_postponed_callback_at_the_same_tick() {
        struct TagCallback {
            tag: &'static str,
            log: Arc<StdMutex<Vec<&'static str>>>,
        }
        impl RegionCallback for TagCallback {
            fn invoke(&mut self, _ctx: &dyn SchedulerCtx) {
                self.log.lock().unwrap().push(self.tag);
            }
        }

        struct OneStepProcess {
            log: Arc<StdMutex<Vec<&'static str>>>,
        }
        impl CompiledProcess for OneStepProcess {
            fn resume(
                &mut self,
                _state: SimHandle,
                arg: ResumeArg,
                ops: &dyn SchedulerOps,
            ) -> Suspend {
                match arg {
                    ResumeArg::Start => Suspend::Delay(0),
                    ResumeArg::TimeElapsed => {
                        // Pinned placement (region_callback's own module
                        // doc): register as a Postponed callback at the
                        // current tick, rather than resuming again via a
                        // second Suspend::Delay(0) (which would land back
                        // in Inactive, the wrong side of Postponed).
                        ops.register_region_callback(
                            Region::Postponed,
                            ops.now(),
                            Box::new(TagCallback {
                                tag: "1step",
                                log: self.log.clone(),
                            }),
                        );
                        Suspend::Done
                    }
                    other => panic!("OneStepProcess unexpected arg {other:?}"),
                }
            }
        }

        let mut engine = new_engine();
        let log: Arc<StdMutex<Vec<&'static str>>> = Arc::new(StdMutex::new(Vec::new()));
        {
            let ops = engine.test_dispatch_ops(engine.now(), Region::Active, 0);
            ops.register_region_callback(
                Region::Postponed,
                engine.now(),
                Box::new(TagCallback {
                    tag: "A",
                    log: log.clone(),
                }),
            );
            ops.register_region_callback(
                Region::Postponed,
                engine.now(),
                Box::new(TagCallback {
                    tag: "B",
                    log: log.clone(),
                }),
            );
        }
        engine.register_process(
            Box::new(OneStepProcess { log: log.clone() }),
            ProcessKind::Initial,
            None,
            0,
        );

        let result = engine.run_until(None);
        assert!(matches!(result, RunResult::Completed));

        let recorded = log.lock().unwrap();
        assert_eq!(
            *recorded,
            vec!["A", "B", "1step"],
            "the 1step continuation must fire strictly after every ordinary \
             Postponed callback registered before it, never interleaved before them"
        );
    }

    /// Review fix (Task 4 r1): `Engine::push_region_callback` must reject
    /// every region `Engine::drain_region_callbacks` never drains
    /// (`Active`/`Inactive`/`Nba`) with a typed
    /// `Err(RegionCallbackRegionError::NotDrained(region))`, never silently
    /// accept it into `region_callbacks` where it would sit forever and
    /// never fire (that was the bug: no error, no panic, no log). Only
    /// `Reactive`/`Postponed` -- the two regions that method's own doc
    /// comment names -- may `Ok(())`.
    #[test]
    fn push_region_callback_rejects_every_region_drain_region_callbacks_never_drains() {
        struct NeverInvoked;
        impl RegionCallback for NeverInvoked {
            fn invoke(&mut self, _ctx: &dyn SchedulerCtx) {
                panic!("must never be invoked -- the registration itself must be rejected");
            }
        }

        let engine = new_engine();

        for &region in &[Region::Active, Region::Inactive, Region::Nba] {
            let result = engine.push_region_callback(engine.now(), region, Box::new(NeverInvoked));
            assert_eq!(
                result,
                Err(RegionCallbackRegionError::NotDrained(region)),
                "{region:?} is never drained by drain_region_callbacks and must be rejected, \
                 not silently accepted"
            );
        }

        // `NeverInvoked` registrations accepted just above are still sitting
        // in `engine.region_callbacks`, undrained (this test never calls
        // `run_until`/`drain_region_callbacks`) -- dropping `engine` here
        // drops them without invoking, so `NeverInvoked::invoke`'s panic
        // never fires. That drain-ordering/exactly-once behavior is already
        // covered end-to-end by `postponed_callback_fires_exactly_once_and_
        // only_after_delta_convergence` and
        // `reactive_and_postponed_callbacks_at_the_same_time_do_not_
        // interfere`; this test's own scope is `push_region_callback`'s
        // accept/reject decision alone.
        for &region in &[Region::Reactive, Region::Postponed] {
            let result = engine.push_region_callback(engine.now(), region, Box::new(NeverInvoked));
            assert_eq!(
                result,
                Ok(()),
                "{region:?} IS drained by drain_region_callbacks and must be accepted"
            );
        }
    }

    /// sva T2 (§1.4, review fix, blocking finding 2), Observed's half:
    /// `Region::Observed` joins `Reactive`/`Postponed` on the ACCEPTING
    /// side -- `drain_observed_and_check_settled` genuinely drains it now,
    /// so a registration for it must no longer be rejected.
    #[test]
    fn push_region_callback_accepts_observed() {
        struct NeverInvoked;
        impl RegionCallback for NeverInvoked {
            fn invoke(&mut self, _ctx: &dyn SchedulerCtx) {
                panic!("must never be invoked -- this test only checks the accept decision");
            }
        }

        let engine = new_engine();
        let result =
            engine.push_region_callback(engine.now(), Region::Observed, Box::new(NeverInvoked));
        assert_eq!(
            result,
            Ok(()),
            "Region::Observed IS now drained (by drain_observed_and_check_settled) and must \
             be accepted"
        );
    }

    /// sva T2 (§1.4, review fix, blocking finding 2): adding `Observed` to
    /// the accepting side must not loosen `Preponed`'s own rejection --
    /// `Preponed` stays a pure inline snapshot push, never a registered
    /// callback, even after `Observed` becomes real (the two Task 1/Task 2
    /// additions could easily have been conflated into one "any SVA
    /// region" acceptance by a careless edit; this proves they weren't).
    #[test]
    fn push_region_callback_still_rejects_preponed_after_observed_added() {
        struct NeverInvoked;
        impl RegionCallback for NeverInvoked {
            fn invoke(&mut self, _ctx: &dyn SchedulerCtx) {
                panic!("must never be invoked -- the registration itself must be rejected");
            }
        }

        let engine = new_engine();
        let result =
            engine.push_region_callback(engine.now(), Region::Preponed, Box::new(NeverInvoked));
        assert_eq!(
            result,
            Err(RegionCallbackRegionError::NotDrained(Region::Preponed)),
            "Region::Preponed must still be rejected -- it is a pure inline snapshot push, \
             never a registered callback, regardless of Observed's own addition"
        );
    }

    /// Review fix (Task 4 r1): the one production trait boundary,
    /// `EngineOps::register_region_callback` (`SchedulerOps`'s impl), must
    /// convert `RegionCallbackRegionError` into an immediate, loud panic --
    /// never a silent, forever-unfired registration -- because the trait
    /// method itself returns `()` and cannot propagate a `Result` (its
    /// signature is a cross-track contract many `#[cfg(test)]` mocks outside
    /// this crate also implement).
    #[test]
    #[should_panic(expected = "region must be Reactive, Postponed, or Observed")]
    fn register_region_callback_panics_loudly_for_an_undrained_region() {
        struct NeverInvoked;
        impl RegionCallback for NeverInvoked {
            fn invoke(&mut self, _ctx: &dyn SchedulerCtx) {
                panic!("must never be invoked -- the registration itself must be rejected");
            }
        }

        let engine = new_engine();
        let ops = engine.test_dispatch_ops(engine.now(), Region::Active, 0);
        ops.register_region_callback(Region::Active, engine.now(), Box::new(NeverInvoked));
    }

    // --- Task 5 (B-sched, ledger SC-065, SC-066, SC-047, SC-048, SC-068):
    // EventTerm/EventWaitSet shape + static/dynamic sensitivity closure ---

    /// SC-068's own literal acceptance, verified against the production
    /// `Engine` dispatch loop (not a hand-simulated one): `@(posedge clk iff
    /// en)` must not call `resume()` at all on a posedge where `en` reads
    /// false at wakeup-consideration time, and must resume it (with
    /// `ResumeArg::EventFired`) on the very next posedge where `en` reads
    /// true. `PredicateFn` is settled (`docs/CONTRACTS.md` D8) as a bare
    /// `extern "C" fn(SimHandle) -> bool`, produced per-callsite by codegen's
    /// `compile_iff_guard` -- it carries no `StateCellId` parameter of its
    /// own, so this mock predicate stands in for a compiled guard by reading
    /// a thread-local flag the test flips directly, exactly mirroring how a
    /// real compiled guard would read `en` through the (here, unused-by-the-
    /// mock) `SimHandle`. Fails if `EngineNotifySink::notify_change` ever
    /// enqueues/resumes without first consulting a `Some(iff)` row's
    /// predicate, or evaluates it backwards (blocking on true, allowing on
    /// false).
    #[test]
    fn engine_never_calls_resume_when_iff_is_false_but_does_on_the_next_true_edge() {
        thread_local! {
            static EN: std::cell::Cell<bool> = const { std::cell::Cell::new(false) };
        }
        extern "C" fn iff_reads_en(_state: SimHandle) -> bool {
            EN.with(|c| c.get())
        }

        struct IffWaiter {
            cell: StateCellId,
            resume_log: Arc<StdMutex<Vec<ResumeArg>>>,
        }
        impl CompiledProcess for IffWaiter {
            fn resume(
                &mut self,
                _state: SimHandle,
                arg: ResumeArg,
                _ops: &dyn SchedulerOps,
            ) -> Suspend {
                self.resume_log.lock().unwrap().push(arg);
                match arg {
                    ResumeArg::Start => Suspend::WaitEvents(EventWaitSet {
                        terms: vec![EventTerm::Edge {
                            cell: self.cell,
                            edge: EdgeKind::Posedge,
                            iff: Some(crate::process::PredicateFn(iff_reads_en)),
                        }],
                        ..Default::default()
                    }),
                    _ => Suspend::Done,
                }
            }
        }

        let cell = StateCellId(210);
        let mut engine = Engine::new(
            Box::new(TestArena::seeded(cell, narrow(0))),
            Box::new(TestIoSink::default()),
        );
        let log: Arc<StdMutex<Vec<ResumeArg>>> = Arc::new(StdMutex::new(Vec::new()));
        engine.register_process(
            Box::new(IffWaiter {
                cell,
                resume_log: log.clone(),
            }),
            ProcessKind::Initial,
            None,
            0,
        );

        // Dispatch Start: installs the iff-guarded Edge dynamic wait.
        let ready = engine.drain_and_sort(Region::Active);
        engine.dispatch_ready_set(Region::Active, ready);
        assert_eq!(
            log.lock().unwrap().len(),
            1,
            "only Start's own resume() call must have happened so far"
        );

        let write = |engine: &Engine, value: FourStateValue| {
            let ops = engine.test_dispatch_ops(engine.now(), Region::Active, 0);
            ops.commit_blocking_write(cell, value);
        };

        // en=0: a real posedge occurs, but the predicate reads false --
        // must NOT resume.
        EN.with(|c| c.set(false));
        write(&engine, narrow(1));
        let ready = engine.drain_and_sort(Region::Active);
        engine.dispatch_ready_set(Region::Active, ready);
        assert_eq!(
            log.lock().unwrap().len(),
            1,
            "a posedge with iff=false must not resume() the waiter at all"
        );

        write(&engine, narrow(0));

        // en=1: the next real posedge, predicate now reads true -- must
        // resume exactly once, with ResumeArg::EventFired.
        EN.with(|c| c.set(true));
        write(&engine, narrow(1));
        let ready = engine.drain_and_sort(Region::Active);
        engine.dispatch_ready_set(Region::Active, ready);
        {
            let log = log.lock().unwrap();
            assert_eq!(
                log.len(),
                2,
                "a posedge with iff=true must resume() the waiter exactly once"
            );
            assert!(
                matches!(log[1], ResumeArg::EventFired { .. }),
                "the resuming call's ResumeArg must be EventFired, got {:?}",
                log[1]
            );
        }
        assert_eq!(
            engine.test_sensitivity_entry_count(),
            0,
            "the dynamic entry must be gone once the waiter has resumed"
        );
    }

    // --- Task 6 (B-sched, ledger SC-049, SC-050, SC-051, SC-054, SC-069):
    // SignalEventControl family construction + named-event trigger + e2e ---

    /// SC-069's own literal acceptance, verified against the production
    /// `Engine`/`advance_time_to` dispatch path (not a hand-simulated one):
    /// a delayed named-event trigger (`->>#5 my_event;`'s scheduler-side
    /// shape -- `TimeWheel::arm_trigger` + `Engine::advance_time_to`'s
    /// `TimerAction::TriggerEvent` arm) must, once it fires, enqueue a
    /// process that is `@(my_event)`-waiting via `EventTerm::
    /// NamedEventTrigger` -- and, distinctly, must NOT create or resume any
    /// activation for "the triggering process" itself (there is deliberately
    /// no second, triggering process registered at all in this test, so any
    /// activation showing up in the post-fire ready set other than the
    /// waiter's own would prove a wrong implementation that conflated
    /// "trigger this event" with "resume whoever armed it" -- the exact
    /// regression this test's own plan text names). This is the real,
    /// production-loop version of `sched::wheel::tests::
    /// resume_and_trigger_entries_at_the_same_tick_both_fire_correctly`'s
    /// wheel-only proof, closing the fan-out half that lives in this module,
    /// not in `wheel.rs`.
    #[test]
    fn sc069_a_delayed_trigger_wakes_a_named_event_waiter_not_a_resumed_process() {
        struct NamedEventWaiter {
            ev: StateCellId,
            resume_log: Arc<StdMutex<Vec<ResumeArg>>>,
        }
        impl CompiledProcess for NamedEventWaiter {
            fn resume(
                &mut self,
                _state: SimHandle,
                arg: ResumeArg,
                _ops: &dyn SchedulerOps,
            ) -> Suspend {
                self.resume_log.lock().unwrap().push(arg);
                match arg {
                    ResumeArg::Start => Suspend::WaitEvents(EventWaitSet {
                        terms: vec![EventTerm::NamedEventTrigger(self.ev)],
                        ..Default::default()
                    }),
                    _ => Suspend::Done,
                }
            }
        }

        let ev = StateCellId(400);
        let mut engine = Engine::new(Box::new(TestArena::new()), Box::new(TestIoSink::default()));
        let log: Arc<StdMutex<Vec<ResumeArg>>> = Arc::new(StdMutex::new(Vec::new()));
        let waiter = engine.register_process(
            Box::new(NamedEventWaiter {
                ev,
                resume_log: log.clone(),
            }),
            ProcessKind::Initial,
            None,
            0,
        );

        // Dispatch Start: installs the NamedEventTrigger dynamic wait.
        let ready = engine.drain_and_sort(Region::Active);
        engine.dispatch_ready_set(Region::Active, ready);
        assert_eq!(
            log.lock().unwrap().len(),
            1,
            "only Start's own resume() call must have happened so far"
        );

        // Arm the delayed trigger directly on the wheel (this task's own
        // production entry point, `TimeWheel::arm_trigger`) and advance time
        // to it through the real `Engine::advance_time_to` -- never a
        // hand-simulated fan-out.
        engine.wheel.arm_trigger(10, ev);
        engine.advance_time_to(10);

        let ready = engine.drain_and_sort(Region::Active);
        assert_eq!(
            ready.len(),
            1,
            "the fired trigger must enqueue exactly one activation -- the \
             waiter's -- never zero (dropped) and never more than one \
             (e.g. also resuming some non-existent 'triggering process')"
        );
        assert_eq!(
            ready[0].proc, waiter,
            "the one enqueued activation must be the NamedEventTrigger \
             waiter's own process id"
        );
        engine.dispatch_ready_set(Region::Active, ready);
        {
            let log = log.lock().unwrap();
            assert_eq!(
                log.len(),
                2,
                "the waiter must resume exactly once after the trigger fires"
            );
            assert!(
                matches!(log[1], ResumeArg::EventFired { .. }),
                "the resuming call's ResumeArg must be EventFired, got {:?}",
                log[1]
            );
        }
        assert_eq!(
            engine.test_sensitivity_entry_count(),
            0,
            "the dynamic NamedEventTrigger entry must be gone once the \
             waiter has resumed"
        );
    }

    /// Companion to the wheel-only
    /// `resume_and_trigger_entries_at_the_same_tick_both_fire_correctly`:
    /// proves the SAME same-tick-FIFO guarantee holds through the full
    /// `Engine::advance_time_to` fan-out, not just at the raw
    /// `TimeWheel::advance_to` level -- a `Resume` entry (an ordinary
    /// `#delay`-suspended process) and a `TriggerEvent` entry armed at the
    /// same tick must both take effect: the delayed process resumes, and the
    /// named-event waiter is woken, from one `advance_time_to` call.
    #[test]
    fn engine_advance_time_to_fires_a_resume_and_a_trigger_armed_at_the_same_tick() {
        struct DoneOnResume {
            resume_log: Arc<StdMutex<Vec<ResumeArg>>>,
        }
        impl CompiledProcess for DoneOnResume {
            fn resume(
                &mut self,
                _state: SimHandle,
                arg: ResumeArg,
                _ops: &dyn SchedulerOps,
            ) -> Suspend {
                self.resume_log.lock().unwrap().push(arg);
                Suspend::Done
            }
        }
        struct NamedEventWaiter {
            ev: StateCellId,
            resume_log: Arc<StdMutex<Vec<ResumeArg>>>,
        }
        impl CompiledProcess for NamedEventWaiter {
            fn resume(
                &mut self,
                _state: SimHandle,
                arg: ResumeArg,
                _ops: &dyn SchedulerOps,
            ) -> Suspend {
                self.resume_log.lock().unwrap().push(arg);
                match arg {
                    ResumeArg::Start => Suspend::WaitEvents(EventWaitSet {
                        terms: vec![EventTerm::NamedEventTrigger(self.ev)],
                        ..Default::default()
                    }),
                    _ => Suspend::Done,
                }
            }
        }

        let ev = StateCellId(401);
        let mut engine = Engine::new(Box::new(TestArena::new()), Box::new(TestIoSink::default()));
        let resume_log: Arc<StdMutex<Vec<ResumeArg>>> = Arc::new(StdMutex::new(Vec::new()));
        let waiter_log: Arc<StdMutex<Vec<ResumeArg>>> = Arc::new(StdMutex::new(Vec::new()));

        let resumed_proc = engine.register_process(
            Box::new(DoneOnResume {
                resume_log: resume_log.clone(),
            }),
            ProcessKind::Initial,
            None,
            0,
        );
        let waiter = engine.register_process(
            Box::new(NamedEventWaiter {
                ev,
                resume_log: waiter_log.clone(),
            }),
            ProcessKind::Initial,
            None,
            0,
        );

        // Dispatch both Start activations: `resumed_proc` returns
        // Suspend::Done (its own Start resume already happened, nothing more
        // to arm for it directly -- this test arms its own `Resume` entry on
        // the wheel below, standing in for a `Suspend::Delay`-suspended
        // continuation), `waiter` installs its NamedEventTrigger wait.
        let ready = engine.drain_and_sort(Region::Active);
        engine.dispatch_ready_set(Region::Active, ready);
        resume_log.lock().unwrap().clear();

        let act = PendingActivation {
            id: 9_999,
            proc: resumed_proc,
            arg: ResumeArg::TimeElapsed,
            reconv_iter: 0,
            logical_order: (SchedTime { ticks: 0 }, Region::Active, 0),
        };
        engine.wheel.arm(
            10,
            TimerEntry {
                action: TimerAction::Resume(act, Region::Active),
            },
        );
        engine.wheel.arm_trigger(10, ev);
        engine.advance_time_to(10);

        let ready = engine.drain_and_sort(Region::Active);
        let mut procs: Vec<ProcessId> = ready.iter().map(|a| a.proc).collect();
        procs.sort_unstable();
        let mut want = vec![resumed_proc, waiter];
        want.sort_unstable();
        assert_eq!(
            procs, want,
            "both the Resume entry's own process and the TriggerEvent's \
             waiter must be enqueued from a single advance_time_to call \
             covering the same tick"
        );
    }

    // --- Task 9 (B-sched wave, ledger SC-125) ---

    /// A `resume()` that always returns `Suspend::Done` and, if ever called
    /// a second time, panics -- standing in for "a process that must never
    /// be resumed again," the sanctioned negative-proof shape for
    /// `disable fork;`'s own "terminates without ever resuming its
    /// descendants" acceptance.
    struct NeverResumedTwice(std::sync::atomic::AtomicUsize);
    impl NeverResumedTwice {
        fn new() -> Self {
            NeverResumedTwice(std::sync::atomic::AtomicUsize::new(0))
        }
    }
    impl CompiledProcess for NeverResumedTwice {
        fn resume(&mut self, _s: SimHandle, _arg: ResumeArg, _ops: &dyn SchedulerOps) -> Suspend {
            let prior = self.0.fetch_add(1, Ordering::SeqCst);
            assert_eq!(
                prior, 0,
                "a fork descendant terminated by disable_fork must never be resumed \
                 at all, let alone more than once"
            );
            Suspend::Done
        }
    }

    fn spawn_from(parent: ProcessId) -> SpawnSpec {
        SpawnSpec {
            kernel: 0,
            entry_pc: 0,
            parent,
            disable_name: Vec::new(),
        }
    }

    /// SC-125's own literal `disable fork;` acceptance: a parent with 2 live
    /// children, one of which (`child_a`) itself has 1 live grandchild (a
    /// nested fork) -- the parent's own mock calls `ops.disable_fork(self)`
    /// mid-`resume()` and, in that same call, returns `Suspend::Done` (never
    /// a suspending variant, proving the call itself does not suspend the
    /// caller). Drives the parent's own `Start` through the real `Engine`/
    /// `SchedulerControl` dispatch loop (`run_until`) -- the fork tree
    /// itself is built directly via `ResumeDriver::handle_fork` before that
    /// (this module's own private fields are reachable from this same-file
    /// `tests` submodule), the same "construct the tree directly, dispatch
    /// only the one process under test" pattern `resume.rs`'s own nested-
    /// fork tests already use, needed here because `Suspend::ForkJoin`'s
    /// real dispatch path (`apply_suspend`'s own `mint_child`) always mints
    /// an immediately-`Done` `UnmintedForkChild` (Task 7's documented
    /// interim placeholder), which could never stay live long enough to
    /// itself fork a grandchild.
    #[test]
    fn sc125_disable_fork_terminates_every_live_descendant_without_suspending_the_caller() {
        struct DisablingParent {
            id_cell: Arc<StdMutex<Option<ProcessId>>>,
            resume_count: Arc<std::sync::atomic::AtomicUsize>,
        }
        impl CompiledProcess for DisablingParent {
            fn resume(&mut self, _s: SimHandle, arg: ResumeArg, ops: &dyn SchedulerOps) -> Suspend {
                self.resume_count.fetch_add(1, Ordering::SeqCst);
                match arg {
                    ResumeArg::Start => {
                        let id = self
                            .id_cell
                            .lock()
                            .unwrap()
                            .expect("id_cell set right after register_process returns");
                        ops.disable_fork(id);
                        Suspend::Done
                    }
                    other => panic!("DisablingParent must only ever see Start, got {other:?}"),
                }
            }
        }

        let mut engine = new_engine();
        let id_cell = Arc::new(StdMutex::new(None));
        let resume_count = Arc::new(std::sync::atomic::AtomicUsize::new(0));
        let parent = engine.register_process(
            Box::new(DisablingParent {
                id_cell: id_cell.clone(),
                resume_count: resume_count.clone(),
            }),
            ProcessKind::Initial,
            None,
            0,
        );
        *id_cell.lock().unwrap() = Some(parent);

        // Build the fork tree directly (see this test's own doc comment):
        // parent -> [child_a, child_b]; child_a -> [grandchild].
        let mut mint =
            |_: &SpawnSpec| -> Box<dyn CompiledProcess> { Box::new(NeverResumedTwice::new()) };
        let children = engine.driver.handle_fork(
            &mut engine.table,
            parent,
            JoinKind::None,
            &[spawn_from(parent), spawn_from(parent)],
            &mut mint,
        );
        assert_eq!(children.len(), 2);
        let (child_a, child_b) = (children[0], children[1]);
        let grandchildren = engine.driver.handle_fork(
            &mut engine.table,
            child_a,
            JoinKind::None,
            &[spawn_from(child_a)],
            &mut mint,
        );
        assert_eq!(grandchildren.len(), 1);
        let grandchild = grandchildren[0];

        assert!(engine.test_process_exists(child_a));
        assert!(engine.test_process_exists(child_b));
        assert!(engine.test_process_exists(grandchild));

        // `register_process` already queued `parent`'s own Start activation
        // -- `child_a`/`child_b`/`grandchild` were registered directly via
        // `handle_fork` above, never given a Start activation of their own,
        // so this `run_until` dispatches exactly one process: `parent`.
        let result = engine.run_until(None);
        assert!(matches!(result, RunResult::Completed));

        assert_eq!(
            resume_count.load(Ordering::SeqCst),
            1,
            "the parent's own single resume() call must never be re-entered a second \
             time by disable_fork itself"
        );
        assert!(
            !engine.test_process_exists(child_a),
            "disable_fork must terminate every live child"
        );
        assert!(
            !engine.test_process_exists(child_b),
            "disable_fork must terminate every live child"
        );
        assert!(
            !engine.test_process_exists(grandchild),
            "disable_fork must terminate a live grandchild too -- one level deep is \
             not enough"
        );
    }

    /// Proves `Suspend::WaitFork`'s own scheduler-side *installation*
    /// (`Engine::apply_wait_fork`) — distinct from `ResumeDriver::
    /// on_child_done`'s own decrement logic, which
    /// `sched::resume::tests::sc125_wait_fork_resumes_only_once_every_live_
    /// child_completes_not_on_the_first` already covers directly. Builds 2
    /// live children via `handle_fork` (bypassing `Suspend::ForkJoin`'s own
    /// hardcoded `UnmintedForkChild` mint, per the disable_fork test's own
    /// doc comment, so the children stay live across this test's own
    /// `apply_suspend` calls), suspends the parent directly via
    /// `apply_suspend(parent, Suspend::WaitFork)`, and drives both children
    /// to completion via `apply_suspend(child, Suspend::Done)`. Fails if
    /// `apply_wait_fork` ever miscounts the live-child set (e.g. counting
    /// `fork_children.len()` instead of filtering for still-live entries).
    #[test]
    fn sc125_apply_suspend_wait_fork_installs_live_count_and_resumes_after_the_last_child() {
        struct WaitForkParent {
            log: Arc<StdMutex<Vec<&'static str>>>,
        }
        impl CompiledProcess for WaitForkParent {
            fn resume(
                &mut self,
                _s: SimHandle,
                arg: ResumeArg,
                _ops: &dyn SchedulerOps,
            ) -> Suspend {
                match arg {
                    ResumeArg::AllForkChildrenDone => {
                        self.log.lock().unwrap().push("resumed");
                        Suspend::Done
                    }
                    other => panic!("WaitForkParent: unexpected {other:?}"),
                }
            }
        }

        let log = Arc::new(StdMutex::new(Vec::new()));
        let mut engine = new_engine();
        // Registered directly via `ProcessTable::register` (not
        // `register_process`, which would also queue an unrelated `Start`
        // activation this test never dispatches -- this test drives `proc`
        // entirely through direct `apply_suspend` calls, standing in for a
        // real process already mid-way through its own `fork ... join_none;
        // ...; wait fork;` sequence).
        let parent = engine
            .table
            .register(Box::new(WaitForkParent { log: log.clone() }));
        let mut mint =
            |_: &SpawnSpec| -> Box<dyn CompiledProcess> { Box::new(NeverResumedTwice::new()) };
        let children = engine.driver.handle_fork(
            &mut engine.table,
            parent,
            JoinKind::None,
            &[spawn_from(parent), spawn_from(parent)],
            &mut mint,
        );
        assert_eq!(children.len(), 2);

        engine.apply_suspend(parent, Suspend::WaitFork);
        assert_eq!(
            engine.table.get(parent).unwrap().wait_fork_remaining,
            Some(2),
            "apply_wait_fork must install the true live-child count"
        );

        engine.apply_suspend(children[0], Suspend::Done);
        assert!(!engine.test_process_exists(children[0]));
        assert!(
            log.lock().unwrap().is_empty(),
            "must not resume the parent after only 1 of 2 live children"
        );
        assert_eq!(
            engine.table.get(parent).unwrap().wait_fork_remaining,
            Some(1)
        );

        engine.apply_suspend(children[1], Suspend::Done);
        assert!(!engine.test_process_exists(children[1]));
        assert!(
            engine
                .table
                .get(parent)
                .unwrap()
                .wait_fork_remaining
                .is_none(),
            "wait_fork_remaining must be cleared the instant the count reaches 0, \
             immediately from apply_suspend itself, before the parent's own \
             AllForkChildrenDone activation is ever dispatched below"
        );
        // The last completion's `Suspend::Done` arm only *enqueues*
        // `parent`'s `AllForkChildrenDone` activation onto `Region::Active`
        // (the same real queue `ForkChildDone` notifications use) -- it does
        // not call `resume()` synchronously, so an actual dispatch pass is
        // needed to observe it, exactly like `fork_join_all_actually_
        // resumes_the_parent_when_its_one_child_completes` above needs
        // `run_until` for the identical reason.
        let ready = engine.drain_and_sort(Region::Active);
        engine.dispatch_ready_set(Region::Active, ready);
        assert_eq!(
            log.lock().unwrap().as_slice(),
            &["resumed"],
            "the parent must resume exactly once, only after the 2nd (last) child"
        );
    }

    /// **Mutation-testing addition (B-sched, SC-125).** `apply_wait_fork`'s
    /// own doc comment names "the common 'every child already finished
    /// before `wait fork;` itself ran' case" -- this proves it directly, for
    /// a *mixed* population (one child already done and reclaimed, one still
    /// live) rather than the "0 live" or "all live" cases the other two
    /// tests here already cover. `live` must be computed by filtering
    /// `fork_children` down to those still present in `table` -- counting
    /// every spawned child regardless of liveness (e.g. `children.len()`)
    /// would install a `wait_fork_remaining` one too high here, and the
    /// parent would never resume: its one remaining live child's own
    /// completion would decrement to 1, not 0, and `on_child_done` would
    /// never report `AllForkChildrenDone`.
    #[test]
    fn sc125_apply_wait_fork_counts_only_still_live_children_not_every_spawned_child() {
        struct WaitForkParent {
            log: Arc<StdMutex<Vec<&'static str>>>,
        }
        impl CompiledProcess for WaitForkParent {
            fn resume(
                &mut self,
                _s: SimHandle,
                arg: ResumeArg,
                _ops: &dyn SchedulerOps,
            ) -> Suspend {
                match arg {
                    ResumeArg::AllForkChildrenDone => {
                        self.log.lock().unwrap().push("resumed");
                        Suspend::Done
                    }
                    other => panic!("WaitForkParent: unexpected {other:?}"),
                }
            }
        }

        let log = Arc::new(StdMutex::new(Vec::new()));
        let mut engine = new_engine();
        let parent = engine
            .table
            .register(Box::new(WaitForkParent { log: log.clone() }));
        let mut mint =
            |_: &SpawnSpec| -> Box<dyn CompiledProcess> { Box::new(NeverResumedTwice::new()) };
        let children = engine.driver.handle_fork(
            &mut engine.table,
            parent,
            JoinKind::None,
            &[spawn_from(parent), spawn_from(parent)],
            &mut mint,
        );
        assert_eq!(children.len(), 2);

        // `children[0]` finishes -- and is reclaimed -- *before* `wait
        // fork;` itself ever runs. `parent`'s own `fork_children` still
        // lists both ids (nothing prunes that list on completion), but only
        // `children[1]` is still a live table entry.
        engine.apply_suspend(children[0], Suspend::Done);
        assert!(!engine.test_process_exists(children[0]));
        assert!(engine.test_process_exists(children[1]));

        engine.apply_suspend(parent, Suspend::WaitFork);
        assert_eq!(
            engine.table.get(parent).unwrap().wait_fork_remaining,
            Some(1),
            "apply_wait_fork must count only the one still-live child, not both \
             spawned children"
        );

        engine.apply_suspend(children[1], Suspend::Done);
        assert!(!engine.test_process_exists(children[1]));
        assert!(
            engine
                .table
                .get(parent)
                .unwrap()
                .wait_fork_remaining
                .is_none(),
            "wait_fork_remaining must reach 0 (and clear) the instant this one live \
             child's completion is reported -- not require a second, nonexistent \
             completion"
        );
        let ready = engine.drain_and_sort(Region::Active);
        engine.dispatch_ready_set(Region::Active, ready);
        assert_eq!(
            log.lock().unwrap().as_slice(),
            &["resumed"],
            "the parent must actually resume once its one live child completes, \
             even though one of its two originally-spawned children finished \
             earlier"
        );
    }

    /// Proves `Suspend::WaitOrder(order)`'s own scheduler-side
    /// *installation* (`Engine::apply_wait_order`) through a real
    /// `apply_suspend` call, complementing `sched::resume::tests`'s own
    /// direct `on_child_done` proofs for the ordering/violation logic
    /// itself.
    #[test]
    fn sc125_apply_suspend_wait_order_resumes_only_after_the_exact_declared_order() {
        struct WaitOrderParent {
            log: Arc<StdMutex<Vec<&'static str>>>,
        }
        impl CompiledProcess for WaitOrderParent {
            fn resume(
                &mut self,
                _s: SimHandle,
                arg: ResumeArg,
                _ops: &dyn SchedulerOps,
            ) -> Suspend {
                match arg {
                    ResumeArg::WaitOrderSatisfied => {
                        self.log.lock().unwrap().push("resumed");
                        Suspend::Done
                    }
                    other => panic!("WaitOrderParent: unexpected {other:?}"),
                }
            }
        }

        let log = Arc::new(StdMutex::new(Vec::new()));
        let mut engine = new_engine();
        // See the identical `ProcessTable::register` note on
        // `sc125_apply_suspend_wait_fork_installs_live_count_and_resumes_
        // after_the_last_child` above.
        let parent = engine
            .table
            .register(Box::new(WaitOrderParent { log: log.clone() }));
        let mut mint =
            |_: &SpawnSpec| -> Box<dyn CompiledProcess> { Box::new(NeverResumedTwice::new()) };
        let children = engine.driver.handle_fork(
            &mut engine.table,
            parent,
            JoinKind::None,
            &[spawn_from(parent), spawn_from(parent), spawn_from(parent)],
            &mut mint,
        );
        assert_eq!(children.len(), 3);
        let order = children.clone();

        engine.apply_suspend(parent, Suspend::WaitOrder(order));
        assert!(engine.table.get(parent).unwrap().wait_order.is_some());

        engine.apply_suspend(children[0], Suspend::Done);
        engine.apply_suspend(children[1], Suspend::Done);
        assert!(
            log.lock().unwrap().is_empty(),
            "must not resume before the 3rd, in-order completion"
        );

        engine.apply_suspend(children[2], Suspend::Done);
        // The 3rd completion's `Suspend::Done` arm only *enqueues* the
        // parent's `WaitOrderSatisfied` activation onto `Region::Active` --
        // see the identical note on `sc125_apply_suspend_wait_fork_
        // installs_live_count_and_resumes_after_the_last_child` above.
        let ready = engine.drain_and_sort(Region::Active);
        engine.dispatch_ready_set(Region::Active, ready);
        assert_eq!(log.lock().unwrap().as_slice(), &["resumed"]);
    }

    // --- Classes plan Task 13 (LRM 9.7): the `process` class ---
    //
    // Every test drives the real `Engine` dispatch loop (`run_until`) with
    // scripted processes that call the same `SchedulerOps` entry points the
    // compiled `process` methods reach (`process_track` / `process_request`
    // / `process_state`), so the Kill/Suspend/Resume/Await/status mechanics
    // are pinned without any codegen. ProcessIds are the registration order
    // (0, 1, 2, ...), asserted on every `register_process` return.

    type ProcLog = Arc<StdMutex<Vec<String>>>;

    /// A process whose every `resume()` is one call of a closure.
    type ScriptFn = Box<dyn FnMut(ResumeArg, &dyn SchedulerOps) -> Suspend + Send>;
    struct Script(ScriptFn);
    impl CompiledProcess for Script {
        fn resume(&mut self, _s: SimHandle, arg: ResumeArg, ops: &dyn SchedulerOps) -> Suspend {
            (self.0)(arg, ops)
        }
    }

    fn script(
        f: impl FnMut(ResumeArg, &dyn SchedulerOps) -> Suspend + Send + 'static,
    ) -> Box<dyn CompiledProcess> {
        Box::new(Script(Box::new(f)))
    }

    fn new_log() -> ProcLog {
        Arc::new(StdMutex::new(Vec::new()))
    }

    fn lines(log: &ProcLog) -> Vec<String> {
        log.lock().unwrap().clone()
    }

    fn state_of(ops: &dyn SchedulerOps, target: ProcessId) -> u8 {
        ops.process_state(target) as u8
    }

    fn register_expecting(engine: &mut Engine, proc: Box<dyn CompiledProcess>, want: u32) {
        let id = engine.register_process(proc, ProcessKind::Initial, None, 0);
        assert_eq!(id, want, "tests hard-code ProcessIds as registration order");
    }

    /// LRM 9.7.3: `kill()` terminates the target and ALL its descendants
    /// before it returns, and `status()` reads KILLED in the very segment
    /// that called `kill()`. The victim's pending timer must never fire.
    #[test]
    fn process_kill_terminates_target_and_descendants_and_reads_killed() {
        let mut engine = new_engine();
        let log = new_log();

        let l0 = log.clone();
        register_expecting(
            &mut engine,
            script(move |arg, ops| match arg {
                ResumeArg::Start => {
                    ops.process_track(0);
                    Suspend::Delay(100)
                }
                other => {
                    l0.lock().unwrap().push(format!("victim resumed {other:?}"));
                    Suspend::Done
                }
            }),
            0,
        );
        let (l1, mut step) = (log.clone(), 0);
        register_expecting(
            &mut engine,
            script(move |arg, ops| {
                step += 1;
                match (step, arg) {
                    (1, ResumeArg::Start) => Suspend::Delay(10),
                    (2, ResumeArg::TimeElapsed) => {
                        let before = state_of(ops, 0);
                        ops.process_request(1, ProcessRequest::Kill(0));
                        let after = state_of(ops, 0);
                        l1.lock().unwrap().push(format!(
                            "t={} before={before} after={after}",
                            ops.now().ticks
                        ));
                        Suspend::ByRequest
                    }
                    (3, ResumeArg::TimeElapsed) => {
                        l1.lock()
                            .unwrap()
                            .push(format!("resumed state={}", state_of(ops, 0)));
                        Suspend::Done
                    }
                    other => panic!("killer: unexpected {other:?}"),
                }
            }),
            1,
        );

        // victim (0) -> child (2) -> grandchild (3), built directly.
        let mut mint =
            |_: &SpawnSpec| -> Box<dyn CompiledProcess> { Box::new(NeverResumedTwice::new()) };
        let kids = engine.driver.handle_fork(
            &mut engine.table,
            0,
            JoinKind::None,
            &[spawn_from(0)],
            &mut mint,
        );
        let grand = engine.driver.handle_fork(
            &mut engine.table,
            kids[0],
            JoinKind::None,
            &[spawn_from(kids[0])],
            &mut mint,
        );
        assert_eq!((kids[0], grand[0]), (2, 3));

        assert!(matches!(engine.run_until(None), RunResult::Completed));
        assert_eq!(
            lines(&log),
            // The victim is WAITING on its delay (2); the overlay turns it
            // KILLED (4) at once.
            ["t=10 before=2 after=4", "resumed state=4"],
            "kill must be visible to status() in the same segment, and the \
             victim's Delay(100) must never resume it"
        );
        for id in [0, 2, 3] {
            assert!(!engine.test_process_exists(id), "process {id} must be gone");
        }
        assert_eq!(engine.effective_process_state(0), ProcessState::Killed);
        assert_eq!(
            engine.effective_process_state(3),
            ProcessState::Killed,
            "an untracked descendant killed with its ancestor reads KILLED too"
        );
    }

    /// LRM 9.7.3: `kill()` on a process that already ended still terminates
    /// its live descendants, and the target keeps reading FINISHED.
    #[test]
    fn process_kill_of_a_finished_process_still_kills_its_live_descendants() {
        let mut engine = new_engine();
        register_expecting(
            &mut engine,
            script(|_, ops| {
                ops.process_track(0);
                Suspend::Done
            }),
            0,
        );
        let mut step = 0;
        register_expecting(
            &mut engine,
            script(move |_, ops| {
                step += 1;
                match step {
                    1 => Suspend::Delay(10),
                    2 => {
                        ops.process_request(1, ProcessRequest::Kill(0));
                        Suspend::ByRequest
                    }
                    _ => Suspend::Done,
                }
            }),
            1,
        );
        let mut mint =
            |_: &SpawnSpec| -> Box<dyn CompiledProcess> { Box::new(NeverResumedTwice::new()) };
        let kids = engine.driver.handle_fork(
            &mut engine.table,
            0,
            JoinKind::None,
            &[spawn_from(0)],
            &mut mint,
        );
        assert!(engine.test_process_exists(kids[0]));
        assert!(matches!(engine.run_until(None), RunResult::Completed));
        assert!(
            !engine.test_process_exists(kids[0]),
            "the live descendant of an already-finished process must die"
        );
        assert_eq!(engine.effective_process_state(0), ProcessState::Finished);
    }

    /// LRM 9.7.1: a process may kill itself; it ends there (`process_request`
    /// reports it so the compiled caller can stop) and reads KILLED.
    #[test]
    fn process_self_kill_ends_the_caller_and_reports_it() {
        let mut engine = new_engine();
        let log = new_log();
        let l0 = log.clone();
        register_expecting(
            &mut engine,
            script(move |_, ops| {
                ops.process_track(0);
                let ends_me = ops.process_request(0, ProcessRequest::Kill(0));
                l0.lock().unwrap().push(format!("ends_requester={ends_me}"));
                Suspend::Done
            }),
            0,
        );
        assert!(matches!(engine.run_until(None), RunResult::Completed));
        assert_eq!(lines(&log), ["ends_requester=true"]);
        assert!(!engine.test_process_exists(0));
        assert_eq!(engine.effective_process_state(0), ProcessState::Killed);
    }

    /// Killing the children a `wait fork` is blocked on releases the parent
    /// exactly once, only after the last one is gone.
    #[test]
    fn process_kill_of_fork_children_releases_the_wait_fork_parent() {
        let mut engine = new_engine();
        let log = new_log();
        let l0 = log.clone();
        register_expecting(
            &mut engine,
            script(move |arg, ops| match arg {
                ResumeArg::Start => Suspend::WaitFork,
                ResumeArg::AllForkChildrenDone => {
                    l0.lock()
                        .unwrap()
                        .push(format!("joined@{}", ops.now().ticks));
                    Suspend::Done
                }
                other => panic!("parent: unexpected {other:?}"),
            }),
            0,
        );
        let mut step = 0;
        register_expecting(
            &mut engine,
            script(move |_, ops| {
                step += 1;
                match step {
                    1 => Suspend::Delay(10),
                    2 => {
                        ops.process_request(1, ProcessRequest::Kill(2));
                        ops.process_request(1, ProcessRequest::Kill(3));
                        Suspend::ByRequest
                    }
                    _ => Suspend::Done,
                }
            }),
            1,
        );
        let mut mint =
            |_: &SpawnSpec| -> Box<dyn CompiledProcess> { Box::new(NeverResumedTwice::new()) };
        let kids = engine.driver.handle_fork(
            &mut engine.table,
            0,
            JoinKind::None,
            &[spawn_from(0), spawn_from(0)],
            &mut mint,
        );
        assert_eq!(kids, vec![2, 3]);
        assert!(matches!(engine.run_until(None), RunResult::Completed));
        assert_eq!(lines(&log), ["joined@10"]);
    }

    /// LRM 9.7: `status()` is RUNNING for the caller, WAITING while the
    /// target is blocked, FINISHED once it ended.
    #[test]
    fn process_status_tracks_running_waiting_finished() {
        let mut engine = new_engine();
        let log = new_log();
        let l0 = log.clone();
        let mut step = 0;
        register_expecting(
            &mut engine,
            script(move |_, ops| {
                step += 1;
                match step {
                    1 => {
                        ops.process_track(0);
                        l0.lock()
                            .unwrap()
                            .push(format!("self={}", state_of(ops, 0)));
                        Suspend::Delay(10)
                    }
                    _ => Suspend::Done,
                }
            }),
            0,
        );
        let l1 = log.clone();
        let mut step = 0;
        register_expecting(
            &mut engine,
            script(move |_, ops| {
                step += 1;
                match step {
                    1 => Suspend::Delay(5),
                    2 => {
                        l1.lock().unwrap().push(format!("@5={}", state_of(ops, 0)));
                        Suspend::Delay(20)
                    }
                    _ => {
                        l1.lock().unwrap().push(format!("@25={}", state_of(ops, 0)));
                        Suspend::Done
                    }
                }
            }),
            1,
        );
        assert!(matches!(engine.run_until(None), RunResult::Completed));
        assert_eq!(lines(&log), ["self=1", "@5=2", "@25=0"]);
    }

    /// LRM 9.7: `await()` blocks until the target ends; a target that
    /// already ended, or the caller itself, blocks nothing.
    #[test]
    fn process_await_blocks_until_the_target_ends() {
        let mut engine = new_engine();
        let log = new_log();
        let mut step = 0;
        register_expecting(
            &mut engine,
            script(move |_, ops| {
                step += 1;
                if step == 1 {
                    ops.process_track(0);
                    Suspend::Delay(30)
                } else {
                    Suspend::Done
                }
            }),
            0,
        );
        let (l1, mut step) = (log.clone(), 0);
        register_expecting(
            &mut engine,
            script(move |_, ops| {
                step += 1;
                match step {
                    1 => Suspend::Delay(5),
                    2 => {
                        ops.process_request(1, ProcessRequest::Await(0));
                        l1.lock().unwrap().push("awaiting@5".into());
                        Suspend::ByRequest
                    }
                    _ => {
                        l1.lock()
                            .unwrap()
                            .push(format!("awaited@{}", ops.now().ticks));
                        Suspend::Done
                    }
                }
            }),
            1,
        );
        let (l2, mut step) = (log.clone(), 0);
        register_expecting(
            &mut engine,
            script(move |_, ops| {
                step += 1;
                match step {
                    1 => Suspend::Delay(50),
                    2 => {
                        ops.process_request(2, ProcessRequest::Await(0));
                        Suspend::ByRequest
                    }
                    _ => {
                        l2.lock().unwrap().push(format!("late@{}", ops.now().ticks));
                        Suspend::Done
                    }
                }
            }),
            2,
        );
        let (l3, mut step) = (log.clone(), 0);
        register_expecting(
            &mut engine,
            script(move |_, ops| {
                step += 1;
                if step == 1 {
                    ops.process_request(3, ProcessRequest::Await(3));
                    Suspend::ByRequest
                } else {
                    l3.lock().unwrap().push(format!("self@{}", ops.now().ticks));
                    Suspend::Done
                }
            }),
            3,
        );
        assert!(matches!(engine.run_until(None), RunResult::Completed));
        assert_eq!(
            lines(&log),
            ["self@0", "awaiting@5", "awaited@30", "late@50"],
            "await wakes exactly at the target's end; an ended target / self \
             never blocks"
        );
    }

    /// LRM 9.7.4 / 9.7.5: a suspended process stops advancing; `resume()`
    /// continues it; `resume()` on a process that is not suspended is a
    /// no-op.
    #[test]
    fn process_suspend_resume_periodic_process() {
        let mut engine = new_engine();
        let log = new_log();
        let (l0, mut ticks) = (log.clone(), 0);
        register_expecting(
            &mut engine,
            script(move |arg, ops| match arg {
                ResumeArg::Start => {
                    ops.process_track(0);
                    Suspend::Delay(10)
                }
                _ => {
                    ticks += 1;
                    l0.lock().unwrap().push(format!("tick@{}", ops.now().ticks));
                    if ticks == 4 {
                        Suspend::Done
                    } else {
                        Suspend::Delay(10)
                    }
                }
            }),
            0,
        );
        let (l1, mut step) = (log.clone(), 0);
        register_expecting(
            &mut engine,
            script(move |_, ops| {
                step += 1;
                match step {
                    1 => Suspend::Delay(5),
                    2 => {
                        // Not suspended: no effect.
                        ops.process_request(1, ProcessRequest::Resume(0));
                        Suspend::ByRequest
                    }
                    3 => Suspend::Delay(10),
                    4 => {
                        ops.process_request(1, ProcessRequest::Suspend(0));
                        l1.lock()
                            .unwrap()
                            .push(format!("@15 state={}", state_of(ops, 0)));
                        Suspend::ByRequest
                    }
                    5 => Suspend::Delay(30),
                    6 => {
                        l1.lock()
                            .unwrap()
                            .push(format!("@45 state={}", state_of(ops, 0)));
                        ops.process_request(1, ProcessRequest::Resume(0));
                        l1.lock()
                            .unwrap()
                            .push(format!("resumed state={}", state_of(ops, 0)));
                        Suspend::ByRequest
                    }
                    _ => Suspend::Done,
                }
            }),
            1,
        );
        assert!(matches!(engine.run_until(None), RunResult::Completed));
        assert_eq!(
            lines(&log),
            [
                "tick@10",
                "@15 state=3",
                "@45 state=3",
                "resumed state=2",
                "tick@45",
                "tick@55",
                "tick@65",
            ],
            "the tick due at 20 is held while suspended and replayed at the \
             resume (45); then the 10-unit period continues"
        );
    }

    /// A process may suspend itself (`ByRequest` parks it) and another
    /// process resumes it.
    #[test]
    fn process_self_suspend_is_released_by_another_process() {
        let mut engine = new_engine();
        let log = new_log();
        let (l0, mut step) = (log.clone(), 0);
        register_expecting(
            &mut engine,
            script(move |_, ops| {
                step += 1;
                if step == 1 {
                    ops.process_track(0);
                    ops.process_request(0, ProcessRequest::Suspend(0));
                    l0.lock().unwrap().push("p0 suspending".into());
                    Suspend::ByRequest
                } else {
                    l0.lock()
                        .unwrap()
                        .push(format!("p0 resumed@{}", ops.now().ticks));
                    Suspend::Done
                }
            }),
            0,
        );
        let (l1, mut step) = (log.clone(), 0);
        register_expecting(
            &mut engine,
            script(move |_, ops| {
                step += 1;
                match step {
                    1 => Suspend::Delay(20),
                    2 => {
                        l1.lock()
                            .unwrap()
                            .push(format!("p0 state={}", state_of(ops, 0)));
                        ops.process_request(1, ProcessRequest::Resume(0));
                        Suspend::ByRequest
                    }
                    _ => Suspend::Done,
                }
            }),
            1,
        );
        assert!(matches!(engine.run_until(None), RunResult::Completed));
        assert_eq!(
            lines(&log),
            ["p0 suspending", "p0 state=3", "p0 resumed@20"]
        );
    }

    fn new_semaphore(keys: i64, template: u32) -> u64 {
        let h = crate::objects::alloc(
            crate::netlist::ClassTemplateId(template),
            semaphore::SEMAPHORE_FIELD_WORDS,
        );
        assert!(semaphore::init(h, keys));
        h
    }

    /// Classes plan Task 17 (LRM 15.3.3/15.3.4): a `get` that must wait yields
    /// (`SyncBlock` + `ByRequest`) and reads WAITING until the `put` that
    /// grants it queues `SyncWake`, which resumes it at the putter's time.
    #[test]
    fn semaphore_get_blocks_until_a_put_grants_it() {
        let h = new_semaphore(0, 17_100);
        let mut engine = new_engine();
        let log = new_log();
        let (l0, mut step) = (log.clone(), 0);
        register_expecting(
            &mut engine,
            script(move |_, ops| {
                step += 1;
                if step == 1 {
                    let blocked = semaphore::get(h, 0, 1, &mut |r| {
                        ops.process_request(0, r);
                    });
                    assert!(blocked, "an empty semaphore blocks");
                    Suspend::ByRequest
                } else {
                    l0.lock().unwrap().push(format!("got@{}", ops.now().ticks));
                    Suspend::Done
                }
            }),
            0,
        );
        let (l1, mut step) = (log.clone(), 0);
        register_expecting(
            &mut engine,
            script(move |_, ops| {
                step += 1;
                match step {
                    1 => Suspend::Delay(7),
                    _ => {
                        l1.lock()
                            .unwrap()
                            .push(format!("before put: state={}", state_of(ops, 0)));
                        assert!(semaphore::put(h, 1, &mut |r| {
                            ops.process_request(1, r);
                        }));
                        Suspend::Done
                    }
                }
            }),
            1,
        );
        assert!(matches!(engine.run_until(None), RunResult::Completed));
        assert_eq!(lines(&log), ["before put: state=2", "got@7"]);
        assert_eq!(semaphore::keys(h), Some(0));
    }

    /// A process killed while blocked leaves the queue; a waiter behind it
    /// that the freed head was holding back is released at the kill.
    #[test]
    fn killing_a_blocked_get_releases_the_waiter_behind_it() {
        let h = new_semaphore(1, 17_101);
        let mut engine = new_engine();
        let log = new_log();
        let mut step = 0;
        register_expecting(
            &mut engine,
            script(move |_, ops| {
                step += 1;
                if step == 1 {
                    assert!(semaphore::get(h, 0, 2, &mut |r| {
                        ops.process_request(0, r);
                    }));
                    Suspend::ByRequest
                } else {
                    panic!("the killed waiter must never resume");
                }
            }),
            0,
        );
        let (l1, mut step) = (log.clone(), 0);
        register_expecting(
            &mut engine,
            script(move |_, ops| {
                step += 1;
                if step == 1 {
                    assert!(
                        semaphore::get(h, 1, 1, &mut |r| {
                            ops.process_request(1, r);
                        }),
                        "FIFO: queued behind the get(2) although a key is free"
                    );
                    Suspend::ByRequest
                } else {
                    l1.lock()
                        .unwrap()
                        .push(format!("p1 got@{}", ops.now().ticks));
                    Suspend::Done
                }
            }),
            1,
        );
        let (l2, mut step) = (log.clone(), 0);
        register_expecting(
            &mut engine,
            script(move |_, ops| {
                step += 1;
                match step {
                    1 => Suspend::Delay(5),
                    2 => {
                        l2.lock()
                            .unwrap()
                            .push(format!("waiters before kill: {}", semaphore::waiters(h)));
                        ops.process_request(2, ProcessRequest::Kill(0));
                        Suspend::ByRequest
                    }
                    _ => {
                        l2.lock()
                            .unwrap()
                            .push(format!("waiters after kill: {}", semaphore::waiters(h)));
                        Suspend::Done
                    }
                }
            }),
            2,
        );
        assert!(matches!(engine.run_until(None), RunResult::Completed));
        assert_eq!(
            lines(&log),
            [
                "waiters before kill: 2",
                // The killer continues at once; the released waiter is
                // activated, so it runs after the killer's segment.
                "waiters after kill: 0",
                "p1 got@5",
            ]
        );
        assert_eq!(semaphore::keys(h), Some(0));
    }

    /// A process suspended with `p.suspend()` while blocked in `get` is
    /// granted its keys but stays parked until `p.resume()`.
    #[test]
    fn a_granted_get_of_a_suspended_process_waits_for_resume() {
        let h = new_semaphore(0, 17_102);
        let mut engine = new_engine();
        let log = new_log();
        let (l0, mut step) = (log.clone(), 0);
        register_expecting(
            &mut engine,
            script(move |_, ops| {
                step += 1;
                if step == 1 {
                    ops.process_track(0);
                    assert!(semaphore::get(h, 0, 1, &mut |r| {
                        ops.process_request(0, r);
                    }));
                    Suspend::ByRequest
                } else {
                    l0.lock().unwrap().push(format!("got@{}", ops.now().ticks));
                    Suspend::Done
                }
            }),
            0,
        );
        let mut step = 0;
        register_expecting(
            &mut engine,
            script(move |_, ops| {
                step += 1;
                match step {
                    1 => Suspend::Delay(1),
                    2 => {
                        ops.process_request(1, ProcessRequest::Suspend(0));
                        Suspend::ByRequest
                    }
                    3 => Suspend::Delay(1),
                    4 => {
                        semaphore::put(h, 1, &mut |r| {
                            ops.process_request(1, r);
                        });
                        Suspend::Delay(8)
                    }
                    5 => {
                        ops.process_request(1, ProcessRequest::Resume(0));
                        Suspend::ByRequest
                    }
                    _ => Suspend::Done,
                }
            }),
            1,
        );
        assert!(matches!(engine.run_until(None), RunResult::Completed));
        assert_eq!(lines(&log), ["got@10"]);
    }
}