vj_sim/sched/

notify.rs

//! SCHED::NOTIFY — the NBA commit pass, `notify_change`, and changed-cell
//! diffing (scheduler.md §3/§4; ARCHITECTURE §6.3; ledger SC-038, SC-039,
//! SC-040, SC-041, SC-042).
//!
//! [`commit_nba`] is the SC-039 algorithm: drain the [`NbaLog`] once, apply
//! the staged updates cell by cell in the log's own insertion `order` (so the
//! committed value tracks push order, SC-039), write each update through
//! [`SignalAccess`], and call [`ValueChangeSink::notify_change`] for every
//! update that actually changed the stored value (SC-040/SC-041; LRM 4.9.4,
//! 10.4.2: every value-changing update is an update event, so `q <= 1;
//! q <= 0;` wakes an `@(q)` waiter even though `q` ends where it started) —
//! never for a write whose value equals the cell's current one. A
//! [`ChangeDelta`] is recorded for the cells whose select waiters need it
//! (`ValueChangeSink::wants_delta`). This is one of this wave's two sanctioned production
//! `.notify_change(` call sites (SC-042; the other is Task 8's
//! `commit_blocking_write`; the crate-wide "exactly two call sites" proof is
//! Task 22's, this task's own contribution is that this file has exactly
//! one).

use crate::sched::SchedTime;
use crate::sched::nba::NbaLog;
use crate::state::{
    ChangeDelta, EdgeKind, FourStateValue, SignalAccess, StateCellId, ValueChangeSink,
};
use vj_abi::{Bit4, Region};

/// Extracts the single reference bit `infer_edge` classifies transitions on:
/// bit 0 of the narrow word's `val`/`unk` planes (ARCHITECTURE §6.2's FIXED
/// layout — `unk` always at bit 64 regardless of width), or bit 0 of the
/// wide planes' LSB word ([`FourStateValue::Wide`]'s `aval`/`bval` word
/// planes). Both decode through [`Bit4`]'s own documented conversions
/// (`vj_abi::bit4`), never a hand-rolled second decode.
fn lsb_bit(value: &FourStateValue) -> Bit4 {
    match value {
        FourStateValue::Narrow(word) => {
            let val = (word & 1) != 0;
            let unk = ((word >> 64) & 1) != 0;
            Bit4::from_val_unk(val, unk)
        }
        FourStateValue::Wide { aval, bval } => {
            let a = aval.first().copied().unwrap_or(0) & 1 != 0;
            let b = bval.first().copied().unwrap_or(0) & 1 != 0;
            Bit4::from_aval_bval(a, b)
        }
    }
}

/// Classifies an old->new transition per IEEE 1800 edge-event semantics
/// (LRM Table 9-1, §9.4.2). The rule is precise and asymmetric, *not* a
/// "conservative union" of Posedge+Negedge for any transition touching X/Z:
/// each X/Z-involving transition is assigned to exactly one specific edge
/// (or none). `0->{1,X,Z}` is `Posedge` (X/Z read as "possibly became 1");
/// `1->{0,X,Z}` is `Negedge` (X/Z read as "possibly became 0"); `X->0` and
/// `Z->0` are `Negedge`; `X->1` and `Z->1` are `Posedge`; and the four
/// "stayed unknown" transitions `X->X`, `X->Z`, `Z->X`, `Z->Z` are `None` —
/// verified directly against iverilog 12 (`x->1` fires only `@(posedge)`,
/// `1->x` only `@(negedge)`, `x->z`/`z->x` fire neither edge at all).
/// `EdgeKind::BothEdges` is never produced by this table; that variant
/// exists in the enum for the unrelated `edge` keyword / any-change case,
/// not for X/Z transitions. Exhaustive over all 16 `(Bit4, Bit4)` pairs,
/// written out rather than collapsed with a wildcard, so the truth table is
/// legible inline as the task brief asks:
///
/// | old \ new | 0        | 1        | X         | Z         |
/// |-----------|----------|----------|-----------|-----------|
/// | 0         | None     | Posedge  | Posedge   | Posedge   |
/// | 1         | Negedge  | None     | Negedge   | Negedge   |
/// | X         | Negedge  | Posedge  | None      | None      |
/// | Z         | Negedge  | Posedge  | None      | None      |
pub fn infer_edge(old: &FourStateValue, new: &FourStateValue) -> EdgeKind {
    edge_between(lsb_bit(old), lsb_bit(new))
}

/// [`infer_edge`]'s table on the two lsb bits themselves (the windowed
/// huge-signal commit knows only bit 0, not the whole old value).
pub fn edge_between(old: Bit4, new: Bit4) -> EdgeKind {
    use Bit4::*;
    match (old, new) {
        (Zero, Zero) => EdgeKind::None,
        (Zero, One) => EdgeKind::Posedge,
        (Zero, X) => EdgeKind::Posedge,
        (Zero, Z) => EdgeKind::Posedge,
        (One, Zero) => EdgeKind::Negedge,
        (One, One) => EdgeKind::None,
        (One, X) => EdgeKind::Negedge,
        (One, Z) => EdgeKind::Negedge,
        (X, Zero) => EdgeKind::Negedge,
        (X, One) => EdgeKind::Posedge,
        (X, X) => EdgeKind::None,
        (X, Z) => EdgeKind::None,
        (Z, Zero) => EdgeKind::Negedge,
        (Z, One) => EdgeKind::Posedge,
        (Z, X) => EdgeKind::None,
        (Z, Z) => EdgeKind::None,
    }
}

