vj_sim/sched/

process_table.rs

//! Process table & descriptor storage — Task 11 ("Ready-set unit of work &
//! `ProcessTable`", `docs/superpowers/plans/2026-09-23-wave-A-sched.md`;
//! ledger SC-057, SC-058, SC-063).
//!
//! `ProcessDescriptor` (per registered process), `ProcessTable::register`/
//! `get`, and the `Mutex`/`AtomicBool` interior-mutability shape were shipped
//! by an earlier commit (`555998c`/`3c69852`, scoped to `["SC-061"]` — making
//! Task 12's `ResumeDriver::resume_for` dispatch `resume()` calls from
//! multiple OS threads while still guaranteeing mutual exclusion per process,
//! via `ProcessDescriptor::compiled`'s own `Mutex`, never an external `&mut`
//! borrow of the whole table; see `resume.rs`'s module doc). This commit adds
//! the rest of Task 11: `ProcessTable` is now genuinely `slotmap::SlotMap`-
//! backed (SC-063's literal requirement, not just behaviorally equivalent to
//! it — a `Vec` did not satisfy it no matter how correct its behavior was),
//! with `register`'s/`get`'s public signatures unchanged so `resume.rs`
//! required no edit; `WorkUnit` (SC-057); and `form_ready_set` (SC-058, with
//! one documented signature deviation — see its own doc comment — since
//! Task 10's `QueuedGate` does not exist yet in this tree).

use std::sync::Mutex;
use std::sync::atomic::AtomicBool;

use vj_abi::Region;

use super::{
    CompiledProcess, JoinKind, PendingActivation, ProcessId, RegionQueuePushError, RegionQueues,
    ResumeArg,
};
use crate::process::WaitSpec;

