vj-sim 0.3.0

//! Resume/suspend orchestration (Task 12,
//! `docs/superpowers/plans/2026-09-23-wave-A-sched.md`).
//!
//! **Scope:** ledger `["SC-059","SC-060","SC-061"]`. `ResumeDriver::resume_for`
//! (SC-061, the multithreaded-stress-test-proof method) shipped in an earlier
//! commit and is unmodified here — see the fix note below, kept for reference.
//! This commit adds the rest of Task 12: `start` (SC-059 — `resume(Start)`
//! exactly once per process, via `ProcessDescriptor::started`'s CAS), the
//! general one-`Suspend`-variant-to-one-matching-`ResumeArg` dispatch proven by
//! `each_suspend_variant_gets_the_matching_resume_arg_exactly_once` (SC-060,
//! hand-orchestrated against `resume_for`/`handle_fork`/`on_child_done` since
//! Task 13/21's real wiring doesn't exist in this tree yet), and
//! `handle_fork`/`on_child_done` (fork/join per Resolved design decision D1,
//! and D1a's v1 sched-internal shadow bookkeeping on `ProcessDescriptor` —
//! `fork_join_kind`/`fork_remaining`/`fork_any_signaled`/`fork_parent`/
//! `fork_children`, all shipped unread/unwritten by Task 11).
//!
//! **The SC-061 fix itself:** an earlier draft of this plan had `resume_for` take
//! `table: &mut ProcessTable`, and its test wrapped every thread's call in one
//! external `Mutex<ProcessTable>` held for the whole call. Combined, those two
//! choices meant Rust's own `&mut`-exclusivity rules — not any logic inside
//! `ResumeDriver` — made concurrent entry impossible: a `ResumeDriver` with *no*
//! reentrancy guard of its own would have passed that test exactly as trivially as
//! a correct one (a gate that cannot fail). The fix (this file): `resume_for` takes
//! a **shared** `&ProcessTable` (via `ProcessTable::get`, `process_table.rs`) and
//! reaches the process's state entirely through `ProcessDescriptor`'s own interior
//! mutability — `compiled: Mutex<Box<dyn CompiledProcess>>` is locked for exactly
//! the duration of the `resume()` call (acquired, `.resume(..)` called while held,
//! released only when the guard drops afterward — never acquired-then-released
//! before calling `resume`, which would reopen the same race). With no table-wide
//! lock left anywhere, only `ResumeDriver` correctly locking the right process's
//! `Mutex` for the whole call is what prevents two threads' `resume()` calls from
//! genuinely overlapping under real OS thread parallelism — which is what the test
//! below actually exercises.

use std::sync::atomic::Ordering;

use vj_abi::SimHandle;

use super::{
    CompiledProcess, JoinKind, ProcessId, ProcessTable, ResumeArg, SchedulerOps, SpawnSpec, Suspend,
};

/// Errors `ResumeDriver` returns instead of ever panicking or silently dropping a
/// call — this crate's correctness posture requires a loud, typed error on every
/// rejected/invalid call, never a silent no-op.
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
pub enum ResumeError {
    UnknownProcess(ProcessId),
    /// SC-059: `ResumeDriver::start` was called a second time for a
    /// `ProcessId` whose `ProcessDescriptor::started` CAS was already `true`
    /// — rejected loudly rather than calling `resume(Start)` twice.
    AlreadyStarted(ProcessId),
    /// **NEW (Task 9, B-sched wave, ledger SC-125).** A `wait_order(...)`-
    /// suspended parent observed `ProcessId` complete out of its own
    /// declared order — returned by [`ResumeDriver::on_child_done`] instead
    /// of ever silently accepting the reordering or resuming the parent via
    /// `ResumeArg::WaitOrderSatisfied` anyway.
    WaitOrderViolated(ProcessId),
}

impl std::fmt::Display for ResumeError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            ResumeError::UnknownProcess(id) => {
                write!(f, "no process registered for ProcessId {id}")
            }
            ResumeError::AlreadyStarted(id) => {
                write!(
                    f,
                    "ProcessId {id} was already started -- resume(Start) may only fire once"
                )
            }
            ResumeError::WaitOrderViolated(id) => {
                write!(
                    f,
                    "ProcessId {id} completed out of its parent's declared wait_order sequence"
                )
            }
        }
    }
}

impl std::error::Error for ResumeError {}

/// Drives `CompiledProcess::resume()` calls (scheduler.md §5): `start`
/// (SC-059), `resume_for` (SC-060/SC-061), and fork/join orchestration
/// (`handle_fork`/`on_child_done`, Resolved design decision D1). See this
/// module's doc comment for scope and the SC-061 fix `resume_for` embodies.
pub struct ResumeDriver;

impl ResumeDriver {
    /// SC-061: "the scheduler must never call resume() concurrently on the same
    /// ProcessId; one process activation is the atomicity unit." See this
    /// module's doc comment for the fix this method embodies.
    pub fn resume_for(
        &self,
        table: &ProcessTable,
        proc: ProcessId,
        arg: ResumeArg,
        state: SimHandle,
        ops: &dyn SchedulerOps,
    ) -> Result<Suspend, ResumeError> {
        let desc = table.get(proc).ok_or(ResumeError::UnknownProcess(proc))?;
        // The lock is held across the whole `.resume(..)` call below (never
        // acquired-and-released before calling it) — this is the actual SC-061
        // guarantee: two threads racing this method for the same `proc` serialize
        // on this one lock, and nothing else in this method (or in `ProcessTable`)
        // provides any exclusivity of its own.
        let mut guard = desc
            .compiled
            .lock()
            .expect("ProcessDescriptor::compiled mutex poisoned by a panicking resume()");
        // Review fix (A2-part2 r1, binding rule 7): `resume_for` is one of
        // the two places (`start` is the other) that knows the real `proc`
        // *before* dispatching into `CompiledProcess::resume` -- exactly the
        // ambient-state hook `kernel_adapter::with_current_process`/
        // `real_nba_enqueue` need to attribute an NBA write to its real
        // `ProcessId` despite the frozen `ProcessFn` ABI carrying none.
        Ok(super::kernel_adapter::with_current_process(proc, || {
            guard.resume(state, arg, ops)
        }))
    }