/// The SC-039 NBA commit pass: `take()`s `log`, applies each cell's updates
/// in staging order, writes each through `arena`, and calls
/// `sink.notify_change(..)` for every update that actually changed the stored
/// value (SC-040/SC-041). Returns the count of cells that changed.
pub fn commit_nba(
    log: &mut NbaLog,
    arena: &mut dyn SignalAccess,
    at: SchedTime,
    sink: &mut dyn ValueChangeSink,
) -> usize {
    let entries = log.take();
    let mut masked = log.take_masked();
    // Class-property writes name no cell: land them in staging order, then
    // commit the cell writes.
    if masked.iter().any(|m| m.obj.is_some()) {
        let (mut objs, rest): (Vec<_>, Vec<_>) = masked.into_iter().partition(|m| m.obj.is_some());
        masked = rest;
        objs.sort_unstable_by_key(|m| m.order);
        for m in objs {
            if let Some(w) = m.obj {
                if w.class_event {
                    // Wakes land in the engine's pending list, which it
                    // applies right after this commit pass.
                    super::class_event::trigger(w.h, at.ticks, &mut |req| {
                        super::class_event::defer_wake(req);
                    });
                } else {
                    w.commit();
                }
            }
        }
    }
    if !masked.is_empty() {
        return commit_nba_with_partial_writes(entries, masked, arena, at, sink);
    }

    // The log's insertion-order `order` is one strictly increasing counter
    // over the whole timestep, whichever process staged an update (SC-039:
    // "the committed value tracks push order, not `ProcessId` order"), so
    // sorting by `(cell, order)` applies every cell's updates in the order
    // they were staged.
    //
    // Each update is applied and notified on its own (LRM 4.9.4, 10.4.2:
    // nonblocking updates take effect in order, and every value-changing
    // update is an update event): `q <= 1; q <= 0;` from `q == 0` wakes an
    // `@(q)` / `@(posedge q)` waiter even though `q` ends where it started.
    // Wake-ups of one process are still deduplicated by the engine's queued
    // gate, so a process wakes once per time step however many updates hit it.
    let mut entries = entries;
    entries.sort_by_key(|e| (e.cell, e.order));

    let mut changed_count = 0usize;
    let mut last_changed: Option<StateCellId> = None;
    for entry in entries {
        let cell = entry.cell;
        let old = arena.read(cell);
        let new_for_edge = entry.value.clone();
        if arena.write(cell, entry.value) {
            if sink.wants_delta(cell) {
                sink.note_delta(cell, ChangeDelta::between(&old, &new_for_edge));
            }
            notify_nba_change(sink, cell, infer_edge(&old, &new_for_edge), at);
            if last_changed != Some(cell) {
                changed_count += 1;
                last_changed = Some(cell);
            }
        }
        for (cell, edge) in alias_edges(arena) {
            notify_nba_change(sink, cell, edge, at);
        }
    }
    changed_count
}

/// The changes of other names for the storage the last write touched
/// (window aliases of ref ports, `state::alias`), with their edges.
fn alias_edges(arena: &mut dyn SignalAccess) -> Vec<(StateCellId, EdgeKind)> {
    arena
        .take_alias_changes()
        .into_iter()
        .map(|c| {
            let edge = if c.no_edge {
                EdgeKind::None
            } else {
                infer_edge(&c.old, &c.new)
            };
            (c.cell, edge)
        })
        .collect()
}

/// The one NBA-region change notification (SC-042's sanctioned call site).
fn notify_nba_change(
    sink: &mut dyn ValueChangeSink,
    cell: StateCellId,
    edge: EdgeKind,
    at: SchedTime,
) {
    sink.notify_change(cell, edge, at.to_abi_display(0), Region::Nba);
}

/// Merges `new` into `cur` on the bits set in `mask` (both value planes;
/// the two encodings -- `Narrow`'s val/unk and `Wide`'s aval/bval -- are
/// both per-bit planes, so the same mask applies to each).
fn merge_masked(cur: &FourStateValue, new: &FourStateValue, mask: &[u64]) -> FourStateValue {
    let pick = |c: u64, n: u64, m: u64| (c & !m) | (n & m);
    match (cur, new) {
        (FourStateValue::Narrow(c), FourStateValue::Narrow(n)) => {
            let m = mask.first().copied().unwrap_or(0);
            let val = pick(*c as u64, *n as u64, m);
            let unk = pick((*c >> 64) as u64, (*n >> 64) as u64, m);
            FourStateValue::Narrow(u128::from(val) | (u128::from(unk) << 64))
        }
        (
            FourStateValue::Wide { aval: ca, bval: cb },
            FourStateValue::Wide { aval: na, bval: nb },
        ) => {
            let words = |c: &[u64], n: &[u64]| -> Vec<u64> {
                c.iter()
                    .enumerate()
                    .map(|(i, &cw)| {
                        let nw = n.get(i).copied().unwrap_or(0);
                        pick(cw, nw, mask.get(i).copied().unwrap_or(0))
                    })
                    .collect()
            };
            FourStateValue::Wide {
                aval: words(ca, na),
                bval: words(cb, nb),
            }
        }
        // A memory row is stored in the wide planes whatever its width, so a
        // <=64-bit staged value meets a `Wide` row: convert it (and the
        // symmetric case) to the stored shape first.
        (FourStateValue::Wide { aval, .. }, FourStateValue::Narrow(n)) => {
            let (val, unk) = (*n as u64, (*n >> 64) as u64);
            let mut na = vec![0u64; aval.len()];
            let mut nb = vec![0u64; aval.len()];
            if let (Some(a), Some(b)) = (na.first_mut(), nb.first_mut()) {
                (*a, *b) = vj_abi::bit4::narrow_word_to_wide(val, unk);
            }
            merge_masked(cur, &FourStateValue::Wide { aval: na, bval: nb }, mask)
        }
        (FourStateValue::Narrow(_), FourStateValue::Wide { aval, bval }) => {
            let a = aval.first().copied().unwrap_or(0);
            let b = bval.first().copied().unwrap_or(0);
            let narrow = FourStateValue::Narrow(u128::from(a ^ b) | (u128::from(b) << 64));
            merge_masked(cur, &narrow, mask)
        }
    }
}