/// One resumable process's scheduler-owned bookkeeping (scheduler.md §5; ledger
/// SC-063's named shape, plus the fork/join fields Task 11/D1a add and the
/// `Mutex`/`AtomicBool` review fix Task 11 records for SC-061 — see the plan's
/// Task 11 "Review fix (SC-061 concurrency, Task 12)" note).
pub struct ProcessDescriptor {
    /// The compiled kernel this descriptor drives. `Mutex`-wrapped (not a bare
    /// `Box`) so `ResumeDriver` can reach it through a *shared* `&ProcessTable`
    /// and still get exclusive `&mut` access for the duration of one `resume()`
    /// call — the actual SC-061 guarantee (`Mutex<T>: Sync` since
    /// `CompiledProcess: Send`).
    pub compiled: Mutex<Box<dyn CompiledProcess>>,
    pub queued_this_pass: AtomicBool,
    /// `ResumeDriver::start`'s once-only CAS (SC-059, Task 12, `resume.rs`).
    /// Not written by this file.
    pub started: AtomicBool,
    pub wait_state: Option<WaitSpec>,
    pub fork_parent: Option<ProcessId>,
    /// The name(s) a `disable <label>;` anywhere may single this live
    /// process out by (LRM 9.6.2) -- set once, at `Suspend::ForkJoin`
    /// handling time, from the spawning `SpawnSpec::disable_name` this
    /// process's own fork branch carried (`sched::control::Engine`'s own
    /// `Suspend::ForkJoin` arm, its one writer); empty for a process no
    /// label could ever single out (an unnamed branch of an unnamed fork,
    /// or a process not spawned by a fork at all). `Engine::
    /// collect_disable_named` is its one reader.
    pub disable_names: Vec<String>,
    /// Every live-or-recently-live immediate child of this process, across
    /// ALL of its forks (LRM 9.6.1: `wait fork` waits for every immediate
    /// child, 9.6.3: `disable fork` ends every descendant), oldest fork
    /// first. `handle_fork` prunes entries already reclaimed.
    pub fork_children: Vec<ProcessId>,
    /// How many forks this process has executed: the generation of its
    /// MOST RECENT fork. The join bookkeeping below (`fork_join_kind`,
    /// `fork_remaining`, `fork_any_signaled`) belongs to that fork only; a
    /// child of an earlier fork (`fork_group` != this) never touches it.
    pub fork_gen: u64,
    /// On a fork child: the parent's `fork_gen` at the moment this child was
    /// spawned, i.e. which fork of the parent it belongs to.
    pub fork_group: u64,
    pub registration_order: u64,
    pub fork_join_kind: Option<JoinKind>,
    pub fork_remaining: u32,
    pub fork_any_signaled: bool,
    /// Task 9 (B-sched wave, SC-125): `Some(n)` while this process is
    /// suspended via `Suspend::WaitFork`, counting how many of its own
    /// `fork_children` were still live (not yet `Suspend::Done` + reclaimed)
    /// at the moment `Engine::apply_suspend` installed this wait
    /// (`apply_wait_fork`, `sched/control.rs`). Independent of
    /// `fork_join_kind`/`fork_remaining` above (a real `wait fork;` is
    /// always preceded by a `fork ... join_none`, which leaves
    /// `fork_join_kind` permanently at `Some(JoinKind::None)` -- see
    /// `handle_fork`'s own doc comment -- so there is no real overlap
    /// between the two mechanisms). `None` when not currently waiting;
    /// cleared back to `None` by `ResumeDriver::on_child_done` the instant
    /// it reaches 0 and the parent has been notified via
    /// `ResumeArg::AllForkChildrenDone`.
    pub wait_fork_remaining: Option<u32>,
    /// Task 9 (SC-125): `Some((order, next_index))` while this process is
    /// suspended via `Suspend::WaitOrder(order)` (`apply_wait_order`,
    /// `sched/control.rs`) -- `next_index` is the index into `order` of the
    /// next completion this parent expects. `None` when not currently
    /// waiting on an ordered join.
    pub wait_order: Option<(Vec<ProcessId>, usize)>,
    /// Task 9 (SC-125): latched `true` the first time a `wait_order`-
    /// suspended parent observes an out-of-order completion
    /// (`ResumeDriver::on_child_done`). Once set, no later completion can
    /// ever satisfy this parent's wait -- every subsequent `on_child_done`
    /// call against it returns `Ok(None)`, never re-fires the violation and
    /// never silently accepts a reordering. The violation itself is
    /// surfaced exactly once, as `Err(ResumeError::WaitOrderViolated(child))`
    /// from the `on_child_done` call that first set this flag.
    pub wait_order_violated: bool,
    /// Task 9 (A2-part2, CG-102): this process's own `ProcessSideTable::
    /// region_id`, threaded from `SchedulerControl::register_process`'s new
    /// parameter (`sched/control.rs`'s own doc comment) -- defaults to `0`
    /// for a process registered directly through `ProcessTable::register`
    /// (fork children, and every Task 11/12-owned unit test in this file/
    /// `finals.rs`/`resume.rs`, none of which has any notion of `region_id`
    /// -- see `set_region_id`'s own doc comment for why this field's
    /// initial value is set here rather than by widening `register`'s own
    /// signature), which is also v1's own real, correct value for every one
    /// of those processes (ARCHITECTURE §4 item 1: a single conservative
    /// partition).
    pub region_id: u32,
    /// Classes plan Task 13 (LRM 9.7): `true` between `p.suspend()` and
    /// `p.resume()`. A suspended process's activations are parked (see
    /// `parked`) instead of dispatched.
    pub suspended: bool,
    /// The activation argument that arrived while `suspended`; replayed by
    /// `p.resume()`. Later arrivals coalesce into it (the last one wins).
    pub parked: Option<ResumeArg>,
    /// `Some(t)` while this process is blocked in `t.await()`.
    pub awaiting: Option<ProcessId>,
    /// Classes plan Task 17 (LRM 15.3): `true` while this process is blocked
    /// in a synchronisation-object call (`semaphore.get`) that the matching
    /// `put` has not yet released ([`super::ProcessRequest::SyncBlock`]/
    /// `SyncWake`).
    pub sync_blocked: bool,
    /// `true` while the process is yielded on a wait (`#`, `@`, `wait`,
    /// `join`, ...) -- `process::WAITING`. Maintained only once some `process`
    /// handle exists (`Engine::any_tracked`), so programs that never use the
    /// class pay nothing.
    pub waiting: bool,
    /// The event wait this process is blocked in (`Suspend::WaitEvents`),
    /// kept while `waiting` so `p.suspend()` can desensitise it (LRM 9.7.4)
    /// and `p.resume()` resensitise it (LRM 9.7.5). Maintained under the
    /// same gate as `waiting`.
    pub wait_set: Option<super::EventWaitSet>,
    /// `true` while `wait_set`'s dynamic sensitivity is removed because the
    /// process is suspended; `resume()` reinstalls it.
    pub desensitized: bool,
    /// Whether this process runs in the Reactive region (LRM 17.7.3: a
    /// checker's procedures, continuous assignments and output formal
    /// connections are "scheduled in the Reactive region, similarly to
    /// programs"). Every wake-up of such a process -- a sensitivity match,
    /// including the one a write made by another Reactive process causes --
    /// lands in the Reactive queue instead of Active. `false` for every
    /// design process.
    pub reactive: bool,
    /// A program `initial` thread or a fork descendant of one (LRM 24.3).
    pub program_thread: bool,
}

