vj_sim/sched/

subscribe.rs

//! SCHED::SUBSCRIBE — the `WaitSpec` engine registration surface and the
//! NBA/Inactive drain destination (scheduler.md; process.md §2/§4; ledger
//! SC-085, SC-086; `docs/superpowers/plans/2026-09-23-wave-A-sched.md` Task
//! 13).
//!
//! [`register_wait`] is the SC-085 "engine registration call every `WaitSpec`
//! variant triggers": an exhaustive `match` over `WaitSpec`'s 4 variants
//! (`Delay`/`Edge`/`EdgeList`/`ImplicitComb`), each dispatched to the one
//! engine primitive that owns that variant's registration (the time wheel for
//! `Delay`, [`SensitivityTable::install_dynamic`] for `Edge`/`EdgeList`,
//! [`SensitivityTable::install_static`] for `ImplicitComb`). No wildcard arm
//! — a 5th `WaitSpec` variant added later fails this match to compile.
//!
//! [`nba_drain_destination`] is the SC-086 "the engine's NBA-region drain
//! must be the destination `nba_enqueue` targets, and the engine owns
//! draining/applying" — made concrete as one function that stages `updates`
//! into a temporary [`NbaLog`] and delegates to Task 7's already-existing
//! [`commit_nba`], never reimplementing its write+diff algorithm a second
//! time (see this module's own commit message / the plan's Task 13 "review
//! fix, sixth review pass" note for why: two physical implementations of one
//! pure write+diff algorithm is exactly the failure class this codebase has
//! hit before).
//!
//! **Deviation from the plan's literal signatures** (both functions), recorded
//! here and in `docs/CONTRACTS.md`'s Deviations section: the plan names a
//! `gate: &QueuedGate` parameter on both `register_wait` and
//! `nba_drain_destination`, and a `queues: &mut RegionQueues` parameter on
//! `nba_drain_destination`. `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 — exactly as Task 11's `process_table.rs::form_ready_set`
//! doc comment already records for the same reason — does not exist in this
//! tree: Task 10 has not been implemented yet, and SC-055/SC-056 are not in
//! this task's assigned ledger scope (`["SC-085","SC-086"]`). Per Task 11's
//! own precedent ("Task 10, when it lands, is expected to extend this
//! function's signature... which this task deliberately leaves untouched
//! rather than inventing Task 10's own type ahead of its task"), this module
//! does the same: `register_wait` never needed `gate` for any of its 4 match
//! arms' described behavior (registration, not dedup-at-notify-time) so it is
//! dropped entirely; `nba_drain_destination`'s wakeup sink calls Task 9's
//! already-existing [`SensitivityTable::matches`] directly instead of Task
//! 10's not-yet-existing `dedup::sensitivity_match_to_enqueue` wrapper — this
//! is still exactly one canonical matching implementation (Task 9's), not a
//! second one, so SC-042's "no duplicate implementation" concern this
//! delegation exists to satisfy is unaffected. Neither of Task 13's own
//! specified tests exercises the case dedup would change (a single process
//! sensitive to two different cells that both change in the same
//! `nba_drain_destination` call), so no acceptance behavior in this task's
//! ledger scope is weakened by the omission. `nba_drain_destination`'s
//! `queues: &mut RegionQueues` parameter is dropped for the same reason: the
//! plan's own prose for this function never describes pushing anything onto
//! `queues` — it only describes building and returning
//! `Vec<PendingActivation>` ("This is what's returned") — so the parameter
//! was unused by the function's own specified algorithm.
//!
//! Also not in the plan's literal signature: `register_wait`'s `WaitSpec::
//! ImplicitComb` arm needs the "caller-supplied `SubscriptionSetId ->
//! StaticSensitivity` lookup" the plan's own prose names but omits from the
//! parameter list — added here as `static_sensitivity_by_set`, exactly the
//! `HashMap` parameter the plan's prose says this function gains.
//! `register_wait` returns `Result<(), RegisterWaitError>` rather than the
//! plan's literal `-> ()`: an unresolvable `SubscriptionSetId` (the fake
//! test-only lookup missing an entry) is a real, catchable condition, not a
//! programmer error safe to `.expect()` away, and this workspace's
//! correctness posture forbids a silent no-op or a panic for a condition a
//! caller can legitimately hit.