/// `commit_nba` for a batch that includes partial (`q[3:0] <= v`) or
/// memory-row (`mem[i] <= v`) writes: every write to one cell (or one
/// memory row) is applied in staging order -- so `q <= 0; q[3] <= 1;`
/// yields 8 -- then the cell is written once and notified once if it
/// changed. Cells are processed in `StateCellId` order, as `commit_nba`'s
/// last-write-wins path does.
fn commit_nba_with_partial_writes(
    entries: Vec<super::nba::NbaUpdate>,
    masked: Vec<super::nba::MaskedNbaUpdate>,
    arena: &mut dyn SignalAccess,
    at: SchedTime,
    sink: &mut dyn ValueChangeSink,
) -> usize {
    enum Op {
        Whole(FourStateValue),
        /// `(value, mask, window)`: a windowed op's `value` and `mask` span
        /// only the plane words from `window` on.
        Part(FourStateValue, Vec<u64>, Option<u32>),
        /// A nonblocking event trigger (`->> e`): bump the event cell's
        /// trigger count and stamp it with this commit's time.
        Pulse,
        /// A write of one element of the collection the cell holds: the
        /// element's words and where it goes (`MaskedNbaUpdate::coll`).
        Coll(Vec<u64>, super::nba::CollWrite),
    }
    /// A write target -- a whole cell, or one row of a memory cell.
    type Target = (StateCellId, Option<u64>);
    // Every staged operation, sorted by target and then staging order: the
    // targets come out in the same `(cell, row)` order a sorted map would
    // give, without a node allocation per operation.
    let mut items: Vec<(Target, u32, Op)> = Vec::with_capacity(entries.len() + masked.len());
    for e in entries {
        items.push(((e.cell, None), e.order, Op::Whole(e.value)));
    }
    for m in masked {
        let op = if m.pulse {
            Op::Pulse
        } else if let Some(c) = m.coll {
            Op::Coll(m.mask, c)
        } else {
            Op::Part(m.value, m.mask, m.window)
        };
        items.push(((m.cell, m.row), m.order, op));
    }
    items.sort_unstable_by_key(|(target, order, _)| (*target, *order));

    // One update event per value-changing operation, in cell order and then
    // staging order (LRM 4.9.4: `v[9] <= 1; v[9] <= 0;` is two updates, each
    // waking an `@(v[9])` waiter, though `v` ends unchanged). A memory row's
    // update has no single edge, so it reports `EdgeKind::None`.
    struct Update {
        cell: StateCellId,
        edge: EdgeKind,
        delta: Option<ChangeDelta>,
    }
    let mut updates: Vec<Update> = Vec::new();
    let mut changed_cells = 0usize;
    let mut alias_notes: Vec<(StateCellId, EdgeKind)> = Vec::new();
    let mut items = items.into_iter().peekable();
    while let Some(((cell, row), _, first)) = items.next() {
        let mut group = vec![first];
        while let Some((_, _, op)) = items.next_if(|(t, _, _)| *t == (cell, row)) {
            group.push(op);
        }
        let mark = updates.len();
        let wants = sink.wants_delta(cell);
        // Writes to a few bits of a huge vector: when every write to the
        // cell is windowed, only the words they touch are read, merged,
        // stored and compared (`O(window)`, not `O(N)`).
        if row.is_none() && group.iter().all(|op| matches!(op, Op::Part(_, _, Some(_)))) {
            // The words each write covers, merged into disjoint segments: a
            // few writes far apart cost their own words, not the distance
            // between them.
            let mut spans: Vec<(usize, usize)> = group
                .iter()
                .filter_map(|op| match op {
                    Op::Part(FourStateValue::Wide { aval, .. }, _, Some(w)) => {
                        Some((*w as usize, *w as usize + aval.len()))
                    }
                    _ => None,
                })
                .collect();
            spans.sort_unstable();
            let mut segments: Vec<(usize, usize)> = Vec::new();
            for (l, h) in spans {
                match segments.last_mut() {
                    Some(seg) if l <= seg.1 => seg.1 = seg.1.max(h),
                    _ => segments.push((l, h)),
                }
            }
            // Read every segment, apply the writes in staging order (each one
            // is its own update event, with the words it covered), then store
            // the segments back.
            let mut segs: Vec<(usize, usize, Vec<u64>, Vec<u64>)> = segments
                .iter()
                .map(|&(lo, hi)| {
                    let (a, b) = arena.read_wide_window(cell, lo, hi - lo);
                    (lo, hi, a, b)
                })
                .collect();
            for op in &group {
                let Op::Part(FourStateValue::Wide { aval, bval }, mask, Some(w)) = op else {
                    continue;
                };
                let w = *w as usize;
                let Some((lo, _, a, b)) = segs
                    .iter_mut()
                    .find(|(lo, hi, _, _)| w >= *lo && w + mask.len() <= *hi)
                else {
                    continue;
                };
                let at = w - *lo;
                let n = mask.len();
                let (old_a, old_b) = (a[at..at + n].to_vec(), b[at..at + n].to_vec());
                for (k, &m) in mask.iter().enumerate() {
                    a[at + k] = (a[at + k] & !m) | (aval[k] & m);
                    b[at + k] = (b[at + k] & !m) | (bval[k] & m);
                }
                if old_a[..] == a[at..at + n] && old_b[..] == b[at..at + n] {
                    continue;
                }
                // Only the lsb decides an edge, and only an update covering
                // word 0 can have changed it.
                let edge = if w == 0 {
                    edge_between(
                        Bit4::from_aval_bval(old_a[0] & 1 != 0, old_b[0] & 1 != 0),
                        Bit4::from_aval_bval(a[0] & 1 != 0, b[0] & 1 != 0),
                    )
                } else {
                    EdgeKind::None
                };
                let delta = wants.then(|| ChangeDelta {
                    lo_word: w,
                    old_a,
                    old_b,
                    new_a: a[at..at + n].to_vec(),
                    new_b: b[at..at + n].to_vec(),
                });
                updates.push(Update { cell, edge, delta });
            }
            let mut changed = false;
            let mut forced = false;
            for (lo, hi, a, b) in &segs {
                if arena.write_wide_window(cell, *lo, a, b) {
                    changed = true;
                } else if updates.len() > mark {
                    // Nothing stored: either the writes cancelled, or the cell
                    // is forced and kept its value.
                    let (ra, rb) = arena.read_wide_window(cell, *lo, hi - lo);
                    forced |= ra != *a || rb != *b;
                }
            }
            if forced {
                updates.truncate(mark);
            }
            if changed || updates.len() > mark {
                changed_cells += 1;
            }
            continue;
        }
        let old = match row {
            Some(r) => arena.read_row(cell, r),
            None => arena.read(cell),
        };
        let apply = |cur: &FourStateValue, op: Op| match op {
            Op::Whole(v) => v,
            Op::Pulse => triggered_event_value(cur, at),
            // The handle is read now, so the write lands in the collection
            // the variable holds at update time, and an out-of-range index
            // is ignored there (`collections::set`); a copy-on-write set may
            // hand back a new handle, which becomes the variable's value.
            Op::Coll(elem, c) => {
                let h = match cur {
                    FourStateValue::Narrow(v) => *v as u64,
                    FourStateValue::Wide { aval, .. } => aval.first().copied().unwrap_or(0),
                };
                let new = match c.key {
                    Some(k) => crate::collections::store(&c.shape, h, k, &elem),
                    None => crate::collections::set(&c.shape, h, c.index, &elem, c.queue),
                };
                match cur {
                    FourStateValue::Narrow(_) => FourStateValue::Narrow(u128::from(new)),
                    FourStateValue::Wide { aval, bval } => {
                        let (mut a, b) = (aval.clone(), vec![0; bval.len()]);
                        a[0] = new;
                        FourStateValue::Wide { aval: a, bval: b }
                    }
                }
            }
            Op::Part(v, mask, None) => merge_masked(cur, &v, &mask),
            Op::Part(v, mask, Some(w)) => {
                // A windowed write meeting a whole-value one: widen it to the
                // full value.
                let (FourStateValue::Wide { aval, bval }, FourStateValue::Wide { aval: ca, .. }) =
                    (&v, cur)
                else {
                    unreachable!("a windowed NBA targets a wide cell")
                };
                let n = ca.len();
                let at = w as usize;
                let (mut fa, mut fb, mut fm) = (vec![0u64; n], vec![0u64; n], vec![0u64; n]);
                fa[at..at + aval.len()].copy_from_slice(aval);
                fb[at..at + bval.len()].copy_from_slice(bval);
                fm[at..at + mask.len()].copy_from_slice(&mask);
                merge_masked(cur, &FourStateValue::Wide { aval: fa, bval: fb }, &fm)
            }
        };
        let mut cur = old.clone();
        for op in group {
            let next = apply(&cur, op);
            if !same_value(&cur, &next) {
                let (edge, delta) = if row.is_some() {
                    (EdgeKind::None, None)
                } else {
                    (
                        infer_edge(&cur, &next),
                        wants.then(|| ChangeDelta::between(&cur, &next)),
                    )
                };
                updates.push(Update { cell, edge, delta });
            }
            cur = next;
        }
        let stored = match row {
            Some(r) => arena.write_row(cell, r, cur.clone()),
            None => arena.write(cell, cur.clone()),
        };
        if updates.len() > mark {
            if !stored {
                // Nothing stored: the writes cancelled, or the cell is forced.
                let now = match row {
                    Some(r) => arena.read_row(cell, r),
                    None => arena.read(cell),
                };
                if !same_value(&now, &cur) {
                    updates.truncate(mark);
                }
            }
            if updates.len() > mark {
                changed_cells += 1;
            }
        }
        alias_notes.extend(alias_edges(arena));
    }
    for u in updates {
        if let Some(d) = u.delta {
            sink.note_delta(u.cell, d);
        }
        notify_nba_change(sink, u.cell, u.edge, at);
    }
    for (cell, edge) in alias_notes {
        notify_nba_change(sink, cell, edge, at);
    }
    changed_cells
}