impl ProcessDescriptor {
    fn new(compiled: Box<dyn CompiledProcess>, registration_order: u64) -> Self {
        ProcessDescriptor {
            compiled: Mutex::new(compiled),
            queued_this_pass: AtomicBool::new(false),
            started: AtomicBool::new(false),
            wait_state: None,
            fork_parent: None,
            disable_names: Vec::new(),
            fork_children: Vec::new(),
            fork_gen: 0,
            fork_group: 0,
            registration_order,
            fork_join_kind: None,
            fork_remaining: 0,
            fork_any_signaled: false,
            wait_fork_remaining: None,
            wait_order: None,
            wait_order_violated: false,
            region_id: 0,
            suspended: false,
            parked: None,
            awaiting: None,
            sync_blocked: false,
            waiting: false,
            wait_set: None,
            desensitized: false,
            reactive: false,
            program_thread: false,
        }
    }
}

slotmap::new_key_type! {
    /// The `SlotMap`'s own internal key type (SC-063's literal "ProcessTable
    /// is a `SlotMap<ProcessId, ProcessDescriptor{...}>`" requirement). Never
    /// exposed outside this module — `ProcessId` (a plain, stable, densely-
    /// assigned `u32`) remains the only process identity every other module
    /// (`resume.rs`, `queues.rs`, `sched/mod.rs`'s own types) sees, via the
    /// side index below.
    pub struct ProcessKey;
}

/// Process storage, genuinely `slotmap::SlotMap`-backed (SC-063). `ProcessId`
/// (`u32`) stays the crate-visible identity: v1 never frees/reuses a slot
/// (fork children get fresh ids from the same monotonic sequence; nothing
/// calls `SlotMap::remove`), so `order_to_key[id as usize]` is a stable,
/// densely-assigned `ProcessId -> ProcessKey` side index — the same "assigned
/// order equals registration order, for the lifetime of a run" invariant the
/// previous `Vec`-backed slice already relied on, preserved exactly across
/// this migration.
#[derive(Default)]
pub struct ProcessTable {
    slots: slotmap::SlotMap<ProcessKey, ProcessDescriptor>,
    order_to_key: Vec<ProcessKey>,
}

impl ProcessTable {
    pub fn new() -> Self {
        ProcessTable {
            slots: slotmap::SlotMap::with_key(),
            order_to_key: Vec::new(),
        }
    }

    /// Registers a new process, assigning it the next `ProcessId` and
    /// `registration_order` (monotonic, never reused — v1 never frees a slot,
    /// matching Task 11's documented assumption that `ProcessId` assignment
    /// order equals `registration_order` for the lifetime of a run).
    /// Signature unchanged by the slotmap migration — `resume.rs` (Task 12)
    /// keeps compiling with no edit of its own.
    pub fn register(&mut self, compiled: Box<dyn CompiledProcess>) -> ProcessId {
        let id = self.order_to_key.len() as ProcessId;
        let key = self
            .slots
            .insert(ProcessDescriptor::new(compiled, id as u64));
        self.order_to_key.push(key);
        id
    }

    /// How many `ProcessId`s were ever assigned: ids are dense, `0..count()`.
    pub fn count(&self) -> usize {
        self.order_to_key.len()
    }

    /// Shared-reference lookup (Task 11's review fix) — lets `ResumeDriver`
    /// reach a descriptor's `Mutex`/`AtomicBool` fields through a shared
    /// `&ProcessTable`, never requiring `&mut` access to the whole table.
    /// Signature unchanged by the slotmap migration.
    pub fn get(&self, id: ProcessId) -> Option<&ProcessDescriptor> {
        let key = *self.order_to_key.get(id as usize)?;
        self.slots.get(key)
    }

    /// Mutable lookup — **added by Task 12** (a minimal, same-track,
    /// additive extension to this Task-11-owned file; recorded in
    /// `docs/CONTRACTS.md`/this commit's deviations, not a rewrite of
    /// anything Task 11 shipped). `handle_fork`/`on_child_done` mutate
    /// `fork_children`/`fork_parent`/`fork_join_kind`/`fork_remaining`/
    /// `fork_any_signaled` directly — plain, non-`Mutex` fields per Task
    /// 11/D1a — which needs exclusive access to the whole table via
    /// `&mut ProcessTable`. This does not reopen SC-061: Task 12's review
    /// fix note (resume.rs) already establishes this wave never calls
    /// `handle_fork`/`on_child_done` concurrently with `resume_for`/`start`
    /// (v1's default `SequentialController` dispatches everything from one
    /// thread; only the SC-061 stress test deliberately drives `resume_for`
    /// from many, and never touches fork state).
    pub fn get_mut(&mut self, id: ProcessId) -> Option<&mut ProcessDescriptor> {
        let key = *self.order_to_key.get(id as usize)?;
        self.slots.get_mut(key)
    }