use std::collections::HashMap;

use vj_abi::{Region, SimTime};

use crate::process::{SubscriptionSetId, WaitSpec};
use crate::sched::{
    ActivationId, EventTerm, EventWaitSet, NbaLog, NbaUpdate, PendingActivation, ProcessId,
    RegionQueues, ResumeArg, SchedTime, SensitivityTable, StaticSensitivity, TimeWheel, commit_nba,
};
use crate::state::{EdgeKind, SignalAccess, StateCellId, ValueChangeSink};

/// `register_wait`'s one failure mode. See this module's doc comment for why
/// this exists instead of the plan's literal `-> ()`.
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
pub enum RegisterWaitError {
    /// `WaitSpec::ImplicitComb`'s `SubscriptionSetId` had no entry in the
    /// caller-supplied lookup (codegen's real static-sensitivity table does
    /// not exist yet in this wave — the plan's own text names this as an
    /// in-crate test fixture stand-in).
    UnknownSubscriptionSet(SubscriptionSetId),
    /// [`TimeWheel::arm_delay`] is documented to return only
    /// `Region::Active` or `Region::Inactive`. This variant exists purely so
    /// this function's own dispatch on that return value stays an
    /// exhaustive match over the full 7-variant `Region` type with no
    /// wildcard arm (this workspace's "loud typed error, never a silent
    /// no-op" posture), catching a future change to `arm_delay`'s contract
    /// at this call site instead of silently mishandling it. sva T1 (§1.4,
    /// review fix, blocking finding 2) folded `Region::Preponed` into this
    /// same defensive arm alongside `Nba`/`Reactive`/`Postponed`; sva T2
    /// (review fix, blocking finding 2), Observed's half, folds
    /// `Region::Observed` into it too, for the identical reason.
    UnexpectedWheelRegion(Region),
}

impl std::fmt::Display for RegisterWaitError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            RegisterWaitError::UnknownSubscriptionSet(id) => {
                write!(f, "no StaticSensitivity registered for {id:?}")
            }
            RegisterWaitError::UnexpectedWheelRegion(r) => {
                write!(f, "TimeWheel::arm_delay returned unexpected region {r:?}")
            }
        }
    }
}

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

/// Converts a `WaitSpec::Delay`'s `SimTime` *duration* into wheel-facing
/// ticks. `WaitSpec::Delay` names a duration (an `#N` delay statement), not
/// an absolute time, so only `sim_units` (the minimum-time-precision tick
/// count) is meaningful here — `delta` (the delta-cycle/region-iteration
/// counter, per `vj_abi::SimTime`'s own doc comment) has no meaning for a
/// duration. Saturates at zero rather than reinterpreting a negative
/// `sim_units` as a huge `u64` via a raw `as` cast, which would silently
/// arm a near-infinite delay instead of surfacing the codegen/elaboration
/// bug a negative delay duration would represent.
fn ticks_from(t: SimTime) -> u64 {
    t.sim_units.max(0) as u64
}