/// An event cell's value (`state::EVENT_CELL_WIDTH` bits, 2-state: the trigger
/// count in the low 64 bits, `tick + 1` of the latest trigger in the high 64)
/// after one more trigger at `at` -- what `-> e` computes in compiled code
/// (`vj_codegen::expr::lower_event_trigger`).
fn triggered_event_value(cur: &FourStateValue, at: SchedTime) -> FourStateValue {
    let count = match cur {
        FourStateValue::Wide { aval, .. } => aval.first().copied().unwrap_or(0),
        FourStateValue::Narrow(n) => *n as u64,
    };
    FourStateValue::Wide {
        aval: vec![count.wrapping_add(1), at.ticks.wrapping_add(1)],
        bval: vec![0, 0],
    }
}

/// Whether two stored values are bit-identical in both planes.
fn same_value(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,
        _ => a.planes() == b.planes(),
    }
}

#[cfg(test)]
mod tests {
    use super::super::nba::NbaLog;
    use crate::sched::{ProcessId, SchedTime};
    use crate::state::{
        EdgeKind, FourStateValue, ListenerRole, SignalAccess, StateCellId, ValueChangeSink,
    };
    use std::cell::RefCell;
    use std::collections::{HashMap, HashSet};
    use std::rc::Rc;
    use vj_abi::{Region, SimTime};