    /// SC-059: "the scheduler must call resume(Start) exactly once to begin
    /// each registered process." The once-only guarantee is
    /// `ProcessDescriptor::started`'s `AtomicBool` CAS: a losing
    /// `compare_exchange` (already `true`) is rejected with
    /// `Err(ResumeError::AlreadyStarted(proc))` *before* `compiled`'s lock is
    /// even acquired, so a second `start` call never reaches `resume()` at
    /// all — never a panic, matching `resume_for`'s typed-error style for the
    /// same reason (misuse rejection, not exhaustive-match fallback, but the
    /// same no-silent-no-op posture).
    pub fn start(
        &self,
        table: &ProcessTable,
        proc: ProcessId,
        state: SimHandle,
        ops: &dyn SchedulerOps,
    ) -> Result<Suspend, ResumeError> {
        let desc = table.get(proc).ok_or(ResumeError::UnknownProcess(proc))?;
        if desc
            .started
            .compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst)
            .is_err()
        {
            return Err(ResumeError::AlreadyStarted(proc));
        }
        let mut guard = desc
            .compiled
            .lock()
            .expect("ProcessDescriptor::compiled mutex poisoned by a panicking resume()");
        // Review fix (A2-part2 r1, binding rule 7): see the identical note
        // on `resume_for` above.
        Ok(super::kernel_adapter::with_current_process(proc, || {
            guard.resume(state, ResumeArg::Start, ops)
        }))
    }

    /// Fork/join per Resolved design decision D1 (`docs/CONTRACTS.md`), using
    /// D1a's v1 sched-internal shadow bookkeeping on `ProcessDescriptor`
    /// (`fork_join_kind`/`fork_remaining`/`fork_any_signaled`, plain fields —
    /// not `process::ForkFrame`, which `sched` does not own/construct this
    /// wave). Registers one child per `SpawnSpec` via `ProcessTable::register`
    /// (`mint_child` stands in for "codegen emits each branch as its own
    /// process-table kernel"), writes each new child's own
    /// `ProcessDescriptor::fork_parent = Some(parent)` (LRM 9.3.2 — this is
    /// `handle_fork`'s one and only writer of `fork_parent`), and sets the
    /// parent's `fork_children`/`fork_join_kind`/`fork_remaining`/
    /// `fork_any_signaled` per `kind` (`All => children.len(), Any =>
    /// 1.min(children.len()), None => 0` — `None` leaves `fork_remaining`
    /// at 0, the caller-visible "parent proceeds now, no `on_child_done`
    /// needed" signal). `table: &mut ProcessTable` (not the shared
    /// `&ProcessTable` `resume_for`/`start` take) is correct here: this
    /// wave never calls `handle_fork` concurrently with anything else (see
    /// this module's doc comment).
    pub fn handle_fork(
        &self,
        table: &mut ProcessTable,
        parent: ProcessId,
        kind: JoinKind,
        spawns: &[SpawnSpec],
        mint_child: &mut dyn FnMut(&SpawnSpec) -> Box<dyn CompiledProcess>,
    ) -> Vec<ProcessId> {
        // Join bookkeeping is per FORK, not per parent process: a parent that
        // runs `fork..join_none; #2; fork..join` has a stray child of the
        // first fork still running when it blocks on the second. Each fork
        // gets a fresh generation, stamped on its children, and
        // `on_child_done` lets only the CURRENT generation's children touch
        // the parent's join counters (LRM 9.3.2: a join waits for the
        // processes spawned by its own fork).
        let group = table.get(parent).map_or(0, |d| d.fork_gen) + 1;
        let program_thread = table.get(parent).is_some_and(|d| d.program_thread);
        let mut children = Vec::with_capacity(spawns.len());
        for spec in spawns {
            let compiled = mint_child(spec);
            let child = table.register(compiled);
            if let Some(child_desc) = table.get_mut(child) {
                child_desc.fork_parent = Some(parent);
                child_desc.fork_group = group;
                child_desc.program_thread = program_thread;
            }
            children.push(child);
        }

        let remaining = match kind {
            JoinKind::All => children.len() as u32,
            JoinKind::Any => 1u32.min(children.len() as u32),
            JoinKind::None => 0,
        };
        // Earlier forks' children stay listed (`wait fork`/`disable fork`
        // cover every immediate child, LRM 9.6.1/9.6.3) except the ones
        // already reclaimed, pruned here so a `forever` loop of
        // `fork..join_none` does not grow the list without bound.
        let mut all_children: Vec<ProcessId> = table
            .get(parent)
            .map(|d| d.fork_children.clone())
            .unwrap_or_default();
        all_children.retain(|&c| table.get(c).is_some());
        all_children.extend(children.iter().copied());
        if let Some(parent_desc) = table.get_mut(parent) {
            parent_desc.fork_children = all_children;
            parent_desc.fork_gen = group;
            parent_desc.fork_join_kind = Some(kind);
            parent_desc.fork_remaining = remaining;
            parent_desc.fork_any_signaled = false;
        }

        children
    }

    /// D1: reports one fork child's completion to its parent, returning the
    /// parent's own next `resume()` argument exactly when the join condition
    /// is satisfied — `None` otherwise (parent not resumed yet, or the
    /// parent's join was already satisfied and this is a later, detached
    /// child under `JoinKind::Any`). Looks up `child`'s parent via the
    /// *child's own* `ProcessDescriptor::fork_parent` (an O(1) read set by
    /// `handle_fork` above), never a linear scan over every parent's
    /// `fork_children`.
    ///
    /// **Task 9 (B-sched wave, SC-125) additions:** before the ordinary
    /// `fork_join_kind` (`All`/`Any`/`None`) computation below, this method
    /// now also checks `child`'s parent for an active `wait_order`
    /// (`ProcessDescriptor::wait_order`, installed by `Engine::
    /// apply_wait_order`) or `wait_fork` (`wait_fork_remaining`, installed
    /// by `Engine::apply_wait_fork`) wait — both genuinely independent of
    /// `fork_join_kind` in practice (a real `wait fork`/`wait_order` is
    /// always preceded by a `fork ... join_none`, which leaves
    /// `fork_join_kind` permanently at `Some(JoinKind::None)`, itself
    /// always a no-op arm below — see `handle_fork`'s own doc comment).
    /// Return type changed from `Option<ResumeArg>` to
    /// `Result<Option<ResumeArg>, ResumeError>` to carry
    /// `ResumeError::WaitOrderViolated` — every pre-existing call site
    /// (this file's own tests, `sched::control::Engine::apply_suspend`) is
    /// updated accordingly; the ordinary `All`/`Any`/`None` paths still
    /// only ever produce `Ok(..)`, never `Err`.
    pub fn on_child_done(
        &self,
        table: &mut ProcessTable,
        child: ProcessId,
    ) -> Result<Option<ResumeArg>, ResumeError> {
        // Review fix (Task 7 r1, SC-062): the parent lookup below is allowed
        // to fail -- e.g. a nested/multi-level fork tree where the
        // intermediate parent is itself a `JoinKind::None` fork child that
        // completed and was reclaimed synchronously (via `control.rs`'s
        // `apply_suspend` recursive `Suspend::ForkJoin`->`Suspend::Done`
        // continuation) *before* this, its own child, ever ran -- so
        // `table.get_mut(parent)` can legitimately return `None` here. That
        // must never prevent `child`'s own slot from being reclaimed below:
        // `child` itself unconditionally returned `Suspend::Done`, and its
        // slot is `child`'s to reclaim regardless of whether its parent is
        // still around to be notified. The join-satisfaction computation is
        // therefore isolated in this inner closure, which is allowed to
        // short-circuit to `Ok(None)` on its own without early-returning out
        // of `on_child_done` itself.
        let result = (|| -> Result<Option<ResumeArg>, ResumeError> {
            let parent = match table.get(child).and_then(|d| d.fork_parent) {
                Some(p) => p,
                None => return Ok(None),
            };

            // Task 9 (SC-125): `wait_order`, checked first.
            if let Some(parent_desc) = table.get_mut(parent)
                && parent_desc.wait_order.is_some()
            {
                if parent_desc.wait_order_violated {
                    // Already-latched violation (see this field's own doc
                    // comment): this parent's wait can never be satisfied
                    // again -- every later completion among its declared
                    // `order` is a no-op here, never a second `Err` and
                    // never a silently-accepted late match.
                    return Ok(None);
                }
                let (order, next_idx) = parent_desc.wait_order.clone().unwrap();
                // Review fix (Task 9 r1): `wait_order(...)` only cares about
                // ordering among the identifiers actually named in `order`.
                // A sibling fork branch that isn't a member of `order` at
                // all (e.g. `fork a; b; c; join_none; wait_order(a, c);` --
                // `b` is not listed) completing early or late is a no-op
                // here, never a violation: it isn't part of the sequence
                // being watched, so it can neither advance nor break it.
                if !order.contains(&child) {
                    return Ok(None);
                }
                if next_idx < order.len() && child == order[next_idx] {
                    let advanced = next_idx + 1;
                    if advanced == order.len() {
                        parent_desc.wait_order = None;
                        return Ok(Some(ResumeArg::WaitOrderSatisfied));
                    }
                    parent_desc.wait_order = Some((order, advanced));
                    return Ok(None);
                }
                // Out-of-order completion -- latch the violation and
                // surface it loudly now, never silently accept it and never
                // panic.
                parent_desc.wait_order_violated = true;
                return Err(ResumeError::WaitOrderViolated(child));
            }

            // Task 9 (SC-125): `wait_fork`, independent of `wait_order`
            // above.
            if let Some(parent_desc) = table.get_mut(parent)
                && let Some(remaining) = parent_desc.wait_fork_remaining
            {
                let remaining = remaining.saturating_sub(1);
                return if remaining == 0 {
                    parent_desc.wait_fork_remaining = None;
                    Ok(Some(ResumeArg::AllForkChildrenDone))
                } else {
                    parent_desc.wait_fork_remaining = Some(remaining);
                    Ok(None)
                };
            }

            // A child of an EARLIER fork (e.g. a `join_none` child still
            // running) must not count against the join of the fork the
            // parent is blocked on now -- see `handle_fork`.
            let child_group = table.get(child).map_or(0, |d| d.fork_group);
            let parent_desc = match table.get_mut(parent) {
                Some(d) => d,
                None => return Ok(None),
            };
            if child_group != parent_desc.fork_gen {
                return Ok(None);
            }
            Ok(match parent_desc.fork_join_kind {
                Some(JoinKind::All) => {
                    parent_desc.fork_remaining = parent_desc.fork_remaining.saturating_sub(1);
                    if parent_desc.fork_remaining == 0 {
                        // Review fix: one-shot latch, symmetric with the `Any` arm's
                        // `fork_any_signaled` guard below. Without this, `fork_remaining`
                        // stays saturated at 0 forever (`saturating_sub` never goes
                        // negative), so a later `on_child_done` call against this same
                        // parent -- a duplicate/retried completion notification, or a
                        // future scheduler bug reporting one child twice -- would still
                        // match `Some(All)` and `fork_remaining == 0` and re-fire
                        // `Some(ForkChildDone)` a second time, handing the caller a
                        // second `resume()` to fire against a parent that already
                        // resumed past its wait (violates SC-060's "exactly one matching
                        // resume() call fires" guarantee). Clearing `fork_join_kind`
                        // makes this match fall to the `None` arm on any subsequent
                        // call for this parent, exactly like `Any`'s `fork_any_signaled`
                        // check does once it has already signaled.
                        parent_desc.fork_join_kind = None;
                        Some(ResumeArg::ForkChildDone { child })
                    } else {
                        None
                    }
                }
                Some(JoinKind::Any) => {
                    if parent_desc.fork_any_signaled {
                        None
                    } else {
                        parent_desc.fork_any_signaled = true;
                        parent_desc.fork_remaining = parent_desc.fork_remaining.saturating_sub(1);
                        Some(ResumeArg::ForkChildDone { child })
                    }
                }
                Some(JoinKind::None) | None => None,
            })
        })();
        // Task 7 (ledger SC-062): "Suspend::Done is terminal: the scheduler
        // drops or recycles the process slot; for fork children, it
        // notifies the parent per the block's join semantics" -- two
        // separate actions, not one. `on_child_done` is only ever called
        // (by `Engine::apply_suspend`'s own `Suspend::Done` arm) once
        // `child` has itself genuinely returned `Suspend::Done`, so `child`
        // is unconditionally reclaimed here regardless of whether this
        // particular completion happens to satisfy/partially-satisfy the
        // parent's join (`result` being `Ok(Some(..))`) or not -- an early
        // `All` completion that does not yet zero `fork_remaining`, a later
        // `Any` straggler arriving after that join already fired, every
        // `JoinKind::None` detached child, and a `wait_order` violation are
        // all still real, one-time `Suspend::Done`s for `child` itself, and
        // each must have its slot reclaimed exactly once. This unconditional
        // reclaim now also covers the case where the `result` closure above
        // bailed out early because `child`'s own parent (or the parent's
        // `fork_join_kind`) was no longer live in the table at all -- e.g.
        // an intermediate parent in a nested fork tree that was itself
        // already reclaimed by a `JoinKind::None` synchronous continuation
        // before `child` ever got a chance to run. Regardless of *why* the
        // join computation above did or didn't fire, `child`'s own slot
        // must not leak. A *repeated* notification for an already-removed
        // `child` (this function's own one-shot-latch tests call
        // `on_child_done` again for a child already reported done) reaches
        // `ProcessTable::remove` a second time, which is a harmless `None`
        // (`slotmap`'s own generational keys), never a panic or a double
        // free.
        table.remove(child);
        result
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::sched::EventWaitSet;
    use crate::state::{FourStateValue, StateCellId};
    use std::hash::{Hash, Hasher};
    use std::sync::{Arc, Mutex};
    use vj_abi::Region;

    /// A no-op `SchedulerOps` double, `Sync` so `&dyn SchedulerOps` can be shared
    /// across the SC-061 test's 8 threads. Never touches `state`/`cell`/`value`
    /// arguments.
    struct NoopOps;
    impl SchedulerOps for NoopOps {
        fn now(&self) -> super::super::SchedTime {
            super::super::SchedTime { ticks: 0 }
        }
        fn current_region(&self) -> Region {
            Region::Active
        }
        fn stage_nba(&self, _cell: StateCellId, _value: FourStateValue, _source: ProcessId) {}
        fn commit_blocking_write(&self, _cell: StateCellId, _value: FourStateValue) {}
        fn request_finish(&self, _code: i32) {}
        fn defer_io(
            &self,
            _at: super::super::SchedTime,
            _region: Region,
            _cb: super::super::DeferredIoCallback,
        ) {
        }
        fn register_region_callback(
            &self,
            _region: Region,
            _at: super::super::SchedTime,
            _cb: Box<dyn super::super::RegionCallback>,
        ) {
        }
    }

    /// The in-crate mock standing in for a real CPS-lowered kernel (Task 12's
    /// `CountingProcess`, this task's slice of it): a reentrancy trap. `inside`
    /// (a plain, non-atomic `Cell<bool>`) is set `true` on entry and back to
    /// `false` on exit; a `resume()` call observing it already `true` on entry, or
    /// finding it flipped back to `false` mid-call, proves a second call
    /// genuinely overlapped this one under real OS-thread parallelism. A small
    /// thread-id-derived sleep between the two checks widens the race window so a
    /// missing/incorrect lock has a real, high-probability chance of tripping the
    /// trap rather than merely a theoretical one.
    struct CountingProcess {
        inside: std::cell::Cell<bool>,
    }

    impl CountingProcess {
        fn new() -> Self {
            CountingProcess {
                inside: std::cell::Cell::new(false),
            }
        }
    }

    impl CompiledProcess for CountingProcess {
        fn resume(
            &mut self,
            _state: SimHandle,
            _arg: ResumeArg,
            _ops: &dyn SchedulerOps,
        ) -> Suspend {
            assert!(
                !self.inside.replace(true),
                "SC-061 violated: resume() called while another resume() for the \
                 same ProcessId was already in progress"
            );
            let mut hasher = std::collections::hash_map::DefaultHasher::new();
            std::thread::current().id().hash(&mut hasher);
            let jitter_micros = hasher.finish() % 500;
            std::thread::sleep(std::time::Duration::from_micros(jitter_micros));
            assert!(
                self.inside.get(),
                "SC-061 violated: the reentrancy flag was clobbered mid-call by a \
                 concurrent resume() for the same ProcessId"
            );
            self.inside.set(false);
            Suspend::Done
        }
    }

    /// SC-061 — review fix (was a gate that could not fail, see this module's doc
    /// comment): register 1 `CountingProcess`, share the table via
    /// `Arc<ProcessTable>` (no external lock of any kind — `resume_for` takes
    /// `&ProcessTable`, so a cloned `Arc` is all each thread needs), and drive 8
    /// OS threads' `resume_for` calls at the *same* `ProcessId` concurrently, with
    /// nothing synchronizing them besides whatever `ResumeDriver`/
    /// `ProcessDescriptor::compiled`'s own `Mutex` provides internally.
    ///
    /// Why a plausible wrong implementation fails this: `resume_for` must route
    /// every call through `desc.compiled.lock().unwrap().resume(..)`, holding the
    /// lock across the call. An implementation that (a) never locks `compiled` at
    /// all, or (b) locks and releases it *before* calling `resume`, lets two
    /// threads' `resume()` calls genuinely overlap under real OS parallelism; the
    /// randomized jitter inside `CountingProcess::resume` makes that overlap trip
    /// the reentrancy trap (an `assert!` panic) with high probability across 8
    /// threads, which surfaces here as a `JoinHandle::join()` returning `Err`.
    #[test]
    fn resume_is_never_called_concurrently_for_the_same_process_id() {
        let mut table = ProcessTable::new();
        let proc = table.register(Box::new(CountingProcess::new()));
        let table = Arc::new(table);
        let ops = Arc::new(NoopOps);

        let mut handles = Vec::with_capacity(8);
        for _ in 0..8 {
            let table = Arc::clone(&table);
            let ops = Arc::clone(&ops);
            handles.push(std::thread::spawn(move || {
                // `vj_abi::SimHandle` wraps a raw `*mut c_void` and derives no
                // `Send`, so it cannot be captured directly by this `'static`
                // thread closure. Neither `NoopOps` nor `CountingProcess::resume`
                // ever dereferences it, so moving its address across as a `usize`
                // and reconstructing it here (test-local only, no edit to the
                // `abi`-owned `vj-abi` crate) is sound for this test.
                let state = SimHandle(std::ptr::null_mut());
                let driver = ResumeDriver;
                driver.resume_for(&table, proc, ResumeArg::TimeElapsed, state, ops.as_ref())
            }));
        }

        let mut ok_count = 0usize;
        for h in handles {
            let result = h
                .join()
                .expect("resume_for thread panicked -- SC-061's reentrancy trap tripped");
            assert!(result.is_ok(), "resume_for returned an error: {result:?}");
            ok_count += 1;
        }
        assert_eq!(
            ok_count, 8,
            "all 8 concurrent resume_for calls must complete"
        );
    }

    #[test]
    fn resume_for_an_unregistered_process_id_returns_a_typed_error_not_a_panic() {
        let table = ProcessTable::new();
        let ops = NoopOps;
        let driver = ResumeDriver;
        let result = driver.resume_for(
            &table,
            0,
            ResumeArg::TimeElapsed,
            SimHandle(std::ptr::null_mut()),
            &ops,
        );
        assert_eq!(result.unwrap_err(), ResumeError::UnknownProcess(0));
    }

    /// Mutation-test kill: an implementation that discards `CompiledProcess::
    /// resume()`'s real return value (e.g. always reporting `Ok(Suspend::Done)`)
    /// passes every other test in this file, since they only check `is_ok()`/
    /// the error path. This process's `resume()` return value is keyed off the
    /// `ResumeArg` it actually receives, so this test also kills a mutant that
    /// forwards the wrong `ResumeArg` to `resume()`.
    struct ArgEchoProcess;
    impl CompiledProcess for ArgEchoProcess {
        fn resume(
            &mut self,
            _state: SimHandle,
            arg: ResumeArg,
            _ops: &dyn SchedulerOps,
        ) -> Suspend {
            let tag = match arg {
                ResumeArg::Start => 1,
                ResumeArg::TimeElapsed => 2,
                ResumeArg::EventFired { .. } => 3,
                ResumeArg::ForkChildDone { .. } => 4,
                ResumeArg::Disabled => 5,
                ResumeArg::AllForkChildrenDone => 6,
                ResumeArg::WaitOrderSatisfied => 7,
            };
            Suspend::Delay(tag)
        }
    }

    #[test]
    fn resume_for_returns_the_real_suspend_value_for_the_resume_arg_it_was_given() {
        let mut table = ProcessTable::new();
        let proc = table.register(Box::new(ArgEchoProcess));
        let ops = NoopOps;
        let driver = ResumeDriver;

        let result = driver.resume_for(
            &table,
            proc,
            ResumeArg::Disabled,
            SimHandle(std::ptr::null_mut()),
            &ops,
        );
        match result {
            Ok(Suspend::Delay(5)) => {}
            other => panic!(
                "resume_for must return the compiled process's actual resume() \
                 result for the ResumeArg it was actually called with (expected \
                 Ok(Suspend::Delay(5)) for ResumeArg::Disabled), got {other:?}"
            ),
        }
    }

    /// Mutation-test kill: a `resume_for` that reacts to lock contention with
    /// `try_lock` + "silently report `Suspend::Done` without calling `resume()`"
    /// still passes `resume_is_never_called_concurrently_for_the_same_process_id`
    /// above (that test only asserts no panic and that all calls return `Ok`,
    /// which a skip trivially satisfies too). This test proves the *other* half
    /// of SC-061: a contended caller must block on the process's own lock and
    /// genuinely invoke `resume()`, never skip it.
    struct RendezvousProcess {
        call_count: Arc<std::sync::atomic::AtomicUsize>,
        started_tx: std::sync::mpsc::Sender<()>,
        release_rx: std::sync::mpsc::Receiver<()>,
    }
    impl CompiledProcess for RendezvousProcess {
        fn resume(
            &mut self,
            _state: SimHandle,
            _arg: ResumeArg,
            _ops: &dyn SchedulerOps,
        ) -> Suspend {
            self.call_count
                .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
            self.started_tx.send(()).expect("test harness rx dropped");
            self.release_rx
                .recv()
                .expect("test harness did not release this call");
            Suspend::Done
        }
    }

    #[test]
    fn resume_for_blocks_a_contended_caller_instead_of_silently_skipping_it() {
        use std::sync::atomic::{AtomicUsize, Ordering};
        use std::sync::mpsc;
        use std::time::Duration;

        let call_count = Arc::new(AtomicUsize::new(0));
        let (started_tx, started_rx) = mpsc::channel();
        let (release_tx, release_rx) = mpsc::channel();

        let mut table = ProcessTable::new();
        let proc = table.register(Box::new(RendezvousProcess {
            call_count: Arc::clone(&call_count),
            started_tx,
            release_rx,
        }));
        let table = Arc::new(table);
        let ops = Arc::new(NoopOps);

        // Thread A takes the process's lock and parks inside resume() until we
        // release it, so it holds the lock for the whole window below.
        let table_a = Arc::clone(&table);
        let ops_a = Arc::clone(&ops);
        let a = std::thread::spawn(move || {
            let driver = ResumeDriver;
            driver.resume_for(
                &table_a,
                proc,
                ResumeArg::TimeElapsed,
                SimHandle(std::ptr::null_mut()),
                ops_a.as_ref(),
            )
        });
        started_rx
            .recv_timeout(Duration::from_secs(2))
            .expect("thread A never entered resume() -- lock never acquired");

        // Thread B races for the SAME ProcessId while A still holds the lock.
        let table_b = Arc::clone(&table);
        let ops_b = Arc::clone(&ops);
        let (b_done_tx, b_done_rx) = mpsc::channel();
        let b = std::thread::spawn(move || {
            let driver = ResumeDriver;
            let result = driver.resume_for(
                &table_b,
                proc,
                ResumeArg::TimeElapsed,
                SimHandle(std::ptr::null_mut()),
                ops_b.as_ref(),
            );
            let _ = b_done_tx.send(());
            result
        });

        // While A still holds the lock, B must NOT have returned yet: a correct
        // `resume_for` blocks on the contended process's own `Mutex` rather than
        // `try_lock`-ing and silently reporting `Suspend::Done` on failure.
        assert!(
            b_done_rx.recv_timeout(Duration::from_millis(200)).is_err(),
            "resume_for returned for a contended ProcessId before the holder \
             released the lock -- it must block on the contended process's own \
             lock, never silently skip the call"
        );

        // Release A, then B; both must complete and both must have genuinely
        // invoked resume() (never overlapping, per SC-061 — that's covered by
        // the other test above).
        release_tx.send(()).expect("release channel closed early");
        let a_result = a.join().expect("thread A panicked");
        assert!(
            a_result.is_ok(),
            "thread A's resume_for errored: {a_result:?}"
        );

        release_tx.send(()).expect("release channel closed early");
        let b_result = b.join().expect("thread B panicked");
        assert!(
            b_result.is_ok(),
            "thread B's resume_for errored: {b_result:?}"
        );

        assert_eq!(
            call_count.load(Ordering::SeqCst),
            2,
            "both contended resume_for calls must have genuinely invoked \
             resume() -- a try_lock-and-skip implementation would leave this at 1"
        );
    }

    /// Mutation-test kill: `ResumeError`'s `Display` impl must report the actual
    /// unknown `ProcessId`, not a fixed placeholder -- nothing else in this file
    /// inspects the message text (only `PartialEq`/`Debug` via `unwrap_err`).
    #[test]
    fn resume_error_display_reports_the_actual_process_id() {
        let err = ResumeError::UnknownProcess(42);
        let msg = err.to_string();
        assert!(
            msg.contains("42"),
            "ResumeError::Display must include the actual unknown ProcessId \
             (42), got: {msg:?}"
        );
    }

    // --- Task 12 remaining work (SC-059, SC-060, fork/join D1) ---

    /// A comparable, owned mirror of `ResumeArg` (which derives neither
    /// `PartialEq` nor `Hash` — it is a cross-track contract type this task
    /// does not own, see `docs/CONTRACTS.md`) — lets the tests below assert
    /// on exactly which `ResumeArg` variant/payload a call log entry was,
    /// without touching `ResumeArg`'s own derive list.
    #[derive(Clone, Copy, PartialEq, Eq, Debug)]
    enum ArgKind {
        Start,
        TimeElapsed,
        EventFired(usize),
        ForkChildDone(ProcessId),
        Disabled,
        AllForkChildrenDone,
        WaitOrderSatisfied,
    }
    impl From<ResumeArg> for ArgKind {
        fn from(arg: ResumeArg) -> Self {
            match arg {
                ResumeArg::Start => ArgKind::Start,
                ResumeArg::TimeElapsed => ArgKind::TimeElapsed,
                ResumeArg::EventFired { term_index } => ArgKind::EventFired(term_index),
                ResumeArg::AllForkChildrenDone => ArgKind::AllForkChildrenDone,
                ResumeArg::WaitOrderSatisfied => ArgKind::WaitOrderSatisfied,
                ResumeArg::ForkChildDone { child } => ArgKind::ForkChildDone(child),
                ResumeArg::Disabled => ArgKind::Disabled,
            }
        }
    }

    /// The in-crate mock standing in for a real CPS-lowered kernel (this
    /// task's own `CountingProcess` sketch, named `ScriptedProcess` here to
    /// avoid colliding with the SC-061 reentrancy-trap mock above, which the
    /// plan's Status note says is the actually-shipped mock under that
    /// name): records every `(ResumeArg, call_index)` it receives into a
    /// shared, externally-inspectable log (a plain `Box<dyn CompiledProcess>`
    /// can't be downcast back out of `ProcessDescriptor::compiled` after
    /// registration, so the log is an `Arc<Mutex<..>>` the test keeps its
    /// own clone of), and returns a scripted sequence of `Suspend`s from a
    /// `VecDeque` pre-loaded at construction (past the end, `Suspend::Done`).
    struct ScriptedProcess {
        log: Arc<Mutex<Vec<(ArgKind, usize)>>>,
        script: std::collections::VecDeque<Suspend>,
    }
    impl ScriptedProcess {
        fn new(log: Arc<Mutex<Vec<(ArgKind, usize)>>>, script: Vec<Suspend>) -> Self {
            ScriptedProcess {
                log,
                script: script.into(),
            }
        }
    }
    impl CompiledProcess for ScriptedProcess {
        fn resume(
            &mut self,
            _state: SimHandle,
            arg: ResumeArg,
            _ops: &dyn SchedulerOps,
        ) -> Suspend {
            let mut log = self.log.lock().unwrap();
            let idx = log.len();
            log.push((ArgKind::from(arg), idx));
            drop(log);
            self.script.pop_front().unwrap_or(Suspend::Done)
        }
    }

    fn null_state() -> SimHandle {
        SimHandle(std::ptr::null_mut())
    }

    /// A `mint_child` double every `handle_fork` test below shares: mints a
    /// fresh `ScriptedProcess` (its own, unshared log — these tests only
    /// inspect the *parent's* log / the `ProcessTable`'s fork-bookkeeping
    /// fields) that immediately reports `Suspend::Done` when resumed.
    fn mint_done_child(_: &SpawnSpec) -> Box<dyn CompiledProcess> {
        Box::new(ScriptedProcess::new(
            Arc::new(Mutex::new(Vec::new())),
            vec![Suspend::Done],
        ))
    }

    /// SC-059 — "the scheduler must call resume(Start) exactly once to begin
    /// each registered process."
    #[test]
    fn resume_start_is_called_exactly_once_per_process() {
        let log = Arc::new(Mutex::new(Vec::new()));
        let mut table = ProcessTable::new();
        let proc = table.register(Box::new(ScriptedProcess::new(
            Arc::clone(&log),
            vec![Suspend::Done],
        )));
        let ops = NoopOps;
        let driver = ResumeDriver;

        let first = driver.start(&table, proc, null_state(), &ops);
        assert!(first.is_ok(), "the first start() must succeed: {first:?}");

        let second = driver.start(&table, proc, null_state(), &ops);
        assert_eq!(
            second.unwrap_err(),
            ResumeError::AlreadyStarted(proc),
            "a second start() for the same process must be rejected, never panic"
        );

        let log = log.lock().unwrap();
        assert_eq!(
            log.as_slice(),
            &[(ArgKind::Start, 0)],
            "the rejected second start() call must never have reached resume() \
             at all -- exactly one Start entry, not two"
        );
    }

    /// SC-060 — "the scheduler must call resume(arg) exactly once per
    /// satisfied Suspend the process last returned, with `arg` describing
    /// why." 4 table-driven sub-cases, one per `Suspend` variant, each with
    /// its own fresh `ProcessTable` (no cross-contamination between
    /// sub-cases run in the same test process).
    #[test]
    fn each_suspend_variant_gets_the_matching_resume_arg_exactly_once() {
        let ops = NoopOps;
        let driver = ResumeDriver;

        // Suspend::Delay -> ResumeArg::TimeElapsed.
        {
            let log = Arc::new(Mutex::new(Vec::new()));
            let mut table = ProcessTable::new();
            let proc = table.register(Box::new(ScriptedProcess::new(
                Arc::clone(&log),
                vec![Suspend::Delay(10), Suspend::Done],
            )));
            let started = driver.start(&table, proc, null_state(), &ops).unwrap();
            assert!(
                matches!(started, Suspend::Delay(10)),
                "expected Suspend::Delay(10), got {started:?}"
            );
            driver
                .resume_for(&table, proc, ResumeArg::TimeElapsed, null_state(), &ops)
                .unwrap();
            let log = log.lock().unwrap();
            assert_eq!(
                log.as_slice(),
                &[(ArgKind::Start, 0), (ArgKind::TimeElapsed, 1)]
            );
        }

        // Suspend::WaitEvents -> ResumeArg::EventFired{term_index}.
        {
            let log = Arc::new(Mutex::new(Vec::new()));
            let mut table = ProcessTable::new();
            let proc = table.register(Box::new(ScriptedProcess::new(
                Arc::clone(&log),
                vec![Suspend::WaitEvents(EventWaitSet::default()), Suspend::Done],
            )));
            driver.start(&table, proc, null_state(), &ops).unwrap();
            driver
                .resume_for(
                    &table,
                    proc,
                    ResumeArg::EventFired { term_index: 2 },
                    null_state(),
                    &ops,
                )
                .unwrap();
            let log = log.lock().unwrap();
            assert_eq!(
                log.as_slice(),
                &[(ArgKind::Start, 0), (ArgKind::EventFired(2), 1)]
            );
        }

        // Suspend::ForkJoin -> (via handle_fork/on_child_done) ResumeArg::ForkChildDone.
        {
            let log = Arc::new(Mutex::new(Vec::new()));
            let mut table = ProcessTable::new();
            let spawns = vec![SpawnSpec {
                kernel: 0,
                entry_pc: 0,
                parent: 0, // the parent below is this table's first registration -> id 0.
                disable_name: Vec::new(),
            }];
            let parent = table.register(Box::new(ScriptedProcess::new(
                Arc::clone(&log),
                vec![Suspend::ForkJoin(JoinKind::All, spawns), Suspend::Done],
            )));
            assert_eq!(
                parent, 0,
                "test assumes the parent is this table's first registration"
            );

            let started = driver.start(&table, parent, null_state(), &ops).unwrap();
            let (kind, spawns) = match started {
                Suspend::ForkJoin(kind, spawns) => (kind, spawns),
                other => panic!("expected Suspend::ForkJoin, got {other:?}"),
            };
            let mut mint = mint_done_child;
            let children = driver.handle_fork(&mut table, parent, kind, &spawns, &mut mint);
            assert_eq!(children.len(), 1);

            let arg = driver
                .on_child_done(&mut table, children[0])
                .expect("an ordinary All join's on_child_done never errors")
                .expect("the only child of an All join must signal the parent on completion");
            driver
                .resume_for(&table, parent, arg, null_state(), &ops)
                .unwrap();

            let log = log.lock().unwrap();
            assert_eq!(log.len(), 2);
            assert_eq!(log[0], (ArgKind::Start, 0));
            match log[1].0 {
                ArgKind::ForkChildDone(child) => assert_eq!(child, children[0]),
                other => panic!("expected ArgKind::ForkChildDone, got {other:?}"),
            }
        }

        // A disabled dynamic wait -> ResumeArg::Disabled.
        {
            let log = Arc::new(Mutex::new(Vec::new()));
            let mut table = ProcessTable::new();
            let proc = table.register(Box::new(ScriptedProcess::new(
                Arc::clone(&log),
                vec![Suspend::WaitEvents(EventWaitSet::default()), Suspend::Done],
            )));
            driver.start(&table, proc, null_state(), &ops).unwrap();
            driver
                .resume_for(&table, proc, ResumeArg::Disabled, null_state(), &ops)
                .unwrap();
            let log = log.lock().unwrap();
            assert_eq!(
                log.as_slice(),
                &[(ArgKind::Start, 0), (ArgKind::Disabled, 1)]
            );
        }
    }

    /// D1 — `JoinKind::All` resumes the parent only after every child
    /// reports done.
    #[test]
    fn join_all_resumes_parent_only_after_every_child_reports_done() {
        let mut table = ProcessTable::new();
        let parent = table.register(Box::new(ScriptedProcess::new(
            Arc::new(Mutex::new(Vec::new())),
            vec![Suspend::Done],
        )));
        let spawns = vec![
            SpawnSpec {
                kernel: 0,
                entry_pc: 0,
                parent,
                disable_name: Vec::new(),
            },
            SpawnSpec {
                kernel: 1,
                entry_pc: 0,
                parent,
                disable_name: Vec::new(),
            },
            SpawnSpec {
                kernel: 2,
                entry_pc: 0,
                parent,
                disable_name: Vec::new(),
            },
        ];
        let driver = ResumeDriver;
        let mut mint = mint_done_child;
        let children = driver.handle_fork(&mut table, parent, JoinKind::All, &spawns, &mut mint);
        assert_eq!(children.len(), 3);

        assert!(
            driver
                .on_child_done(&mut table, children[0])
                .unwrap()
                .is_none(),
            "must not resume the parent after only 1 of 3 children"
        );
        assert!(
            driver
                .on_child_done(&mut table, children[1])
                .unwrap()
                .is_none(),
            "must not resume the parent after only 2 of 3 children"
        );
        match driver.on_child_done(&mut table, children[2]) {
            Ok(Some(ResumeArg::ForkChildDone { child })) => assert_eq!(child, children[2]),
            other => panic!(
                "expected Ok(Some(ForkChildDone{{child: children[2]}})) once all 3 \
                 children are done, got {other:?}"
            ),
        }
    }

    /// D1, review fix — a repeated/extra `on_child_done` call against a
    /// parent whose `All` join has already fired must return `None`, never
    /// re-signal. Mirrors `join_any_resumes_parent_on_first_child_others_
    /// continue_detached`'s double-call proof for `Any`, but for `All`:
    /// before the fix, `fork_remaining` stayed saturated at 0 forever
    /// (`saturating_sub` never underflows) and `fork_join_kind` was never
    /// cleared, so this 4th call would still match `Some(All)` +
    /// `fork_remaining == 0` and wrongly return
    /// `Some(ForkChildDone { child: children[2] })` a second time.
    #[test]
    fn join_all_on_child_done_is_a_one_shot_latch_after_the_join_already_fired() {
        let mut table = ProcessTable::new();
        let parent = table.register(Box::new(ScriptedProcess::new(
            Arc::new(Mutex::new(Vec::new())),
            vec![Suspend::Done],
        )));
        let spawns = vec![
            SpawnSpec {
                kernel: 0,
                entry_pc: 0,
                parent,
                disable_name: Vec::new(),
            },
            SpawnSpec {
                kernel: 1,
                entry_pc: 0,
                parent,
                disable_name: Vec::new(),
            },
            SpawnSpec {
                kernel: 2,
                entry_pc: 0,
                parent,
                disable_name: Vec::new(),
            },
        ];
        let driver = ResumeDriver;
        let mut mint = mint_done_child;
        let children = driver.handle_fork(&mut table, parent, JoinKind::All, &spawns, &mut mint);
        assert_eq!(children.len(), 3);

        assert!(
            driver
                .on_child_done(&mut table, children[0])
                .unwrap()
                .is_none()
        );
        assert!(
            driver
                .on_child_done(&mut table, children[1])
                .unwrap()
                .is_none()
        );
        match driver.on_child_done(&mut table, children[2]) {
            Ok(Some(ResumeArg::ForkChildDone { child })) => assert_eq!(child, children[2]),
            other => panic!(
                "expected Ok(Some(ForkChildDone{{child: children[2]}})) once all 3 \
                 children are done, got {other:?}"
            ),
        }

        // A repeated/extra completion notification for a child of this same,
        // already-satisfied parent (e.g. `children[2]` reported done twice,
        // or a duplicate delivery of `children[1]`'s completion) must not
        // hand the caller a second `resume()` to fire.
        assert!(
            driver
                .on_child_done(&mut table, children[2])
                .unwrap()
                .is_none(),
            "All must not double-signal the parent once its join has already \
             fired -- a repeated on_child_done call must return None"
        );
        assert!(
            driver
                .on_child_done(&mut table, children[1])
                .unwrap()
                .is_none(),
            "a duplicate completion report for a different already-done child \
             must also return None, not re-fire the parent's resume"
        );
    }

    /// D1 — `JoinKind::Any` resumes the parent on the first child; later
    /// children finish detached, with no double-signal.
    #[test]
    fn join_any_resumes_parent_on_first_child_others_continue_detached() {
        let mut table = ProcessTable::new();
        let parent = table.register(Box::new(ScriptedProcess::new(
            Arc::new(Mutex::new(Vec::new())),
            vec![Suspend::Done],
        )));
        let spawns = vec![
            SpawnSpec {
                kernel: 0,
                entry_pc: 0,
                parent,
                disable_name: Vec::new(),
            },
            SpawnSpec {
                kernel: 1,
                entry_pc: 0,
                parent,
                disable_name: Vec::new(),
            },
            SpawnSpec {
                kernel: 2,
                entry_pc: 0,
                parent,
                disable_name: Vec::new(),
            },
        ];
        let driver = ResumeDriver;
        let mut mint = mint_done_child;
        let children = driver.handle_fork(&mut table, parent, JoinKind::Any, &spawns, &mut mint);
        assert_eq!(children.len(), 3);

        match driver.on_child_done(&mut table, children[0]) {
            Ok(Some(ResumeArg::ForkChildDone { child })) => assert_eq!(child, children[0]),
            other => panic!(
                "expected Ok(Some(ForkChildDone{{child: children[0]}})) on the first \
                 Any child, got {other:?}"
            ),
        }
        assert!(
            driver
                .on_child_done(&mut table, children[1])
                .unwrap()
                .is_none(),
            "Any must not double-signal the parent once it has already resumed"
        );

        assert_eq!(
            table.get(parent).unwrap().fork_children,
            children,
            "detached children must not be forgotten from fork_children, just \
             not blocking"
        );
    }

    /// LRM 9.3.2: a join waits for the processes of ITS OWN fork. A child of
    /// an earlier `join_none` fork that is still running when the parent
    /// blocks on a later `join`/`join_any` must not decrement that fork's
    /// counters (the join bookkeeping is per fork, not per parent), while
    /// `fork_children` keeps both forks' children for `wait fork`.
    #[test]
    fn stray_child_of_an_earlier_fork_never_satisfies_a_later_join() {
        for kind in [JoinKind::All, JoinKind::Any] {
            let mut table = ProcessTable::new();
            let parent = table.register(Box::new(ScriptedProcess::new(
                Arc::new(Mutex::new(Vec::new())),
                vec![Suspend::Done],
            )));
            let spec = |kernel| SpawnSpec {
                kernel,
                entry_pc: 0,
                parent,
                disable_name: Vec::new(),
            };
            let driver = ResumeDriver;
            let mut mint = mint_done_child;
            let stray =
                driver.handle_fork(&mut table, parent, JoinKind::None, &[spec(0)], &mut mint);
            let own = driver.handle_fork(&mut table, parent, kind, &[spec(1)], &mut mint);
            assert_eq!(
                table.get(parent).unwrap().fork_children,
                [stray[0], own[0]],
                "{kind:?}: wait fork / disable fork must still see the earlier fork's child"
            );
            assert!(
                driver
                    .on_child_done(&mut table, stray[0])
                    .unwrap()
                    .is_none(),
                "{kind:?}: a stray child of an earlier fork must not release the later join"
            );
            match driver.on_child_done(&mut table, own[0]) {
                Ok(Some(ResumeArg::ForkChildDone { child })) => assert_eq!(child, own[0]),
                other => panic!("{kind:?}: the fork's own child must release it, got {other:?}"),
            }
        }
    }

    /// Mutation-test kill: `on_child_done`'s `Any` arm gates on
    /// `fork_any_signaled`, never on `fork_remaining` reaching a particular
    /// value, so a `handle_fork` that set `fork_remaining` to
    /// `children.len()` instead of `1.min(children.len())` would pass every
    /// other test in this file (`join_any_resumes_parent_on_first_child_
    /// others_continue_detached` never reads `fork_remaining` directly).
    /// This test asserts `handle_fork`'s own documented field directly,
    /// immediately after the call, before any `on_child_done`.
    #[test]
    fn handle_fork_any_sets_fork_remaining_to_one_regardless_of_child_count() {
        let mut table = ProcessTable::new();
        let parent = table.register(Box::new(ScriptedProcess::new(
            Arc::new(Mutex::new(Vec::new())),
            vec![Suspend::Done],
        )));
        let spawns = vec![
            SpawnSpec {
                kernel: 0,
                entry_pc: 0,
                parent,
                disable_name: Vec::new(),
            },
            SpawnSpec {
                kernel: 1,
                entry_pc: 0,
                parent,
                disable_name: Vec::new(),
            },
            SpawnSpec {
                kernel: 2,
                entry_pc: 0,
                parent,
                disable_name: Vec::new(),
            },
        ];
        let driver = ResumeDriver;
        let mut mint = mint_done_child;
        let children = driver.handle_fork(&mut table, parent, JoinKind::Any, &spawns, &mut mint);
        assert_eq!(children.len(), 3);

        assert_eq!(
            table.get(parent).unwrap().fork_remaining,
            1,
            "JoinKind::Any must set fork_remaining to 1.min(children.len()) == 1, \
             never children.len() -- Any only ever waits for the first child"
        );
    }

    /// D1 — `JoinKind::None` never blocks the parent: `handle_fork` returns
    /// with `fork_remaining == 0` immediately, with no `on_child_done` call
    /// required to unblock it.
    #[test]
    fn join_none_never_blocks_the_parent() {
        let mut table = ProcessTable::new();
        let parent = table.register(Box::new(ScriptedProcess::new(
            Arc::new(Mutex::new(Vec::new())),
            vec![Suspend::Done],
        )));
        let spawns = vec![SpawnSpec {
            kernel: 0,
            entry_pc: 0,
            parent,
            disable_name: Vec::new(),
        }];
        let driver = ResumeDriver;
        let mut mint = mint_done_child;
        driver.handle_fork(&mut table, parent, JoinKind::None, &spawns, &mut mint);

        let desc = table.get(parent).unwrap();
        assert_eq!(desc.fork_join_kind, Some(JoinKind::None));
        assert_eq!(
            desc.fork_remaining, 0,
            "JoinKind::None must leave fork_remaining at 0 -- the caller-visible \
             'parent proceeds now' signal, with no on_child_done call needed"
        );
    }

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

    /// SC-062's own literal acceptance: "A JoinAll fork child returning
    /// Suspend::Done correctly satisfies... the parent's join-wait; the
    /// child's slot is reclaimed." A `JoinAll` parent with a single child:
    /// once that child reports done, both halves must hold -- (a) the
    /// existing `ForkChildDone` signal (unchanged behavior), and (b) the
    /// child's own `ProcessTable` entry must be gone. Fails if slot reclaim
    /// is never wired at all, or if it removes the wrong process (e.g. the
    /// parent instead of the child).
    #[test]
    fn on_child_done_reclaims_the_childs_slot_when_the_join_is_satisfied() {
        let mut table = ProcessTable::new();
        let parent = table.register(Box::new(ScriptedProcess::new(
            Arc::new(Mutex::new(Vec::new())),
            vec![Suspend::Done],
        )));
        let spawns = vec![SpawnSpec {
            kernel: 0,
            entry_pc: 0,
            parent,
            disable_name: Vec::new(),
        }];
        let driver = ResumeDriver;
        let mut mint = mint_done_child;
        let children = driver.handle_fork(&mut table, parent, JoinKind::All, &spawns, &mut mint);
        assert_eq!(children.len(), 1);
        let child = children[0];
        assert!(
            table.get(child).is_some(),
            "the freshly minted child must be live before it ever completes"
        );

        match driver.on_child_done(&mut table, child) {
            Ok(Some(ResumeArg::ForkChildDone { child: signaled })) => {
                assert_eq!(signaled, child, "the join must signal for the real child")
            }
            other => panic!("expected Ok(Some(ForkChildDone{{child}})), got {other:?}"),
        }
        assert!(
            table.get(child).is_none(),
            "the child's own ProcessTable slot must be reclaimed once its \
             Suspend::Done has satisfied the parent's join"
        );
        assert!(
            table.get(parent).is_some(),
            "reclaiming the child must never remove the parent"
        );
    }

    /// A `JoinKind::None` child is never awaited (`on_child_done` always
    /// returns `None` for it), but it is still a genuine `Suspend::Done`
    /// for the child itself and must still be reclaimed -- proving reclaim
    /// is unconditional on the *child's* own completion, not gated on
    /// whether this particular completion happened to signal the parent.
    #[test]
    fn on_child_done_reclaims_a_detached_join_none_childs_slot_even_though_it_never_signals() {
        let mut table = ProcessTable::new();
        let parent = table.register(Box::new(ScriptedProcess::new(
            Arc::new(Mutex::new(Vec::new())),
            vec![Suspend::Done],
        )));
        let spawns = vec![SpawnSpec {
            kernel: 0,
            entry_pc: 0,
            parent,
            disable_name: Vec::new(),
        }];
        let driver = ResumeDriver;
        let mut mint = mint_done_child;
        let children = driver.handle_fork(&mut table, parent, JoinKind::None, &spawns, &mut mint);
        let child = children[0];

        assert!(
            driver.on_child_done(&mut table, child).unwrap().is_none(),
            "JoinKind::None never signals the parent"
        );
        assert!(
            table.get(child).is_none(),
            "a JoinKind::None child's slot must still be reclaimed on its own \
             Suspend::Done, even though it never satisfies any join"
        );
    }

    /// D1, review fix — an earlier draft of this test let an implementation
    /// that never writes `fork_parent` pass anyway via an "or the SpawnSpec
    /// fed to mint_child" escape hatch that only checked the test's own
    /// construction input, trivially true by construction. This version
    /// asserts directly on each minted child's *own table entry*, with no
    /// fallback to the test's own `SpawnSpec.parent` input.
    #[test]
    fn spawn_spec_children_reach_parent_scope_via_the_parent_field() {
        let mut table = ProcessTable::new();
        let parent = table.register(Box::new(ScriptedProcess::new(
            Arc::new(Mutex::new(Vec::new())),
            vec![Suspend::Done],
        )));
        let spawns = vec![
            SpawnSpec {
                kernel: 0,
                entry_pc: 0,
                parent,
                disable_name: Vec::new(),
            },
            SpawnSpec {
                kernel: 1,
                entry_pc: 0,
                parent,
                disable_name: Vec::new(),
            },
        ];
        let driver = ResumeDriver;
        let mut mint = mint_done_child;
        let children = driver.handle_fork(&mut table, parent, JoinKind::All, &spawns, &mut mint);
        assert_eq!(children.len(), 2);

        for child in children {
            assert_eq!(
                table.get(child).unwrap().fork_parent,
                Some(parent),
                "handle_fork must write each minted child's own fork_parent \
                 field -- checked directly on the child's table entry, not via \
                 the test's own SpawnSpec.parent input"
            );
        }
    }

    /// Review fix (Task 7 r1, SC-062) — regression for the leak found in
    /// review: a nested fork tree where the *intermediate* parent is itself
    /// reclaimed (its own `on_child_done` call against *its* parent
    /// succeeds and removes it from the table) before its own child ever
    /// reports done. When that child later completes, `on_child_done` must
    /// still reclaim the child's own slot even though `table.get_mut
    /// (parent)` now fails -- this is exactly the shape `control.rs`'s
    /// `apply_suspend` produces for a `JoinKind::None` fork: the forking
    /// process's own next segment is resumed synchronously, in the same
    /// call, and can itself return `Suspend::Done` (triggering its own
    /// `on_child_done` and reclaim) before any of the children it just
    /// spawned have been dispatched at all. Built from `handle_fork`/
    /// `on_child_done` calls alone, no codegen/`Engine`, mirroring this
    /// module's other fork-tree tests.
    ///
    /// Before the fix: `on_child_done`'s `table.get(child).and_then(|d|
    /// d.fork_parent)?` and `table.get_mut(parent)?` early-returned `None`
    /// as soon as the grandchild's parent (`intermediate`) was missing from
    /// the table, so the final `table.remove(child)` was never reached and
    /// the grandchild's slot leaked forever.
    #[test]
    fn on_child_done_reclaims_the_childs_slot_even_when_its_own_parent_was_already_reclaimed() {
        let mut table = ProcessTable::new();
        let driver = ResumeDriver;

        // Grandparent `G`, joined `All` on a single child: the intermediate
        // process `P`.
        let grandparent = table.register(Box::new(ScriptedProcess::new(
            Arc::new(Mutex::new(Vec::new())),
            vec![Suspend::Done],
        )));
        let spawns_for_intermediate = vec![SpawnSpec {
            kernel: 0,
            entry_pc: 0,
            parent: grandparent,
            disable_name: Vec::new(),
        }];
        let mut mint = mint_done_child;
        let intermediate_children = driver.handle_fork(
            &mut table,
            grandparent,
            JoinKind::All,
            &spawns_for_intermediate,
            &mut mint,
        );
        assert_eq!(intermediate_children.len(), 1);
        let intermediate = intermediate_children[0];

        // `P` (the intermediate) itself forks a `JoinKind::None` child --
        // the grandchild whose slot this test is protecting.
        let spawns_for_grandchild = vec![SpawnSpec {
            kernel: 0,
            entry_pc: 0,
            parent: intermediate,
            disable_name: Vec::new(),
        }];
        let grandchildren = driver.handle_fork(
            &mut table,
            intermediate,
            JoinKind::None,
            &spawns_for_grandchild,
            &mut mint,
        );
        assert_eq!(grandchildren.len(), 1);
        let grandchild = grandchildren[0];
        assert!(
            table.get(grandchild).is_some(),
            "the grandchild must be live immediately after being forked"
        );

        // `P` now completes its own `Suspend::Done` -- exactly what
        // `apply_suspend`'s synchronous `JoinKind::None` continuation
        // produces when the forking process has nothing left to do right
        // after spawning. This satisfies `G`'s `All` join (its only child
        // was `P`) and reclaims `P`'s own slot -- *before* the grandchild
        // above has been dispatched at all.
        match driver.on_child_done(&mut table, intermediate) {
            Ok(Some(ResumeArg::ForkChildDone { child })) => assert_eq!(child, intermediate),
            other => panic!(
                "expected Ok(Some(...)) for P's own completion to satisfy G's \
                 single-child All join, got {other:?}"
            ),
        }
        assert!(
            table.get(intermediate).is_none(),
            "P's own slot must be reclaimed once its Suspend::Done satisfied G's join"
        );

        // The grandchild now completes. Its own parent (`intermediate`) is
        // already gone from the table -- this must not stop its own slot
        // from being reclaimed.
        assert!(
            driver
                .on_child_done(&mut table, grandchild)
                .unwrap()
                .is_none(),
            "the grandchild's parent is gone, so there is no join left to \
             satisfy -- but that must not stop the grandchild's own slot \
             from being reclaimed below"
        );
        assert!(
            table.get(grandchild).is_none(),
            "the grandchild's own ProcessTable slot must still be reclaimed \
             on its own Suspend::Done even though its parent was already \
             removed from the table by the time it completed -- this is the \
             leak found in review"
        );
    }

    // --- Task 9 (B-sched wave, ledger SC-125) ---
    //
    // These exercise `ResumeDriver::on_child_done`'s new `wait_fork`/
    // `wait_order` arms directly against a bare `ProcessTable` (this file's
    // own established style for `on_child_done`, e.g. the D1 join tests
    // above), with `ProcessDescriptor::wait_fork_remaining`/`wait_order` set
    // by hand exactly as `sched::control::Engine::apply_wait_fork`/
    // `apply_wait_order` would install them -- those two installers are
    // separately, directly tested against a real `Engine` in
    // `sched::control::tests` (this task's own Engine-integration tests),
    // since they need `Engine::apply_suspend`, which this file does not
    // have access to.

    /// SC-125's own literal `wait fork;` acceptance: resumes the parent only
    /// once every live child has completed, never on the first or second.
    /// 3 children forked `JoinKind::None` (the only real-world precursor to
    /// `wait fork;` -- see `handle_fork`'s own doc comment), with the
    /// parent's `wait_fork_remaining` set to `3` (what `apply_wait_fork`
    /// would have computed from `fork_children`'s live count at suspend
    /// time). Fails if `wait_fork` were wired to the ordinary
    /// `fork_join_kind`/`fork_remaining` counter instead of this fresh field
    /// -- `fork_join_kind` here is `Some(JoinKind::None)` (handle_fork's own
    /// permanent value for that kind), whose arm always returns `None`, so a
    /// wrong implementation reusing it would report `None` even after the
    /// 3rd completion, never `Some(AllForkChildrenDone)`.
    #[test]
    fn sc125_wait_fork_resumes_only_once_every_live_child_completes_not_on_the_first() {
        let mut table = ProcessTable::new();
        let parent = table.register(Box::new(ScriptedProcess::new(
            Arc::new(Mutex::new(Vec::new())),
            vec![Suspend::Done],
        )));
        let spawns = vec![
            SpawnSpec {
                kernel: 0,
                entry_pc: 0,
                parent,
                disable_name: Vec::new(),
            },
            SpawnSpec {
                kernel: 1,
                entry_pc: 0,
                parent,
                disable_name: Vec::new(),
            },
            SpawnSpec {
                kernel: 2,
                entry_pc: 0,
                parent,
                disable_name: Vec::new(),
            },
        ];
        let driver = ResumeDriver;
        let mut mint = mint_done_child;
        let children = driver.handle_fork(&mut table, parent, JoinKind::None, &spawns, &mut mint);
        assert_eq!(children.len(), 3);
        table.get_mut(parent).unwrap().wait_fork_remaining = Some(3);

        assert!(
            driver
                .on_child_done(&mut table, children[0])
                .unwrap()
                .is_none(),
            "must not resume the parent after only 1 of 3 live children"
        );
        assert!(
            driver
                .on_child_done(&mut table, children[1])
                .unwrap()
                .is_none(),
            "must not resume the parent after only 2 of 3 live children"
        );
        match driver.on_child_done(&mut table, children[2]) {
            Ok(Some(ResumeArg::AllForkChildrenDone)) => {}
            other => panic!(
                "expected Ok(Some(AllForkChildrenDone)) once all 3 live children have \
                 completed, got {other:?}"
            ),
        }
        assert!(
            table.get(parent).unwrap().wait_fork_remaining.is_none(),
            "wait_fork_remaining must be cleared once the wait is satisfied, matching \
             fork_join_kind's own one-shot-latch clearing"
        );
    }

    /// SC-125's `wait_order(...)` acceptance: resumes the parent only once
    /// every declared `ProcessId` has completed, in exactly that order.
    #[test]
    fn sc125_wait_order_resumes_only_when_completions_are_observed_in_the_exact_declared_order() {
        let mut table = ProcessTable::new();
        let parent = table.register(Box::new(ScriptedProcess::new(
            Arc::new(Mutex::new(Vec::new())),
            vec![Suspend::Done],
        )));
        let spawns = vec![
            SpawnSpec {
                kernel: 0,
                entry_pc: 0,
                parent,
                disable_name: Vec::new(),
            },
            SpawnSpec {
                kernel: 1,
                entry_pc: 0,
                parent,
                disable_name: Vec::new(),
            },
            SpawnSpec {
                kernel: 2,
                entry_pc: 0,
                parent,
                disable_name: Vec::new(),
            },
        ];
        let driver = ResumeDriver;
        let mut mint = mint_done_child;
        let children = driver.handle_fork(&mut table, parent, JoinKind::None, &spawns, &mut mint);
        let (c1, c2, c3) = (children[0], children[1], children[2]);
        table.get_mut(parent).unwrap().wait_order = Some((vec![c1, c2, c3], 0));

        assert!(
            driver.on_child_done(&mut table, c1).unwrap().is_none(),
            "must not resume after only the 1st of 3 declared completions"
        );
        assert!(
            driver.on_child_done(&mut table, c2).unwrap().is_none(),
            "must not resume after only the 1st and 2nd of 3 declared completions"
        );
        match driver.on_child_done(&mut table, c3) {
            Ok(Some(ResumeArg::WaitOrderSatisfied)) => {}
            other => panic!(
                "expected Ok(Some(WaitOrderSatisfied)) once all 3 completed in exactly \
                 their declared order, got {other:?}"
            ),
        }
    }

    /// SC-125's own literal "not a panic, not a silently-accepted
    /// reordering" acceptance for `wait_order(...)`: `c2` completes before
    /// `c1` (out of the declared `[c1, c2, c3]` order). The parent must
    /// never be resumed with `WaitOrderSatisfied` -- not for the misordered
    /// completion itself, and not for any later completion either, even one
    /// that would otherwise have completed the sequence.
    #[test]
    fn sc125_wait_order_flags_a_misordered_completion_rather_than_silently_accepting_it() {
        let mut table = ProcessTable::new();
        let parent = table.register(Box::new(ScriptedProcess::new(
            Arc::new(Mutex::new(Vec::new())),
            vec![Suspend::Done],
        )));
        let spawns = vec![
            SpawnSpec {
                kernel: 0,
                entry_pc: 0,
                parent,
                disable_name: Vec::new(),
            },
            SpawnSpec {
                kernel: 1,
                entry_pc: 0,
                parent,
                disable_name: Vec::new(),
            },
            SpawnSpec {
                kernel: 2,
                entry_pc: 0,
                parent,
                disable_name: Vec::new(),
            },
        ];
        let driver = ResumeDriver;
        let mut mint = mint_done_child;
        let children = driver.handle_fork(&mut table, parent, JoinKind::None, &spawns, &mut mint);
        let (c1, c2, c3) = (children[0], children[1], children[2]);
        table.get_mut(parent).unwrap().wait_order = Some((vec![c1, c2, c3], 0));

        // c2 completes before c1 -- out of the declared order.
        match driver.on_child_done(&mut table, c2) {
            Err(ResumeError::WaitOrderViolated(child)) => assert_eq!(child, c2),
            other => panic!(
                "expected Err(ResumeError::WaitOrderViolated(c2)) for the very completion \
                 that broke the declared order, got {other:?}"
            ),
        }
        assert!(
            table.get(parent).unwrap().wait_order_violated,
            "the violation must be latched on the parent's own descriptor"
        );

        // c1 and c3 now complete in a way that would otherwise finish the
        // sequence -- the parent must never be resumed via
        // WaitOrderSatisfied once the order has already been broken.
        assert!(
            driver.on_child_done(&mut table, c1).unwrap().is_none(),
            "a violated wait_order must never be silently satisfied by a later \
             completion, even one matching what the sequence originally expected"
        );
        assert!(
            driver.on_child_done(&mut table, c3).unwrap().is_none(),
            "a violated wait_order must never be silently satisfied by the final \
             completion either"
        );
    }

    /// Review fix (Task 9 r1, SC-125): `wait_order(...)` only watches the
    /// identifiers actually named in its `order` list. `fork a; b; c;
    /// join_none; wait_order(a, c);` -- `b` is a sibling fork child but is
    /// *not* part of `order`. `b` completing before `a` must be a pure
    /// no-op: it can neither advance the `a`, `c` sequence nor violate it,
    /// and `a` then `c` completing in their declared order must still
    /// resume the parent normally. Before this fix, `on_child_done`
    /// compared every completing fork child against `order[next_idx]`
    /// unconditionally, so `b` (never equal to `order[0] == a`) fell into
    /// the "out-of-order" arm and permanently latched
    /// `WaitOrderViolated(b)`, deadlocking a fully LRM-legal program.
    #[test]
    fn sc125_wait_order_ignores_a_sibling_child_not_named_in_order() {
        let mut table = ProcessTable::new();
        let parent = table.register(Box::new(ScriptedProcess::new(
            Arc::new(Mutex::new(Vec::new())),
            vec![Suspend::Done],
        )));
        let spawns = vec![
            SpawnSpec {
                kernel: 0,
                entry_pc: 0,
                parent,
                disable_name: Vec::new(),
            },
            SpawnSpec {
                kernel: 1,
                entry_pc: 0,
                parent,
                disable_name: Vec::new(),
            },
            SpawnSpec {
                kernel: 2,
                entry_pc: 0,
                parent,
                disable_name: Vec::new(),
            },
        ];
        let driver = ResumeDriver;
        let mut mint = mint_done_child;
        let children = driver.handle_fork(&mut table, parent, JoinKind::None, &spawns, &mut mint);
        let (a, b, c) = (children[0], children[1], children[2]);
        // `order` only names `a` and `c` -- `b` is a sibling fork child, not
        // part of the declared sequence.
        table.get_mut(parent).unwrap().wait_order = Some((vec![a, c], 0));

        // `b` completes first, out of turn relative to `a`/`c` -- but `b`
        // isn't in `order` at all, so this must be a silent no-op, never a
        // violation.
        assert!(
            driver.on_child_done(&mut table, b).unwrap().is_none(),
            "a sibling fork child that isn't named in `order` must never trigger \
             WaitOrderViolated, and must not resume the parent either"
        );
        assert!(
            !table.get(parent).unwrap().wait_order_violated,
            "an unrelated sibling's completion must never latch wait_order_violated"
        );

        // `a` then `c` complete in exactly their declared order -- the wait
        // must still be satisfiable normally, unaffected by `b`.
        assert!(
            driver.on_child_done(&mut table, a).unwrap().is_none(),
            "must not resume after only the 1st of 2 declared completions"
        );
        match driver.on_child_done(&mut table, c) {
            Ok(Some(ResumeArg::WaitOrderSatisfied)) => {}
            other => panic!(
                "expected Ok(Some(WaitOrderSatisfied)) once `a` then `c` completed in \
                 their declared order, unaffected by `b`'s earlier, unrelated \
                 completion, got {other:?}"
            ),
        }
    }
}