/// The SC-085 engine registration call every `WaitSpec` variant triggers, on
/// `proc`'s behalf, at `now`. See this module's doc comment for the
/// deviations from the plan's literal signature.
pub fn register_wait(
    table: &mut SensitivityTable,
    wheel: &mut TimeWheel,
    queues: &mut RegionQueues,
    proc: ProcessId,
    spec: &WaitSpec,
    now: SchedTime,
    static_sensitivity_by_set: &HashMap<SubscriptionSetId, StaticSensitivity>,
) -> Result<(), RegisterWaitError> {
    match spec {
        WaitSpec::Delay(t) => {
            let delay_ticks = ticks_from(*t);
            let act = PendingActivation {
                id: proc as ActivationId,
                proc,
                arg: ResumeArg::TimeElapsed,
                reconv_iter: 0,
                logical_order: (
                    SchedTime {
                        ticks: now.ticks + delay_ticks,
                    },
                    Region::Active,
                    0,
                ),
            };
            match wheel.arm_delay(now.ticks, delay_ticks, act) {
                Region::Inactive => {
                    // `delay_ticks == 0`: `arm_delay` deliberately never
                    // touches the wheel for this case (SC-029) — the caller
                    // (here) must land `act` directly in `Region::Inactive`.
                    // `RegionQueues::push` only ever rejects `Region::Nba`
                    // (Task 4), so pushing to `Inactive` cannot fail.
                    //
                    // `act` was built above with `logical_order.1` set to
                    // `Region::Active` (the common case, where `arm_delay`
                    // arms the wheel instead). That tag must agree with the
                    // queue the activation actually lands in — `order.rs`'s
                    // `canonical_key_of`/`sort_ready_set` read
                    // `logical_order.1` directly to build the canonical key
                    // (SC-010/DT-005/DT-010), and `RegionQueues::push`'s own
                    // contract is to route `act` to *its region's* queue —
                    // so re-tag it to `Region::Inactive` before pushing.
                    let act = PendingActivation {
                        logical_order: (act.logical_order.0, Region::Inactive, act.logical_order.2),
                        ..act
                    };
                    queues
                        .push(Region::Inactive, act)
                        .expect("Region::Inactive is always accepted by RegionQueues::push");
                }
                Region::Active => {
                    // Armed onto the wheel by `arm_delay` itself (nonzero
                    // delay) — nothing further to do here.
                }
                // sva T1 (§1.4, review fix, blocking finding 2): `Region::
                // Preponed` joins this defensive arm for the identical
                // reason `Nba`/`Reactive`/`Postponed` are already here --
                // `TimeWheel::arm_delay` can no more legitimately return
                // `Preponed` than any of those three (it is a pure
                // snapshot-push point, never a wheel-armed delay target).
                // sva T2 (review fix, blocking finding 2), Observed's half:
                // `Region::Observed` joins it too -- a process's delay/edge
                // wait can no more land in Observed than in Preponed (it is
                // reached only through the region-callback mechanism, never
                // a wheel-armed delay target).
                other @ (Region::Nba
                | Region::Reactive
                | Region::Postponed
                | Region::Preponed
                | Region::Observed) => {
                    return Err(RegisterWaitError::UnexpectedWheelRegion(other));
                }
            }
        }
        WaitSpec::Edge { edge, signal, iff } => {
            let wait = EventWaitSet {
                terms: vec![EventTerm::Edge {
                    cell: *signal,
                    edge: *edge,
                    iff: *iff,
                }],
                ..Default::default()
            };
            table.install_dynamic(&wait, proc);
        }
        WaitSpec::EdgeList(terms) => {
            let wait = EventWaitSet {
                terms: terms
                    .iter()
                    .map(|(edge, signal, iff)| EventTerm::Edge {
                        cell: *signal,
                        edge: *edge,
                        iff: *iff,
                    })
                    .collect(),
                ..Default::default()
            };
            table.install_dynamic(&wait, proc);
        }
        WaitSpec::ImplicitComb(set_id) => {
            let sensitivity = static_sensitivity_by_set
                .get(set_id)
                .ok_or(RegisterWaitError::UnknownSubscriptionSet(*set_id))?;
            table.install_static(sensitivity, proc);
        }
    }
    Ok(())
}

/// The [`ValueChangeSink`] [`nba_drain_destination`] hands to Task 7's
/// [`commit_nba`] instead of calling the sink method directly (or any other
/// fan-out) itself. Its own `notify_change` body routes straight through
/// Task 9's already-existing [`SensitivityTable::matches`] — no new matching
/// logic, no second implementation of the sensitivity lookup Task 18's
/// `EngineNotifySink` also reuses (see this module's doc comment for why
/// this is `matches` directly rather than Task 10's not-yet-existing
/// `dedup::sensitivity_match_to_enqueue` wrapper).
struct NbaWakeupSink<'a> {
    sensitivity: &'a SensitivityTable,
    /// Always `Region::Active` for every activation this sink produces
    /// (SC-086 — NBA-triggered wakeups land in Active, scheduler.md §1 step
    /// 3), regardless of the `region` `commit_nba` itself reports on each
    /// `notify_change` call (`Region::Nba`, the region the *write* committed
    /// in — not the region the *woken process* must next run in).
    region: Region,
    /// The scheduler time `nba_drain_destination` was called at (its own
    /// `at: SchedTime` parameter), threaded through so every
    /// `PendingActivation` this sink produces carries the *real* vtime as
    /// `logical_order.0` — the primary field `order.rs`'s
    /// `canonical_key_of`/`sort_ready_set` sort by (SC-010/DT-005/DT-010) —
    /// instead of an unconditional `SchedTime { ticks: 0 }`.
    /// `commit_nba`'s own `notify_change` callback receives `_at: SimTime`
    /// (the write's timestamp, which is `at` reinterpreted, per this
    /// module's existing convention of discarding it here — see below) but
    /// that value is per-write and always equal to this sink's own `now`
    /// for every call `commit_nba` makes within one `nba_drain_destination`
    /// invocation, so capturing it once at construction is equivalent and
    /// avoids re-deriving `SchedTime` from `SimTime` per call.
    now: SchedTime,
    out: &'a mut Vec<PendingActivation>,
}