    // `FakeArena` per the task brief: one shared piece of state (a
    // `HashMap<StateCellId, FourStateValue>` plus a notify-call log) that
    // backs both a `SignalAccess` view and a `ValueChangeSink` view. Split
    // into two thin handle types sharing one `Rc<RefCell<..>>` rather than one
    // type implementing both traits directly, because `commit_nba` takes
    // `arena: &mut dyn SignalAccess` and `sink: &mut dyn ValueChangeSink` as
    // two *simultaneous* `&mut` parameters — two `&mut` borrows of the exact
    // same value in one call is rejected by the borrow checker regardless of
    // which fields each trait impl touches, so the shared state has to live
    // behind interior mutability and be reached through two distinct handle
    // values instead. This is a mechanical adaptation only: both handles see
    // and mutate the same underlying cells/notifications, so the test's
    // observable behavior (arena state + notify log) is exactly what a single
    // dual-trait `FakeArena` would produce.
    struct FakeArenaState {
        cells: HashMap<StateCellId, FourStateValue>,
        notifications: Vec<(StateCellId, EdgeKind, SimTime, Region)>,
        deltas: Vec<super::ChangeDelta>,
    }

    #[derive(Clone)]
    struct FakeArena(Rc<RefCell<FakeArenaState>>);

    struct ArenaHandle(Rc<RefCell<FakeArenaState>>);
    struct SinkHandle(Rc<RefCell<FakeArenaState>>);

    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,
        }
    }

    impl FakeArena {
        fn new() -> Self {
            Self(Rc::new(RefCell::new(FakeArenaState {
                cells: HashMap::new(),
                notifications: Vec::new(),
                deltas: Vec::new(),
            })))
        }

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

        fn read_cell(&self, cell: StateCellId) -> FourStateValue {
            self.0
                .borrow()
                .cells
                .get(&cell)
                .cloned()
                .unwrap_or(FourStateValue::Narrow(0))
        }

        fn notifications(&self) -> Vec<(StateCellId, EdgeKind, SimTime, Region)> {
            self.0.borrow().notifications.clone()
        }

        fn arena_handle(&self) -> ArenaHandle {
            ArenaHandle(self.0.clone())
        }

        fn sink_handle(&self) -> SinkHandle {
            SinkHandle(self.0.clone())
        }
    }

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

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

        fn subscribe(
            &mut self,
            _cell: StateCellId,
            _role: ListenerRole,
            _sink: Box<dyn ValueChangeSink + Send>,
        ) {
            // Not exercised by Task 7's tests — Task 9 owns SensitivityTable
            // wiring.
        }
    }

    impl ValueChangeSink for SinkHandle {
        fn notify_change(
            &mut self,
            cell: StateCellId,
            edge: EdgeKind,
            at: SimTime,
            region: Region,
        ) {
            self.0
                .borrow_mut()
                .notifications
                .push((cell, edge, at, region));
        }

        fn wants_delta(&self, _cell: StateCellId) -> bool {
            true
        }

        fn note_delta(&mut self, _cell: StateCellId, delta: super::ChangeDelta) {
            self.0.borrow_mut().deltas.push(delta);
        }
    }

    const PROC_A: ProcessId = 10;
    const PROC_B: ProcessId = 5;

    #[test]
    fn two_nba_writes_same_cell_deterministically_commit_the_last_logged_one() {
        let cell = StateCellId(1);

        // A (ProcessId 10, numerically larger) pushes first (order 0), B
        // (ProcessId 5, numerically smaller) pushes second (order 1). If the
        // tiebreak were numeric ProcessId order rather than push order, A
        // (the larger id) would win; the spec says push order wins, so B
        // (pushed last) must win here.
        let fake1 = FakeArena::new();
        let mut arena1 = fake1.arena_handle();
        let mut sink1 = fake1.sink_handle();
        let mut log1 = NbaLog::new();
        log1.stage(cell, FourStateValue::Narrow(1), PROC_A);
        log1.stage(cell, FourStateValue::Narrow(2), PROC_B);
        super::commit_nba(&mut log1, &mut arena1, SchedTime { ticks: 0 }, &mut sink1);
        match fake1.read_cell(cell) {
            FourStateValue::Narrow(v) => assert_eq!(v, 2, "last-pushed write (B) must win"),
            FourStateValue::Wide { .. } => panic!("expected Narrow"),
        }

        // Swap the push order: B first, A second. The committed value must
        // track the new push order (A wins now), proving the tiebreak is
        // push/insertion order, not ProcessId numeric order (which is
        // unchanged between the two sub-tests).
        let fake2 = FakeArena::new();
        let mut arena2 = fake2.arena_handle();
        let mut sink2 = fake2.sink_handle();
        let mut log2 = NbaLog::new();
        log2.stage(cell, FourStateValue::Narrow(2), PROC_B);
        log2.stage(cell, FourStateValue::Narrow(1), PROC_A);
        super::commit_nba(&mut log2, &mut arena2, SchedTime { ticks: 0 }, &mut sink2);
        match fake2.read_cell(cell) {
            FourStateValue::Narrow(v) => assert_eq!(v, 1, "last-pushed write (A) must win"),
            FourStateValue::Wide { .. } => panic!("expected Narrow"),
        }
    }

    #[test]
    fn cancelling_nbas_are_two_update_events_in_staging_order() {
        // LRM 4.9.4 / 10.4.2: `q <= 1; q <= 0;` applies in order and each
        // value-changing update is an update event, though `q` ends where it
        // began.
        let cell = StateCellId(3);
        let fake = FakeArena::new();
        fake.seed(cell, FourStateValue::Narrow(0));
        let mut arena = fake.arena_handle();
        let mut sink = fake.sink_handle();
        let mut log = NbaLog::new();
        log.stage(cell, FourStateValue::Narrow(1), PROC_A);
        log.stage(cell, FourStateValue::Narrow(0), PROC_A);
        super::commit_nba(&mut log, &mut arena, SchedTime { ticks: 0 }, &mut sink);
        let edges: Vec<EdgeKind> = fake.notifications().into_iter().map(|n| n.1).collect();
        assert_eq!(edges, vec![EdgeKind::Posedge, EdgeKind::Negedge]);
        match fake.read_cell(cell) {
            FourStateValue::Narrow(v) => assert_eq!(v, 0),
            FourStateValue::Wide { .. } => panic!("expected Narrow"),
        }
    }

    #[test]
    fn cancelling_partial_nbas_are_two_update_events_each_with_its_bits() {
        // `v[9] <= 1; v[9] <= 0;` on a 128-bit vector: two events, whose
        // deltas show bit 9 rising and then falling.
        let cell = StateCellId(4);
        let fake = FakeArena::new();
        fake.seed(
            cell,
            FourStateValue::Wide {
                aval: vec![0, 0],
                bval: vec![0, 0],
            },
        );
        let mut arena = fake.arena_handle();
        let mut sink = fake.sink_handle();
        let mut log = NbaLog::new();
        for bit in [1u64, 0] {
            log.stage_masked(
                cell,
                None,
                FourStateValue::Wide {
                    aval: vec![bit << 9, 0],
                    bval: vec![0, 0],
                },
                vec![1 << 9, 0],
                None,
                PROC_A,
            );
        }
        super::commit_nba(&mut log, &mut arena, SchedTime { ticks: 0 }, &mut sink);
        let n = fake.notifications();
        assert_eq!(n.len(), 2, "one update event per value-changing write");
        let deltas = fake.0.borrow().deltas.clone();
        assert_eq!(deltas.len(), 2);
        assert_eq!(deltas[0].bit(9), Some(((false, false), (true, false))));
        assert_eq!(deltas[1].bit(9), Some(((true, false), (false, false))));
    }

    #[test]
    fn a_staged_pulse_triggers_the_event_cell_in_the_nba_region() {
        // LRM 15.5.2: `->> e` (twice) bumps the event's trigger count and
        // stamps it with the commit's `tick + 1`, notifying once per trigger
        // from the NBA region, in order with a data write staged before it.
        let ev = StateCellId(6);
        let data = StateCellId(7);
        let fake = FakeArena::new();
        fake.seed(
            ev,
            FourStateValue::Wide {
                aval: vec![3, 4],
                bval: vec![0, 0],
            },
        );
        fake.seed(data, FourStateValue::Narrow(0));
        let mut arena = fake.arena_handle();
        let mut sink = fake.sink_handle();
        let mut log = NbaLog::new();
        log.stage(data, FourStateValue::Narrow(1), PROC_A);
        log.stage_pulse(ev, None, PROC_A);
        log.stage_pulse(ev, None, PROC_B);
        super::commit_nba(&mut log, &mut arena, SchedTime { ticks: 9 }, &mut sink);
        match fake.read_cell(ev) {
            FourStateValue::Wide { aval, bval } => {
                assert_eq!(aval, vec![5, 10], "count 3 -> 5, stamped tick 9 + 1");
                assert_eq!(bval, vec![0, 0]);
            }
            FourStateValue::Narrow(_) => panic!("an event cell is wide"),
        }
        let n = fake.notifications();
        assert_eq!(n.iter().filter(|x| x.0 == ev).count(), 2);
        assert!(n.iter().all(|x| x.3 == Region::Nba));
        assert_eq!(n.iter().filter(|x| x.0 == data).count(), 1);
    }

    #[test]
    fn unchanged_commit_produces_no_notify_change_call() {
        let cell = StateCellId(2);
        let fake = FakeArena::new();
        fake.seed(cell, FourStateValue::Narrow(5));
        let mut arena = fake.arena_handle();
        let mut sink = fake.sink_handle();

        let mut log = NbaLog::new();
        log.stage(cell, FourStateValue::Narrow(5), PROC_A);

        let changed = super::commit_nba(&mut log, &mut arena, SchedTime { ticks: 3 }, &mut sink);

        assert_eq!(changed, 0);
        assert!(
            fake.notifications().is_empty(),
            "an NBA write equal to the current value must not call notify_change"
        );
    }

    #[test]
    fn changed_commit_calls_notify_change_exactly_once_per_changed_cell() {
        let cell_changed_1 = StateCellId(10);
        let cell_changed_2 = StateCellId(11);
        let cell_unchanged = StateCellId(12);

        let fake = FakeArena::new();
        fake.seed(cell_changed_1, FourStateValue::Narrow(0));
        fake.seed(cell_changed_2, FourStateValue::Narrow(0));
        fake.seed(cell_unchanged, FourStateValue::Narrow(7));
        let mut arena = fake.arena_handle();
        let mut sink = fake.sink_handle();

        let mut log = NbaLog::new();
        log.stage(cell_changed_1, FourStateValue::Narrow(1), PROC_A);
        log.stage(cell_changed_2, FourStateValue::Narrow(1), PROC_A);
        log.stage(cell_unchanged, FourStateValue::Narrow(7), PROC_A);

        let changed = super::commit_nba(&mut log, &mut arena, SchedTime { ticks: 0 }, &mut sink);

        assert_eq!(changed, 2);
        let notified: Vec<StateCellId> = fake.notifications().into_iter().map(|n| n.0).collect();
        assert_eq!(notified.len(), 2, "no duplicate notify_change calls");
        let notified_set: HashSet<StateCellId> = notified.into_iter().collect();
        assert!(notified_set.contains(&cell_changed_1));
        assert!(notified_set.contains(&cell_changed_2));
        assert!(!notified_set.contains(&cell_unchanged));
    }

    #[test]
    fn changed_cell_diffing_is_the_sole_wakeup_source() {
        let cell_a = StateCellId(20);
        let cell_b = StateCellId(21);
        let cell_c = StateCellId(22);
        let cell_d = StateCellId(23);

        let fake = FakeArena::new();
        fake.seed(cell_a, FourStateValue::Narrow(0));
        fake.seed(cell_b, FourStateValue::Narrow(9));
        fake.seed(cell_c, FourStateValue::Narrow(0));
        fake.seed(cell_d, FourStateValue::Narrow(4));
        let mut arena = fake.arena_handle();
        let mut sink = fake.sink_handle();

        let mut log = NbaLog::new();
        // a and c actually change; b and d are staged with their current
        // value and must not change.
        log.stage(cell_a, FourStateValue::Narrow(1), PROC_A);
        log.stage(cell_b, FourStateValue::Narrow(9), PROC_A);
        log.stage(cell_c, FourStateValue::Narrow(2), PROC_A);
        log.stage(cell_d, FourStateValue::Narrow(4), PROC_A);

        super::commit_nba(&mut log, &mut arena, SchedTime { ticks: 0 }, &mut sink);

        let notified_set: HashSet<StateCellId> =
            fake.notifications().into_iter().map(|n| n.0).collect();
        let expected: HashSet<StateCellId> = [cell_a, cell_c].into_iter().collect();
        // Exact-equality (not a subset check): a buggy "notify everything
        // staged" implementation would produce {a, b, c, d} here, which
        // fails this assertion.
        assert_eq!(notified_set, expected);
    }

    #[test]
    fn multi_bit_change_with_unchanged_bit0_still_wakes_an_any_change_static_entry() {
        // Regression for the review finding on SensitivityTable::matches:
        // infer_edge only classifies bit 0 of the old/new value (Table 9-1's
        // scalar posedge/negedge rule), so a real, multi-bit change that
        // happens to leave bit 0 fixed (Narrow(0b10) -> Narrow(0b110): bit 0
        // is 0 in both) is a genuine `SignalAccess::write` change but
        // `infer_edge` classifies it as `EdgeKind::None`. An always_comb-style
        // "any change" Static entry (SC-046) must still be woken by this,
        // through the real `commit_nba` -> `notify_change` -> `matches`
        // pipeline, not a hand-picked call into `matches` directly.
        use crate::sched::StaticSensitivity;
        use crate::sched::sensitivity::{SensitivityLookupSink, SensitivityTable};

        let cell = StateCellId(30);
        let fake = FakeArena::new();
        fake.seed(cell, FourStateValue::Narrow(0b10));
        let mut arena = fake.arena_handle();

        let mut table = SensitivityTable::new();
        const PROC_COMB: ProcessId = 99;
        table.install_static(
            &StaticSensitivity {
                edges: vec![(cell, None)],
            },
            PROC_COMB,
        );
        let mut woken = Vec::new();
        let mut sink = SensitivityLookupSink::new(&table, &mut woken);

        let mut log = NbaLog::new();
        log.stage(cell, FourStateValue::Narrow(0b110), PROC_A);

        let changed = super::commit_nba(&mut log, &mut arena, SchedTime { ticks: 0 }, &mut sink);

        assert_eq!(changed, 1, "Narrow(0b10) -> Narrow(0b110) is a real change");
        assert_eq!(
            woken,
            vec![PROC_COMB],
            "an any-change Static entry must fire on a real multi-bit change even when bit 0 (infer_edge's only input) doesn't toggle"
        );
    }

    #[test]
    fn infer_edge_covers_the_four_state_transition_table() {
        use vj_abi::Bit4;

        fn bit_value(b: Bit4) -> FourStateValue {
            let (val, unk) = b.to_val_unk();
            FourStateValue::Narrow((val as u128) | ((unk as u128) << 64))
        }

        let cases: &[(Bit4, Bit4, EdgeKind)] = &[
            (Bit4::Zero, Bit4::Zero, EdgeKind::None),
            (Bit4::One, Bit4::One, EdgeKind::None),
            (Bit4::X, Bit4::X, EdgeKind::None),
            (Bit4::Z, Bit4::Z, EdgeKind::None),
            (Bit4::Zero, Bit4::One, EdgeKind::Posedge),
            (Bit4::One, Bit4::Zero, EdgeKind::Negedge),
            // LRM Table 9-1 (§9.4.2): X/Z-involving transitions are
            // asymmetric, each landing on exactly one specific edge (or
            // none) — never `BothEdges`. Verified against iverilog 12.
            (Bit4::Zero, Bit4::X, EdgeKind::Posedge),
            (Bit4::X, Bit4::One, EdgeKind::Posedge),
            (Bit4::Z, Bit4::One, EdgeKind::Posedge),
            (Bit4::Zero, Bit4::Z, EdgeKind::Posedge),
            (Bit4::One, Bit4::X, EdgeKind::Negedge),
            (Bit4::One, Bit4::Z, EdgeKind::Negedge),
            (Bit4::X, Bit4::Zero, EdgeKind::Negedge),
            (Bit4::Z, Bit4::Zero, EdgeKind::Negedge),
            // The four "stayed unknown" transitions produce no edge event
            // at all per the LRM — a buggy "any X/Z touch = BothEdges"
            // implementation (the bug this regression guards against)
            // would report `BothEdges` here instead of `None`.
            (Bit4::X, Bit4::Z, EdgeKind::None),
            (Bit4::Z, Bit4::X, EdgeKind::None),
        ];

        for (old, new, expected) in cases {
            let old_v = bit_value(*old);
            let new_v = bit_value(*new);
            let got = super::infer_edge(&old_v, &new_v);
            assert_eq!(
                got, *expected,
                "infer_edge({:?} -> {:?}) should be {:?}, got {:?}",
                old, new, expected, got
            );
        }
    }

    #[test]
    fn nba_write_lowering_shapes_both_append_to_the_log() {
        // (a) simulates lowering a `ProceduralAssignStatement`-style
        // nonblocking write (`x <= expr;`).
        let mut log_a = NbaLog::new();
        log_a.stage(StateCellId(9), FourStateValue::Narrow(42), 3);
        let from_procedural_assign = log_a.take();

        // (b) simulates lowering an `Assignment` expression with
        // `isNonBlocking()==true` nested inside an `ExpressionStatement`.
        let mut log_b = NbaLog::new();
        log_b.stage(StateCellId(9), FourStateValue::Narrow(42), 3);
        let from_expression_statement = log_b.take();

        assert_eq!(from_procedural_assign.len(), 1);
        assert_eq!(from_expression_statement.len(), 1);

        let a = &from_procedural_assign[0];
        let b = &from_expression_statement[0];
        // Both AST shapes must lower to representationally identical
        // NbaUpdate values — the log format does not (and must not)
        // distinguish which AST shape produced an entry.
        assert_eq!(a.cell, b.cell);
        assert_eq!(a.source, b.source);
        assert_eq!(a.order, b.order);
        match (&a.value, &b.value) {
            (FourStateValue::Narrow(x), FourStateValue::Narrow(y)) => assert_eq!(x, y),
            _ => panic!("expected Narrow"),
        }
    }
}