    /// **Task 9 (A2-part2, CG-102).** Sets `id`'s own `region_id`, called by
    /// `SchedulerControl::register_process` (`sched/control.rs`) right after
    /// `register` -- a small, additive, same-track extension of this Task-
    /// 11-owned file (mirroring Task 12's own `get_mut` precedent, recorded
    /// there identically), rather than widening `register`'s own signature:
    /// `register`'s direct callers (`finals.rs`, `resume.rs`, this file's
    /// own unit tests, fork-child minting) have no notion of `region_id` at
    /// all and are unrelated to CG-102's scheduler-dispatch scope, so
    /// leaving `register`'s arity untouched keeps this a minimal, targeted
    /// addition instead of churning every one of those call sites for a
    /// field they'd only ever pass `0` for anyway (`ProcessDescriptor::new`
    /// already defaults `region_id` to that same real v1 value).
    pub fn set_region_id(&mut self, id: ProcessId, region_id: u32) {
        if let Some(desc) = self.get_mut(id) {
            desc.region_id = region_id;
        }
    }

    /// Marks `id` as a Reactive-region process ([`ProcessDescriptor::reactive`]).
    pub fn set_reactive(&mut self, id: ProcessId) {
        if let Some(desc) = self.get_mut(id) {
            desc.reactive = true;
        }
    }

    /// Whether any live fork child still belongs to a program thread
    /// (LRM 24.3).
    pub fn any_live_program_child(&self) -> bool {
        self.slots
            .values()
            .any(|d| d.program_thread && d.fork_parent.is_some())
    }

    /// Whether `id` runs in the Reactive region. `false` for an unknown id.
    pub fn is_reactive(&self, id: ProcessId) -> bool {
        self.get(id).is_some_and(|d| d.reactive)
    }

    /// **Task 9 (A2-part2, CG-102).** `region_id(&self, id: ProcessId) ->
    /// u32` -- the plan's own literal getter shape. Returns `0` (v1's own
    /// real, correct single-conservative-partition value, not a sentinel)
    /// for an unknown `id` rather than panicking or returning `Option`: a
    /// caller reaching this (e.g. `dispatch_ready_set`) only ever holds
    /// `ProcessId`s this same `Engine` minted, so an unknown id here would
    /// already indicate a bug in the caller's own bookkeeping, not a
    /// legitimate "no region_id set" state this getter needs to
    /// distinguish.
    pub fn region_id(&self, id: ProcessId) -> u32 {
        self.get(id).map(|d| d.region_id).unwrap_or(0)
    }

    /// **Task 7 (B-sched wave, ledger SC-062): the `SlotMap::remove`
    /// wrapper this file's own module doc comment already promised** ("...
    /// stable... until eventually calls `SlotMap::remove` -- a forward
    /// reference to work that was never done" -- now done). Removes `id`'s
    /// descriptor and returns it; a later `get`/`get_mut` for the same `id`
    /// returns `None`.
    ///
    /// **Deviation from the plan's literal text (recorded here and in
    /// `docs/CONTRACTS.md`):** the plan additionally describes tombstoning
    /// `id`'s own `order_to_key` side-index slot. That is unnecessary and
    /// deliberately not done: `slotmap::SlotMap::remove` already bumps the
    /// removed slot's internal generation, so the *same* `ProcessKey`
    /// value still sitting in `order_to_key[id]` no longer resolves via
    /// `self.slots.get`/`get_mut` -- this is the entire reason Task 11
    /// required a genuine `SlotMap` (SC-063) rather than a `Vec` in the
    /// first place. Adding a second, hand-maintained tombstone layer on
    /// top of that would only be able to drift out of sync with the
    /// `SlotMap`'s own source of truth; leaving `order_to_key` untouched
    /// keeps exactly one source of truth for liveness, and a stale/reused
    /// `ProcessId` still fails loudly via `get`'s existing `Option` return,
    /// exactly as the plan's own acceptance requires.
    pub fn remove(&mut self, id: ProcessId) -> Option<ProcessDescriptor> {
        let key = *self.order_to_key.get(id as usize)?;
        self.slots.remove(key)
    }
}

/// The scheduler's work-item type: wraps exactly one `resume()` invocation
/// (SC-057 — "the unit of parallel/schedulable work is one process
/// activation... within one activation the compiled kernel runs to
/// completion on one thread and never yields mid-statement"). Distinct from
/// `determinism::WorkItem` (which additionally carries the ordering key) —
/// `WorkUnit` is the pre-ordering, "what to run" half.
#[derive(Clone, Copy, Debug)]
pub struct WorkUnit {
    pub proc: ProcessId,
    pub arg: ResumeArg,
}