impl ValueChangeSink for NbaWakeupSink<'_> {
    fn notify_change(&mut self, cell: StateCellId, edge: EdgeKind, _at: SimTime, _region: Region) {
        // Task 3 widened `matches`'s item type to `(ProcessId, usize,
        // Option<u64>)`, and Task 5 (SC-068) widened it again to add a 4th
        // `Option<PredicateFn>` element; this sink has no repeat-counting or
        // `iff`-evaluation role (see this module's own doc comment for why
        // it calls `matches` directly rather than
        // `dedup::sensitivity_match_to_enqueue`) and only ever needed the
        // `ProcessId`.
        for (proc, _, _, _) in self.sensitivity.matches(cell, edge) {
            // Placeholder `id`/tiebreak values (this task's own scope is
            // SC-085/SC-086, which name nothing about global `ActivationId`
            // assignment or a real per-process `registration_order`; both
            // remain Task 21 outer-driver concerns, matching the existing
            // precedent of `notify.rs`'s `NBA_COMMIT_KEY_PROCESS_ID`/
            // `reconv_iter: 0` placeholders for the same not-yet-wired
            // concern). `self.out.len()` at push time is nonetheless a
            // real, strictly increasing counter that exactly matches
            // `commit_nba`'s own deterministic (canonical, cell-id-ordered)
            // `notify_change` call order — not a caller-`stage()`-order
            // artifact — which is what SC-086's canonical-order tests need.
            let seq = self.out.len() as u64;
            self.out.push(PendingActivation {
                id: seq,
                proc,
                // `SensitivityTable::matches` (Task 9) returns only
                // `ProcessId`, not the originating `Dynamic` entry's
                // `term_index` — that information is discarded by `matches`
                // itself and is not recoverable here without changing Task
                // 9's own contract, which this task does not own. `0` is a
                // placeholder, matching this module's established pattern
                // for information not yet threaded end-to-end.
                arg: ResumeArg::EventFired { term_index: 0 },
                reconv_iter: 0,
                logical_order: (self.now, self.region, seq),
            });
        }
    }
}