/// Collects every activation ready at one `(SchedTime, Region)` into a single
/// ready-set object before dispatch (SC-058). Drains `region`'s queue (Task
/// 4's `RegionQueues`) and hands back the un-sorted `Vec` — Task 2's
/// `sort_ready_set` is applied by the caller, kept separate so this
/// function's contract is purely "collect what's ready".
///
/// `Region::Nba` has no queue of its own (Task 4/6 — NBA is a staging log,
/// not a `RegionQueues` member), so it is rejected with the same typed
/// `RegionQueuePushError::NbaHasNoQueue` `RegionQueues::push` already uses,
/// exhaustive over all 7 `Region` variants with no wildcard arm — never a
/// silently-returned empty `Vec`, which a caller could not distinguish from
/// "genuinely nothing ready this pass".
///
/// **Deviation from the plan's literal signature** (recorded in
/// `docs/CONTRACTS.md` and this task's commit): the plan names a third
/// parameter, `gate: &QueuedGate`, to reset each drained activation's dedup
/// flag as it's pulled. `QueuedGate` is Task 10's own type
/// (`docs/superpowers/plans/2026-09-23-wave-A-sched.md` Task 10, ledger
/// SC-055/SC-056) and does not exist in this tree as of this commit — Task
/// 10 has not been implemented yet, and SC-055/SC-056 are not in this task's
/// assigned ledger scope (`["SC-057","SC-058","SC-063"]`). SC-058's own
/// literal acceptance text ("all processes ready at the same (time,region)
/// are collected into one ready-set object before dispatch") says nothing
/// about a dedup-flag reset, so this task implements exactly that, self-
/// contained. Task 10, when it lands, is expected to extend this function's
/// signature (or wrap it) to also reset each drained activation's dedup
/// entry — either via its own `QueuedGate` or via the `queued_this_pass`
/// `AtomicBool` already present on each activation's `ProcessDescriptor`
/// (`ProcessTable::get(act.proc)`), which this task deliberately leaves
/// untouched rather than inventing Task 10's own type ahead of its task.
pub fn form_ready_set(
    queues: &mut RegionQueues,
    region: Region,
) -> Result<Vec<PendingActivation>, RegionQueuePushError> {
    match region {
        Region::Active => Ok(queues.drain_active()),
        Region::Inactive => Ok(queues.drain_inactive()),
        Region::Reactive => Ok(queues.drain_reactive_once()),
        Region::Postponed => Ok(queues.drain_postponed_once()),
        Region::Nba => Err(RegionQueuePushError::NbaHasNoQueue),
        // sva T1 (§1.4, review fix, blocking finding 2): `Region::Preponed`
        // is queue-less for the identical reason `Region::Nba` is -- a live
        // SVA thread/snapshot push is never dispatched through
        // `RegionQueues`/`ProcessTable` as a `PendingActivation` at all.
        Region::Preponed => Err(RegionQueuePushError::PreponedHasNoQueue),
        // sva T2 (review fix, blocking finding 2), Observed's half: equally
        // queue-less -- a live SVA thread is owned by the automaton engine's
        // own thread table (a later task), never dispatched here either.
        Region::Observed => Err(RegionQueuePushError::ObservedHasNoQueue),
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::sched::{ResumeArg, SchedulerOps, Suspend};
    use std::sync::atomic::Ordering;
    use vj_abi::SimHandle;

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

    #[test]
    fn register_assigns_ids_in_call_order() {
        let mut table = ProcessTable::new();
        let a = table.register(Box::new(NoopProcess));
        let b = table.register(Box::new(NoopProcess));
        let c = table.register(Box::new(NoopProcess));
        assert_eq!([a, b, c], [0, 1, 2]);
        assert_eq!(table.get(a).unwrap().registration_order, 0);
        assert_eq!(table.get(b).unwrap().registration_order, 1);
        assert_eq!(table.get(c).unwrap().registration_order, 2);
    }

    #[test]
    fn get_returns_none_for_an_unregistered_id() {
        let mut table = ProcessTable::new();
        let only = table.register(Box::new(NoopProcess));
        assert!(table.get(only).is_some());
        assert!(table.get(only + 1).is_none());
    }

    #[test]
    fn new_descriptor_starts_unqueued_not_started_and_with_no_fork_state() {
        let mut table = ProcessTable::new();
        let id = table.register(Box::new(NoopProcess));
        let desc = table.get(id).unwrap();
        assert!(!desc.queued_this_pass.load(Ordering::Relaxed));
        assert!(!desc.started.load(Ordering::Relaxed));
        assert!(desc.wait_state.is_none());
        assert!(desc.fork_parent.is_none());
        assert!(desc.fork_children.is_empty());
        assert!(desc.fork_join_kind.is_none());
        assert_eq!(desc.fork_remaining, 0);
        assert!(!desc.fork_any_signaled);
        // Task 9 (SC-125): a freshly registered process is not mid-`wait
        // fork`/`wait_order` either.
        assert!(desc.wait_fork_remaining.is_none());
        assert!(desc.wait_order.is_none());
        assert!(!desc.wait_order_violated);
        // `compiled` is reachable only through the lock, never as a bare
        // `Box` — proves it is `Mutex<Box<dyn CompiledProcess>>`.
        let _guard = desc.compiled.lock().unwrap();
    }

    // --- Task 11 remaining work (SC-057, SC-058, SC-063) ---

    /// SC-063 — the slotmap migration's own regression proof beyond re-running
    /// the tests above: register 5 processes, assert the returned `ProcessId`s
    /// are still usable as stable, densely-assigned keys matching the
    /// pre-migration `Vec`-index behavior exactly. Fails if the
    /// `ProcessId <-> ProcessKey` side-index has an off-by-one or reordering
    /// bug the `Vec`-backed version couldn't have had.
    #[test]
    fn slotmap_migration_preserves_ids_and_registration_order() {
        let mut table = ProcessTable::new();
        let ids: Vec<ProcessId> = (0..5)
            .map(|_| table.register(Box::new(NoopProcess)))
            .collect();
        assert_eq!(ids, vec![0, 1, 2, 3, 4]);
        for id in ids {
            assert_eq!(
                table.get(id).unwrap().registration_order,
                id as u64,
                "ProcessId {id} must map to the descriptor registered at that order"
            );
        }
    }

    /// SC-063, review fix — a behavior-preserving migration alone doesn't
    /// prove the literal type requirement, since a `Vec` and a fresh
    /// `SlotMap` are behaviorally indistinguishable from outside when
    /// nothing is ever removed. Source-text scan (no shell plumbing, pure
    /// Rust `std::fs::read_to_string`), and — unlike the first version of
    /// this test — genuinely non-tautological:
    ///
    /// - It strips every `#[cfg(test)]` body before scanning, via the
    ///   shared `determinism::scan_support::strip_cfg_test_blocks` (the
    ///   same technique, reused rather than reimplemented, that Task 3's
    ///   `dt006_no_thread_spawn_or_rollback_code_outside_determinism_module`
    ///   and Task 22's `notify_change` scan already use, for the identical
    ///   reason). Without this, the scan reads its *own* assertion's string
    ///   literal back and is trivially true regardless of `ProcessTable`'s
    ///   real storage.
    /// - It searches for the field declaration's exact type signature
    ///   (`"slots: slotmap::SlotMap<ProcessKey"`), not the generic phrase
    ///   `"slotmap::SlotMap"`. That phrase also appears, correctly, in this
    ///   module's own doc comments describing the migration (this doc
    ///   comment included) — a scan for just the phrase would keep passing
    ///   even after `slots` regressed to `Vec<ProcessDescriptor>`, as long
    ///   as the prose describing the (no-longer-true) migration was left
    ///   stale, which is exactly the kind of edit that "happens to keep all
    ///   other tests green".
    ///
    /// `scan_is_genuinely_slotmap_backed` below is shared with
    /// `slotmap_scan_rejects_a_reverted_vec_even_with_stale_doc_comment_mentions`,
    /// which proves this check actually fails on a reverted file — the
    /// thing the original version of this test could never do.
    fn scan_is_genuinely_slotmap_backed(src: &str) -> bool {
        let production_only = crate::determinism::scan_support::strip_cfg_test_blocks(src);
        production_only.contains("slots: slotmap::SlotMap<ProcessKey")
    }

    #[test]
    fn process_table_is_genuinely_slotmap_backed() {
        let src = std::fs::read_to_string(concat!(
            env!("CARGO_MANIFEST_DIR"),
            "/src/sched/process_table.rs"
        ))
        .expect("must be able to read process_table.rs's own source");
        assert!(
            scan_is_genuinely_slotmap_backed(&src),
            "ProcessTable's `slots` field must be genuinely SlotMap-backed (SC-063), \
             not just behaviorally equivalent to one -- scanning production code \
             only (test bodies stripped) for the field's actual declared type"
        );
    }

    /// Review-fix regression proof: confirms the check above is not
    /// tautological by feeding `scan_is_genuinely_slotmap_backed` a
    /// synthetic "reverted to `Vec`" source that still contains the
    /// generic phrase `"slotmap::SlotMap"` both in a stale doc comment and
    /// inside a `#[cfg(test)]` body (exactly the shape that made the
    /// pre-fix version of this gate unable to ever fail) and asserts the
    /// real check correctly rejects it.
    #[test]
    fn slotmap_scan_rejects_a_reverted_vec_even_with_stale_doc_comment_mentions() {
        let reverted_src = concat!(
            "//! Process storage, genuinely `slotmap::SlotMap`-backed (SC-063).\n",
            "pub struct ProcessTable {\n",
            "    slots: Vec<ProcessDescriptor>,\n",
            "}\n",
            "#[cfg(test)]\n",
            "mod tests {\n",
            "    #[test]\n",
            "    fn t() {\n",
            "        assert!(\"x\".contains(\"slotmap::SlotMap\"));\n",
            "    }\n",
            "}\n",
        );
        // Sanity check on the fixture itself: the generic phrase IS present
        // (in both the doc comment and the #[cfg(test)] body) -- this is
        // exactly what made the pre-fix gate tautologically true.
        assert!(reverted_src.contains("slotmap::SlotMap"));
        assert!(
            !scan_is_genuinely_slotmap_backed(reverted_src),
            "a source with `slots: Vec<...>` must be rejected even though the \
             generic phrase \"slotmap::SlotMap\" still appears in a doc comment \
             and in a #[cfg(test)] body"
        );
    }

    /// Positive control for the same helper, kept separate from the
    /// filesystem-reading test above: guards against the marker string
    /// itself silently drifting out of sync with the real field
    /// declaration on line 92 (e.g. a whitespace/formatting change) by
    /// checking it against that exact declaration text in isolation.
    #[test]
    fn slotmap_scan_accepts_the_actual_current_field_declaration() {
        assert!(scan_is_genuinely_slotmap_backed(
            "slots: slotmap::SlotMap<ProcessKey, ProcessDescriptor>,"
        ));
    }

    /// SC-058 — push 4 activations at the same `(SchedTime, Region::Active)`
    /// via `RegionQueues::push` (Task 4); calling `form_ready_set` must
    /// collect all of them into one `Vec` (not 4 separate calls/objects).
    #[test]
    fn form_ready_set_collects_all_activations_at_one_time_region_as_one_object() {
        use crate::sched::{PendingActivation, RegionQueues, ResumeArg, SchedTime};
        use vj_abi::Region;

        let mut queues = RegionQueues::new();
        for i in 0..4u64 {
            queues
                .push(
                    Region::Active,
                    PendingActivation {
                        id: i,
                        proc: i as u32,
                        arg: ResumeArg::Start,
                        reconv_iter: 0,
                        logical_order: (SchedTime { ticks: 5 }, Region::Active, i),
                    },
                )
                .unwrap();
        }

        let ready = form_ready_set(&mut queues, Region::Active).unwrap();
        assert_eq!(
            ready.len(),
            4,
            "all processes ready at the same (time, region) must be collected \
             into one ready-set object before dispatch"
        );
    }

    /// SC-058 negative half — `Region::Nba` has no queue of its own (it is
    /// the NBA staging log, Task 6), so `form_ready_set` must reject it with
    /// the same typed error `RegionQueues::push` already uses, never a
    /// silent empty `Vec` that could be confused with "genuinely nothing
    /// ready this pass".
    #[test]
    fn form_ready_set_rejects_nba_with_a_typed_error_not_a_silent_empty_vec() {
        use crate::sched::{RegionQueuePushError, RegionQueues};
        use vj_abi::Region;

        let mut queues = RegionQueues::new();
        let result = form_ready_set(&mut queues, Region::Nba);
        assert_eq!(result.unwrap_err(), RegionQueuePushError::NbaHasNoQueue);
    }

    /// sva T1 (§1.4, review fix, blocking finding 2): the identical negative
    /// half for `Region::Preponed` -- a pure snapshot push with no ready
    /// processes of its own, never dispatched through `form_ready_set`.
    #[test]
    fn form_ready_set_preponed_returns_named_error() {
        use crate::sched::{RegionQueuePushError, RegionQueues};
        use vj_abi::Region;

        let mut queues = RegionQueues::new();
        let result = form_ready_set(&mut queues, Region::Preponed);
        assert_eq!(
            result.unwrap_err(),
            RegionQueuePushError::PreponedHasNoQueue
        );
    }

    /// sva T2 (review fix, blocking finding 2), Observed's half: the
    /// identical negative half for `Region::Observed` -- a live SVA thread
    /// owned by the automaton engine's own thread table, never dispatched
    /// through `form_ready_set`.
    #[test]
    fn form_ready_set_observed_returns_named_error() {
        use crate::sched::{RegionQueuePushError, RegionQueues};
        use vj_abi::Region;

        let mut queues = RegionQueues::new();
        let result = form_ready_set(&mut queues, Region::Observed);
        assert_eq!(
            result.unwrap_err(),
            RegionQueuePushError::ObservedHasNoQueue
        );
    }

    /// SC-057 — a type-level test: `WorkUnit` has no field representing a
    /// partial statement offset (only `proc`+`arg`); combined with
    /// `CompiledProcess::resume`'s own signature (unchanged: `&mut self,
    /// state, arg, ops) -> Suspend`), there is no API in this module that
    /// could invoke half of a `resume()` call. Checked by `WorkUnit`'s field
    /// list via exhaustive destructuring (a third field added later without
    /// updating this test would fail to compile).
    #[test]
    fn work_unit_wraps_exactly_one_resume_invocation_no_mid_statement_split() {
        use crate::sched::ResumeArg;

        let unit = WorkUnit {
            proc: 7,
            arg: ResumeArg::Start,
        };
        let WorkUnit { proc, arg } = unit;
        assert_eq!(proc, 7);
        assert!(matches!(arg, ResumeArg::Start));
    }

    /// Task 9 (A2-part2, CG-102): a freshly registered process defaults to
    /// `region_id == 0` (v1's own real value, not a placeholder -- see
    /// `ProcessDescriptor::region_id`'s own doc comment), and `set_region_id`
    /// genuinely stores whatever value it is given, read back unchanged by
    /// `region_id` -- fails if either getter/setter ignores its argument.
    #[test]
    fn region_id_defaults_to_zero_and_set_region_id_is_read_back_by_region_id() {
        let mut table = ProcessTable::new();
        let id = table.register(Box::new(NoopProcess));
        assert_eq!(table.region_id(id), 0);

        table.set_region_id(id, 9);
        assert_eq!(
            table.region_id(id),
            9,
            "set_region_id/region_id must round-trip a real value, not stay pinned at 0"
        );
    }

    /// `region_id` on an unregistered id returns the same real v1 default
    /// (`0`) rather than panicking -- see its own doc comment.
    #[test]
    fn region_id_of_an_unregistered_id_is_zero_not_a_panic() {
        let table = ProcessTable::new();
        assert_eq!(table.region_id(42), 0);
    }

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

    /// The literal SC-062 acceptance clause this task closes: "the child's
    /// slot is reclaimed" -- `remove` must drop the descriptor from the
    /// `SlotMap` itself (not merely mark it dead some other way), proven by
    /// a `get` afterward returning `None`. Fails if `remove` is a no-op, or
    /// if it only clears some other field while leaving the slot live.
    #[test]
    fn remove_drops_the_descriptor_and_a_later_get_returns_none() {
        let mut table = ProcessTable::new();
        let a = table.register(Box::new(NoopProcess));
        let b = table.register(Box::new(NoopProcess));

        assert!(table.get(a).is_some());
        let removed = table.remove(a);
        assert!(
            removed.is_some(),
            "remove must return the removed descriptor, not None, for a live id"
        );
        assert!(
            table.get(a).is_none(),
            "a later get() for a removed ProcessId must return None -- the slot \
             itself must be gone from the SlotMap, not just logically ignored"
        );
        // A sibling process registered before the removed one must be
        // completely unaffected -- proves `remove` does not disturb the
        // rest of the table (e.g. by shifting indices the way a `Vec::
        // remove` would).
        assert!(
            table.get(b).is_some(),
            "removing one process must not affect an unrelated, still-live one"
        );
    }

    /// A second `remove` of the same, already-removed id is a harmless
    /// `None`, never a panic -- `SlotMap::remove`'s own generational-key
    /// semantics already guarantee this, but this pins it as this crate's
    /// own contract too (never a silent double-free-shaped bug surfacing
    /// as a panic instead of a typed `None`).
    #[test]
    fn removing_an_already_removed_id_returns_none_not_a_panic() {
        let mut table = ProcessTable::new();
        let a = table.register(Box::new(NoopProcess));
        assert!(table.remove(a).is_some());
        assert!(table.remove(a).is_none());
    }

    /// `remove` of an id that was never registered at all returns `None`,
    /// never a panic -- mirrors `get`'s own established posture for an
    /// out-of-range id.
    #[test]
    fn removing_an_unregistered_id_returns_none_not_a_panic() {
        let mut table = ProcessTable::new();
        assert!(table.remove(42).is_none());
    }
}