/// The SC-086 NBA/Inactive-region drain destination: stages `updates` into a
/// temporary [`NbaLog`] (preserving their original `order`, since `updates`
/// is itself [`NbaLog::take`]'s output and `stage` assigns `order` from push
/// position — Task 6's own contract), then delegates to Task 7's
/// [`commit_nba`] rather than reimplementing its write+diff algorithm.
/// Returns one [`PendingActivation`] per process woken by an actually-changed
/// cell, each landing in `Region::Active` (SC-086).
pub fn nba_drain_destination(
    updates: Vec<NbaUpdate>,
    arena: &mut dyn SignalAccess,
    sensitivity: &SensitivityTable,
    at: SchedTime,
) -> Vec<PendingActivation> {
    let mut temp_log = NbaLog::new();
    for u in updates {
        temp_log.stage(u.cell, u.value, u.source);
    }

    let mut out: Vec<PendingActivation> = Vec::new();
    {
        let mut sink = NbaWakeupSink {
            sensitivity,
            region: Region::Active,
            now: at,
            out: &mut out,
        };
        commit_nba(&mut temp_log, arena, at, &mut sink);
    }
    out
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::sched::{ProcessId as SchedProcessId, StaticSensitivity, sort_ready_set};
    use crate::state::{FourStateValue, ListenerRole};
    use std::collections::HashMap;

    // --- register_wait / SC-085 ---

    fn four_state_eq(a: &FourStateValue, b: &FourStateValue) -> bool {
        match (a, b) {
            (FourStateValue::Narrow(x), FourStateValue::Narrow(y)) => x == y,
            (
                FourStateValue::Wide { aval: a1, bval: b1 },
                FourStateValue::Wide { aval: a2, bval: b2 },
            ) => a1 == a2 && b1 == b2,
            _ => false,
        }
    }

    struct FakeArena {
        cells: HashMap<StateCellId, FourStateValue>,
    }

    impl FakeArena {
        fn new() -> Self {
            Self {
                cells: HashMap::new(),
            }
        }

        fn seed(&mut self, cell: StateCellId, value: FourStateValue) {
            self.cells.insert(cell, value);
        }
    }

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

        fn write(&mut self, cell: StateCellId, value: FourStateValue) -> bool {
            let changed = match self.cells.get(&cell) {
                Some(existing) => !four_state_eq(existing, &value),
                None => true,
            };
            self.cells.insert(cell, value);
            changed
        }

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

    #[test]
    fn every_wait_spec_variant_has_a_registration_arm() {
        let empty_lookup: HashMap<SubscriptionSetId, StaticSensitivity> = HashMap::new();

        // --- Delay ---
        {
            let mut table = SensitivityTable::new();
            let mut wheel = TimeWheel::new();
            let mut queues = RegionQueues::new();
            let proc: SchedProcessId = 1;
            let spec = WaitSpec::Delay(SimTime {
                sim_units: 5,
                delta: 0,
            });
            register_wait(
                &mut table,
                &mut wheel,
                &mut queues,
                proc,
                &spec,
                SchedTime { ticks: 10 },
                &empty_lookup,
            )
            .expect("Delay registration must succeed");
            assert_eq!(
                wheel.next_tick(),
                Some(15),
                "Delay(5) registered at tick 10 must arm the wheel to fire at tick 15"
            );
        }

        // --- Edge ---
        {
            let mut table = SensitivityTable::new();
            let mut wheel = TimeWheel::new();
            let mut queues = RegionQueues::new();
            let proc: SchedProcessId = 2;
            let cell = StateCellId(100);
            let spec = WaitSpec::Edge {
                edge: EdgeKind::Posedge,
                signal: cell,
                iff: None,
            };
            register_wait(
                &mut table,
                &mut wheel,
                &mut queues,
                proc,
                &spec,
                SchedTime { ticks: 0 },
                &empty_lookup,
            )
            .expect("Edge registration must succeed");
            assert_eq!(
                table
                    .matches(cell, EdgeKind::Posedge)
                    .map(|(p, _, _, _)| p)
                    .collect::<Vec<_>>(),
                vec![proc],
                "Edge registration must install a dynamic entry the table finds"
            );
        }

        // --- EdgeList ---
        {
            let mut table = SensitivityTable::new();
            let mut wheel = TimeWheel::new();
            let mut queues = RegionQueues::new();
            let proc: SchedProcessId = 3;
            let cell_a = StateCellId(101);
            let cell_b = StateCellId(102);
            let spec = WaitSpec::EdgeList(vec![
                (EdgeKind::Posedge, cell_a, None),
                (EdgeKind::Negedge, cell_b, None),
            ]);
            register_wait(
                &mut table,
                &mut wheel,
                &mut queues,
                proc,
                &spec,
                SchedTime { ticks: 0 },
                &empty_lookup,
            )
            .expect("EdgeList registration must succeed");
            assert_eq!(
                table
                    .matches(cell_a, EdgeKind::Posedge)
                    .map(|(p, _, _, _)| p)
                    .collect::<Vec<_>>(),
                vec![proc],
                "EdgeList must install a dynamic entry per (edge, signal) pair (first pair)"
            );
            assert_eq!(
                table
                    .matches(cell_b, EdgeKind::Negedge)
                    .map(|(p, _, _, _)| p)
                    .collect::<Vec<_>>(),
                vec![proc],
                "EdgeList must install a dynamic entry per (edge, signal) pair (second pair)"
            );
        }

        // --- ImplicitComb ---
        {
            let mut table = SensitivityTable::new();
            let mut wheel = TimeWheel::new();
            let mut queues = RegionQueues::new();
            let proc: SchedProcessId = 4;
            let cell = StateCellId(103);
            let set_id = SubscriptionSetId(7);
            let mut lookup: HashMap<SubscriptionSetId, StaticSensitivity> = HashMap::new();
            lookup.insert(
                set_id,
                StaticSensitivity {
                    edges: vec![(cell, None)],
                },
            );
            let spec = WaitSpec::ImplicitComb(set_id);
            register_wait(
                &mut table,
                &mut wheel,
                &mut queues,
                proc,
                &spec,
                SchedTime { ticks: 0 },
                &lookup,
            )
            .expect("ImplicitComb registration must succeed via the fake lookup");
            assert_eq!(
                table
                    .matches(cell, EdgeKind::Negedge)
                    .map(|(p, _, _, _)| p)
                    .collect::<Vec<_>>(),
                vec![proc],
                "ImplicitComb must install a static any-change entry via the lookup"
            );
        }
    }

    #[test]
    fn delay_zero_lands_directly_in_inactive_not_the_wheel() {
        let mut table = SensitivityTable::new();
        let mut wheel = TimeWheel::new();
        let mut queues = RegionQueues::new();
        let empty_lookup: HashMap<SubscriptionSetId, StaticSensitivity> = HashMap::new();
        let proc: SchedProcessId = 9;
        let spec = WaitSpec::Delay(SimTime {
            sim_units: 0,
            delta: 0,
        });

        register_wait(
            &mut table,
            &mut wheel,
            &mut queues,
            proc,
            &spec,
            SchedTime { ticks: 3 },
            &empty_lookup,
        )
        .expect("Delay(0) registration must succeed");

        assert_eq!(
            wheel.next_tick(),
            None,
            "Delay(0) must never touch the wheel"
        );
        let drained = queues.drain_inactive();
        assert_eq!(drained.len(), 1);
        assert_eq!(drained[0].proc, proc);
        assert_eq!(
            drained[0].logical_order.1,
            Region::Inactive,
            "an activation landed in the Inactive queue must carry Region::Inactive as its \
             own logical_order.1 tag, agreeing with the queue it actually lives in — \
             canonical_key_of/sort_ready_set (order.rs) read logical_order.1 directly to \
             build the canonical key that SC-010/DT-005/DT-010 depend on"
        );
    }

    /// sva T1 (§1.4, review fix, blocking finding 2): `register_wait`'s
    /// `match wheel.arm_delay(..)` defensive arm must still compile
    /// exhaustively over `Region`'s new 6th variant, with `Preponed` folded
    /// into the SAME `other @ (Nba | Reactive | Postponed | Preponed)` arm
    /// as the pre-existing three (not a fresh wildcard, which this
    /// workspace's "loud typed error" posture forbids). `arm_delay` itself
    /// is documented to return only `Active`/`Inactive`, so this arm is
    /// never reachable through the real code path above (there is no mock
    /// `TimeWheel` seam this function could inject one through) -- this
    /// test instead proves, by direct construction, that
    /// `RegisterWaitError::UnexpectedWheelRegion` can represent
    /// `Region::Preponed` and names it correctly in its `Display` text, the
    /// same proof-by-construction this type's own doc comment already
    /// relies on for `Nba`/`Reactive`/`Postponed`.
    #[test]
    fn register_wait_defensive_arm_still_exhaustive_with_preponed() {
        let err = RegisterWaitError::UnexpectedWheelRegion(Region::Preponed);
        assert_eq!(
            err,
            RegisterWaitError::UnexpectedWheelRegion(Region::Preponed)
        );
        assert!(
            err.to_string().contains("Preponed"),
            "the Display text must name the offending region: {err}"
        );
    }

    /// sva T2 (review fix, blocking finding 2), Observed's half: the
    /// identical proof-by-construction as
    /// `register_wait_defensive_arm_still_exhaustive_with_preponed`, now for
    /// `Region::Observed`.
    #[test]
    fn register_wait_defensive_arm_still_exhaustive_with_observed() {
        let err = RegisterWaitError::UnexpectedWheelRegion(Region::Observed);
        assert_eq!(
            err,
            RegisterWaitError::UnexpectedWheelRegion(Region::Observed)
        );
        assert!(
            err.to_string().contains("Observed"),
            "the Display text must name the offending region: {err}"
        );
    }

    #[test]
    fn implicit_comb_with_an_unresolvable_set_id_is_a_typed_error_not_a_panic() {
        let mut table = SensitivityTable::new();
        let mut wheel = TimeWheel::new();
        let mut queues = RegionQueues::new();
        let empty_lookup: HashMap<SubscriptionSetId, StaticSensitivity> = HashMap::new();
        let spec = WaitSpec::ImplicitComb(SubscriptionSetId(999));

        let result = register_wait(
            &mut table,
            &mut wheel,
            &mut queues,
            1,
            &spec,
            SchedTime { ticks: 0 },
            &empty_lookup,
        );

        assert_eq!(
            result,
            Err(RegisterWaitError::UnknownSubscriptionSet(
                SubscriptionSetId(999)
            ))
        );
    }

    // --- nba_drain_destination / SC-086 ---

    #[test]
    fn nba_drain_produces_activations_only_for_changed_cells_in_canonical_form() {
        let cell_changed = StateCellId(200);
        let cell_unchanged = StateCellId(201);
        let proc_changed: SchedProcessId = 10;
        let proc_unchanged: SchedProcessId = 20;

        let mut arena = FakeArena::new();
        arena.seed(cell_changed, FourStateValue::Narrow(0));
        arena.seed(cell_unchanged, FourStateValue::Narrow(7));

        let mut sensitivity = SensitivityTable::new();
        sensitivity.install_static(
            &StaticSensitivity {
                edges: vec![(cell_changed, None)],
            },
            proc_changed,
        );
        sensitivity.install_static(
            &StaticSensitivity {
                edges: vec![(cell_unchanged, None)],
            },
            proc_unchanged,
        );

        let updates = vec![
            NbaUpdate {
                cell: cell_changed,
                value: FourStateValue::Narrow(1),
                source: 1,
                order: 0,
            },
            NbaUpdate {
                cell: cell_unchanged,
                value: FourStateValue::Narrow(7),
                source: 1,
                order: 1,
            },
        ];

        let activations =
            nba_drain_destination(updates, &mut arena, &sensitivity, SchedTime { ticks: 5 });

        assert_eq!(
            activations.len(),
            1,
            "only the process sensitive to the genuinely-changed cell may be activated"
        );
        assert_eq!(activations[0].proc, proc_changed);
        assert_eq!(
            activations[0].logical_order.1,
            Region::Active,
            "NBA-triggered wakeups must land in Region::Active"
        );
        assert_eq!(
            activations[0].logical_order.0,
            SchedTime { ticks: 5 },
            "an NBA-triggered wakeup's logical_order.0 must be the `at` scheduler time \
             nba_drain_destination was called with, not a hardcoded tick"
        );
    }

    #[test]
    fn nba_drain_at_nonzero_time_stamps_activations_with_that_time_not_zero() {
        // SC-010/DT-005/DT-010 regression: `NbaWakeupSink` used to hardcode
        // `logical_order.0` to `SchedTime { ticks: 0 }` for every activation
        // it produced, regardless of `nba_drain_destination`'s own `at`
        // parameter. A call at a nonzero time must stamp its activations
        // with that time.
        let cell = StateCellId(230);
        let proc: SchedProcessId = 40;

        let mut arena = FakeArena::new();
        arena.seed(cell, FourStateValue::Narrow(0));

        let mut sensitivity = SensitivityTable::new();
        sensitivity.install_static(
            &StaticSensitivity {
                edges: vec![(cell, None)],
            },
            proc,
        );

        let updates = vec![NbaUpdate {
            cell,
            value: FourStateValue::Narrow(1),
            source: 1,
            order: 0,
        }];

        let at = SchedTime { ticks: 7 };
        let activations = nba_drain_destination(updates, &mut arena, &sensitivity, at);

        assert_eq!(activations.len(), 1);
        assert_eq!(
            activations[0].logical_order.0, at,
            "logical_order.0 must equal the `at` time nba_drain_destination was called \
             with, not SchedTime {{ ticks: 0 }}"
        );
    }

    #[test]
    fn nba_drain_applies_exactly_the_queued_tuples_in_canonical_order() {
        let cell_a = StateCellId(210);
        let cell_b = StateCellId(211);
        let cell_c = StateCellId(212);
        let proc_a: SchedProcessId = 30;
        let proc_b: SchedProcessId = 31;
        let proc_c: SchedProcessId = 32;

        let mut sensitivity = SensitivityTable::new();
        sensitivity.install_static(
            &StaticSensitivity {
                edges: vec![(cell_a, None)],
            },
            proc_a,
        );
        sensitivity.install_static(
            &StaticSensitivity {
                edges: vec![(cell_b, None)],
            },
            proc_b,
        );
        sensitivity.install_static(
            &StaticSensitivity {
                edges: vec![(cell_c, None)],
            },
            proc_c,
        );

        // Out-of-order relative to the cells' own ordering.
        let updates = vec![
            NbaUpdate {
                cell: cell_c,
                value: FourStateValue::Narrow(1),
                source: 1,
                order: 0,
            },
            NbaUpdate {
                cell: cell_a,
                value: FourStateValue::Narrow(1),
                source: 1,
                order: 1,
            },
            NbaUpdate {
                cell: cell_b,
                value: FourStateValue::Narrow(1),
                source: 1,
                order: 2,
            },
        ];

        let mut arena1 = FakeArena::new();
        let acts1 = nba_drain_destination(
            updates.clone(),
            &mut arena1,
            &sensitivity,
            SchedTime { ticks: 7 },
        );

        // A second run with a different stage/arrival order for the same
        // logical set of updates (only the arena is fresh; the sensitivity
        // table and target cells are identical).
        let updates2 = vec![updates[1].clone(), updates[2].clone(), updates[0].clone()];
        let mut arena2 = FakeArena::new();
        let acts2 =
            nba_drain_destination(updates2, &mut arena2, &sensitivity, SchedTime { ticks: 7 });

        let mut sorted1 = acts1;
        let mut sorted2 = acts2;
        sort_ready_set(&mut sorted1);
        sort_ready_set(&mut sorted2);

        let procs1: Vec<SchedProcessId> = sorted1.iter().map(|a| a.proc).collect();
        let procs2: Vec<SchedProcessId> = sorted2.iter().map(|a| a.proc).collect();
        assert_eq!(
            procs1, procs2,
            "canonical order must not depend on the original stage/arrival order"
        );
        assert_eq!(procs1.len(), 3);
    }

    #[test]
    fn nba_drain_destination_delegates_to_commit_nba_not_a_second_diff_implementation() {
        let cell = StateCellId(220);
        let mut arena = FakeArena::new();
        arena.seed(cell, FourStateValue::Narrow(5));

        let sensitivity = SensitivityTable::new();

        let updates = vec![NbaUpdate {
            cell,
            value: FourStateValue::Narrow(5),
            source: 1,
            order: 0,
        }];

        let activations =
            nba_drain_destination(updates, &mut arena, &sensitivity, SchedTime { ticks: 0 });

        assert!(
            activations.is_empty(),
            "an NBA write equal to the current value must produce no activation — a \
             reimplementation that diffs against a stale/default value instead of \
             genuinely calling commit_nba's arena-compare would produce a spurious one here"
        );

        // Source-text companion check (same technique as process_table.rs's
        // slotmap scan / determinism's DT-006 scan): production code (test
        // bodies stripped) must call `commit_nba(` and must never call
        // `.notify_change(` itself.
        let src = std::fs::read_to_string(concat!(
            env!("CARGO_MANIFEST_DIR"),
            "/src/sched/subscribe.rs"
        ))
        .expect("must be able to read subscribe.rs's own source");
        let production_only = crate::determinism::scan_support::strip_cfg_test_blocks(&src);
        assert!(
            production_only.contains("commit_nba("),
            "nba_drain_destination must call Task 7's commit_nba"
        );
        assert!(
            !production_only.contains(".notify_change("),
            "subscribe.rs must not call .notify_change( itself — it must delegate to \
             commit_nba, not reimplement NBA commit semantics"
        );
    }
}